Improvement: TypeInformation support for scala.Option (#6952)

TouK · Oct 4, 2024 · 26a9e12 · 26a9e12
1 parent 1f0495b
commit 26a9e12
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 1 deletion.
diff --git a/docs/Changelog.md b/docs/Changelog.md
@@ -4,6 +4,9 @@
 
 #### Highlights
 
+### 1.17-esp
+* [#6952](https://github.com/TouK/nussknacker/pull/6952) Improvement: TypeInformation support for scala.Option
+
 ### 1.17.1 (Not released yet)
 
 * [#6880](https://github.com/TouK/nussknacker/pull/6880) Performance optimization of generating Avro messages with unions

diff --git a/docs/MigrationGuide.md b/docs/MigrationGuide.md
@@ -2,6 +2,10 @@
 
 To see the biggest differences please consult the [changelog](Changelog.md).
 
+## In version 1.17-esp
+* [#6952](https://github.com/TouK/nussknacker/pull/6952) Improvement: TypeInformation support for scala.Option:
+  If you used CaseClassTypeInfoFactory with case classes that contain the Option type, the state won't be restored after the upgrade.
+
 ## In version 1.17.0
 
 ### Code API changes

diff --git a/...la/pl/touk/nussknacker/engine/flink/api/typeinfo/caseclass/CaseClassTypeInfoFactory.scala b/...la/pl/touk/nussknacker/engine/flink/api/typeinfo/caseclass/CaseClassTypeInfoFactory.scala
@@ -5,6 +5,7 @@ import org.apache.flink.api.common.typeinfo.{TypeInfoFactory, TypeInformation}
 import org.apache.flink.api.common.typeutils.TypeSerializer
 import org.apache.flink.api.java.typeutils.TypeExtractor
 import org.apache.flink.api.java.typeutils.runtime.NullableSerializer
+import pl.touk.nussknacker.engine.flink.api.typeinfo.option.OptionTypeInfo
 
 import java.lang.reflect.Type
 import scala.reflect._
@@ -41,7 +42,13 @@ abstract class CaseClassTypeInfoFactory[T <: Product: ClassTag] extends TypeInfo
     val fieldNames = fields.map(_.name.decodedName.toString)
     val fieldTypes = fields.map { field =>
       val fieldClass = mirror.runtimeClass(field.typeSignature)
-      TypeExtractor.getForClass(fieldClass)
+
+      if (classOf[Option[_]].isAssignableFrom(fieldClass)) {
+        val optionTypeClass = mirror.runtimeClass(field.typeSignature.typeArgs.head)
+        new OptionTypeInfo(TypeExtractor.getForClass(optionTypeClass))
+      } else {
+        TypeExtractor.getForClass(fieldClass)
+      }
     }
     (fieldNames, fieldTypes)
   }