Merge branch 'master' into update/sbt-mdoc-2.8.0

Author: novakov-alexey

.git-blame-ignore-revs

@@ -45,3 +45,6 @@ b83264042216c7d83b73247b8b7a998902ec26d5
 
 # Scala Steward: Reformat with scalafmt 3.9.10
 7d900f00d9e3f23f45fca7989ae6a4c1c75580c7
+
+# Scala Steward: Reformat with scalafmt 3.10.0
+12eaae6b3f4f112a41e4e69dc5c7109ba0296dcd

.github/workflows/ci.yml

@@ -1,7 +1,7 @@
 # This workflow will build a Java project with Maven
 # For more information see: https://help.github.com/actions/language-and-framework-guides/building-and-testing-java-with-maven
 
-name: CI
+name: All Tests
 
 on:
   push:
@@ -15,15 +15,15 @@ jobs:
     strategy:
       matrix:
         java: [11, 17]
-        scala: [2.13.17, 3.3.6]
+        scala: [2.13.17, 3.3.7]
         flink: [1.19.3, 1.20.2]
         sbt-module: ['flink-1-api', 'scala-api-common']
         include:
           - scala: 2.13.17
             java: 17
             flink: 2.0.0
             sbt-module: 'flink-2-api'
-          - scala: 3.3.6
+          - scala: 3.3.7
             java: 21
             flink: 2.0.0
             sbt-module: 'flink-2-api'

.github/workflows/release.yml

@@ -1,8 +1,7 @@
-name: release
+name: Release
 concurrency: release
 on:
   push:
-    branches: [master, main]
     tags: ["*"]
 jobs:
   release:
@@ -22,7 +21,7 @@ jobs:
       - run: echo $PGP_SECRET | base64 --decode | gpg --batch --import
         env:
           PGP_SECRET: ${{ secrets.PGP_SECRET }}
-      - run: sbt test ciReleaseTagNextVersion ciReleaseSonatype
+      - run: sbt test ciReleaseSonatype
         env:
           SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
           SONATYPE_USERNAME: ${{ secrets.SONATYPE_USERNAME }}

.scalafmt.conf

@@ -1,6 +1,6 @@
 style       = defaultWithAlign
 maxColumn   = 120
-version = 3.9.10
+version = 3.10.0
 assumeStandardLibraryStripMargin = true
 align.stripMargin = true
 runner.dialect = scala3

README.md

