diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java index a8cdc031e134..0358e75df6a1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java @@ -87,7 +87,11 @@ public interface DataFileMeta { new DataField(17, "_EXTERNAL_PATH", newStringType(true)), new DataField(18, "_FIRST_ROW_ID", new BigIntType(true)), new DataField( - 19, "_WRITE_COLS", new ArrayType(true, newStringType(false))))); + 19, "_WRITE_COLS", new ArrayType(true, newStringType(false))), + new DataField( + 20, + "_WRITTEN_FIELD_IDS", + new ArrayType(true, new IntType(false))))); BinaryRow EMPTY_MIN_KEY = EMPTY_ROW; BinaryRow EMPTY_MAX_KEY = EMPTY_ROW; @@ -108,6 +112,40 @@ static DataFileMeta forAppend( @Nullable String externalPath, @Nullable Long firstRowId, @Nullable List writeCols) { + return forAppend( + fileName, + fileSize, + rowCount, + rowStats, + minSequenceNumber, + maxSequenceNumber, + schemaId, + extraFiles, + embeddedIndex, + fileSource, + valueStatsCols, + externalPath, + firstRowId, + writeCols, + null); + } + + static DataFileMeta forAppend( + String fileName, + long fileSize, + long rowCount, + SimpleStats rowStats, + long minSequenceNumber, + long maxSequenceNumber, + long schemaId, + List extraFiles, + @Nullable byte[] embeddedIndex, + @Nullable FileSource fileSource, + @Nullable List valueStatsCols, + @Nullable String externalPath, + @Nullable Long firstRowId, + @Nullable List writeCols, + @Nullable int[] writtenFieldIds) { return new PojoDataFileMeta( fileName, fileSize, @@ -128,7 +166,8 @@ static DataFileMeta forAppend( valueStatsCols, externalPath, firstRowId, - writeCols); + writeCols, + writtenFieldIds); } static DataFileMeta create( @@ -171,7 +210,8 @@ static DataFileMeta create( valueStatsCols, externalPath, firstRowId, - writeCols); + writeCols, + null); } static DataFileMeta create( @@ -212,7 +252,8 @@ static DataFileMeta create( valueStatsCols, null, firstRowId, - writeCols); + writeCols, + null); } static DataFileMeta create( @@ -236,6 +277,52 @@ static DataFileMeta create( @Nullable String externalPath, @Nullable Long firstRowId, @Nullable List writeCols) { + return create( + fileName, + fileSize, + rowCount, + minKey, + maxKey, + keyStats, + valueStats, + minSequenceNumber, + maxSequenceNumber, + schemaId, + level, + extraFiles, + creationTime, + deleteRowCount, + embeddedIndex, + fileSource, + valueStatsCols, + externalPath, + firstRowId, + writeCols, + null); + } + + static DataFileMeta create( + String fileName, + long fileSize, + long rowCount, + BinaryRow minKey, + BinaryRow maxKey, + SimpleStats keyStats, + SimpleStats valueStats, + long minSequenceNumber, + long maxSequenceNumber, + long schemaId, + int level, + List extraFiles, + Timestamp creationTime, + @Nullable Long deleteRowCount, + @Nullable byte[] embeddedIndex, + @Nullable FileSource fileSource, + @Nullable List valueStatsCols, + @Nullable String externalPath, + @Nullable Long firstRowId, + @Nullable List writeCols, + @Nullable int[] writtenFieldIds) { return new PojoDataFileMeta( fileName, fileSize, @@ -256,7 +343,8 @@ static DataFileMeta create( valueStatsCols, externalPath, firstRowId, - writeCols); + writeCols, + writtenFieldIds); } String fileName(); @@ -326,6 +414,18 @@ default Range nonNullRowIdRange() { @Nullable List writeCols(); + /** + * The field ids of the columns written in this file, or {@code null} if all columns are written + * (or the file was created by an older version that only recorded {@link #writeCols()} names). + * This is the id-based counterpart of {@link #writeCols()}: field ids are stable across column + * renames and can also address nested fields uniformly, since every (nested) field has a + * globally unique id in the table schema. + */ + @Nullable + default int[] writtenFieldIds() { + return null; + } + DataFileMeta upgrade(int newLevel); DataFileMeta rename(String newFileName); diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java index afed7265d476..a0d14ed6fbd2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java @@ -19,6 +19,7 @@ package org.apache.paimon.io; import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.manifest.FileSource; @@ -61,7 +62,8 @@ public InternalRow toRow(DataFileMeta meta) { toStringArrayData(meta.valueStatsCols()), meta.externalPath().map(BinaryString::fromString).orElse(null), meta.firstRowId(), - meta.writeCols() == null ? null : toStringArrayData(meta.writeCols())); + meta.writeCols() == null ? null : toStringArrayData(meta.writeCols()), + meta.writtenFieldIds() == null ? null : new GenericArray(meta.writtenFieldIds())); } @Override @@ -86,6 +88,7 @@ public DataFileMeta fromRow(InternalRow row) { row.isNullAt(16) ? null : fromStringArrayData(row.getArray(16)), row.isNullAt(17) ? null : row.getString(17).toString(), row.isNullAt(18) ? null : row.getLong(18), - row.isNullAt(19) ? null : fromStringArrayData(row.getArray(19))); + row.isNullAt(19) ? null : fromStringArrayData(row.getArray(19)), + row.isNullAt(20) ? null : row.getArray(20).toIntArray()); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaWriteColsLegacySerializer.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaWriteColsLegacySerializer.java new file mode 100644 index 000000000000..6f35e43ac957 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaWriteColsLegacySerializer.java @@ -0,0 +1,137 @@ +/* + * 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.paimon.io; + +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.manifest.FileSource; +import org.apache.paimon.stats.SimpleStats; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.BigIntType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.IntType; +import org.apache.paimon.types.RowType; +import org.apache.paimon.types.TinyIntType; +import org.apache.paimon.utils.ObjectSerializer; + +import java.util.Arrays; + +import static org.apache.paimon.utils.InternalRowUtils.fromStringArrayData; +import static org.apache.paimon.utils.InternalRowUtils.toStringArrayData; +import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow; +import static org.apache.paimon.utils.SerializationUtils.newBytesType; +import static org.apache.paimon.utils.SerializationUtils.newStringType; +import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow; + +/** + * Serializer for {@link DataFileMeta} with the legacy 20-field layout, i.e. up to {@code + * _WRITE_COLS} and before {@code _WRITTEN_FIELD_IDS} was appended. It freezes that layout so + * streams written by older versions keep deserializing correctly. + */ +public class DataFileMetaWriteColsLegacySerializer extends ObjectSerializer { + + private static final long serialVersionUID = 1L; + + /** The frozen {@link DataFileMeta} schema of the {@code _WRITE_COLS} era (fields 0-19). */ + public static final RowType SCHEMA = + new RowType( + false, + Arrays.asList( + new DataField(0, "_FILE_NAME", newStringType(false)), + new DataField(1, "_FILE_SIZE", new BigIntType(false)), + new DataField(2, "_ROW_COUNT", new BigIntType(false)), + new DataField(3, "_MIN_KEY", newBytesType(false)), + new DataField(4, "_MAX_KEY", newBytesType(false)), + new DataField(5, "_KEY_STATS", SimpleStats.SCHEMA), + new DataField(6, "_VALUE_STATS", SimpleStats.SCHEMA), + new DataField(7, "_MIN_SEQUENCE_NUMBER", new BigIntType(false)), + new DataField(8, "_MAX_SEQUENCE_NUMBER", new BigIntType(false)), + new DataField(9, "_SCHEMA_ID", new BigIntType(false)), + new DataField(10, "_LEVEL", new IntType(false)), + new DataField( + 11, "_EXTRA_FILES", new ArrayType(false, newStringType(false))), + new DataField(12, "_CREATION_TIME", DataTypes.TIMESTAMP_MILLIS()), + new DataField(13, "_DELETE_ROW_COUNT", new BigIntType(true)), + new DataField(14, "_EMBEDDED_FILE_INDEX", newBytesType(true)), + new DataField(15, "_FILE_SOURCE", new TinyIntType(true)), + new DataField( + 16, + "_VALUE_STATS_COLS", + DataTypes.ARRAY(DataTypes.STRING().notNull())), + new DataField(17, "_EXTERNAL_PATH", newStringType(true)), + new DataField(18, "_FIRST_ROW_ID", new BigIntType(true)), + new DataField( + 19, "_WRITE_COLS", new ArrayType(true, newStringType(false))))); + + public DataFileMetaWriteColsLegacySerializer() { + super(SCHEMA); + } + + @Override + public InternalRow toRow(DataFileMeta meta) { + return GenericRow.of( + BinaryString.fromString(meta.fileName()), + meta.fileSize(), + meta.rowCount(), + serializeBinaryRow(meta.minKey()), + serializeBinaryRow(meta.maxKey()), + meta.keyStats().toRow(), + meta.valueStats().toRow(), + meta.minSequenceNumber(), + meta.maxSequenceNumber(), + meta.schemaId(), + meta.level(), + toStringArrayData(meta.extraFiles()), + meta.creationTime(), + meta.deleteRowCount().orElse(null), + meta.embeddedIndex(), + meta.fileSource().map(FileSource::toByteValue).orElse(null), + toStringArrayData(meta.valueStatsCols()), + meta.externalPath().map(BinaryString::fromString).orElse(null), + meta.firstRowId(), + meta.writeCols() == null ? null : toStringArrayData(meta.writeCols())); + } + + @Override + public DataFileMeta fromRow(InternalRow row) { + return DataFileMeta.create( + row.getString(0).toString(), + row.getLong(1), + row.getLong(2), + deserializeBinaryRow(row.getBinary(3)), + deserializeBinaryRow(row.getBinary(4)), + SimpleStats.fromRow(row.getRow(5, 3)), + SimpleStats.fromRow(row.getRow(6, 3)), + row.getLong(7), + row.getLong(8), + row.getLong(9), + row.getInt(10), + fromStringArrayData(row.getArray(11)), + row.getTimestamp(12, 3), + row.isNullAt(13) ? null : row.getLong(13), + row.isNullAt(14) ? null : row.getBinary(14), + row.isNullAt(15) ? null : FileSource.fromByteValue(row.getByte(15)), + row.isNullAt(16) ? null : fromStringArrayData(row.getArray(16)), + row.isNullAt(17) ? null : row.getString(17).toString(), + row.isNullAt(18) ? null : row.getLong(18), + row.isNullAt(19) ? null : fromStringArrayData(row.getArray(19))); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/io/PojoDataFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/io/PojoDataFileMeta.java index 9e845b26fe14..3152c9ddcd0e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/PojoDataFileMeta.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/PojoDataFileMeta.java @@ -82,6 +82,8 @@ public class PojoDataFileMeta implements DataFileMeta { private final @Nullable List writeCols; + private final @Nullable int[] writtenFieldIds; + public PojoDataFileMeta( String fileName, long fileSize, @@ -103,6 +105,52 @@ public PojoDataFileMeta( @Nullable String externalPath, @Nullable Long firstRowId, @Nullable List writeCols) { + this( + fileName, + fileSize, + rowCount, + minKey, + maxKey, + keyStats, + valueStats, + minSequenceNumber, + maxSequenceNumber, + schemaId, + level, + extraFiles, + creationTime, + deleteRowCount, + embeddedIndex, + fileSource, + valueStatsCols, + externalPath, + firstRowId, + writeCols, + null); + } + + public PojoDataFileMeta( + String fileName, + long fileSize, + long rowCount, + BinaryRow minKey, + BinaryRow maxKey, + SimpleStats keyStats, + SimpleStats valueStats, + long minSequenceNumber, + long maxSequenceNumber, + long schemaId, + int level, + List extraFiles, + Timestamp creationTime, + @Nullable Long deleteRowCount, + @Nullable byte[] embeddedIndex, + @Nullable FileSource fileSource, + @Nullable List valueStatsCols, + @Nullable String externalPath, + @Nullable Long firstRowId, + @Nullable List writeCols, + @Nullable int[] writtenFieldIds) { this.fileName = fileName; this.fileSize = fileSize; @@ -127,6 +175,7 @@ public PojoDataFileMeta( this.externalPath = externalPath; this.firstRowId = firstRowId; this.writeCols = writeCols; + this.writtenFieldIds = writtenFieldIds; } @Override @@ -252,6 +301,12 @@ public List writeCols() { return writeCols; } + @Nullable + @Override + public int[] writtenFieldIds() { + return writtenFieldIds; + } + @Override public PojoDataFileMeta upgrade(int newLevel) { checkArgument(newLevel > this.level); @@ -275,7 +330,8 @@ public PojoDataFileMeta upgrade(int newLevel) { valueStatsCols, externalPath, firstRowId, - writeCols); + writeCols, + writtenFieldIds); } @Override @@ -301,7 +357,8 @@ public PojoDataFileMeta rename(String newFileName) { valueStatsCols, newExternalPath, firstRowId, - writeCols); + writeCols, + writtenFieldIds); } @Override @@ -326,7 +383,8 @@ public PojoDataFileMeta copyWithoutStats() { Collections.emptyList(), externalPath, firstRowId, - writeCols); + writeCols, + writtenFieldIds); } @Override @@ -351,7 +409,8 @@ public PojoDataFileMeta assignSequenceNumber(long minSequenceNumber, long maxSeq valueStatsCols, externalPath, firstRowId, - writeCols); + writeCols, + writtenFieldIds); } @Override @@ -376,7 +435,8 @@ public PojoDataFileMeta assignFirstRowId(long firstRowId) { valueStatsCols, externalPath, firstRowId, - writeCols); + writeCols, + writtenFieldIds); } @Override @@ -401,7 +461,8 @@ public PojoDataFileMeta newFirstRowId(@Nullable Long newFirstRowId) { valueStatsCols, externalPath, newFirstRowId, - writeCols); + writeCols, + writtenFieldIds); } @Override @@ -426,7 +487,8 @@ public PojoDataFileMeta copy(List newExtraFiles) { valueStatsCols, externalPath, firstRowId, - writeCols); + writeCols, + writtenFieldIds); } @Override @@ -451,7 +513,8 @@ public PojoDataFileMeta newExternalPath(String newExternalPath) { valueStatsCols, newExternalPath, firstRowId, - writeCols); + writeCols, + writtenFieldIds); } @Override @@ -476,7 +539,8 @@ public PojoDataFileMeta copy(byte[] newEmbeddedIndex) { valueStatsCols, externalPath, firstRowId, - writeCols); + writeCols, + writtenFieldIds); } @Override @@ -538,7 +602,8 @@ public boolean equals(Object o) { && Objects.equals(valueStatsCols, that.valueStatsCols()) && Objects.equals(externalPath, that.externalPath().orElse(null)) && Objects.equals(firstRowId, that.firstRowId()) - && Objects.equals(writeCols, that.writeCols()); + && Objects.equals(writeCols, that.writeCols()) + && Arrays.equals(writtenFieldIds, that.writtenFieldIds()); } @Override @@ -563,7 +628,8 @@ public int hashCode() { valueStatsCols, externalPath, firstRowId, - writeCols); + writeCols, + Arrays.hashCode(writtenFieldIds)); } @Override @@ -573,7 +639,8 @@ public String toString() { + "minKey: %s, maxKey: %s, keyStats: %s, valueStats: %s, " + "minSequenceNumber: %d, maxSequenceNumber: %d, " + "schemaId: %d, level: %d, extraFiles: %s, creationTime: %s, " - + "deleteRowCount: %d, fileSource: %s, valueStatsCols: %s, externalPath: %s, firstRowId: %s, writeCols: %s}", + + "deleteRowCount: %d, fileSource: %s, valueStatsCols: %s, externalPath: %s, " + + "firstRowId: %s, writeCols: %s, writtenFieldIds: %s}", fileName, fileSize, rowCount, @@ -593,6 +660,7 @@ public String toString() { valueStatsCols, externalPath, firstRowId, - writeCols); + writeCols, + Arrays.toString(writtenFieldIds)); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java index 02dff0fd8233..e2d718a9bef1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java @@ -55,6 +55,7 @@ public class RowDataFileWriter extends StatsCollectingSingleFileWriter auxiliaryFileWriters; private final FileSource fileSource; @Nullable private final List writeCols; + @Nullable private final int[] writtenFieldIds; private final RowDataFileSequenceNumberTracker sequenceNumberTracker; public RowDataFileWriter( @@ -130,6 +131,14 @@ public RowDataFileWriter( this.auxiliaryFileWriters = Collections.unmodifiableList(auxiliaryFileWriters); this.fileSource = fileSource; this.writeCols = writeCols; + // the id-based counterpart of writeCols: writeCols always names fields of writeSchema, so + // the field ids (stable across renames) can be derived right here + this.writtenFieldIds = + writeCols == null + ? null + : writeCols.stream() + .mapToInt(col -> writeSchema.getField(col).id()) + .toArray(); this.sequenceNumberTracker = new RowDataFileSequenceNumberTracker( writeSchema, seqNumCounterSupplier, super::recordCount); @@ -220,7 +229,8 @@ public DataFileMeta result() throws IOException { statsPair.getKey(), externalPath, null, - writeCols); + writeCols, + writtenFieldIds); } private interface DataFileAuxiliaryWriter { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageSerializer.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageSerializer.java index 5e621a9f789c..a71df74919f6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageSerializer.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageSerializer.java @@ -33,6 +33,7 @@ import org.apache.paimon.io.DataFileMeta12LegacySerializer; import org.apache.paimon.io.DataFileMetaFirstRowIdLegacySerializer; import org.apache.paimon.io.DataFileMetaSerializer; +import org.apache.paimon.io.DataFileMetaWriteColsLegacySerializer; import org.apache.paimon.io.DataIncrement; import org.apache.paimon.io.DataInputDeserializer; import org.apache.paimon.io.DataInputView; @@ -51,11 +52,12 @@ /** {@link VersionedSerializer} for {@link CommitMessage}. */ public class CommitMessageSerializer implements VersionedSerializer { - public static final int CURRENT_VERSION = 11; + public static final int CURRENT_VERSION = 12; private final DataFileMetaSerializer dataFileSerializer; private final IndexFileMetaSerializer indexEntrySerializer; + private DataFileMetaWriteColsLegacySerializer dataFileMetaWriteColsLegacySerializer; private DataFileMetaFirstRowIdLegacySerializer dataFileMetaFirstRowIdLegacySerializer; private DataFileMeta12LegacySerializer dataFileMeta12LegacySerializer; private DataFileMeta10LegacySerializer dataFileMeta10LegacySerializer; @@ -184,8 +186,13 @@ private CommitMessage deserialize(int version, DataInputView view) throws IOExce private IOExceptionSupplier> fileDeserializer( int version, DataInputView view) { - if (version >= 9) { + if (version >= 12) { return () -> dataFileSerializer.deserializeList(view); + } else if (version >= 9) { + if (dataFileMetaWriteColsLegacySerializer == null) { + dataFileMetaWriteColsLegacySerializer = new DataFileMetaWriteColsLegacySerializer(); + } + return () -> dataFileMetaWriteColsLegacySerializer.deserializeList(view); } else if (version == 8) { if (dataFileMetaFirstRowIdLegacySerializer == null) { dataFileMetaFirstRowIdLegacySerializer = diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java index df31763a3001..88bf60f019c8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java @@ -28,6 +28,7 @@ import org.apache.paimon.io.DataFileMeta12LegacySerializer; import org.apache.paimon.io.DataFileMetaFirstRowIdLegacySerializer; import org.apache.paimon.io.DataFileMetaSerializer; +import org.apache.paimon.io.DataFileMetaWriteColsLegacySerializer; import org.apache.paimon.io.DataInputView; import org.apache.paimon.io.DataInputViewStreamWrapper; import org.apache.paimon.io.DataOutputView; @@ -63,7 +64,7 @@ public class DataSplit implements Split { private static final long serialVersionUID = 7L; private static final long MAGIC = -2394839472490812314L; - private static final int VERSION = 8; + private static final int VERSION = 9; private long snapshotId = 0; private BinaryRow partition; @@ -509,6 +510,10 @@ private static FunctionWithIOException getFileMetaS new DataFileMetaFirstRowIdLegacySerializer(); return serializer::deserialize; } else if (version == 8) { + DataFileMetaWriteColsLegacySerializer serializer = + new DataFileMetaWriteColsLegacySerializer(); + return serializer::deserialize; + } else if (version == 9) { DataFileMetaSerializer serializer = new DataFileMetaSerializer(); return serializer::deserialize; } else { diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java index 57ba93ae199d..2bcd1a908234 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java @@ -19,6 +19,10 @@ package org.apache.paimon.utils; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.types.RowType; + +import javax.annotation.Nullable; import java.util.Collection; import java.util.Comparator; @@ -27,6 +31,7 @@ import java.util.stream.Collectors; import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile; +import static org.apache.paimon.table.SpecialFields.rowTypeWithRowTracking; import static org.apache.paimon.types.VectorType.isVectorStoreFile; import static org.apache.paimon.utils.Preconditions.checkArgument; import static org.apache.paimon.utils.Preconditions.checkState; @@ -65,6 +70,28 @@ public static T retrieveAnchorFile( return anchor; } + /** + * Resolve the field ids of the columns written in the given file. Prefers the id-based {@link + * DataFileMeta#writtenFieldIds()} when present; otherwise falls back to resolving the legacy + * name-based {@link DataFileMeta#writeCols()} against the file's schema (with row-tracking + * fields, since writeCols may contain {@code _ROW_ID}/{@code _SEQUENCE_NUMBER}). Returns {@code + * null} if the file wrote all columns. + */ + @Nullable + public static int[] writtenFieldIds( + DataFileMeta file, Function schemaFetcher) { + if (file.writtenFieldIds() != null) { + return file.writtenFieldIds(); + } + List writeCols = file.writeCols(); + if (writeCols == null) { + return null; + } + RowType rowType = + rowTypeWithRowTracking(schemaFetcher.apply(file.schemaId()).logicalRowType()); + return writeCols.stream().mapToInt(col -> rowType.getField(col).id()).toArray(); + } + /** Check files row ranges. */ public static Range checkContiguousRowRange(List files) { checkArgument(!files.isEmpty(), "%s should not be empty.", "Data evolution compact files"); diff --git a/paimon-core/src/test/java/org/apache/paimon/io/DataFileMetaWrittenFieldIdsCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/io/DataFileMetaWrittenFieldIdsCompatibilityTest.java new file mode 100644 index 000000000000..0869c08f8e87 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/io/DataFileMetaWrittenFieldIdsCompatibilityTest.java @@ -0,0 +1,183 @@ +/* + * 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.paimon.io; + +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.stats.SimpleStats; +import org.apache.paimon.table.SpecialFields; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.DataEvolutionUtils; + +import org.junit.jupiter.api.Test; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.function.Function; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Compatibility tests for the id-based {@link DataFileMeta#writtenFieldIds()}: new files carry both + * name-based {@code writeCols} and id-based {@code writtenFieldIds}; streams in the legacy 20-field + * layout (before {@code _WRITTEN_FIELD_IDS}) keep deserializing and resolve to the same field ids + * via {@link DataEvolutionUtils#writtenFieldIds}. + */ +public class DataFileMetaWrittenFieldIdsCompatibilityTest { + + private static DataFileMeta file(@Nullable List writeCols, @Nullable int[] ids) { + return DataFileMeta.forAppend( + "f0.parquet", + 100L, + 10L, + SimpleStats.EMPTY_STATS, + 0L, + 9L, + 0L, + Collections.emptyList(), + null, + null, + null, + null, + 0L, + writeCols, + ids); + } + + @Test + public void testRoundTripWithWrittenFieldIds() throws IOException { + DataFileMeta meta = file(Arrays.asList("a", "b"), new int[] {0, 1}); + DataFileMetaSerializer serializer = new DataFileMetaSerializer(); + + DataOutputSerializer out = new DataOutputSerializer(128); + serializer.serialize(meta, out); + DataFileMeta actual = + serializer.deserialize(new DataInputDeserializer(out.getCopyOfBuffer())); + + assertThat(actual.writeCols()).containsExactly("a", "b"); + assertThat(actual.writtenFieldIds()).containsExactly(0, 1); + assertThat(actual).isEqualTo(meta); + } + + @Test + public void testRoundTripWithoutWrittenFieldIds() throws IOException { + DataFileMeta meta = file(Arrays.asList("a", "b"), null); + DataFileMetaSerializer serializer = new DataFileMetaSerializer(); + + DataOutputSerializer out = new DataOutputSerializer(128); + serializer.serialize(meta, out); + DataFileMeta actual = + serializer.deserialize(new DataInputDeserializer(out.getCopyOfBuffer())); + + assertThat(actual.writtenFieldIds()).isNull(); + assertThat(actual).isEqualTo(meta); + } + + @Test + public void testLegacyLayoutStreamStillDeserializes() throws IOException { + // a stream written by an old version in the 20-field (_WRITE_COLS era) layout + DataFileMeta legacyMeta = file(Arrays.asList("a", "b"), null); + DataFileMetaWriteColsLegacySerializer legacySerializer = + new DataFileMetaWriteColsLegacySerializer(); + DataOutputSerializer out = new DataOutputSerializer(128); + legacySerializer.serialize(legacyMeta, out); + + // the new code reads it through the legacy serializer: names preserved, ids absent + DataFileMeta actual = + legacySerializer.deserialize(new DataInputDeserializer(out.getCopyOfBuffer())); + assertThat(actual.writeCols()).containsExactly("a", "b"); + assertThat(actual.writtenFieldIds()).isNull(); + } + + @Test + public void testNewStreamReadByOldSerializer() throws IOException { + // forward-compat of the binary (DataSplit/CommitMessage) row layout: a stream written by + // the NEW 21-field serializer must still be readable by the OLD 20-field serializer, which + // should just drop the trailing _WRITTEN_FIELD_IDS field + DataFileMeta meta = file(Arrays.asList("a", "b"), new int[] {0, 1}); + DataOutputSerializer out = new DataOutputSerializer(128); + new DataFileMetaSerializer().serialize(meta, out); + + DataFileMeta readByOld = + new DataFileMetaWriteColsLegacySerializer() + .deserialize(new DataInputDeserializer(out.getCopyOfBuffer())); + assertThat(readByOld.writeCols()).containsExactly("a", "b"); + assertThat(readByOld.writtenFieldIds()).isNull(); + } + + @Test + public void testLegacySerializerDropsWrittenFieldIds() throws IOException { + // writing a new meta through the legacy layout (e.g. for an old consumer) must not fail + // and simply drops the id-based representation + DataFileMeta meta = file(Arrays.asList("a", "b"), new int[] {0, 1}); + DataFileMetaWriteColsLegacySerializer legacySerializer = + new DataFileMetaWriteColsLegacySerializer(); + DataOutputSerializer out = new DataOutputSerializer(128); + legacySerializer.serialize(meta, out); + + DataFileMeta actual = + legacySerializer.deserialize(new DataInputDeserializer(out.getCopyOfBuffer())); + assertThat(actual.writeCols()).containsExactly("a", "b"); + assertThat(actual.writtenFieldIds()).isNull(); + } + + @Test + public void testResolveWrittenFieldIds() { + Function schemaFetcher = + schemaId -> + TableSchema.create( + 0L, + new Schema( + Arrays.asList( + new DataField(0, "a", DataTypes.INT()), + new DataField(1, "b", DataTypes.STRING()), + new DataField(2, "c", DataTypes.INT())), + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyMap(), + "")); + + // new file: ids used directly + assertThat( + DataEvolutionUtils.writtenFieldIds( + file(Arrays.asList("b", "c"), new int[] {1, 2}), schemaFetcher)) + .containsExactly(1, 2); + + // old file: names resolved against the schema, same result + assertThat( + DataEvolutionUtils.writtenFieldIds( + file(Arrays.asList("b", "c"), null), schemaFetcher)) + .containsExactly(1, 2); + + // old file containing row-tracking system fields resolves via the row-tracked schema + assertThat( + DataEvolutionUtils.writtenFieldIds( + file(Arrays.asList("a", SpecialFields.ROW_ID.name()), null), + schemaFetcher)) + .containsExactly(0, SpecialFields.ROW_ID.id()); + + // full write: null in both representations + assertThat(DataEvolutionUtils.writtenFieldIds(file(null, null), schemaFetcher)).isNull(); + } +} diff --git a/paimon-core/src/test/resources/compatibility/split-v1-chain b/paimon-core/src/test/resources/compatibility/split-v1-chain index d9f12c976540..1a58286aa413 100644 Binary files a/paimon-core/src/test/resources/compatibility/split-v1-chain and b/paimon-core/src/test/resources/compatibility/split-v1-chain differ diff --git a/paimon-core/src/test/resources/compatibility/split-v1-data b/paimon-core/src/test/resources/compatibility/split-v1-data index 6cbade9c58a6..9d2f6c085f68 100644 Binary files a/paimon-core/src/test/resources/compatibility/split-v1-data and b/paimon-core/src/test/resources/compatibility/split-v1-data differ diff --git a/paimon-core/src/test/resources/compatibility/split-v1-fallback b/paimon-core/src/test/resources/compatibility/split-v1-fallback index 976a41d75fd1..c8033b69139a 100644 Binary files a/paimon-core/src/test/resources/compatibility/split-v1-fallback and b/paimon-core/src/test/resources/compatibility/split-v1-fallback differ diff --git a/paimon-core/src/test/resources/compatibility/split-v1-fallback-data b/paimon-core/src/test/resources/compatibility/split-v1-fallback-data index 24be9e0ad56b..42d174704c79 100644 Binary files a/paimon-core/src/test/resources/compatibility/split-v1-fallback-data and b/paimon-core/src/test/resources/compatibility/split-v1-fallback-data differ diff --git a/paimon-core/src/test/resources/compatibility/split-v1-incremental b/paimon-core/src/test/resources/compatibility/split-v1-incremental index f52e8022ec09..50fd72f825be 100644 Binary files a/paimon-core/src/test/resources/compatibility/split-v1-incremental and b/paimon-core/src/test/resources/compatibility/split-v1-incremental differ diff --git a/paimon-core/src/test/resources/compatibility/split-v1-indexed b/paimon-core/src/test/resources/compatibility/split-v1-indexed index 0d20df101234..eab6fbe8f82a 100644 Binary files a/paimon-core/src/test/resources/compatibility/split-v1-indexed and b/paimon-core/src/test/resources/compatibility/split-v1-indexed differ diff --git a/paimon-core/src/test/resources/compatibility/split-v1-query-auth b/paimon-core/src/test/resources/compatibility/split-v1-query-auth index fce1aa5fb9ae..5e5acea55d31 100644 Binary files a/paimon-core/src/test/resources/compatibility/split-v1-query-auth and b/paimon-core/src/test/resources/compatibility/split-v1-query-auth differ