diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java index bd846cec28d..3982e7474b7 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java @@ -62,17 +62,19 @@ public class PaimonLakeCatalog implements LakeCatalog { private static final Logger LOG = LoggerFactory.getLogger(PaimonLakeCatalog.class); private static final String PAIMON_PATH_KEY = "paimon.path"; - public static final LinkedHashMap SYSTEM_COLUMNS = new LinkedHashMap<>(); + public static final LinkedHashMap LEGACY_SYSTEM_COLUMNS = + new LinkedHashMap<>(); static { // We need __bucket system column to filter out the given bucket // for paimon bucket-unaware append only table. // It's not required for paimon bucket-aware table like primary key table - // and bucket-aware append only table, but we always add the system column - // for consistent behavior - SYSTEM_COLUMNS.put(BUCKET_COLUMN_NAME, DataTypes.INT()); - SYSTEM_COLUMNS.put(OFFSET_COLUMN_NAME, DataTypes.BIGINT()); - SYSTEM_COLUMNS.put(TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP_LTZ_MILLIS()); + // and bucket-aware append only table, but legacy tables always carry the system column + // for consistent behavior. Under FIP-27 these columns are no longer added to newly created + // (clean) tables; they only remain on legacy tables created before FIP-27. + LEGACY_SYSTEM_COLUMNS.put(BUCKET_COLUMN_NAME, DataTypes.INT()); + LEGACY_SYSTEM_COLUMNS.put(OFFSET_COLUMN_NAME, DataTypes.BIGINT()); + LEGACY_SYSTEM_COLUMNS.put(TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP_LTZ_MILLIS()); } private final Catalog paimonCatalog; @@ -133,12 +135,13 @@ public void alterTable(TablePath tablePath, List tableChanges, Cont currentPaimonSchema, toPaimonSchema(context.getCurrentTable()))) { // if the paimon schema is same as current fluss schema, directly apply all the // changes. - paimonSchemaChanges = toPaimonSchemaChanges(changesToApply); + paimonSchemaChanges = toPaimonSchemaChanges(table, changesToApply); } else if (isPaimonSchemaCompatible( currentPaimonSchema, toPaimonSchema(context.getExpectedTable()))) { // if the schema is same as applied fluss schema , skip adding columns. paimonSchemaChanges = toPaimonSchemaChanges( + table, changesToApply.stream() .filter( tableChange -> diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java index 18afe2d982e..9e04ce15ee8 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java @@ -68,7 +68,7 @@ import java.util.Set; import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; -import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS; +import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonPartition; import static org.apache.fluss.utils.Preconditions.checkArgument; @@ -280,7 +280,7 @@ private static int[] businessFieldProjection(FileStoreTable fileStoreTable) { List fields = fileStoreTable.schema().logicalRowType().getFields(); List projectedFields = new ArrayList<>(); for (int i = 0; i < fields.size(); i++) { - if (!SYSTEM_COLUMNS.containsKey(fields.get(i).name())) { + if (!LEGACY_SYSTEM_COLUMNS.containsKey(fields.get(i).name())) { projectedFields.add(i); } } diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonRecordReader.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonRecordReader.java index 3af7467cfab..dd2fca065ee 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonRecordReader.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonRecordReader.java @@ -19,6 +19,7 @@ package org.apache.fluss.lake.paimon.source; import org.apache.fluss.lake.paimon.utils.PaimonRowAsFlussRow; +import org.apache.fluss.lake.paimon.utils.PaimonUtils; import org.apache.fluss.lake.source.RecordReader; import org.apache.fluss.record.ChangeType; import org.apache.fluss.record.GenericRecord; @@ -39,13 +40,20 @@ import java.util.Arrays; import java.util.stream.IntStream; +import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toChangeType; -import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; -import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; /** Record reader for paimon table. */ public class PaimonRecordReader implements RecordReader { + /** + * Sentinel log offset / timestamp emitted for rows read from a lake table. The lake table does + * not carry a per-record log offset (a clean table has no system columns, and for a legacy + * table we no longer read them), so a negative value is emitted and interpreted downstream as + * "no valid offset" (snapshot phase), see {@code LakeRecordRecordEmitter}. + */ + private static final long NO_SYSTEM_COLUMN_VALUE = -1L; + protected PaimonRowAsFlussRecordIterator iterator; protected @Nullable int[][] project; protected RowType paimonRowType; @@ -57,9 +65,8 @@ public PaimonRecordReader( @Nullable Predicate predicate) throws IOException { ReadBuilder readBuilder = fileStoreTable.newReadBuilder(); - RowType paimonFullRowType = fileStoreTable.rowType(); if (project != null) { - readBuilder = applyProject(readBuilder, project, paimonFullRowType); + readBuilder = applyProject(readBuilder, project); } if (predicate != null) { @@ -86,20 +93,11 @@ public CloseableIterator read() throws IOException { return iterator; } - private ReadBuilder applyProject( - ReadBuilder readBuilder, int[][] projects, RowType paimonFullRowType) { + private ReadBuilder applyProject(ReadBuilder readBuilder, int[][] projects) { + // The projected column ids reference the user (business) columns; the log offset / + // timestamp are not read from the lake table, so no system column needs to be projected. int[] projectIds = Arrays.stream(projects).mapToInt(project -> project[0]).toArray(); - - int offsetFieldPos = paimonFullRowType.getFieldIndex(OFFSET_COLUMN_NAME); - int timestampFieldPos = paimonFullRowType.getFieldIndex(TIMESTAMP_COLUMN_NAME); - - int[] paimonProject = - IntStream.concat( - IntStream.of(projectIds), - IntStream.of(offsetFieldPos, timestampFieldPos)) - .toArray(); - - return readBuilder.withProjection(paimonProject); + return readBuilder.withProjection(projectIds); } /** Iterator for paimon row as fluss record. */ @@ -110,18 +108,20 @@ public static class PaimonRowAsFlussRecordIterator implements CloseableIterator< private final ProjectedRow projectedRow; private final PaimonRowAsFlussRow paimonRowAsFlussRow; - private final int logOffsetColIndex; - private final int timestampColIndex; - public PaimonRowAsFlussRecordIterator( org.apache.paimon.utils.CloseableIterator paimonRowIterator, RowType paimonRowType) { this.paimonRowIterator = paimonRowIterator; - this.logOffsetColIndex = paimonRowType.getFieldIndex(OFFSET_COLUMN_NAME); - this.timestampColIndex = paimonRowType.getFieldIndex(TIMESTAMP_COLUMN_NAME); - int[] project = IntStream.range(0, paimonRowType.getFieldCount() - 2).toArray(); - projectedRow = ProjectedRow.from(project); + // A legacy table read without projection still exposes its three trailing system + // columns; trim them so only the business columns are emitted. A clean table (or any + // projected read) has no system columns to trim. + int fieldCount = paimonRowType.getFieldCount(); + int businessFieldCount = + PaimonUtils.isLegacyTable(paimonRowType) + ? fieldCount - LEGACY_SYSTEM_COLUMNS.size() + : fieldCount; + projectedRow = ProjectedRow.from(IntStream.range(0, businessFieldCount).toArray()); paimonRowAsFlussRow = new PaimonRowAsFlussRow(); } @@ -143,12 +143,9 @@ public boolean hasNext() { public LogRecord next() { InternalRow paimonRow = paimonRowIterator.next(); ChangeType changeType = toChangeType(paimonRow.getRowKind()); - long offset = paimonRow.getLong(logOffsetColIndex); - long timestamp = paimonRow.getTimestamp(timestampColIndex, 6).getMillisecond(); - return new GenericRecord( - offset, - timestamp, + NO_SYSTEM_COLUMN_VALUE, + NO_SYSTEM_COLUMN_VALUE, changeType, projectedRow.replaceRow(paimonRowAsFlussRow.replaceRow(paimonRow))); } diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java index bc030301037..78ddf30450b 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java @@ -25,7 +25,7 @@ import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; -import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS; +import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toRowKind; import static org.apache.fluss.utils.Preconditions.checkNotNull; import static org.apache.fluss.utils.Preconditions.checkState; @@ -33,6 +33,7 @@ /** To wrap Fluss {@link LogRecord} as paimon {@link InternalRow}. */ public class FlussRecordAsPaimonRow extends FlussRowAsPaimonRow { + private final boolean paimonIncludingSystemColumns; private final int bucket; private LogRecord logRecord; private int originRowFieldCount; @@ -41,10 +42,19 @@ public class FlussRecordAsPaimonRow extends FlussRowAsPaimonRow { private final int offsetFieldIndex; private final int timestampFieldIndex; - public FlussRecordAsPaimonRow(int bucket, RowType tableTowType) { - super(tableTowType); + public FlussRecordAsPaimonRow(int bucket, RowType tableRowType) { + this(bucket, tableRowType, false); + } + + public FlussRecordAsPaimonRow( + int bucket, RowType tableRowType, boolean paimonIncludingSystemColumns) { + super(tableRowType); this.bucket = bucket; - this.businessFieldCount = tableRowType.getFieldCount() - SYSTEM_COLUMNS.size(); + this.paimonIncludingSystemColumns = paimonIncludingSystemColumns; + this.businessFieldCount = + tableRowType.getFieldCount() + - (paimonIncludingSystemColumns ? LEGACY_SYSTEM_COLUMNS.size() : 0); + // only valid when paimon includes the system columns this.bucketFieldIndex = businessFieldCount; this.offsetFieldIndex = businessFieldCount + 1; this.timestampFieldIndex = businessFieldCount + 2; @@ -97,7 +107,7 @@ public boolean isNullAt(int pos) { @Override public int getInt(int pos) { - if (pos == bucketFieldIndex) { + if (paimonIncludingSystemColumns && pos == bucketFieldIndex) { // bucket system column return bucket; } @@ -114,12 +124,14 @@ public int getInt(int pos) { @Override public long getLong(int pos) { checkState(logRecord != null, "setFlussRecord() must be called before accessing the row."); - if (pos == offsetFieldIndex) { - // offset system column - return logRecord.logOffset(); - } else if (pos == timestampFieldIndex) { - // timestamp system column - return logRecord.timestamp(); + if (paimonIncludingSystemColumns) { + if (pos == offsetFieldIndex) { + // offset system column + return logRecord.logOffset(); + } else if (pos == timestampFieldIndex) { + // timestamp system column + return logRecord.timestamp(); + } } if (pos >= originRowFieldCount) { throw new IllegalStateException( @@ -135,7 +147,7 @@ public long getLong(int pos) { public Timestamp getTimestamp(int pos, int precision) { checkState(logRecord != null, "setFlussRecord() must be called before accessing the row."); // it's timestamp system column - if (pos == timestampFieldIndex) { + if (paimonIncludingSystemColumns && pos == timestampFieldIndex) { return Timestamp.fromEpochMillis(logRecord.timestamp()); } if (pos >= originRowFieldCount) { diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java index 2a38e388a06..9b82403b6ce 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java @@ -21,6 +21,7 @@ import org.apache.fluss.lake.batch.RecordBatch; import org.apache.fluss.lake.paimon.tiering.append.AppendOnlyWriter; import org.apache.fluss.lake.paimon.tiering.mergetree.MergeTreeWriter; +import org.apache.fluss.lake.paimon.utils.PaimonUtils; import org.apache.fluss.lake.writer.LakeWriter; import org.apache.fluss.lake.writer.SupportsRecordBatchWrite; import org.apache.fluss.lake.writer.WriterInitContext; @@ -58,6 +59,11 @@ public PaimonLakeWriter( List partitionKeys = fileStoreTable.partitionKeys(); RowType flussRowType = writerInitContext.tableInfo().getRowType(); + // FIP-27: detect whether the target Paimon table is a clean table (only user columns) or a + // legacy table (carrying the three Fluss system columns). Writers emit system columns only + // for legacy tables. + boolean paimonIncludingSystemColumns = PaimonUtils.isLegacyTable(fileStoreTable.rowType()); + this.recordWriter = fileStoreTable.primaryKeys().isEmpty() ? new AppendOnlyWriter( @@ -65,14 +71,16 @@ public PaimonLakeWriter( writerInitContext.tableBucket(), writerInitContext.partition(), partitionKeys, - flussRowType) + flussRowType, + paimonIncludingSystemColumns) : new MergeTreeWriter( fileStoreTable, writerInitContext.tableBucket(), writerInitContext.partition(), partitionKeys, flussRowType, - writerInitContext.ioTmpDirs()); + writerInitContext.ioTmpDirs(), + paimonIncludingSystemColumns); } @Override diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java index 173407bb9c7..2260d553bc9 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java @@ -49,7 +49,8 @@ public RecordWriter( TableBucket tableBucket, @Nullable String partition, List partitionKeys, - org.apache.fluss.types.RowType flussRowType) { + org.apache.fluss.types.RowType flussRowType, + boolean paimonIncludingSystemColumns) { this.tableWrite = tableWrite; this.tableRowType = tableRowType; this.bucket = tableBucket.getBucket(); @@ -62,7 +63,8 @@ public RecordWriter( this.partition = resolvePartition(partition, partitionKeys, flussRowType); } this.flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket.getBucket(), tableRowType); + new FlussRecordAsPaimonRow( + tableBucket.getBucket(), tableRowType, paimonIncludingSystemColumns); } public abstract void write(LogRecord record) throws Exception; diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyArrowBatchHelper.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyArrowBatchHelper.java index 4fdb8bce79b..a19c2628b14 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyArrowBatchHelper.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyArrowBatchHelper.java @@ -57,6 +57,7 @@ class AppendOnlyArrowBatchHelper implements AutoCloseable { private final TableWriteImpl tableWrite; private final RowType tableRowType; private final int bucket; + private final boolean paimonIncludingSystemColumns; private static final Field BUCKET_FIELD = new Field( @@ -88,11 +89,13 @@ class AppendOnlyArrowBatchHelper implements AutoCloseable { FileStoreTable fileStoreTable, TableWriteImpl tableWrite, RowType tableRowType, - int bucket) { + int bucket, + boolean paimonIncludingSystemColumns) { this.fileStoreTable = fileStoreTable; this.tableWrite = tableWrite; this.tableRowType = tableRowType; this.bucket = bucket; + this.paimonIncludingSystemColumns = paimonIncludingSystemColumns; } /** @@ -107,6 +110,16 @@ void writeArrowBatch(ArrowBatchData arrowBatchData, BinaryRow partition) throws } VectorSchemaRoot originalRoot = arrowBatchData.getVectorSchemaRoot(); + + if (!paimonIncludingSystemColumns) { + // Clean tables contain only user columns, so the incoming Arrow batch already matches + // the Paimon table schema. Write it directly without enriching system columns. + ArrowBundleRecords cleanRecords = + new ArrowBundleRecords(originalRoot, tableRowType, false); + tableWrite.writeBundle(partition, writtenBucket, cleanRecords); + return; + } + long baseOffset = arrowBatchData.getBaseLogOffset(); long timestamp = arrowBatchData.getTimestamp(); int rowCount = originalRoot.getRowCount(); diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java index 0caaed97c4b..23f61a33171 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java @@ -46,12 +46,15 @@ public class AppendOnlyWriter extends RecordWriter { */ @Nullable private AutoCloseable arrowBatchHelper; + private final boolean paimonIncludingSystemColumns; + public AppendOnlyWriter( FileStoreTable fileStoreTable, TableBucket tableBucket, @Nullable String partition, List partitionKeys, - RowType flussRowType) { + RowType flussRowType, + boolean paimonIncludingSystemColumns) { //noinspection unchecked super( (TableWriteImpl) @@ -61,8 +64,10 @@ public AppendOnlyWriter( tableBucket, partition, partitionKeys, - flussRowType); + flussRowType, + paimonIncludingSystemColumns); this.fileStoreTable = fileStoreTable; + this.paimonIncludingSystemColumns = paimonIncludingSystemColumns; } @Override @@ -90,7 +95,11 @@ public void writeArrowBatch(ArrowBatchData arrowBatchData) throws Exception { if (arrowBatchHelper == null) { helper = new AppendOnlyArrowBatchHelper( - fileStoreTable, tableWrite, tableRowType, bucket); + fileStoreTable, + tableWrite, + tableRowType, + bucket, + paimonIncludingSystemColumns); arrowBatchHelper = helper; } else { helper = (AppendOnlyArrowBatchHelper) arrowBatchHelper; diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java index b3e1ecdeed8..37aeef7afe6 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java @@ -49,8 +49,16 @@ public MergeTreeWriter( TableBucket tableBucket, @Nullable String partition, List partitionKeys, - RowType flussRowType) { - this(fileStoreTable, tableBucket, partition, partitionKeys, flussRowType, (String[]) null); + RowType flussRowType, + boolean paimonIncludingSystemColumns) { + this( + fileStoreTable, + tableBucket, + partition, + partitionKeys, + flussRowType, + (String[]) null, + paimonIncludingSystemColumns); } public MergeTreeWriter( @@ -59,14 +67,16 @@ public MergeTreeWriter( @Nullable String partition, List partitionKeys, RowType flussRowType, - @Nullable String[] ioTmpDirs) { + @Nullable String[] ioTmpDirs, + boolean paimonIncludingSystemColumns) { this( fileStoreTable, createIOManager(ioTmpDirs), tableBucket, partition, partitionKeys, - flussRowType); + flussRowType, + paimonIncludingSystemColumns); } MergeTreeWriter( @@ -75,14 +85,16 @@ public MergeTreeWriter( TableBucket tableBucket, @Nullable String partition, List partitionKeys, - RowType flussRowType) { + RowType flussRowType, + boolean paimonIncludingSystemColumns) { super( createTableWrite(fileStoreTable, ioManager), fileStoreTable.rowType(), tableBucket, partition, partitionKeys, - flussRowType); + flussRowType, + paimonIncludingSystemColumns); this.rowKeyExtractor = fileStoreTable.createRowKeyExtractor(); this.ioManager = ioManager; } diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java index becf8a2f056..cf3fc034827 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java @@ -37,6 +37,7 @@ import org.apache.paimon.options.Options; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.table.Table; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; import org.apache.paimon.types.RowKind; @@ -50,7 +51,7 @@ import java.util.Set; import java.util.function.Function; -import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS; +import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; import static org.apache.fluss.utils.Preconditions.checkState; /** Utils for conversion between Paimon and Fluss. */ @@ -172,7 +173,11 @@ public static BinaryRow toPaimonPartition( return partitionExtractor.apply(new FlussRowAsPaimonRow(partitionRow, paimonRowType)); } - public static List toPaimonSchemaChanges(List tableChanges) { + public static List toPaimonSchemaChanges( + Table paimonTable, List tableChanges) { + // A legacy table (created before FIP-27) still carries the three trailing system columns, + // recognisable by the presence of the __timestamp column. A clean table has none of them. + boolean paimonIncludingSystemColumns = PaimonUtils.isLegacyTable(paimonTable.rowType()); List schemaChanges = new ArrayList<>(tableChanges.size()); for (TableChange tableChange : tableChanges) { @@ -189,6 +194,13 @@ public static List toPaimonSchemaChanges(List tableCh } else if (tableChange instanceof TableChange.AddColumn) { TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange; + if (LEGACY_SYSTEM_COLUMNS.containsKey(addColumn.getName())) { + throw new InvalidTableException( + "Column " + + addColumn.getName() + + " conflicts with a system column name of paimon table, please rename the column."); + } + if (!(addColumn.getPosition() instanceof TableChange.Last)) { throw new UnsupportedOperationException( "Only support to add column at last for paimon table."); @@ -203,14 +215,27 @@ public static List toPaimonSchemaChanges(List tableCh org.apache.paimon.types.DataType paimonDataType = flussDataType.accept(FlussDataTypeToPaimonDataType.INSTANCE); - String firstSystemColumnName = SYSTEM_COLUMNS.keySet().iterator().next(); - schemaChanges.add( - SchemaChange.addColumn( - addColumn.getName(), - paimonDataType, - addColumn.getComment(), - SchemaChange.Move.before( - addColumn.getName(), firstSystemColumnName))); + if (paimonIncludingSystemColumns) { + // Legacy tables keep the three system columns as the last physical columns, so + // a new business column must be inserted right before the first system column. + String firstSystemColumnName = LEGACY_SYSTEM_COLUMNS.keySet().iterator().next(); + schemaChanges.add( + SchemaChange.addColumn( + addColumn.getName(), + paimonDataType, + addColumn.getComment(), + SchemaChange.Move.before( + addColumn.getName(), firstSystemColumnName))); + } else { + // Clean tables have no trailing system columns, so a new business column is + // simply appended at the end. + schemaChanges.add( + SchemaChange.addColumn( + addColumn.getName(), + paimonDataType, + addColumn.getComment(), + SchemaChange.Move.last(addColumn.getName()))); + } } else { throw new UnsupportedOperationException( "Unsupported table change: " + tableChange.getClass()); @@ -252,7 +277,7 @@ public static Schema toPaimonSchema(TableDescriptor tableDescriptor) { for (org.apache.fluss.metadata.Schema.Column column : tableDescriptor.getSchema().getColumns()) { String columnName = column.getName(); - if (SYSTEM_COLUMNS.containsKey(columnName)) { + if (LEGACY_SYSTEM_COLUMNS.containsKey(columnName)) { throw new InvalidTableException( "Column " + columnName @@ -264,11 +289,6 @@ public static Schema toPaimonSchema(TableDescriptor tableDescriptor) { column.getComment().orElse(null)); } - // add system metadata columns to schema - for (Map.Entry systemColumn : SYSTEM_COLUMNS.entrySet()) { - schemaBuilder.column(systemColumn.getKey(), systemColumn.getValue()); - } - // set pk if (tableDescriptor.hasPrimaryKey()) { schemaBuilder.primaryKey( diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonRowAsFlussRow.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonRowAsFlussRow.java index fe956561a69..5fa5a2feed6 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonRowAsFlussRow.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonRowAsFlussRow.java @@ -27,8 +27,6 @@ import org.apache.paimon.data.Timestamp; -import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS; - /** Adapter for paimon row as fluss row. */ public class PaimonRowAsFlussRow implements InternalRow { @@ -47,7 +45,7 @@ public PaimonRowAsFlussRow replaceRow(org.apache.paimon.data.InternalRow paimonR @Override public int getFieldCount() { - return paimonRow.getFieldCount() - SYSTEM_COLUMNS.size(); + return paimonRow.getFieldCount(); } @Override diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonTableValidation.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonTableValidation.java index 0b8f56052f9..c6254084c17 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonTableValidation.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonTableValidation.java @@ -32,6 +32,7 @@ import java.util.Map; import java.util.Objects; +import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PAIMON_UNSETTABLE_OPTIONS; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY; import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; @@ -40,6 +41,15 @@ public class PaimonTableValidation { public static boolean isPaimonSchemaCompatible(Schema existingSchema, Schema newSchema) { + // FIP-27: newly generated schemas are always clean (no system columns). When the existing + // table is a legacy table that still carries the three system columns (recognisable by the + // presence of the __timestamp column), re-enabling lake tiering must keep that physical + // layout. Enrich the clean new schema with the trailing system columns before comparison, + // so an existing legacy table is recognised as compatible and its layout is preserved. + if (PaimonUtils.isLegacyTable(existingSchema.rowType())) { + newSchema = appendSystemColumns(newSchema); + } + if (!equalPhysicalSchema(existingSchema, newSchema)) { // Allow different precisions for __timestamp column for backward compatibility, // old cluster will use precision 6, but new cluster will use precision 3, @@ -50,6 +60,32 @@ public static boolean isPaimonSchemaCompatible(Schema existingSchema, Schema new return true; } + /** + * Returns a copy of {@code cleanSchema} with the three Fluss system columns appended as the + * last physical columns, so a clean schema generated by the current cluster can be compared + * against an existing legacy table. Partition keys, primary keys, options and comment are + * preserved. System-column field ids continue the existing id sequence to avoid collisions; + * they are irrelevant to {@link #equalPhysicalSchema}, which ignores field ids. + */ + private static Schema appendSystemColumns(Schema cleanSchema) { + List fields = new ArrayList<>(cleanSchema.fields()); + int nextFieldId = 0; + for (DataField field : fields) { + nextFieldId = Math.max(nextFieldId, field.id() + 1); + } + for (Map.Entry systemColumn : + LEGACY_SYSTEM_COLUMNS.entrySet()) { + fields.add( + new DataField(nextFieldId++, systemColumn.getKey(), systemColumn.getValue())); + } + return new Schema( + fields, + cleanSchema.partitionKeys(), + cleanSchema.primaryKeys(), + cleanSchema.options(), + cleanSchema.comment()); + } + /** * Check if the {@code existingSchema} is compatible with {@code newSchema} by ignoring the * precision difference of the system column {@code __timestamp}. @@ -67,6 +103,16 @@ public static boolean isPaimonSchemaCompatible(Schema existingSchema, Schema new public static boolean equalIgnoreSystemColumnTimestampPrecision( Schema existingSchema, Schema newSchema) { List existingFields = new ArrayList<>(existingSchema.fields()); + // The precision relaxation only applies to a legacy table's trailing __timestamp system + // column. A clean table (or any table whose last column is not __timestamp) has no such + // column, so there is nothing to relax and we compare the schemas directly. + if (existingFields.isEmpty() + || !existingFields + .get(existingFields.size() - 1) + .name() + .equals(TIMESTAMP_COLUMN_NAME)) { + return equalPhysicalSchema(existingSchema, newSchema); + } DataField systemTimestampField = existingFields.get(existingFields.size() - 1); if (systemTimestampField.name().equals(TIMESTAMP_COLUMN_NAME) && systemTimestampField diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonUtils.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonUtils.java new file mode 100644 index 00000000000..6b58f3f2b29 --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonUtils.java @@ -0,0 +1,40 @@ +/* + * 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.fluss.lake.paimon.utils; + +import org.apache.paimon.types.RowType; + +import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; + +/** Common utilities for the Paimon lake integration. */ +public final class PaimonUtils { + + private PaimonUtils() {} + + /** + * Returns whether the physical Paimon table uses the legacy layout, i.e. it still carries the + * three trailing Fluss system columns ({@code __bucket}, {@code __offset}, {@code + * __timestamp}). + * + *

A legacy table is identified by the presence of the {@code __timestamp} system column; + * under FIP-27 newly created tables are clean and carry none of these columns. + */ + public static boolean isLegacyTable(RowType tableRowType) { + return tableRowType.getFieldIndex(TIMESTAMP_COLUMN_NAME) >= 0; + } +} diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java index f1d75d27810..0a0c07d9735 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java @@ -44,7 +44,6 @@ import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.GenericRow; -import org.apache.paimon.data.Timestamp; import org.apache.paimon.options.Options; import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.table.FileStoreTable; @@ -74,6 +73,7 @@ import java.util.Set; import java.util.stream.Stream; +import static org.apache.fluss.lake.paimon.testutils.PaimonTestUtils.adjustToLegacyV1Table; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PAIMON_UNSETTABLE_OPTIONS; import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; @@ -172,19 +172,9 @@ void testCreateLakeEnabledTable() throws Exception { RowType.of( new DataType[] { org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.STRING(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.STRING() }, - new String[] { - "log_c1", - "log_c2", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME - }), + new String[] {"log_c1", "log_c2"}), "log_c1,log_c2", BUCKET_NUM); @@ -210,19 +200,9 @@ void testCreateLakeEnabledTable() throws Exception { RowType.of( new DataType[] { org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.STRING(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.STRING() }, - new String[] { - "log_c1", - "log_c2", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME - }), + new String[] {"log_c1", "log_c2"}), null, BUCKET_NUM); @@ -249,19 +229,9 @@ void testCreateLakeEnabledTable() throws Exception { RowType.of( new DataType[] { org.apache.paimon.types.DataTypes.INT().notNull(), - org.apache.paimon.types.DataTypes.STRING(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.STRING() }, - new String[] { - "pk_c1", - "pk_c2", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME - }), + new String[] {"pk_c1", "pk_c2"}), "pk_c1", BUCKET_NUM); @@ -292,20 +262,9 @@ void testCreateLakeEnabledTable() throws Exception { new DataType[] { org.apache.paimon.types.DataTypes.INT().notNull(), org.apache.paimon.types.DataTypes.STRING(), - org.apache.paimon.types.DataTypes.STRING().notNull(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.STRING().notNull() }, - new String[] { - "c1", - "c2", - "c3", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME - }), + new String[] {"c1", "c2", "c3"}), "c1", BUCKET_NUM); } @@ -359,32 +318,12 @@ void testCreateLakeEnabledTableWithAllTypes() throws Exception { org.apache.paimon.types.DataTypes.DATE(), org.apache.paimon.types.DataTypes.TIME(), org.apache.paimon.types.DataTypes.TIMESTAMP(), - org.apache.paimon.types.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE() }, new String[] { - "log_c1", - "log_c2", - "log_c3", - "log_c4", - "log_c5", - "log_c6", - "log_c7", - "log_c8", - "log_c9", - "log_c10", - "log_c11", - "log_c12", - "log_c13", - "log_c14", - "log_c15", - "log_c16", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME + "log_c1", "log_c2", "log_c3", "log_c4", "log_c5", "log_c6", "log_c7", + "log_c8", "log_c9", "log_c10", "log_c11", "log_c12", "log_c13", + "log_c14", "log_c15", "log_c16" }), null, BUCKET_NUM); @@ -599,10 +538,9 @@ void testCreateLakeEnableTableWithExistLakeTable() throws Exception { .hasMessageContaining( "The table `fluss`.`log_table_with_exist_lake_table` already exists in Paimon catalog, but the table schema is not compatible.") .hasMessageContaining( - "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE]") + "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT]") .hasMessageContaining("options={bucket=-1") - .hasMessageContaining( - "new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE]") + .hasMessageContaining("new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT]") .hasMessageContaining("options={bucket=3") .hasMessageContaining("bucket-key=c1,c2") .hasMessageEndingWith( @@ -624,9 +562,9 @@ void testCreateLakeEnableTableWithExistLakeTable() throws Exception { .hasMessageContaining( "The table `fluss`.`log_table_with_exist_lake_table` already exists in Paimon catalog, but the table schema is not compatible.") .hasMessageContaining( - "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE]") + "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT]") .hasMessageContaining( - "new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `c3` STRING, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE]") + "new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `c3` STRING]") .hasMessageEndingWith( "Please first drop the table in Paimon catalog or use a new table name."); @@ -781,19 +719,9 @@ void testAlterLakeEnabledLogTable() throws Exception { RowType.of( new DataType[] { org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.STRING(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.STRING() }, - new String[] { - "log_c1", - "log_c2", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME - }), + new String[] {"log_c1", "log_c2"}), "log_c1,log_c2", BUCKET_NUM); @@ -888,19 +816,9 @@ void testAlterLakeEnabledTableProperties() throws Exception { RowType.of( new DataType[] { org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.STRING(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.STRING() }, - new String[] { - "c1", - "c2", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME - }), + new String[] {"c1", "c2"}), "c1,c2", BUCKET_NUM); @@ -919,19 +837,9 @@ void testAlterLakeEnabledTableProperties() throws Exception { RowType.of( new DataType[] { org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.STRING(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.STRING() }, - new String[] { - "c1", - "c2", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME - }), + new String[] {"c1", "c2"}), "c1,c2", BUCKET_NUM); @@ -966,19 +874,9 @@ void testAlterLakeEnabledTableProperties() throws Exception { RowType.of( new DataType[] { org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.STRING(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.STRING() }, - new String[] { - "c1", - "c2", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME - }), + new String[] {"c1", "c2"}), "c1,c2", BUCKET_NUM); @@ -1047,19 +945,9 @@ void testEnableLakeTableAfterAlterTableProperties() throws Exception { RowType.of( new DataType[] { org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.STRING(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.STRING() }, - new String[] { - "c1", - "c2", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME - }), + new String[] {"c1", "c2"}), "c1,c2", BUCKET_NUM); @@ -1100,19 +988,9 @@ void testAlterLakeEnabledTableSchema() throws Exception { RowType.of( new DataType[] { org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.STRING(), - // for __bucket, __offset, __timestamp - org.apache.paimon.types.DataTypes.INT(), - org.apache.paimon.types.DataTypes.BIGINT(), - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS() + org.apache.paimon.types.DataTypes.STRING() }, - new String[] { - "c1", - "c2", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME - }), + new String[] {"c1", "c2"}), "c1,c2", BUCKET_NUM); @@ -1131,15 +1009,8 @@ void testAlterLakeEnabledTableSchema() throws Exception { paimonCatalog.getTable(Identifier.create(DATABASE, tablePath.getTableName())); // Verify the new column c3 with comment was added to Paimon table RowType alteredRowType = alteredPaimonTable.rowType(); - assertThat(alteredRowType.getFieldCount()).isEqualTo(6); - assertThat(alteredRowType.getFieldNames()) - .containsExactly( - "c1", - "c2", - "c3", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME); + assertThat(alteredRowType.getFieldCount()).isEqualTo(3); + assertThat(alteredRowType.getFieldNames()).containsExactly("c1", "c2", "c3"); // Verify c3 column has the correct type and comment assertThat(alteredRowType.getField("c3").type()) .isEqualTo(org.apache.paimon.types.DataTypes.INT()); @@ -1159,12 +1030,21 @@ void testEnableLakeTableWithLegacySystemTimestampColumn() throws Exception { Identifier paimonIdentifier = Identifier.create(DATABASE, tablePath.getTableName()); - // alter to TIMESTAMP_WITH_LOCAL_TIME_ZONE to mock the legacy behavior + // FIP-27: a newly created table is clean (no system columns). To exercise the legacy + // compatibility path, first turn it into a legacy table by appending the three trailing + // system columns, using TIMESTAMP_WITH_LOCAL_TIME_ZONE for __timestamp to mock the + // precision-6 layout written by an old cluster. paimonCatalog.alterTable( paimonIdentifier, - SchemaChange.updateColumnType( - TIMESTAMP_COLUMN_NAME, - org.apache.paimon.types.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()), + Arrays.asList( + SchemaChange.addColumn( + BUCKET_COLUMN_NAME, org.apache.paimon.types.DataTypes.INT()), + SchemaChange.addColumn( + OFFSET_COLUMN_NAME, org.apache.paimon.types.DataTypes.BIGINT()), + SchemaChange.addColumn( + TIMESTAMP_COLUMN_NAME, + org.apache.paimon.types.DataTypes + .TIMESTAMP_WITH_LOCAL_TIME_ZONE())), false); // disable data lake @@ -1190,6 +1070,49 @@ void testEnableLakeTableWithLegacySystemTimestampColumn() throws Exception { .isTrue(); } + @Test + void testAddColumnForLegacyTableWithSystemColumns() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "legacy_add_column_table"); + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("c1", DataTypes.INT()) + .column("c2", DataTypes.STRING()) + .build()) + .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true) + .build(); + admin.createTable(tablePath, tableDescriptor, false).get(); + + // turn the freshly created clean table into a legacy table carrying the three system + // columns + adjustToLegacyV1Table(tablePath, paimonCatalog); + + // adding a business column to a legacy table must still work, and the new column must be + // inserted before the trailing system columns so the legacy physical layout is preserved. + admin.alterTable( + tablePath, + Collections.singletonList( + TableChange.addColumn( + "c3", + DataTypes.INT(), + "c3 comment", + TableChange.ColumnPosition.last())), + false) + .get(); + + Identifier identifier = Identifier.create(DATABASE, tablePath.getTableName()); + RowType rowType = paimonCatalog.getTable(identifier).rowType(); + assertThat(rowType.getFieldNames()) + .containsExactly( + "c1", + "c2", + "c3", + BUCKET_COLUMN_NAME, + OFFSET_COLUMN_NAME, + TIMESTAMP_COLUMN_NAME); + } + @Test void testCreatePaimonDvTableWithNonStringPartitionColumn() throws Exception { TablePath tablePath = TablePath.of(DATABASE, "invalid_dv_table"); @@ -1310,13 +1233,7 @@ private void writeData(Table table) throws Exception { BatchTableCommit commit = writeBuilder.newCommit()) { for (int i = 0; i < 10; i++) { - GenericRow row = - GenericRow.of( - i, - BinaryString.fromString("row-" + i), - 0, - (long) i, - Timestamp.fromEpochMillis(System.currentTimeMillis())); + GenericRow row = GenericRow.of(i, BinaryString.fromString("row-" + i)); write.write(row); } diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java index 1683533e506..1e67a92a281 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java @@ -160,15 +160,7 @@ void testAlterTableAddColumnLastNullable() throws Exception { Table table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier); assertThat(table.rowType().getFieldNames()) - .containsSequence( - "id", - "name", - "amount", - "address", - "new_col", - "__bucket", - "__offset", - "__timestamp"); + .containsSequence("id", "name", "amount", "address", "new_col"); } @Test @@ -205,15 +197,7 @@ void testAlterTableAddColumnIgnoresPaimonCommentsAndOptions() throws Exception { assertThat(((FileStoreTable) table).schema().toSchema().comment()).isEqualTo(""); assertThat(table.options().get("fluss.key")).isEqualTo("value"); assertThat(table.rowType().getFieldNames()) - .containsSequence( - "id", - "name", - "amount", - "address", - "is_direct_play", - "__bucket", - "__offset", - "__timestamp"); + .containsSequence("id", "name", "amount", "address", "is_direct_play"); } @Test diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java index 81fa9b05018..9efa832e23d 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java @@ -70,6 +70,7 @@ import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertRowResultsIgnoreOrder; import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectBatchRows; import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectRowsWithTimeout; +import static org.apache.fluss.lake.paimon.testutils.PaimonTestUtils.adjustToLegacyV1Table; import static org.apache.fluss.testutils.DataTestUtils.row; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; @@ -123,7 +124,9 @@ void testUnionReadFullType(Boolean isPartitioned) throws Exception { CollectionUtil.iteratorToList(tableResult.collect()).stream() .map( row -> { - int userColumnCount = row.getArity() - 3; + // FIP-27: a clean lake table exposes only user columns via + // $lake, so there are no trailing system columns to strip. + int userColumnCount = row.getArity(); Object[] fields = new Object[userColumnCount]; for (int i = 0; i < userColumnCount; i++) { fields[i] = row.getField(i); @@ -277,7 +280,9 @@ void testUnionReadFullType(Boolean isPartitioned) throws Exception { .stream() .map( row -> { - int columnCount = row.getArity() - 3; + // FIP-27: a clean lake table exposes only user columns via + // $lake, so there are no trailing system columns to strip. + int columnCount = row.getArity(); Object[] fields = new Object[columnCount]; for (int i = 0; i < columnCount; i++) { fields[i] = row.getField(i); @@ -546,6 +551,48 @@ void testUnionReadWhenSomeBucketNotTiered() throws Exception { .isEqualTo("[+I[0, v01, v02], +I[1, v11, v12], +I[2, v21, v22], +I[3, v31, v32]]"); } + @Test + void testUnionReadLegacyTable() throws Exception { + // FIP-27: union read (lake snapshot merged with fresh Fluss log) must still work for a + // legacy table that carries the three trailing system columns. + String tableName = "pk_table_union_read_legacy"; + TablePath t1 = TablePath.of(DEFAULT_DB, tableName); + int bucketNum = 3; + long tableId = createSimplePkTable(t1, bucketNum, false, true); + + // turn the freshly created clean table into a legacy table before any tiering happens + adjustToLegacyV1Table(t1, paimonCatalog); + + // tier an initial version of every row into the lake + JobClient jobClient = buildTieringJob(execEnv); + List rows = new ArrayList<>(); + Map bucketLogEndOffset = new HashMap<>(); + for (int i = 0; i < bucketNum; i++) { + rows.add( + GenericRow.of( + i, BinaryString.fromString("v1_1"), BinaryString.fromString("v1_2"))); + bucketLogEndOffset.put(new TableBucket(tableId, i), 1L); + } + writeRows(t1, rows, false); + waitUntilBucketsSynced(bucketLogEndOffset.keySet()); + assertReplicaStatus(bucketLogEndOffset); + + // stop tiering, then overwrite every row so the fresh values only exist in the Fluss log + jobClient.cancel().get(); + rows.clear(); + for (int i = 0; i < bucketNum; i++) { + rows.add( + GenericRow.of( + i, BinaryString.fromString("v2_1"), BinaryString.fromString("v2_2"))); + } + writeRows(t1, rows, false); + + // union read must merge the lake snapshot with the fresh log and return the latest values + List result = toSortedRows(batchTEnv.executeSql("select * from " + tableName)); + assertThat(result.toString()) + .isEqualTo("[+I[0, v2_1, v2_2], +I[1, v2_1, v2_2], +I[2, v2_1, v2_2]]"); + } + @ParameterizedTest @ValueSource(booleans = {false, true}) void testUnionReadInStreamMode(Boolean isPartitioned) throws Exception { diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/PaimonTestUtils.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/PaimonTestUtils.java new file mode 100644 index 00000000000..590ff4886cd --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/PaimonTestUtils.java @@ -0,0 +1,53 @@ +/* + * 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.fluss.lake.paimon.testutils; + +import org.apache.fluss.metadata.TablePath; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.types.DataTypes; + +import java.util.Arrays; + +import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; +import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; + +/** The utils for paimon testing. */ +public class PaimonTestUtils { + + /** + * Adjusts the Paimon table to a legacy table by appending the three trailing Fluss system + * columns ({@code __bucket}, {@code __offset}, {@code __timestamp}). This simulates a table + * created before FIP-27, when the lake table always carried these system columns, so tests can + * verify that legacy tables remain readable and writable. + */ + public static void adjustToLegacyV1Table(TablePath tablePath, Catalog paimonCatalog) + throws Exception { + paimonCatalog.alterTable( + toPaimon(tablePath), + Arrays.asList( + SchemaChange.addColumn(BUCKET_COLUMN_NAME, DataTypes.INT()), + SchemaChange.addColumn(OFFSET_COLUMN_NAME, DataTypes.BIGINT()), + SchemaChange.addColumn( + TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP_LTZ_MILLIS())), + false); + } +} diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java index e085b694ba0..891ce12c55a 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java @@ -49,6 +49,38 @@ /** Test for {@link FlussRecordAsPaimonRow}. */ class FlussRecordAsPaimonRowTest { + @Test + void testCleanTableWritesNoSystemColumns() { + // A clean table (FIP-27 default) contains only user columns; the writer must not emit any + // of the three system columns. + int tableBucket = 3; + RowType tableRowType = + RowType.of( + new org.apache.paimon.types.IntType(), + new org.apache.paimon.types.VarCharType()); + + FlussRecordAsPaimonRow flussRecordAsPaimonRow = + new FlussRecordAsPaimonRow(tableBucket, tableRowType, false); + + long logOffset = 7L; + long timeStamp = System.currentTimeMillis(); + GenericRow genericRow = new GenericRow(2); + genericRow.setField(0, 1); + genericRow.setField(1, BinaryString.fromString("v1")); + LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow); + flussRecordAsPaimonRow.setFlussRecord(logRecord); + + // exposes exactly the two business columns, no trailing system columns + assertThat(flussRecordAsPaimonRow.getFieldCount()).isEqualTo(2); + assertThat(flussRecordAsPaimonRow.getInt(0)).isEqualTo(1); + assertThat(flussRecordAsPaimonRow.getString(1).toString()).isEqualTo("v1"); + + // the no-arg-layout constructor defaults to the clean layout + FlussRecordAsPaimonRow defaultRow = new FlussRecordAsPaimonRow(tableBucket, tableRowType); + defaultRow.setFlussRecord(logRecord); + assertThat(defaultRow.getFieldCount()).isEqualTo(2); + } + @Test void testLogTableRecordAllTypes() { // Construct a FlussRecordAsPaimonRow instance @@ -75,7 +107,7 @@ void testLogTableRecordAllTypes() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(14); @@ -143,7 +175,7 @@ void testPrimaryKeyTableRecord() { new org.apache.paimon.types.BigIntType(), new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -185,7 +217,7 @@ void testArrayTypeWithIntElements() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 10; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(2); @@ -224,7 +256,7 @@ void testArrayTypeWithStringElements() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 5; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -259,7 +291,7 @@ void testNestedArrayType() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -313,7 +345,7 @@ void testArrayWithAllPrimitiveTypes() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(7); @@ -386,7 +418,7 @@ void testArrayWithDecimalElements() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -420,7 +452,7 @@ void testArrayWithTimestampElements() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -454,7 +486,7 @@ void testArrayWithBinaryElements() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -483,7 +515,7 @@ void testNullArray() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -508,7 +540,7 @@ void testArrayWithNullableElements() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -537,7 +569,7 @@ void testEmptyArray() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -564,7 +596,7 @@ void testPaimonSchemaWiderThanFlussRecord() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 7L; long timeStamp = System.currentTimeMillis(); @@ -595,7 +627,7 @@ void testFlussRecordWiderThanPaimonSchema() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 7L; long timeStamp = System.currentTimeMillis(); @@ -690,7 +722,7 @@ void testNestedRowType() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(8); @@ -918,7 +950,7 @@ void testMapWithAllTypes() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, nestedMapRowType); + new FlussRecordAsPaimonRow(tableBucket, nestedMapRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow nestedMapGenericRow = new GenericRow(1); @@ -971,7 +1003,7 @@ private void testMapType( new org.apache.paimon.types.BigIntType(), new org.apache.paimon.types.LocalZonedTimestampType(3)); - FlussRecordAsPaimonRow flussRow = new FlussRecordAsPaimonRow(tableBucket, rowType); + FlussRecordAsPaimonRow flussRow = new FlussRecordAsPaimonRow(tableBucket, rowType, true); GenericRow genericRow = new GenericRow(1); genericRow.setField(0, new GenericMap(mapData)); LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow); @@ -997,7 +1029,7 @@ void testNullMap() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -1023,7 +1055,7 @@ void testMapWithNullableValues() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -1058,7 +1090,7 @@ void testEmptyMap() { new org.apache.paimon.types.LocalZonedTimestampType(3)); FlussRecordAsPaimonRow flussRecordAsPaimonRow = - new FlussRecordAsPaimonRow(tableBucket, tableRowType); + new FlussRecordAsPaimonRow(tableBucket, tableRowType, true); long logOffset = 0; long timeStamp = System.currentTimeMillis(); GenericRow genericRow = new GenericRow(1); @@ -1082,7 +1114,7 @@ void testAccessRowBeforeSetThrowsIllegalState() { new org.apache.paimon.types.IntType(), new org.apache.paimon.types.BigIntType(), new org.apache.paimon.types.LocalZonedTimestampType(3)); - FlussRecordAsPaimonRow row = new FlussRecordAsPaimonRow(0, rowType); + FlussRecordAsPaimonRow row = new FlussRecordAsPaimonRow(0, rowType, true); assertThatThrownBy(row::getRowKind) .isInstanceOf(IllegalStateException.class) .hasMessageContaining(expectedMsg); diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java index 3d4da4fe50c..6d035aec098 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java @@ -67,6 +67,7 @@ import java.util.Map; import java.util.stream.Stream; +import static org.apache.fluss.lake.paimon.testutils.PaimonTestUtils.adjustToLegacyV1Table; import static org.apache.fluss.testutils.DataTestUtils.row; import static org.assertj.core.api.Assertions.assertThat; @@ -188,6 +189,37 @@ void testTiering() throws Exception { } } + @Test + void testTieringForLegacyTable() throws Exception { + // FIP-27: verify that tiering still works for a legacy table that carries the three + // trailing + // system columns (__bucket, __offset, __timestamp). + TablePath t = TablePath.of(DEFAULT_DB, "legacyLogTable"); + long tId = createLogTable(t); + + // turn the freshly created clean table into a legacy table before writing/tiering + adjustToLegacyV1Table(t, paimonCatalog); + + TableBucket tBucket = new TableBucket(tId, 0); + List flussRows = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + List rows = Arrays.asList(row(1, "v1"), row(2, "v2"), row(3, "v3")); + flussRows.addAll(rows); + writeRows(t, rows, true); + } + + JobClient jobClient = buildTieringJob(execEnv); + try { + assertReplicaStatus(tBucket, 30); + assertThat(getLeaderReplica(tBucket).getLogTablet().getLakeMaxTimestamp()) + .isGreaterThan(-1); + // the legacy table keeps the system columns, and the tiered __offset must be correct + checkDataInPaimonLegacyAppendOnlyTable(t, flussRows, 0); + } finally { + jobClient.cancel().get(); + } + } + private static Stream tieringAllTypesWriteArgs() { return Stream.of(Arguments.of(true), Arguments.of(false)); } @@ -504,7 +536,28 @@ private void checkDataInPaimonAppendOnlyTable( InternalRow flussRow = flussRowIterator.next(); assertThat(row.getInt(0)).isEqualTo(flussRow.getInt(0)); assertThat(row.getString(1).toString()).isEqualTo(flussRow.getString(1).toString()); - // system columns are always the last three: __bucket, __offset, __timestamp + // FIP-27: a clean lake table only stores user columns, so the Paimon row has exactly + // the same field count as the expected Fluss row (no trailing + // __bucket/__offset/__timestamp columns). + assertThat(row.getFieldCount()).isEqualTo(flussRow.getFieldCount()); + } + assertThat(flussRowIterator.hasNext()).isFalse(); + } + + private void checkDataInPaimonLegacyAppendOnlyTable( + TablePath tablePath, List expectedRows, long startingOffset) + throws Exception { + Iterator paimonRowIterator = + getPaimonRowCloseableIterator(tablePath); + Iterator flussRowIterator = expectedRows.iterator(); + while (paimonRowIterator.hasNext()) { + org.apache.paimon.data.InternalRow row = paimonRowIterator.next(); + InternalRow flussRow = flussRowIterator.next(); + assertThat(row.getInt(0)).isEqualTo(flussRow.getInt(0)); + assertThat(row.getString(1).toString()).isEqualTo(flussRow.getString(1).toString()); + // A legacy table keeps the two business columns plus the three trailing system columns + // __bucket/__offset/__timestamp, so the tiered offset is stored at fieldCount - 2. + assertThat(row.getFieldCount()).isEqualTo(flussRow.getFieldCount() + 3); int offsetIndex = row.getFieldCount() - 2; assertThat(row.getLong(offsetIndex)).isEqualTo(startingOffset++); } @@ -526,8 +579,10 @@ private void checkDataInPaimonAppendOnlyPartitionedTable( assertThat(row.getInt(0)).isEqualTo(flussRow.getInt(0)); assertThat(row.getString(1).toString()).isEqualTo(flussRow.getString(1).toString()); assertThat(row.getString(2).toString()).isEqualTo(flussRow.getString(2).toString()); - // the idx 3 is __bucket, so use 4 - assertThat(row.getLong(4)).isEqualTo(startingOffset++); + // FIP-27: a clean lake table only stores user columns, so the Paimon row has exactly + // the same field count as the expected Fluss row (no trailing + // __bucket/__offset/__timestamp columns). + assertThat(row.getFieldCount()).isEqualTo(flussRow.getFieldCount()); } assertThat(flussRowIterator.hasNext()).isFalse(); } @@ -594,9 +649,8 @@ void testTieringWithAddColumn() throws Exception { FileStoreTable paimonTable = (FileStoreTable) paimonCatalog.getTable(tableIdentifier); List fieldNames = paimonTable.rowType().getFieldNames(); - // Should have exact fields in order: a, b, c3, __bucket, __offset, __timestamp - assertThat(fieldNames) - .containsExactly("a", "b", "c3", "__bucket", "__offset", "__timestamp"); + // FIP-27: a clean lake table only stores user columns, in order: a, b, c3. + assertThat(fieldNames).containsExactly("a", "b", "c3"); // 9. Verify both schema evolution and data correctness // For initial rows (before ADD COLUMN), c3 should be NULL diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java index e8ba9714dd6..e898420e1e9 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java @@ -137,6 +137,7 @@ public static void validateTableDescriptor( /** Validates the schema after altering table columns. */ @Internal public static void validateAlterTableSchema(TableInfo table, Schema newSchema) { + checkSystemColumns(newSchema.getRowType()); if (table.getTableConfig() .getMergeEngineType() .map(MergeEngineType.AGGREGATION::equals) diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/paimon/SparkLakePaimonCatalogTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/paimon/SparkLakePaimonCatalogTest.scala index fe5d3eb5d6e..c08fe5aa780 100644 --- a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/paimon/SparkLakePaimonCatalogTest.scala +++ b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/paimon/SparkLakePaimonCatalogTest.scala @@ -21,7 +21,6 @@ import org.apache.fluss.config.{ConfigOptions, Configuration, FlussConfigUtils} import org.apache.fluss.exception.FlussRuntimeException import org.apache.fluss.lake.paimon.utils.PaimonConversions import org.apache.fluss.metadata._ -import org.apache.fluss.metadata.TableDescriptor.{BUCKET_COLUMN_NAME, OFFSET_COLUMN_NAME, TIMESTAMP_COLUMN_NAME} import org.apache.fluss.server.utils.LakeStorageUtils import org.apache.fluss.spark.SparkCatalogTest import org.apache.fluss.spark.SparkConnectorOptions.{BUCKET_KEY, BUCKET_NUMBER, PRIMARY_KEY} @@ -154,12 +153,9 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest { .of( Array.apply( org.apache.paimon.types.DataTypes.INT, - org.apache.paimon.types.DataTypes.STRING, - org.apache.paimon.types.DataTypes.INT, - org.apache.paimon.types.DataTypes.BIGINT, - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS + org.apache.paimon.types.DataTypes.STRING ), - Array.apply("id", "name", BUCKET_COLUMN_NAME, OFFSET_COLUMN_NAME, TIMESTAMP_COLUMN_NAME) + Array.apply("id", "name") ), "id", 2 @@ -187,12 +183,9 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest { .of( Array.apply( org.apache.paimon.types.DataTypes.INT, - org.apache.paimon.types.DataTypes.STRING, - org.apache.paimon.types.DataTypes.INT, - org.apache.paimon.types.DataTypes.BIGINT, - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS + org.apache.paimon.types.DataTypes.STRING ), - Array.apply("id", "name", BUCKET_COLUMN_NAME, OFFSET_COLUMN_NAME, TIMESTAMP_COLUMN_NAME) + Array.apply("id", "name") ), null, 2 @@ -222,18 +215,9 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest { Array.apply( org.apache.paimon.types.DataTypes.INT.notNull(), org.apache.paimon.types.DataTypes.STRING, - org.apache.paimon.types.DataTypes.STRING.notNull(), - org.apache.paimon.types.DataTypes.INT, - org.apache.paimon.types.DataTypes.BIGINT, - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS + org.apache.paimon.types.DataTypes.STRING.notNull() ), - Array.apply( - "id", - "name", - "pk1", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME) + Array.apply("id", "name", "pk1") ), "id", 2 @@ -265,19 +249,9 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest { org.apache.paimon.types.DataTypes.INT.notNull(), org.apache.paimon.types.DataTypes.STRING, org.apache.paimon.types.DataTypes.STRING.notNull(), - org.apache.paimon.types.DataTypes.STRING.notNull(), - org.apache.paimon.types.DataTypes.INT, - org.apache.paimon.types.DataTypes.BIGINT, - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS + org.apache.paimon.types.DataTypes.STRING.notNull() ), - Array.apply( - "id", - "name", - "pk1", - "pt1", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME) + Array.apply("id", "name", "pk1", "pt1") ), "id", 2 @@ -328,10 +302,7 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest { ), org.apache.paimon.types.DataTypes.MAP( org.apache.paimon.types.DataTypes.STRING.notNull(), - org.apache.paimon.types.DataTypes.INT), - org.apache.paimon.types.DataTypes.INT, - org.apache.paimon.types.DataTypes.BIGINT, - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS + org.apache.paimon.types.DataTypes.INT) ), Array.apply( "c1", @@ -349,10 +320,7 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest { "c13", "c14", "c15", - "c16", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME) + "c16") ), null, 2 @@ -402,12 +370,9 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest { .of( Array.apply( org.apache.paimon.types.DataTypes.INT, - org.apache.paimon.types.DataTypes.STRING, - org.apache.paimon.types.DataTypes.INT, - org.apache.paimon.types.DataTypes.BIGINT, - org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS + org.apache.paimon.types.DataTypes.STRING ), - Array.apply("id", "name", BUCKET_COLUMN_NAME, OFFSET_COLUMN_NAME, TIMESTAMP_COLUMN_NAME) + Array.apply("id", "name") ), null, 2