Port To.safe macro to Scala 3

.github/workflows/migration.yml

@@ -23,7 +23,7 @@ jobs:
           # - scio-extra/compile
           # - scio-extra/test
           - scio-core/compile
-          # - scio-core/test
+          - scio-core/test
           # - scio-examples/compile
           # - scio-examples/test
           # - scio-redis/compile
@@ -54,5 +54,7 @@ jobs:
         uses: coursier/cache-action@v5
       - name: java 8 setup
         uses: olafurpg/setup-scala@v10
-      - name: Compile
+#      - name: Scala 2
+#        run: sbt "++2.13.5;${{ matrix.task }}"
+      - name: Scala 3
         run: sbt "++3.0.0-RC2;${{ matrix.task }}"

build.sbt

@@ -584,7 +584,8 @@ lazy val `scio-macros`: Project = project
     libraryDependencies ++= Seq(
       "com.esotericsoftware" % "kryo-shaded" % kryoVersion,
       "org.apache.beam" % "beam-sdks-java-extensions-sql" % beamVersion,
-      "org.apache.avro" % "avro" % avroVersion
+      "org.apache.avro" % "avro" % avroVersion,
+      "org.scalatest" %% "scalatest" % scalatestVersion % Test
     ),
     // Scala 2 dependencies
     libraryDependencies ++= {

scio-core/src/main/scala-3/com/spotify/scio/schemas/To.scala

@@ -18,15 +18,165 @@
 package com.spotify.scio.schemas
 
 import org.apache.beam.sdk.schemas.{SchemaCoder, Schema => BSchema}
+import BSchema.{ FieldType => BFieldType }
 
 import scala.compiletime._
 import scala.deriving._
 import scala.quoted._
+import scala.reflect.ClassTag
+import scala.collection.mutable
+import com.spotify.scio.IsJavaBean.checkGetterAndSetters
 
 object ToMacro {
-  def safeImpl[I, O](si: Expr[Schema[I]])(implicit q: Quotes): Expr[To[I, O]] = {
-    ???
+
+  def safeImpl[I: scala.quoted.Type, O: scala.quoted.Type](
+    iSchema: Expr[Schema[I]],
+    oSchema: Expr[Schema[O]]
+  )(using Quotes): Expr[To[I, O]] = {
+    import scala.quoted.quotes.reflect.{report, TypeRepr}
+
+    (interpret[I] , interpret[O]) match {
+      case (None, None) => report.throwError(
+        s"""
+        |Could not interpret input schema:
+        |  ${iSchema.show}
+        |Could not interpret output schema:
+        |  ${oSchema.show}
+        |""".stripMargin
+      )
+      case (None, _) => report.throwError("Could not interpret input schema: " + iSchema.show)
+      case (_, None) => report.throwError("Could not interpret output schema: " + oSchema.show)
+      case (Some(sIn), Some(sOut)) =>
+        val schemaOut: BSchema = SchemaMaterializer.fieldType(sOut).getRowSchema()
+        val schemaIn: BSchema = SchemaMaterializer.fieldType(sIn).getRowSchema()
+        val classTagOpt = Expr.summon[ClassTag[O]]
+        if (classTagOpt.isEmpty) {
+          report.throwError(s"Could not summon Expr[ClassTag[${TypeRepr.of[O].show}]]")
+        }
+        val classTag = classTagOpt.get
+        To.checkCompatibility(schemaIn, schemaOut)('{ To.unchecked[I, O](using $iSchema, $oSchema, $classTag) })
+          .fold(message => report.throwError(message), identity)
+    }
   }
+
+  private def sequence[T](ls: List[Option[T]]): Option[List[T]] =
+    if ls.exists(_.isEmpty) then None
+    else Some(ls.collect { case Some(x) => x })
+
+  private def interpret[T: scala.quoted.Type](using Quotes): Option[Schema[T]] = 
+    Type.of[T] match {
+      case '[java.lang.Byte]        => Some(Schema.jByteSchema.asInstanceOf[Schema[T]])
+      case '[Array[java.lang.Byte]] => Some(Schema.jBytesSchema.asInstanceOf[Schema[T]])
+      case '[java.lang.Short]       => Some(Schema.jShortSchema.asInstanceOf[Schema[T]])
+      case '[java.lang.Integer]     => Some(Schema.jIntegerSchema.asInstanceOf[Schema[T]])
+      case '[java.lang.Long]        => Some(Schema.jLongSchema.asInstanceOf[Schema[T]])
+      case '[java.lang.Float]       => Some(Schema.jFloatSchema.asInstanceOf[Schema[T]])
+      case '[java.lang.Double]      => Some(Schema.jDoubleSchema.asInstanceOf[Schema[T]])
+      case '[java.math.BigDecimal]  => Some(Schema.jBigDecimalSchema.asInstanceOf[Schema[T]])
+      case '[java.lang.Boolean]     => Some(Schema.jBooleanSchema.asInstanceOf[Schema[T]])
+      case '[java.util.List[u]] =>
+        for (itemSchema) <- interpret[u]
+        yield Schema.jListSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case '[java.util.ArrayList[u]] =>
+        for (itemSchema) <- interpret[u]
+        yield Schema.jArrayListSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case '[java.util.Map[k, v]] =>
+        for {
+          keySchema <- interpret[k]
+          valueSchema <- interpret[v]
+        } yield Schema.jMapSchema(using keySchema, valueSchema).asInstanceOf[Schema[T]]
+      // TODO javaBeanSchema
+      // TODO javaEnumSchema
+      case '[java.time.LocalDate] => Some(Schema.jLocalDate.asInstanceOf[Schema[T]])
+
+      case '[String]      => Some(Schema.stringSchema.asInstanceOf[Schema[T]])
+      case '[Byte]        => Some(Schema.byteSchema.asInstanceOf[Schema[T]])
+      case '[Array[Byte]] => Some(Schema.bytesSchema.asInstanceOf[Schema[T]])
+      case '[Short]       => Some(Schema.sortSchema.asInstanceOf[Schema[T]])
+      case '[Int]         => Some(Schema.intSchema.asInstanceOf[Schema[T]])
+      case '[Long]        => Some(Schema.longSchema.asInstanceOf[Schema[T]])
+      case '[Float]       => Some(Schema.floatSchema.asInstanceOf[Schema[T]])
+      case '[Double]      => Some(Schema.doubleSchema.asInstanceOf[Schema[T]])
+      case '[BigDecimal]  => Some(Schema.bigDecimalSchema.asInstanceOf[Schema[T]])
+      case '[Boolean]     => Some(Schema.booleanSchema.asInstanceOf[Schema[T]])
+      case '[Option[u]] =>
+        for (itemSchema <- interpret[u])
+        yield Schema.optionSchema(using itemSchema).asInstanceOf[Schema[T]]
+      // TODO Array[T]
+      case '[List[u]] =>
+        for (itemSchema <- interpret[u])
+        yield Schema.listSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case '[mutable.ArrayBuffer[u]] =>
+        for (itemSchema <- interpret[u])
+        yield Schema.arrayBufferSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case '[mutable.Buffer[u]] =>
+        for (itemSchema <- interpret[u])
+        yield Schema.bufferSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case '[mutable.Set[u]] =>
+        for (itemSchema <- interpret[u])
+        yield Schema.mutableSetSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case '[Set[u]] =>
+        for (itemSchema <- interpret[u])
+        yield Schema.setSchema(using itemSchema).asInstanceOf[Schema[T]]
+      // TODO SortedSet[T]
+      case '[mutable.ListBuffer[u]] =>
+        for (itemSchema <- interpret[u])
+        yield Schema.listBufferSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case '[Vector[u]] =>
+        for (itemSchema <- interpret[u])
+        yield Schema.vectorSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case '[mutable.Map[k, v]] =>
+        for {
+          keySchema <- interpret[k]
+          valueSchema <- interpret[v]
+        } yield Schema.mutableMapSchema(using keySchema, valueSchema).asInstanceOf[Schema[T]]
+      case '[Map[k, v]] =>
+        for {
+          keySchema <- interpret[k]
+          valueSchema <- interpret[v]
+        } yield Schema.mapSchema(using keySchema, valueSchema).asInstanceOf[Schema[T]]
+      case '[Seq[u]] => 
+        for (itemSchema <- interpret[u])
+        yield Schema.seqSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case '[TraversableOnce[u]] =>
+        for (itemSchema <- interpret[u])
+        yield Schema.traversableOnceSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case '[Iterable[u]] =>
+        for (itemSchema <- interpret[u])
+        yield Schema.iterableSchema(using itemSchema).asInstanceOf[Schema[T]]
+      case _ =>
+        import quotes.reflect._
+        val tp = TypeRepr.of[T]
+        val tpSymbol: Symbol = tp.typeSymbol
+
+        // if case class iterate and recurse, else sorry
+        if tp <:< TypeRepr.of[Product] && tpSymbol.caseFields.nonEmpty then {
+          val schemasOpt: List[Option[(String, Schema[Any])]] = tpSymbol.caseFields.map { (f: Symbol) =>
+            assert(f.isValDef)
+            val fieldName = f.name
+            val fieldType: TypeRepr = tp.memberType(f)
+            fieldType.asType match {
+              // mattern match to create a bind <3
+              case '[u] => interpret[u].asInstanceOf[Option[Schema[Any]]].map(s => (fieldName, s))
+            }
+          }
+          sequence(schemasOpt).map(schemas => Record(schemas.toArray, null, null))
+        } else if tpSymbol.flags.is(Flags.JavaDefined) && scala.util.Try(checkGetterAndSetters[T]).isSuccess then {
+          val schemasOpt = tpSymbol.declaredMethods.collect {
+            case s if s.name.toString.startsWith("get") && s.isDefDef=>
+              val fieldName: String = s.name.toString.drop(3)
+              val fieldType: TypeRepr = tp.memberType(s)
+              fieldType match {
+                case MethodType(_, _, returnTpt) => 
+                  returnTpt.asType match {
+                    case '[u] => interpret[u].asInstanceOf[Option[Schema[Any]]].map(s => (fieldName, s))
+                  }
+              }
+          }
+          // RawRecord is used for JavaBeans, not Record
+          sequence(schemasOpt).map(schemas => Record(schemas.toArray, null, null))
+        } else None
+    }
 }
 
 trait ToMacro {
@@ -36,7 +186,6 @@ trait ToMacro {
    * at compile time.
    * @see To#unsafe
    */
-  // TODO: scala3
-  inline def safe[I, O](inline si: Schema[I], inline so: Schema[O]): To[I, O] =
-    ???
+  inline def safe[I, O](using inline iSchema: Schema[I], inline oSchema: Schema[O]): To[I, O] = 
+    ${ ToMacro.safeImpl('iSchema, 'oSchema) }
 }

scio-core/src/test/scala/com/spotify/scio/JavaBeanA.java

@@ -0,0 +1,28 @@
+package com.spotify.scio;
+
+class JavaBeanA implements java.io.Serializable {
+   private String firstName = null;
+   private String lastName = null;
+   private int age = 0;
+
+   public JavaBeanA() {
+   }
+   public String getFirstName(){
+      return firstName;
+   }
+   public String getLastName(){
+      return lastName;
+   }
+   public int getAge(){
+      return age;
+   }
+   public void setFirstName(String firstName){
+      this.firstName = firstName;
+   }
+   public void setLastName(String lastName){
+      this.lastName = lastName;
+   }
+   public void setAge(int age){
+      this.age = age;
+   }
+}

scio-core/src/test/scala/com/spotify/scio/JavaBeanB.java

@@ -0,0 +1,28 @@
+package com.spotify.scio;
+
+class JavaBeanB implements java.io.Serializable {
+   private String name = null;
+   private String uuid = null;
+   private int money = 0;
+
+   public JavaBeanB() {
+   }
+   public String getName(){
+      return name;
+   }
+   public String getUuid(){
+      return uuid;
+   }
+   public int getMoney(){
+      return money;
+   }
+   public void setName(String name){
+      this.name = name;
+   }
+   public void setUuid(String uuid){
+      this.uuid = uuid;
+   }
+   public void setMoney(int money){
+      this.money = money;
+   }
+}

scio-core/src/test/scala/com/spotify/scio/JavaBeanC.java

@@ -0,0 +1,28 @@
+package com.spotify.scio;
+
+class JavaBeanC implements java.io.Serializable {
+   private String firstName = null;
+   private String lastName = null;
+   private int age = 0;
+
+   public JavaBeanC() {
+   }
+   public String getFirstName(){
+      return firstName;
+   }
+   public String getLastName(){
+      return lastName;
+   }
+   public int getAge(){
+      return age;
+   }
+   public void setFirstName(String firstName){
+      this.firstName = firstName;
+   }
+   public void setLastName(String lastName){
+      this.lastName = lastName;
+   }
+   public void setAge(int age){
+      this.age = age;
+   }
+}

scio-core/src/test/scala/com/spotify/scio/ToSafeSuite.scala

@@ -0,0 +1,64 @@
+package com.spotify.scio
+
+import com.spotify.scio.schemas.To
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+case class JavaListInt(l: java.util.List[java.lang.Integer])
+case class JavaListString(l: java.util.List[java.lang.String])
+case class ListInt(l: List[Int])
+case class JavaSource(b: java.lang.Boolean)
+case class Source(b: Boolean)
+case class Dest(b: Boolean)
+case class Mistake(b: Int)
+case class Mistake2(c: Boolean)
+
+case class Sources(name: String, links: List[Array[Byte]])
+case class Destinations(name: String, links: List[Array[Byte]])
+case class DestinationsWrong(name: String, links: List[Array[Int]])
+
+class ToSafeTest extends AnyFlatSpec with Matchers {
+  "To.safe" should "generate a conversion on compatible flat case class schemas" in {
+    To.safe[Source, Dest]
+  }
+
+  it should "generate a conversion between java.lang.Boolean and Boolean" in {
+    To.safe[JavaSource, Source]
+    To.safe[Source, JavaSource]
+  }
+
+  it should "generate a conversion between java.util.List[java.lang.Integer] and List[Int]" in {
+    To.safe[JavaListInt, ListInt]
+    To.safe[ListInt, JavaListInt]
+  }
+
+  it should "fail on incompatible Java types" in {
+    "To.safe[JavaListString, JavaListInt]" shouldNot compile
+    "To.safe[JavaListString, ListInt]" shouldNot compile
+  }
+
+  it should "fail on incompatible flat case class schemas" in {
+    "To.safe[Source, Mistake2]" shouldNot compile
+    "To.safe[Source, Mistake]" shouldNot compile
+  }
+
+  it should "generate a conversion on compatible nested case class schemas" in {
+    To.safe[Sources, Destinations]
+  }
+
+  it should "fail on incompatible nested case class schemas" in {
+    "To.safe[Sources, DestinationsWrong]" shouldNot compile
+  }
+
+  it should "work with java beans" in {
+    "To.safe[JavaBeanA, JavaBeanB]" shouldNot compile
+    "To.safe[JavaBeanB, JavaBeanA]" shouldNot compile
+
+    "To.safe[JavaBeanB, JavaBeanC]" shouldNot compile
+    "To.safe[JavaBeanC, JavaBeanB]" shouldNot compile
+
+    To.safe[JavaBeanA, JavaBeanC]
+    To.safe[JavaBeanC, JavaBeanA]
+  }
+}