@@ -352,6 +352,56 @@ implicit val mapper2: TypeMapper[WrappedString, String] = new TypeMapper[Wrapped
 }  
 ```
 
+### Ordering
+
+`SortedSet` requires a type-information for its elements and also for the ordering of the elements. Type-information of default orderings are not implicitly available in the context because we cannot make the assumption the user wants to use the natural ordering or a custom one.
+
+Type-information of default ordering are available in `org.apache.flinkx.api.serializer.OrderingTypeInfo` and can be used as follows:
+```scala mdoc:reset-object
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flinkx.api._
+import org.apache.flinkx.api.serializer.OrderingTypeInfo
+import org.apache.flinkx.api.serializers._
+import scala.collection.immutable.SortedSet
+
+case class Foo(bars: SortedSet[String])
+
+object Foo {
+  implicit val fooInfo: TypeInformation[Foo] = {
+    // type-information for Ordering need to be explicitly put in the context
+    implicit val orderingStringInfo: TypeInformation[Ordering[String]] =
+      OrderingTypeInfo.DefaultStringOrderingInfo
+    deriveTypeInformation
+  }
+}
+```
+
+It's also possible to derive the type-information of a custom ordering if it's an ADT:
+```scala mdoc:reset-object
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flinkx.api._
+import org.apache.flinkx.api.serializer.OrderingTypeInfo
+import org.apache.flinkx.api.serializers._
+import scala.collection.immutable.SortedSet
+
+case class Bar(a: Int, b: String)
+
+case object BarOrdering extends Ordering[Bar] {
+  override def compare(x: Bar, y: Bar): Int = x.a.compare(y.a)
+}
+
+case class Foo(bar: SortedSet[Bar])
+
+object Foo {
+  implicit val fooInfo: TypeInformation[Foo] = {
+    // Derive the type-information of custom Bar ordering
+    implicit val barOrderingInfo: TypeInformation[Ordering[Bar]] =
+      OrderingTypeInfo.deriveOrdering[BarOrdering.type, Bar]
+    deriveTypeInformation
+  }
+}
+```
+
 ### Schema evolution
 
 #### ADT

build.sbt

@@ -3,8 +3,8 @@ import sbtrelease.ReleaseStateTransformations.*
 Global / onChangedBuildSource := ReloadOnSourceChanges
 Global / excludeLintKeys      := Set(crossScalaVersions)
 
-lazy val rootScalaVersion = "3.3.6"
-lazy val crossVersions    = Seq("2.13.16", rootScalaVersion)
+lazy val rootScalaVersion = "3.3.7"
+lazy val crossVersions    = Seq("2.13.17", rootScalaVersion)
 lazy val flinkVersion1    = System.getProperty("flinkVersion1", "1.20.2")
 lazy val flinkVersion2    = System.getProperty("flinkVersion2", "2.0.0")
 

modules/examples/scripts/gen-csv-file.sc

@@ -1,4 +1,4 @@
-//> using dep "org.flinkextended::flink-scala-api-1:1.2.9"
+//> using dep "org.flinkextended::flink-scala-api-1:2.0.4"
 //> using dep "org.apache.flink:flink-clients:1.20.1"
 //> using dep "org.apache.flink:flink-csv:1.20.1"
 //> using dep "org.apache.flink:flink-connector-files:1.20.1"

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/CaseClassSerializer.scala

@@ -22,6 +22,7 @@ import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeSerializerSnap
 import org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase
 import org.apache.flink.core.memory.{DataInputView, DataOutputView}
 import org.apache.flink.types.NullFieldException
+import org.apache.flinkx.api.{NullMarker, VariableLengthDataType}
 import org.apache.flinkx.api.serializer.CaseClassSerializer.EmptyByteArray
 import org.slf4j.{Logger, LoggerFactory}
 
@@ -103,11 +104,13 @@ class CaseClassSerializer[T <: Product](
       createInstance(fields)
     }
 
-  override val getLength: Int = if (super.getLength == -1) -1 else super.getLength + 4 // +4 bytes for the arity field
+  override val getLength: Int =
+    if (super.getLength == VariableLengthDataType) VariableLengthDataType
+    else super.getLength + 4 // +4 bytes for the arity field
 
   def serialize(value: T, target: DataOutputView): Unit = {
-    // Write an arity of -1 to indicate null value
-    val sourceArity = if (value == null) -1 else arity
+    // Write a negative arity to indicate null value
+    val sourceArity = if (value == null) NullMarker else arity
     target.writeInt(sourceArity)
     if (value == null) target.write(nullPadding)
 
@@ -129,7 +132,7 @@ class CaseClassSerializer[T <: Product](
 
   def deserialize(source: DataInputView): T = {
     val sourceArity = source.readInt()
-    if (sourceArity == -1) {
+    if (sourceArity < 0) {
       source.skipBytesToRead(nullPadding.length)
       null.asInstanceOf[T]
     } else {
@@ -146,7 +149,7 @@ class CaseClassSerializer[T <: Product](
   override def copy(source: DataInputView, target: DataOutputView): Unit = {
     val sourceArity = source.readInt()
     target.writeInt(sourceArity)
-    if (sourceArity == -1) {
+    if (sourceArity < 0) {
       source.skipBytesToRead(nullPadding.length)
       target.skipBytesToWrite(nullPadding.length)
     } else {

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/CollectionSerializerSnapshot.scala

@@ -69,8 +69,9 @@ class CollectionSerializerSnapshot[F[_], T, S <: TypeSerializer[F[T]]](
       case "long"    => out.writeUTF("java.lang.Long")
       case "byte"    => out.writeUTF("java.lang.Byte")
       case "short"   => out.writeUTF("java.lang.Short")
-      case "char"    => out.writeUTF("java.lang.Char")
+      case "char"    => out.writeUTF("java.lang.Character")
       case "boolean" => out.writeUTF("java.lang.Boolean")
+      case "void"    => out.writeUTF("java.lang.Void")
       case other     => out.writeUTF(other)
     }
     TypeSerializerSnapshot.writeVersionedSnapshot(out, nestedSerializer.snapshotConfiguration())

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/SortedCollectionSerializerSnapshot.scala

@@ -0,0 +1,50 @@
+package org.apache.flinkx.api.serializer
+
+import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeSerializerSnapshot}
+import org.apache.flink.core.memory.{DataInputView, DataOutputView}
+
+/** Generic serializer snapshot for sorted collection.
+  * @param aSerializer
+  *   the serializer of `A`
+  * @param aOrderingSerializer
+  *   the serializer of `Ordering[A]`
+  * @param sClass
+  *   the class of `S`
+  * @param aClass
+  *   the class of `A`
+  * @tparam F
+  *   the type of the serialized collection
+  * @tparam A
+  *   the type of the collection's elements
+  * @tparam S
+  *   the type of the collection serializer
+  */
+class SortedCollectionSerializerSnapshot[F[_], A, S <: TypeSerializer[F[A]]](
+    aSerializer: TypeSerializer[A],
+    sClass: Class[S],
+    aClass: Class[A],
+    private var aOrderingSerializer: TypeSerializer[Ordering[A]]
+) extends CollectionSerializerSnapshot[F, A, S](aSerializer, sClass, aClass) {
+
+  // Empty constructor is required to instantiate this class during deserialization.
+  def this() = this(null, null, null, null)
+
+  override def getCurrentVersion: Int = super.getCurrentVersion // Must be aligned with CollectionSerializerSnapshot
+
+  override def writeSnapshot(out: DataOutputView): Unit = {
+    super.writeSnapshot(out)
+    TypeSerializerSnapshot.writeVersionedSnapshot(out, aOrderingSerializer.snapshotConfiguration())
+  }
+
+  override def readSnapshot(readVersion: Int, in: DataInputView, userCodeClassLoader: ClassLoader): Unit = {
+    super.readSnapshot(readVersion, in, userCodeClassLoader)
+    aOrderingSerializer =
+      TypeSerializerSnapshot.readVersionedSnapshot[Ordering[A]](in, userCodeClassLoader).restoreSerializer()
+  }
+
+  override def restoreSerializer(): TypeSerializer[F[A]] = {
+    val constructor = clazz.getConstructors()(0)
+    constructor.newInstance(nestedSerializer, vclazz, aOrderingSerializer).asInstanceOf[TypeSerializer[F[A]]]
+  }
+
+}