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 371dcd136e1d56..4334591cbd15b3 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 @@ private static TypeSerializer createInternal(LogicalType type) { 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 7a634f67efe047..b399d1884de38c 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.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 { 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 RowDataSerializer(RowType rowType) { 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 RowDataSerializer(LogicalType... types) { } 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 RowDataSerializer(LogicalType[] types, TypeSerializer[] fieldSerialize .toArray(RowData.FieldGetter[]::new); } + @VisibleForTesting + @Nullable + String[] getFieldNames() { + return fieldNames; + } + + @VisibleForTesting + TypeSerializer[] fieldSerializers() { + return fieldSerializers; + } + @Override public TypeSerializer 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 int getLength() { @Override public TypeSerializerSnapshot snapshotConfiguration() { - return new RowDataSerializerSnapshot(types, fieldSerializers); + return new RowDataSerializerSnapshot(types, fieldSerializers, fieldNames); } /** {@link TypeSerializerSnapshot} for {@link BinaryRowDataSerializer}. */ public static final class RowDataSerializerSnapshot implements TypeSerializerSnapshot { - 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 RowDataSerializerSnapshot() { // 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); } @@ -300,6 +327,20 @@ public int getCurrentVersion() { return CURRENT_VERSION; } + /** + * Returns the top-level {@link RowData} field names captured when this snapshot was + * written. + * + *

Returns {@code null} for legacy snapshots written before field-name persistence was + * introduced (snapshot version 3 and earlier). When present, the names are the top-level + * {@link RowType} field names and correspond one-to-one, in order, with the field types. + * + * @return the top-level field names, or {@code null} if the snapshot did not persist them + */ + public @Nullable String[] getFieldNames() { + return fieldNames; + } + @Override public void writeSnapshot(DataOutputView out) throws IOException { out.writeInt(types.length); @@ -308,6 +349,14 @@ public void writeSnapshot(DataOutputView out) throws IOException { 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 +376,22 @@ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCode this.nestedSerializersSnapshotDelegate = NestedSerializersSnapshotDelegate.readNestedSerializerSnapshots( in, userCodeClassLoader); + if (readVersion >= 4) { + if (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 00000000000000..745d9d07aba9bc --- /dev/null +++ b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerFieldNamesTest.java @@ -0,0 +1,210 @@ +/* + * 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 snapshotPublicGetterReturnsPersistedNames() { + RowType rowType = rowType(new String[] {"f0", "f1"}, new IntType(), new BigIntType()); + + RowDataSerializerSnapshot snapshot = + (RowDataSerializerSnapshot) + InternalSerializers.create(rowType).snapshotConfiguration(); + + assertThat(snapshot.getFieldNames()).containsExactly("f0", "f1"); + } + + @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 restored = + TypeSerializerSnapshot.readVersionedSnapshot(in, getClass().getClassLoader()); + + assertThat(((RowDataSerializerSnapshot) restored).getFieldNames()).isNull(); + + 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 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 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 resolve( + RowType newType, RowType oldType) { + TypeSerializerSnapshot newSnapshot = + InternalSerializers.create(newType).snapshotConfiguration(); + TypeSerializerSnapshot oldSnapshot = + InternalSerializers.create(oldType).snapshotConfiguration(); + return newSnapshot.resolveSchemaCompatibility(oldSnapshot); + } + + private static TypeSerializerSnapshot roundTrip( + TypeSerializerSnapshot 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()); + } +}