Copilot commented on code in PR #28973:
URL: https://github.com/apache/flink/pull/28973#discussion_r4282111884
##########
flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java:
##########
@@ -387,25 +483,236 @@ public TypeSerializerSchemaCompatibility<RowData>
resolveSchemaCompatibility(
RowDataSerializerSnapshot oldRowDataSerializerSnapshot =
(RowDataSerializerSnapshot) oldSerializerSnapshot;
- if (!Arrays.equals(types, oldRowDataSerializerSnapshot.types)) {
+ // A side that carries no names cannot disagree with anything:
without names a reorder
+ // is undetectable, so identical types keep meaning "compatible as
is" there, exactly as
+ // they do unarmed.
+ boolean namesDisagree =
+ stateSchemaEvolutionEnabled
+ && fieldNames != null
+ && oldRowDataSerializerSnapshot.fieldNames != null
+ && !Arrays.equals(fieldNames,
oldRowDataSerializerSnapshot.fieldNames);
+
+ // Identical positional layout: the nested composite path. Equal
types at equal
+ // positions say nothing about which field is which, so an armed
resolution has to see
+ // the names agree as well before it can treat the layout as
unchanged -- otherwise a
+ // reorder or a rename among same-typed fields resolves here as
needing no migration
+ // and leaves every value sitting under a neighbour's name.
+ if (Arrays.equals(types, oldRowDataSerializerSnapshot.types) &&
!namesDisagree) {
+
CompositeTypeSerializerUtil.IntermediateCompatibilityResult<RowData>
+ intermediateResult =
+ CompositeTypeSerializerUtil
+
.constructIntermediateCompatibilityResult(
+
nestedSerializersSnapshotDelegate
+
.getNestedSerializerSnapshots(),
+ oldRowDataSerializerSnapshot
+
.nestedSerializersSnapshotDelegate
+
.getNestedSerializerSnapshots());
+
+ if
(intermediateResult.isCompatibleWithReconfiguredSerializer()) {
+ RowDataSerializer reconfiguredCompositeSerializer =
restoreSerializer();
+ return
TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
+ reconfiguredCompositeSerializer);
+ }
+
+ return intermediateResult.getFinalResult();
+ }
+
+ if (!stateSchemaEvolutionEnabled) {
return TypeSerializerSchemaCompatibility.incompatible();
}
-
CompositeTypeSerializerUtil.IntermediateCompatibilityResult<RowData>
- intermediateResult =
-
CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(
- nestedSerializersSnapshotDelegate
- .getNestedSerializerSnapshots(),
-
oldRowDataSerializerSnapshot.nestedSerializersSnapshotDelegate
- .getNestedSerializerSnapshots());
-
- if (intermediateResult.isCompatibleWithReconfiguredSerializer()) {
- RowDataSerializer reconfiguredCompositeSerializer =
restoreSerializer();
- return
TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
- reconfiguredCompositeSerializer);
+ // The new side must carry field names. A name-less new serializer
is reachable -- a
+ // structured type resolves to one -- and admitting it would open
a permanent name-less
+ // evolution channel rather than a ramp for savepoints taken
before names were stored.
+ if (fieldNames == null) {
+ return TypeSerializerSchemaCompatibility.incompatible();
}
- return intermediateResult.getFinalResult();
+ return oldRowDataSerializerSnapshot.fieldNames != null
+ ? checkNameBasedEvolution(oldRowDataSerializerSnapshot)
+ : checkPositionalEvolution(oldRowDataSerializerSnapshot);
+ }
+
+ private TypeSerializerSchemaCompatibility<RowData>
checkNameBasedEvolution(
+ RowDataSerializerSnapshot oldSnapshot) {
+ int[] oldToNew = buildNameMapping(oldSnapshot.fieldNames,
this.fieldNames);
+ int[] newToOld = buildNameMapping(this.fieldNames,
oldSnapshot.fieldNames);
+
+ // (A) Every new-only field (no matching old field) must be
nullable.
+ for (int newPos = 0; newPos < newToOld.length; newPos++) {
+ if (newToOld[newPos] == -1 && !types[newPos].isNullable()) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+
+ // (B) Every old field must survive with a compatible type, and
nested snapshots are
+ // aligned old->new so nested ROW evolution can recurse. Leaf
(non-ROW) fields
+ // require an exactly equal type; ROW fields defer to the
nested recursion in (C).
+ TypeSerializerSnapshot<?>[] newNested =
+
nestedSerializersSnapshotDelegate.getNestedSerializerSnapshots();
+ TypeSerializerSnapshot<?>[] alignedNewNested =
+ new TypeSerializerSnapshot<?>[oldSnapshot.types.length];
+ for (int oldPos = 0; oldPos < oldToNew.length; oldPos++) {
+ int newPos = oldToNew[oldPos];
+ if (newPos == -1) {
+ return TypeSerializerSchemaCompatibility.incompatible();
// field removed
+ }
+ LogicalType oldType = oldSnapshot.types[oldPos];
+ LogicalType newType = types[newPos];
+ if (!bothRowWithSameNullability(oldType, newType) &&
!oldType.equals(newType)) {
+ return TypeSerializerSchemaCompatibility.incompatible();
// leaf type changed
+ }
+ alignedNewNested[oldPos] = newNested[newPos];
+ }
+
+ // (C) Recurse into the aligned nested snapshot pairs.
+ return resolveAlignedNested(alignedNewNested, oldSnapshot);
+ }
+
+ /**
+ * Resolves against a prior snapshot that carries no field names,
matching fields by
+ * position.
+ *
+ * <p>Position is a stable identity only for an append. An insertion
in the middle is
+ * indistinguishable from a retype plus an append, and the two demand
opposite migrations,
+ * so the old layout has to be a prefix of the new one.
+ */
+ private TypeSerializerSchemaCompatibility<RowData>
checkPositionalEvolution(
+ RowDataSerializerSnapshot oldSnapshot) {
+ if (types.length < oldSnapshot.types.length) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ for (int i = 0; i < oldSnapshot.types.length; i++) {
+ LogicalType oldType = oldSnapshot.types[i];
+ LogicalType newType = types[i];
+ if (!bothRowWithSameNullability(oldType, newType) &&
!oldType.equals(newType)) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+ for (int i = oldSnapshot.types.length; i < types.length; i++) {
+ if (!types[i].isNullable()) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+
+ // constructIntermediateCompatibilityResult requires both arrays
to have the same
+ // length, so only the prefix the old layout covers is handed to
it.
+ TypeSerializerSnapshot<?>[] alignedNewNested =
+ Arrays.copyOf(
+
nestedSerializersSnapshotDelegate.getNestedSerializerSnapshots(),
+ oldSnapshot.types.length);
+ return resolveAlignedNested(alignedNewNested, oldSnapshot);
+ }
+
+ private TypeSerializerSchemaCompatibility<RowData>
resolveAlignedNested(
+ TypeSerializerSnapshot<?>[] alignedNewNested,
+ RowDataSerializerSnapshot oldSnapshot) {
+
CompositeTypeSerializerUtil.IntermediateCompatibilityResult<RowData> nested =
+
CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(
+ alignedNewNested,
+ oldSnapshot.nestedSerializersSnapshotDelegate
+ .getNestedSerializerSnapshots());
+ // A reconfigured nested serializer is deliberately not propagated
here, unlike on the
+ // identical-layout path. Reconfiguration exists so a new
serializer can read old bytes;
+ // once the values have been remapped there are no old bytes left,
because the migrated
+ // row is re-encoded by the state's own new serializer.
+ return nested.isIncompatible()
+ ? TypeSerializerSchemaCompatibility.incompatible()
+ :
TypeSerializerSchemaCompatibility.compatibleAfterMigration();
Review Comment:
Rewriting the outer row does not necessarily re-encode its leaf payloads.
For example, adding a nullable field while changing a RAW field's Kryo
registration order passes logical-type equality (`KryoSerializer.equals`
ignores map order), but its nested snapshot requires reconfiguration.
`getNewRowData` retains the byte-backed `BinaryRawValueData`, and
`AbstractBinaryWriter.writeRawValue` copies its already-materialized bytes
unchanged. Accepting migration with the unreconfigured target serializer
therefore leaves old registration IDs that can make RAW values unreadable or
misinterpreted. Reject unsupported leaf compatibility results, including when
another field's migration masks a reconfiguration result, or decode the
payloads with the old serializer and re-encode them with the target serializer.
Cover this with a binary-row regression test.
##########
flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java:
##########
@@ -387,25 +483,236 @@ public TypeSerializerSchemaCompatibility<RowData>
resolveSchemaCompatibility(
RowDataSerializerSnapshot oldRowDataSerializerSnapshot =
(RowDataSerializerSnapshot) oldSerializerSnapshot;
- if (!Arrays.equals(types, oldRowDataSerializerSnapshot.types)) {
+ // A side that carries no names cannot disagree with anything:
without names a reorder
+ // is undetectable, so identical types keep meaning "compatible as
is" there, exactly as
+ // they do unarmed.
+ boolean namesDisagree =
+ stateSchemaEvolutionEnabled
+ && fieldNames != null
+ && oldRowDataSerializerSnapshot.fieldNames != null
+ && !Arrays.equals(fieldNames,
oldRowDataSerializerSnapshot.fieldNames);
+
+ // Identical positional layout: the nested composite path. Equal
types at equal
+ // positions say nothing about which field is which, so an armed
resolution has to see
+ // the names agree as well before it can treat the layout as
unchanged -- otherwise a
+ // reorder or a rename among same-typed fields resolves here as
needing no migration
+ // and leaves every value sitting under a neighbour's name.
+ if (Arrays.equals(types, oldRowDataSerializerSnapshot.types) &&
!namesDisagree) {
+
CompositeTypeSerializerUtil.IntermediateCompatibilityResult<RowData>
+ intermediateResult =
+ CompositeTypeSerializerUtil
+
.constructIntermediateCompatibilityResult(
+
nestedSerializersSnapshotDelegate
+
.getNestedSerializerSnapshots(),
+ oldRowDataSerializerSnapshot
+
.nestedSerializersSnapshotDelegate
+
.getNestedSerializerSnapshots());
+
+ if
(intermediateResult.isCompatibleWithReconfiguredSerializer()) {
+ RowDataSerializer reconfiguredCompositeSerializer =
restoreSerializer();
+ return
TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
+ reconfiguredCompositeSerializer);
+ }
+
+ return intermediateResult.getFinalResult();
+ }
+
+ if (!stateSchemaEvolutionEnabled) {
return TypeSerializerSchemaCompatibility.incompatible();
}
-
CompositeTypeSerializerUtil.IntermediateCompatibilityResult<RowData>
- intermediateResult =
-
CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(
- nestedSerializersSnapshotDelegate
- .getNestedSerializerSnapshots(),
-
oldRowDataSerializerSnapshot.nestedSerializersSnapshotDelegate
- .getNestedSerializerSnapshots());
-
- if (intermediateResult.isCompatibleWithReconfiguredSerializer()) {
- RowDataSerializer reconfiguredCompositeSerializer =
restoreSerializer();
- return
TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
- reconfiguredCompositeSerializer);
+ // The new side must carry field names. A name-less new serializer
is reachable -- a
+ // structured type resolves to one -- and admitting it would open
a permanent name-less
+ // evolution channel rather than a ramp for savepoints taken
before names were stored.
+ if (fieldNames == null) {
+ return TypeSerializerSchemaCompatibility.incompatible();
}
- return intermediateResult.getFinalResult();
+ return oldRowDataSerializerSnapshot.fieldNames != null
+ ? checkNameBasedEvolution(oldRowDataSerializerSnapshot)
+ : checkPositionalEvolution(oldRowDataSerializerSnapshot);
+ }
+
+ private TypeSerializerSchemaCompatibility<RowData>
checkNameBasedEvolution(
+ RowDataSerializerSnapshot oldSnapshot) {
+ int[] oldToNew = buildNameMapping(oldSnapshot.fieldNames,
this.fieldNames);
+ int[] newToOld = buildNameMapping(this.fieldNames,
oldSnapshot.fieldNames);
+
+ // (A) Every new-only field (no matching old field) must be
nullable.
+ for (int newPos = 0; newPos < newToOld.length; newPos++) {
+ if (newToOld[newPos] == -1 && !types[newPos].isNullable()) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+
+ // (B) Every old field must survive with a compatible type, and
nested snapshots are
+ // aligned old->new so nested ROW evolution can recurse. Leaf
(non-ROW) fields
+ // require an exactly equal type; ROW fields defer to the
nested recursion in (C).
+ TypeSerializerSnapshot<?>[] newNested =
+
nestedSerializersSnapshotDelegate.getNestedSerializerSnapshots();
+ TypeSerializerSnapshot<?>[] alignedNewNested =
+ new TypeSerializerSnapshot<?>[oldSnapshot.types.length];
+ for (int oldPos = 0; oldPos < oldToNew.length; oldPos++) {
+ int newPos = oldToNew[oldPos];
+ if (newPos == -1) {
+ return TypeSerializerSchemaCompatibility.incompatible();
// field removed
+ }
+ LogicalType oldType = oldSnapshot.types[oldPos];
+ LogicalType newType = types[newPos];
+ if (!bothRowWithSameNullability(oldType, newType) &&
!oldType.equals(newType)) {
+ return TypeSerializerSchemaCompatibility.incompatible();
// leaf type changed
+ }
+ alignedNewNested[oldPos] = newNested[newPos];
+ }
+
+ // (C) Recurse into the aligned nested snapshot pairs.
+ return resolveAlignedNested(alignedNewNested, oldSnapshot);
+ }
+
+ /**
+ * Resolves against a prior snapshot that carries no field names,
matching fields by
+ * position.
+ *
+ * <p>Position is a stable identity only for an append. An insertion
in the middle is
+ * indistinguishable from a retype plus an append, and the two demand
opposite migrations,
+ * so the old layout has to be a prefix of the new one.
+ */
+ private TypeSerializerSchemaCompatibility<RowData>
checkPositionalEvolution(
+ RowDataSerializerSnapshot oldSnapshot) {
+ if (types.length < oldSnapshot.types.length) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ for (int i = 0; i < oldSnapshot.types.length; i++) {
+ LogicalType oldType = oldSnapshot.types[i];
+ LogicalType newType = types[i];
+ if (!bothRowWithSameNullability(oldType, newType) &&
!oldType.equals(newType)) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+ for (int i = oldSnapshot.types.length; i < types.length; i++) {
+ if (!types[i].isNullable()) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+
+ // constructIntermediateCompatibilityResult requires both arrays
to have the same
+ // length, so only the prefix the old layout covers is handed to
it.
+ TypeSerializerSnapshot<?>[] alignedNewNested =
+ Arrays.copyOf(
+
nestedSerializersSnapshotDelegate.getNestedSerializerSnapshots(),
+ oldSnapshot.types.length);
+ return resolveAlignedNested(alignedNewNested, oldSnapshot);
+ }
+
+ private TypeSerializerSchemaCompatibility<RowData>
resolveAlignedNested(
+ TypeSerializerSnapshot<?>[] alignedNewNested,
+ RowDataSerializerSnapshot oldSnapshot) {
+
CompositeTypeSerializerUtil.IntermediateCompatibilityResult<RowData> nested =
+
CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(
+ alignedNewNested,
+ oldSnapshot.nestedSerializersSnapshotDelegate
+ .getNestedSerializerSnapshots());
+ // A reconfigured nested serializer is deliberately not propagated
here, unlike on the
+ // identical-layout path. Reconfiguration exists so a new
serializer can read old bytes;
+ // once the values have been remapped there are no old bytes left,
because the migrated
+ // row is re-encoded by the state's own new serializer.
+ return nested.isIncompatible()
+ ? TypeSerializerSchemaCompatibility.incompatible()
+ :
TypeSerializerSchemaCompatibility.compatibleAfterMigration();
+ }
+
+ /**
+ * Whether a field pair may skip the exact type check and have its
contents compared by the
+ * nested snapshot recursion instead.
+ *
+ * <p>That recursion compares the ROW's fields and never the ROW's own
nullability, so the
+ * nullability has to be settled here. Without it, narrowing a
nullable ROW field to NOT
+ * NULL is accepted and {@code migrate} then carries an existing null
straight into the
+ * non-null field. Requiring equality matches how a leaf field is
treated, since {@link
+ * LogicalType#equals} compares nullability too.
+ */
+ private static boolean bothRowWithSameNullability(
+ LogicalType oldType, LogicalType newType) {
+ return oldType.getTypeRoot() == LogicalTypeRoot.ROW
+ && newType.getTypeRoot() == LogicalTypeRoot.ROW
+ && oldType.isNullable() == newType.isNullable();
+ }
+
+ @Override
+ public RowData migrate(
+ TypeSerializerSnapshot<RowData> oldSerializerSnapshot, RowData
value) {
+ if (value == null) {
+ return null;
+ }
Review Comment:
`TtlAwareSerializer.migrateValueFromPriorSerializer` also invokes this hook
when TTL is enabled or disabled during restore, even with schema evolution off.
A same-typed rename such as `(a INT)` → `(renamed INT)` passes positional
compatibility, but this unconditional name-based remap replaces the stored
value with null. Same-typed reorders also change values unexpectedly. Return
the input unchanged when `stateSchemaEvolutionEnabled` is false, and add a
regression test combining a TTL transition with a same-typed rename or reorder.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
