[GitHub] igalshilman commented on a change in pull request #7562: [FLINK-11329][core] Migrate CRowSerializerConfigSnapshot to new TypeSerializerSnapshot interface

[GitBox <gi...@apache.org>]

igalshilman commented on a change in pull request #7562: [FLINK-11329][core] Migrate CRowSerializerConfigSnapshot to new TypeSerializerSnapshot interface
URL: https://github.com/apache/flink/pull/7562#discussion_r250113361
 
 

 ##########
 File path: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/types/CRowSerializer.scala
 ##########
 @@ -80,41 +81,21 @@ class CRowSerializer(val rowSerializer: TypeSerializer[Row]) extends TypeSeriali
   // Serializer configuration snapshotting & compatibility
   // --------------------------------------------------------------------------------------------
 
-  override def snapshotConfiguration(): TypeSerializerConfigSnapshot[CRow] = {
-    new CRowSerializer.CRowSerializerConfigSnapshot(Array(rowSerializer))
-  }
-
-  override def ensureCompatibility(
-      configSnapshot: TypeSerializerConfigSnapshot[_]): CompatibilityResult[CRow] = {
-
-    configSnapshot match {
-      case crowSerializerConfigSnapshot: CRowSerializer.CRowSerializerConfigSnapshot =>
-        val compatResult = CompatibilityUtil.resolveCompatibilityResult(
-          crowSerializerConfigSnapshot.getSingleNestedSerializerAndConfig.f0,
-          classOf[UnloadableDummyTypeSerializer[_]],
-          crowSerializerConfigSnapshot.getSingleNestedSerializerAndConfig.f1,
-          rowSerializer)
-
-        if (compatResult.isRequiresMigration) {
-          if (compatResult.getConvertDeserializer != null) {
-            CompatibilityResult.requiresMigration(
-              new CRowSerializer(
-                new TypeDeserializerAdapter(compatResult.getConvertDeserializer))
-            )
-          } else {
-            CompatibilityResult.requiresMigration()
-          }
-        } else {
-          CompatibilityResult.compatible()
-        }
-
-      case _ => CompatibilityResult.requiresMigration()
-    }
+  override def snapshotConfiguration(): TypeSerializerSnapshot[CRow] = {
+    new CRowSerializerSnapshot(this)
   }
 }
 
 object CRowSerializer {
 
+  /**
+    * [[CRowSerializer]] is not meant to be used for persisting state. In versions 1.6+ there
+    * were changes introduced that resulted in incompatibility in java serialization. Thus one
+    * cannot read state in 1.8+ from snapshot written with previous versions of Flink.
+    *
+    * Moreover this serializer is meant to be dropped once we migrate to the new planner
+    * implementation.
+    */
   class CRowSerializerConfigSnapshot(rowSerializers: Array[TypeSerializer[Row]])
 
 Review comment:
   can you add a `@deprecated` annotation?

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services