[FLINK-17520] [core] Rework all implementations of CompositeTypeSeria…

…lizerSnapshot#resolveOuterSchemaCompatibility
tongcheng-elong · May 19, 2020 · 55a21bc · 55a21bc
1 parent 71e3339
commit 55a21bc
Show file tree

Hide file tree

Showing 5 changed files with 20 additions and 10 deletions.
diff --git a/.../main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializerSnapshot.java b/.../main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializerSnapshot.java
@@ -78,8 +78,10 @@ protected void readOuterSnapshot(int readOuterSnapshotVersion, DataInputView in,
  }
 
  @Override
- protected boolean isOuterSnapshotCompatible(GenericArraySerializer<C> newSerializer) {
- return this.componentClass == newSerializer.getComponentClass();
+ protected OuterSchemaCompatibility resolveOuterSchemaCompatibility(GenericArraySerializer<C> newSerializer) {
+ return (this.componentClass == newSerializer.getComponentClass())
+ ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
+ : OuterSchemaCompatibility.INCOMPATIBLE;
  }
 
  @Override

diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java
@@ -348,8 +348,10 @@ protected void readOuterSnapshot(int readOuterSnapshotVersion, DataInputView in,
  }
 
  @Override
- protected boolean isOuterSnapshotCompatible(NullableSerializer<T> newSerializer) {
- return nullPaddingLength == newSerializer.nullPaddingLength();
+ protected OuterSchemaCompatibility resolveOuterSchemaCompatibility(NullableSerializer<T> newSerializer) {
+ return (nullPaddingLength == newSerializer.nullPaddingLength())
+ ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
+ : OuterSchemaCompatibility.INCOMPATIBLE;
  }
  }
 

diff --git a/.../src/main/java/org/apache/flink/api/scala/typeutils/ScalaCaseClassSerializerSnapshot.java b/.../src/main/java/org/apache/flink/api/scala/typeutils/ScalaCaseClassSerializerSnapshot.java
@@ -103,7 +103,9 @@ protected void readOuterSnapshot(int readOuterSnapshotVersion, DataInputView in,
  }
 
  @Override
- protected boolean isOuterSnapshotCompatible(ScalaCaseClassSerializer<T> newSerializer) {
- return Objects.equals(type, newSerializer.getTupleClass());
+ protected OuterSchemaCompatibility resolveOuterSchemaCompatibility(ScalaCaseClassSerializer<T> newSerializer) {
+ return (Objects.equals(type, newSerializer.getTupleClass()))
+ ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
+ : OuterSchemaCompatibility.INCOMPATIBLE;
  }
 }
diff --git a/...ala/src/main/java/org/apache/flink/api/scala/typeutils/TraversableSerializerSnapshot.java b/...ala/src/main/java/org/apache/flink/api/scala/typeutils/TraversableSerializerSnapshot.java
@@ -98,7 +98,9 @@ protected void readOuterSnapshot(
  }
 
  @Override
- protected boolean isOuterSnapshotCompatible(TraversableSerializer<T, E> newSerializer) {
- return cbfCode.equals(newSerializer.cbfCode());
+ protected OuterSchemaCompatibility resolveOuterSchemaCompatibility(TraversableSerializer<T, E> newSerializer) {
+ return (cbfCode.equals(newSerializer.cbfCode()))
+ ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
+ : OuterSchemaCompatibility.INCOMPATIBLE;
  }
 }
diff --git a/...src/main/java/org/apache/flink/api/scala/typeutils/Tuple2CaseClassSerializerSnapshot.java b/...src/main/java/org/apache/flink/api/scala/typeutils/Tuple2CaseClassSerializerSnapshot.java
@@ -95,8 +95,10 @@ protected void readOuterSnapshot(int readOuterSnapshotVersion, DataInputView in,
  }
 
  @Override
- protected boolean isOuterSnapshotCompatible(ScalaCaseClassSerializer<Tuple2<T1, T2>> newSerializer) {
- return Objects.equals(type, newSerializer.getTupleClass());
+ protected OuterSchemaCompatibility resolveOuterSchemaCompatibility(ScalaCaseClassSerializer<Tuple2<T1, T2>> newSerializer) {
+ return (Objects.equals(type, newSerializer.getTupleClass()))
+ ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
+ : OuterSchemaCompatibility.INCOMPATIBLE;
  }
 
  private static <T1, T2> Class<ScalaCaseClassSerializer<scala.Tuple2<T1, T2>>> correspondingSerializerClass() {