This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new c830df80ee5 [FLINK-40120][table] Persist RowData field names in
RowDataSerializerSnapshot (#28717)
c830df80ee5 is described below
commit c830df80ee5723cd05bd6cba3e8a384faa8d022e
Author: Weiqing Yang <[email protected]>
AuthorDate: Fri Jul 17 00:38:20 2026 -0700
[FLINK-40120][table] Persist RowData field names in
RowDataSerializerSnapshot (#28717)
---
.../runtime/typeutils/InternalSerializers.java | 1 +
.../table/runtime/typeutils/RowDataSerializer.java | 55 +++++-
.../typeutils/RowDataSerializerFieldNamesTest.java | 197 +++++++++++++++++++++
3 files changed, 247 insertions(+), 6 deletions(-)
diff --git
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java
index 371dcd136e1..4334591cbd1 100644
---
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java
+++
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalSerializers.java
@@ -108,6 +108,7 @@ public final class InternalSerializers {
MapType mapType = (MapType) type;
return new MapDataSerializer(mapType.getKeyType(),
mapType.getValueType());
case ROW:
+ return new RowDataSerializer((RowType) type);
case STRUCTURED_TYPE:
return new RowDataSerializer(type.getChildren().toArray(new
LogicalType[0]));
case DISTINCT_TYPE:
diff --git
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java
index 7a634f67efe..718fbc20be9 100644
---
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java
+++
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.runtime.typeutils;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil;
import org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate;
import org.apache.flink.api.common.typeutils.TypeSerializer;
@@ -40,6 +41,8 @@ import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.InstantiationUtil;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.Arrays;
import java.util.stream.IntStream;
@@ -51,6 +54,7 @@ public class RowDataSerializer extends
AbstractRowDataSerializer<RowData> {
private BinaryRowDataSerializer binarySerializer;
private final LogicalType[] types;
+ private final @Nullable String[] fieldNames;
private final TypeSerializer[] fieldSerializers;
private final RowData.FieldGetter[] fieldGetters;
@@ -62,7 +66,8 @@ public class RowDataSerializer extends
AbstractRowDataSerializer<RowData> {
rowType.getChildren().toArray(new LogicalType[0]),
rowType.getChildren().stream()
.map(InternalSerializers::create)
- .toArray(TypeSerializer[]::new));
+ .toArray(TypeSerializer[]::new),
+ rowType.getFieldNames().toArray(new String[0]));
}
public RowDataSerializer(LogicalType... types) {
@@ -74,7 +79,15 @@ public class RowDataSerializer extends
AbstractRowDataSerializer<RowData> {
}
public RowDataSerializer(LogicalType[] types, TypeSerializer<?>[]
fieldSerializers) {
+ this(types, fieldSerializers, null);
+ }
+
+ private RowDataSerializer(
+ LogicalType[] types,
+ TypeSerializer<?>[] fieldSerializers,
+ @Nullable String[] fieldNames) {
this.types = types;
+ this.fieldNames = fieldNames;
this.fieldSerializers = fieldSerializers;
this.binarySerializer = new BinaryRowDataSerializer(types.length);
this.fieldGetters =
@@ -83,13 +96,24 @@ public class RowDataSerializer extends
AbstractRowDataSerializer<RowData> {
.toArray(RowData.FieldGetter[]::new);
}
+ @VisibleForTesting
+ @Nullable
+ String[] getFieldNames() {
+ return fieldNames;
+ }
+
+ @VisibleForTesting
+ TypeSerializer[] fieldSerializers() {
+ return fieldSerializers;
+ }
+
@Override
public TypeSerializer<RowData> duplicate() {
TypeSerializer<?>[] duplicateFieldSerializers = new
TypeSerializer[fieldSerializers.length];
for (int i = 0; i < fieldSerializers.length; i++) {
duplicateFieldSerializers[i] = fieldSerializers[i].duplicate();
}
- return new RowDataSerializer(types, duplicateFieldSerializers);
+ return new RowDataSerializer(types, duplicateFieldSerializers,
fieldNames);
}
@Override
@@ -274,14 +298,15 @@ public class RowDataSerializer extends
AbstractRowDataSerializer<RowData> {
@Override
public TypeSerializerSnapshot<RowData> snapshotConfiguration() {
- return new RowDataSerializerSnapshot(types, fieldSerializers);
+ return new RowDataSerializerSnapshot(types, fieldSerializers,
fieldNames);
}
/** {@link TypeSerializerSnapshot} for {@link BinaryRowDataSerializer}. */
public static final class RowDataSerializerSnapshot implements
TypeSerializerSnapshot<RowData> {
- private static final int CURRENT_VERSION = 3;
+ private static final int CURRENT_VERSION = 4;
private LogicalType[] types;
+ private @Nullable String[] fieldNames;
private NestedSerializersSnapshotDelegate
nestedSerializersSnapshotDelegate;
@SuppressWarnings("unused")
@@ -289,8 +314,10 @@ public class RowDataSerializer extends
AbstractRowDataSerializer<RowData> {
// this constructor is used when restoring from a
checkpoint/savepoint.
}
- RowDataSerializerSnapshot(LogicalType[] types, TypeSerializer[]
serializers) {
+ RowDataSerializerSnapshot(
+ LogicalType[] types, TypeSerializer[] serializers, @Nullable
String[] fieldNames) {
this.types = types;
+ this.fieldNames = fieldNames;
this.nestedSerializersSnapshotDelegate =
new NestedSerializersSnapshotDelegate(serializers);
}
@@ -308,6 +335,14 @@ public class RowDataSerializer extends
AbstractRowDataSerializer<RowData> {
InstantiationUtil.serializeObject(stream, previousType);
}
nestedSerializersSnapshotDelegate.writeNestedSerializerSnapshots(out);
+ // A false flag means no field names (null); the count is not
stored because it always
+ // equals the number of fields.
+ out.writeBoolean(fieldNames != null);
+ if (fieldNames != null) {
+ for (String fieldName : fieldNames) {
+ out.writeUTF(fieldName);
+ }
+ }
}
@Override
@@ -327,12 +362,20 @@ public class RowDataSerializer extends
AbstractRowDataSerializer<RowData> {
this.nestedSerializersSnapshotDelegate =
NestedSerializersSnapshotDelegate.readNestedSerializerSnapshots(
in, userCodeClassLoader);
+ if (readVersion >= 4 && in.readBoolean()) {
+ fieldNames = new String[types.length];
+ for (int i = 0; i < types.length; i++) {
+ fieldNames[i] = in.readUTF();
+ }
+ }
}
@Override
public RowDataSerializer restoreSerializer() {
return new RowDataSerializer(
- types,
nestedSerializersSnapshotDelegate.getRestoredNestedSerializers());
+ types,
+
nestedSerializersSnapshotDelegate.getRestoredNestedSerializers(),
+ fieldNames);
}
@Override
diff --git
a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerFieldNamesTest.java
b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerFieldNamesTest.java
new file mode 100644
index 00000000000..7f76786c9a2
--- /dev/null
+++
b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerFieldNamesTest.java
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.typeutils;
+
+import org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import
org.apache.flink.table.runtime.typeutils.RowDataSerializer.RowDataSerializerSnapshot;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.util.InstantiationUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for the top-level field names carried by {@link RowDataSerializer}
and the V4 {@link
+ * RowDataSerializerSnapshot} format that persists them unconditionally. These
assert metadata
+ * (field names, snapshot version, V3 backward compatibility, compatibility
invariance) rather than
+ * per-record serialization round trips, which are covered by {@link
RowDataSerializerTest}.
+ */
+class RowDataSerializerFieldNamesTest {
+
+ private static RowType rowType(String[] names, LogicalType... types) {
+ return RowType.of(types, names);
+ }
+
+ @Test
+ void internalSerializersCreateCarriesTopLevelNames() {
+ RowType rowType = rowType(new String[] {"f0", "f1"}, new IntType(),
new BigIntType());
+
+
assertThat(InternalSerializers.create(rowType).getFieldNames()).containsExactly("f0",
"f1");
+ }
+
+ @Test
+ void typesOnlyConstructorsLeaveNamesNull() {
+ assertThat(new RowDataSerializer(new IntType(), new
BigIntType()).getFieldNames()).isNull();
+ assertThat(
+ new RowDataSerializer(
+ new LogicalType[] {new IntType(), new
BigIntType()},
+ new TypeSerializer[] {
+ InternalSerializers.create(new
IntType()),
+ InternalSerializers.create(new
BigIntType())
+ })
+ .getFieldNames())
+ .isNull();
+ }
+
+ @Test
+ void duplicatePreservesNames() {
+ RowType rowType = rowType(new String[] {"f0", "f1"}, new IntType(),
new BigIntType());
+ RowDataSerializer duplicate =
+ (RowDataSerializer)
InternalSerializers.create(rowType).duplicate();
+
+ assertThat(duplicate.getFieldNames()).containsExactly("f0", "f1");
+ }
+
+ @Test
+ void snapshotVersionIsFour() {
+ RowType rowType = rowType(new String[] {"f0", "f1"}, new IntType(),
new BigIntType());
+
+
assertThat(InternalSerializers.create(rowType).snapshotConfiguration().getCurrentVersion())
+ .isEqualTo(4);
+ }
+
+ @Test
+ void namesSurviveSnapshotRoundTripIncludingNestedRow() throws IOException {
+ RowType nested = rowType(new String[] {"a", "b"}, new IntType(),
VarCharType.STRING_TYPE);
+ RowType rowType = rowType(new String[] {"outer", "tail"}, nested, new
IntType());
+
+ RowDataSerializer restored =
+ (RowDataSerializer)
+
roundTrip(InternalSerializers.create(rowType).snapshotConfiguration())
+ .restoreSerializer();
+
+ assertThat(restored.getFieldNames()).containsExactly("outer", "tail");
+ TypeSerializer<?> nestedChild = restored.fieldSerializers()[0];
+ assertThat(nestedChild).isInstanceOf(RowDataSerializer.class);
+ assertThat(((RowDataSerializer)
nestedChild).getFieldNames()).containsExactly("a", "b");
+ }
+
+ @Test
+ void readsV3SnapshotWithoutFieldNames() throws IOException {
+ LogicalType[] types = {new IntType(), new BigIntType()};
+ TypeSerializer<?>[] fieldSerializers = {
+ InternalSerializers.create(types[0]),
InternalSerializers.create(types[1])
+ };
+
+ // Hand-craft a genuine V3 stream: versioned header + V3 body, where
the V3 body is the
+ // types section followed directly by the nested-delegate section,
with no names block.
+ DataOutputSerializer out = new DataOutputSerializer(128);
+ out.writeUTF(RowDataSerializerSnapshot.class.getName());
+ out.writeInt(3);
+ out.writeInt(types.length);
+ DataOutputViewStream stream = new DataOutputViewStream(out);
+ for (LogicalType type : types) {
+ InstantiationUtil.serializeObject(stream, type);
+ }
+ new
NestedSerializersSnapshotDelegate(fieldSerializers).writeNestedSerializerSnapshots(out);
+
+ DataInputDeserializer in = new
DataInputDeserializer(out.getCopyOfBuffer());
+ TypeSerializerSnapshot<RowData> restored =
+ TypeSerializerSnapshot.readVersionedSnapshot(in,
getClass().getClassLoader());
+
+ RowDataSerializer serializer = (RowDataSerializer)
restored.restoreSerializer();
+ assertThat(serializer.getFieldNames()).isNull();
+
+ // The nested delegate was read at the correct offset (no phantom
names consumed): the
+ // restored serializer round-trips a sample row.
+ GenericRowData sample = GenericRowData.of(7, 11L);
+ DataOutputSerializer rowOut = new DataOutputSerializer(32);
+ serializer.serialize(sample, rowOut);
+ RowData deserialized =
+ serializer.deserialize(new
DataInputDeserializer(rowOut.getCopyOfBuffer()));
+ assertThat(deserialized.getInt(0)).isEqualTo(7);
+ assertThat(deserialized.getLong(1)).isEqualTo(11L);
+ }
+
+ @Test
+ void equalsAndHashCodeIgnoreFieldNames() {
+ RowDataSerializer named =
+ InternalSerializers.create(
+ rowType(new String[] {"f0", "f1"}, new IntType(), new
BigIntType()));
+ RowDataSerializer differentlyNamed =
+ InternalSerializers.create(
+ rowType(new String[] {"x", "y"}, new IntType(), new
BigIntType()));
+
+ assertThat(named).isEqualTo(differentlyNamed);
+ assertThat(named.hashCode()).isEqualTo(differentlyNamed.hashCode());
+ }
+
+ @Test
+ void differentFieldNamesRemainCompatibleWhenTypesMatch() {
+ TypeSerializerSchemaCompatibility<RowData> compatibility =
+ resolve(
+ rowType(new String[] {"f0", "f1"}, new IntType(), new
BigIntType()),
+ rowType(new String[] {"x", "y"}, new IntType(), new
BigIntType()));
+
+ assertThat(compatibility.isCompatibleAsIs()).isTrue();
+ }
+
+ @Test
+ void differentTypesRemainIncompatibleRegardlessOfNames() {
+ TypeSerializerSchemaCompatibility<RowData> compatibility =
+ resolve(
+ rowType(new String[] {"f0", "f1"}, new IntType(), new
BigIntType()),
+ rowType(new String[] {"f0", "f1"}, new IntType(), new
IntType()));
+
+ assertThat(compatibility.isIncompatible()).isTrue();
+ }
+
+ private static TypeSerializerSchemaCompatibility<RowData> resolve(
+ RowType newType, RowType oldType) {
+ TypeSerializerSnapshot<RowData> newSnapshot =
+ InternalSerializers.create(newType).snapshotConfiguration();
+ TypeSerializerSnapshot<RowData> oldSnapshot =
+ InternalSerializers.create(oldType).snapshotConfiguration();
+ return newSnapshot.resolveSchemaCompatibility(oldSnapshot);
+ }
+
+ private static TypeSerializerSnapshot<RowData> roundTrip(
+ TypeSerializerSnapshot<RowData> snapshot) throws IOException {
+ DataOutputSerializer out = new DataOutputSerializer(128);
+ TypeSerializerSnapshot.writeVersionedSnapshot(out, snapshot);
+ DataInputDeserializer in = new
DataInputDeserializer(out.getCopyOfBuffer());
+ return TypeSerializerSnapshot.readVersionedSnapshot(
+ in, RowDataSerializerFieldNamesTest.class.getClassLoader());
+ }
+}
