diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java index 7db0a1957a0..8c9590bb377 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java @@ -20,10 +20,12 @@ import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.InvalidAlterTableException; +import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.exception.TableAlreadyExistException; import org.apache.fluss.exception.TableNotExistException; import org.apache.fluss.lake.iceberg.utils.IcebergCatalogUtils; import org.apache.fluss.lake.iceberg.utils.IcebergPartitionSpecUtils; +import org.apache.fluss.lake.iceberg.utils.IcebergUtils; import org.apache.fluss.lake.lakestorage.LakeCatalog; import org.apache.fluss.metadata.TableChange; import org.apache.fluss.metadata.TableDescriptor; @@ -60,7 +62,7 @@ import java.util.Map; import java.util.Set; -import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.SYSTEM_COLUMNS; +import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.LEGACY_SYSTEM_COLUMNS; import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; import static org.apache.fluss.utils.Preconditions.checkArgument; @@ -223,7 +225,8 @@ private void applySchemaChanges(Table table, List schemaChanges, Co } UpdateSchema updateSchema = table.updateSchema(); - String firstSystemColumnName = SYSTEM_COLUMNS.keySet().iterator().next(); + boolean isLegacyTable = IcebergUtils.isLegacyTable(currentIcebergSchema); + String firstSystemColumnName = LEGACY_SYSTEM_COLUMNS.keySet().iterator().next(); boolean hasChanges = false; for (TableChange tableChange : schemaChanges) { @@ -233,6 +236,13 @@ private void applySchemaChanges(Table table, List schemaChanges, Co } TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange; + if (LEGACY_SYSTEM_COLUMNS.containsKey(addColumn.getName())) { + throw new InvalidTableException( + "Column '" + + addColumn.getName() + + "' conflicts with a reserved system column name."); + } + if (!(addColumn.getPosition() instanceof TableChange.Last)) { throw new UnsupportedOperationException( "Only support to add column at last for iceberg table."); @@ -246,7 +256,12 @@ private void applySchemaChanges(Table table, List schemaChanges, Co Type icebergType = flussDataType.accept(new FlussDataTypeToIcebergDataType()); updateSchema.addColumn(addColumn.getName(), icebergType, addColumn.getComment()); - updateSchema.moveBefore(addColumn.getName(), firstSystemColumnName); + if (isLegacyTable) { + // Legacy tables keep the three system columns as the trailing columns, so a new + // business column must be inserted right before the first system column. Clean + // tables have no system columns, so the new column is simply appended last. + updateSchema.moveBefore(addColumn.getName(), firstSystemColumnName); + } hasChanges = true; } else { throw new UnsupportedOperationException( @@ -273,8 +288,13 @@ boolean isIcebergSchemaCompatible( if (flussTableDescriptor == null) { return false; } - // Identifier fields don't affect the comparison. - Schema expectedSchema = IcebergSchemaUtils.createIcebergSchema(flussTableDescriptor, false); + // FIP-27: newly created tables are clean (no system columns). Legacy tables still carry the + // three trailing system columns. To handle re-enabling lake tiering on a legacy table, we + // build the expected schema to match what the physical table actually has. + Schema expectedSchema = + IcebergUtils.isLegacyTable(icebergSchema) + ? IcebergSchemaUtils.createLegacyIcebergSchema(flussTableDescriptor, false) + : IcebergSchemaUtils.createIcebergSchema(flussTableDescriptor, false); return IcebergSchemaUtils.compatibleWith(icebergSchema, expectedSchema); } @@ -508,7 +528,11 @@ private void createDatabase(String databaseName) { } private SortOrder createSortOrder(Schema icebergSchema) { - // Sort by __offset system column for deterministic ordering + if (icebergSchema.findField(OFFSET_COLUMN_NAME) == null) { + // Clean tables (FIP-27) have no __offset system column; no sort order is needed. + return SortOrder.unsorted(); + } + // Legacy tables: sort by __offset for deterministic ordering SortOrder.Builder builder = SortOrder.builderFor(icebergSchema); builder.asc(OFFSET_COLUMN_NAME); return builder.build(); diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergSchemaUtils.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergSchemaUtils.java index 7559625f919..21ffd32173a 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergSchemaUtils.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergSchemaUtils.java @@ -41,34 +41,58 @@ @Internal public final class IcebergSchemaUtils { - /** The Iceberg-only system columns appended to every Fluss-managed Iceberg table. */ - public static final Map SYSTEM_COLUMNS; + /** + * System columns that legacy Iceberg tables (created before FIP-27) carry as trailing fields. + * Under FIP-27 these columns are no longer added to newly created (clean) tables. + */ + public static final Map LEGACY_SYSTEM_COLUMNS; static { LinkedHashMap systemColumns = new LinkedHashMap<>(); systemColumns.put(BUCKET_COLUMN_NAME, Types.IntegerType.get()); systemColumns.put(OFFSET_COLUMN_NAME, Types.LongType.get()); systemColumns.put(TIMESTAMP_COLUMN_NAME, Types.TimestampType.withZone()); - SYSTEM_COLUMNS = Collections.unmodifiableMap(systemColumns); + LEGACY_SYSTEM_COLUMNS = Collections.unmodifiableMap(systemColumns); } private IcebergSchemaUtils() {} - /** Creates the Iceberg schema managed by Fluss from a Fluss table descriptor. */ + /** + * Creates a clean Iceberg schema from a Fluss table descriptor. + * + *

FIP-27: newly created tables contain only user columns; no system columns are appended. + */ public static Schema createIcebergSchema( TableDescriptor tableDescriptor, boolean isPrimaryKeyTable) { + return buildIcebergSchema(tableDescriptor, isPrimaryKeyTable, false); + } + + /** + * Creates a legacy Iceberg schema from a Fluss table descriptor, appending the three system + * columns (__bucket, __offset, __timestamp) after the user columns. + * + *

Used only for compatibility checks against existing legacy tables (FIP-27 pre-existing). + */ + public static Schema createLegacyIcebergSchema( + TableDescriptor tableDescriptor, boolean isPrimaryKeyTable) { + return buildIcebergSchema(tableDescriptor, isPrimaryKeyTable, true); + } + + private static Schema buildIcebergSchema( + TableDescriptor tableDescriptor, boolean isPrimaryKeyTable, boolean includeSystemCols) { List fields = new ArrayList<>(); int fieldId = 0; + int userColCount = tableDescriptor.getSchema().getColumns().size(); int totalTopLevelFields = - tableDescriptor.getSchema().getColumns().size() + SYSTEM_COLUMNS.size(); + includeSystemCols ? userColCount + LEGACY_SYSTEM_COLUMNS.size() : userColCount; FlussDataTypeToIcebergDataType converter = new FlussDataTypeToIcebergDataType(totalTopLevelFields); 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 IllegalArgumentException( "Column '" + columnName @@ -93,10 +117,12 @@ public static Schema createIcebergSchema( fields.add(field); } - for (Map.Entry systemColumn : SYSTEM_COLUMNS.entrySet()) { - fields.add( - Types.NestedField.required( - fieldId++, systemColumn.getKey(), systemColumn.getValue())); + if (includeSystemCols) { + for (Map.Entry systemColumn : LEGACY_SYSTEM_COLUMNS.entrySet()) { + fields.add( + Types.NestedField.required( + fieldId++, systemColumn.getKey(), systemColumn.getValue())); + } } if (isPrimaryKeyTable) { diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteDataFiles.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteDataFiles.java index 570c056032d..96c1e44b821 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteDataFiles.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteDataFiles.java @@ -106,9 +106,13 @@ private List planRewriteFileGroups(long snapshotId) throws IOE // then, pack the fileScanTasks into compaction units which contains compactable // fileScanTasks, after compaction, we want to it still keep order by __offset column, - // so, let's first sort by __offset column - int offsetFieldId = table.schema().findField(OFFSET_COLUMN_NAME).fieldId(); - fileScanTasks.sort(sortFileScanTask(offsetFieldId)); + // so, let's first sort by __offset column. FIP-27: a clean table has no __offset column, + // so this ordering is skipped (files are packed as-is). + org.apache.iceberg.types.Types.NestedField offsetField = + table.schema().findField(OFFSET_COLUMN_NAME); + if (offsetField != null) { + fileScanTasks.sort(sortFileScanTask(offsetField.fieldId())); + } // do package now BinPacking.ListPacker packer = diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRow.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRow.java index a618df40e35..7bf3d7d15fb 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRow.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRow.java @@ -38,27 +38,39 @@ import java.util.List; import java.util.Map; -import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.SYSTEM_COLUMNS; +import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.LEGACY_SYSTEM_COLUMNS; +import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; /** Adapter for Iceberg Record as fluss row. */ public class IcebergRecordAsFlussRow implements InternalRow { private Record icebergRecord; + // cached business field count, recomputed when the record changes + private int businessFieldCount; public IcebergRecordAsFlussRow() {} public IcebergRecordAsFlussRow(Record icebergRecord) { this.icebergRecord = icebergRecord; + this.businessFieldCount = computeBusinessFieldCount(icebergRecord); } public IcebergRecordAsFlussRow replaceIcebergRecord(Record icebergRecord) { this.icebergRecord = icebergRecord; + this.businessFieldCount = computeBusinessFieldCount(icebergRecord); return this; } + private static int computeBusinessFieldCount(Record record) { + // A legacy table has __timestamp as the last column; a clean table has no system columns. + int total = record.struct().fields().size(); + boolean isLegacy = record.struct().field(TIMESTAMP_COLUMN_NAME) != null; + return isLegacy ? total - LEGACY_SYSTEM_COLUMNS.size() : total; + } + @Override public int getFieldCount() { - return icebergRecord.struct().fields().size() - SYSTEM_COLUMNS.size(); + return businessFieldCount; } @Override diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordReader.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordReader.java index f654fb5d040..63e63c988c7 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordReader.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordReader.java @@ -18,6 +18,7 @@ package org.apache.fluss.lake.iceberg.source; +import org.apache.fluss.lake.iceberg.utils.IcebergUtils; import org.apache.fluss.lake.source.RecordReader; import org.apache.fluss.record.ChangeType; import org.apache.fluss.record.GenericRecord; @@ -53,18 +54,24 @@ * org.apache.iceberg.Scan#ignoreResiduals()} for details. */ public class IcebergRecordReader implements RecordReader { + + /** Sentinel value emitted when the lake table has no per-record offset/timestamp. */ + private static final long NO_SYSTEM_COLUMN_VALUE = -1L; + protected IcebergRecordAsFlussRecordIterator iterator; protected @Nullable int[][] project; protected Types.StructType struct; public IcebergRecordReader(FileScanTask fileScanTask, Table table, @Nullable int[][] project) { + boolean isLegacy = IcebergUtils.isLegacyTable(table.schema()); TableScan tableScan = table.newScan(); if (project != null) { - tableScan = applyProject(tableScan, project); + tableScan = applyProject(tableScan, project, isLegacy); } IcebergGenericReader reader = new IcebergGenericReader(tableScan, true); struct = tableScan.schema().asStruct(); - this.iterator = new IcebergRecordAsFlussRecordIterator(reader.open(fileScanTask), struct); + this.iterator = + new IcebergRecordAsFlussRecordIterator(reader.open(fileScanTask), struct, isLegacy); } @Override @@ -72,7 +79,7 @@ public CloseableIterator read() throws IOException { return iterator; } - private TableScan applyProject(TableScan tableScan, int[][] projects) { + private TableScan applyProject(TableScan tableScan, int[][] projects, boolean isLegacy) { Types.StructType structType = tableScan.schema().asStruct(); List cols = new ArrayList<>(projects.length + 2); @@ -80,8 +87,12 @@ private TableScan applyProject(TableScan tableScan, int[][] projects) { cols.add(structType.fields().get(project[0])); } - cols.add(structType.field(OFFSET_COLUMN_NAME)); - cols.add(structType.field(TIMESTAMP_COLUMN_NAME)); + if (isLegacy) { + // Legacy tables carry __offset and __timestamp; project them so the iterator can + // read the actual per-record offset and timestamp values. + cols.add(structType.field(OFFSET_COLUMN_NAME)); + cols.add(structType.field(TIMESTAMP_COLUMN_NAME)); + } return tableScan.project(new Schema(cols)); } @@ -97,13 +108,25 @@ public static class IcebergRecordAsFlussRecordIterator implements CloseableItera private final int timestampColIndex; public IcebergRecordAsFlussRecordIterator( - CloseableIterable icebergRecordIterator, Types.StructType struct) { + CloseableIterable icebergRecordIterator, + Types.StructType struct, + boolean isLegacy) { this.icebergRecordIterator = icebergRecordIterator.iterator(); - this.logOffsetColIndex = struct.fields().indexOf(struct.field(OFFSET_COLUMN_NAME)); - this.timestampColIndex = struct.fields().indexOf(struct.field(TIMESTAMP_COLUMN_NAME)); - int[] project = IntStream.range(0, struct.fields().size() - 2).toArray(); - projectedRow = ProjectedRow.from(project); + if (isLegacy) { + this.logOffsetColIndex = struct.fields().indexOf(struct.field(OFFSET_COLUMN_NAME)); + this.timestampColIndex = + struct.fields().indexOf(struct.field(TIMESTAMP_COLUMN_NAME)); + // The last two projected columns are __offset and __timestamp; strip them from the + // business row. + int[] project = IntStream.range(0, struct.fields().size() - 2).toArray(); + projectedRow = ProjectedRow.from(project); + } else { + this.logOffsetColIndex = -1; + this.timestampColIndex = -1; + projectedRow = + ProjectedRow.from(IntStream.range(0, struct.fields().size()).toArray()); + } icebergRecordAsFlussRow = new IcebergRecordAsFlussRow(); } @@ -124,12 +147,19 @@ public boolean hasNext() { @Override public LogRecord next() { Record icebergRecord = icebergRecordIterator.next(); - long offset = icebergRecord.get(logOffsetColIndex, Long.class); - long timestamp = - icebergRecord - .get(timestampColIndex, OffsetDateTime.class) - .toInstant() - .toEpochMilli(); + long offset; + long timestamp; + if (logOffsetColIndex >= 0) { + offset = icebergRecord.get(logOffsetColIndex, Long.class); + timestamp = + icebergRecord + .get(timestampColIndex, OffsetDateTime.class) + .toInstant() + .toEpochMilli(); + } else { + offset = NO_SYSTEM_COLUMN_VALUE; + timestamp = NO_SYSTEM_COLUMN_VALUE; + } return new GenericRecord( offset, diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java index a397a5af00e..afb66aecb19 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java @@ -35,6 +35,7 @@ import org.apache.iceberg.expressions.ExpressionVisitors; import org.apache.iceberg.expressions.UnboundPredicate; import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.types.Types; import javax.annotation.Nullable; @@ -146,16 +147,25 @@ private Function createBucketExtractor(Table table) { PartitionSpec partitionSpec = table.spec(); List partitionFields = partitionSpec.fields(); - // the last one must be partition by fluss bucket - PartitionField bucketField = partitionFields.get(partitionFields.size() - 1); + // FIP-27: a clean bucket-unaware table has an empty (unpartitioned) spec, so there is no + // bucket to extract. + if (partitionFields.isEmpty()) { + return task -> -1; + } - if (table.schema() - .asStruct() - .field(bucketField.sourceId()) - .name() - .equals(BUCKET_COLUMN_NAME)) { - // partition by __bucket column, should be fluss log table without bucket key, - // we don't care about the bucket since it's bucket un-aware + // The bucket is the last partition field only when it is a Fluss-bucketed field: either the + // legacy identity(__bucket) partition, or a bucket(bucketKey) transform. If the last field + // is an identity partition on a user column (bucket-unaware partitioned table), there is no + // bucket to extract. + PartitionField lastField = partitionFields.get(partitionFields.size() - 1); + Types.NestedField lastSourceField = table.schema().findField(lastField.sourceId()); + boolean lastIsLegacyBucket = + lastSourceField != null && lastSourceField.name().equals(BUCKET_COLUMN_NAME); + boolean lastIsBucketTransform = lastField.transform().toString().startsWith("bucket["); + + if (lastIsLegacyBucket || !lastIsBucketTransform) { + // legacy bucket-unaware (identity __bucket), or clean bucket-unaware partitioned + // (last field is an identity partition column) -> no meaningful bucket. return task -> -1; } else { int bucketFieldIndex = partitionFields.size() - 1; @@ -167,23 +177,30 @@ private Function> createPartitionExtractor(Table tabl PartitionSpec partitionSpec = table.spec(); List partitionFields = partitionSpec.fields(); - // if only one partition, it must not be partitioned table since we will always use - // partition by fluss bucket - if (partitionSpec.fields().size() <= 1) { + if (partitionFields.isEmpty()) { + return task -> Collections.emptyList(); + } + + // The trailing partition field is the Fluss bucket (legacy identity(__bucket) or a + // bucket(bucketKey) transform); everything before it is the Fluss partition columns. A + // clean bucket-unaware partitioned table has no trailing bucket field, so all fields are + // partition columns. + PartitionField lastField = partitionFields.get(partitionFields.size() - 1); + Types.NestedField lastSourceField = table.schema().findField(lastField.sourceId()); + boolean lastIsBucket = + (lastSourceField != null && lastSourceField.name().equals(BUCKET_COLUMN_NAME)) + || lastField.transform().toString().startsWith("bucket["); + int partitionColCount = lastIsBucket ? partitionFields.size() - 1 : partitionFields.size(); + + if (partitionColCount == 0) { return task -> Collections.emptyList(); - } else { - List partitionFieldIndices = - // since will always first partition by fluss partition columns, then fluss - // bucket, - // just ignore the last partition column of iceberg - IntStream.range(0, partitionFields.size() - 1) - .boxed() - .collect(Collectors.toList()); - return task -> - partitionFieldIndices.stream() - // since currently, only string partition is supported - .map(index -> task.partition().get(index, String.class)) - .collect(Collectors.toList()); } + List partitionFieldIndices = + IntStream.range(0, partitionColCount).boxed().collect(Collectors.toList()); + return task -> + partitionFieldIndices.stream() + // since currently, only string partition is supported + .map(index -> task.partition().get(index, String.class)) + .collect(Collectors.toList()); } } diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/FlussRecordAsIcebergRecord.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/FlussRecordAsIcebergRecord.java index 60f400575d9..77b2ff19b25 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/FlussRecordAsIcebergRecord.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/FlussRecordAsIcebergRecord.java @@ -28,11 +28,10 @@ import java.time.OffsetDateTime; import java.time.ZoneOffset; -import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.SYSTEM_COLUMNS; +import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.LEGACY_SYSTEM_COLUMNS; 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; -import static org.apache.fluss.utils.Preconditions.checkState; /** * Wrap Fluss {@link LogRecord} as Iceberg {@link Record}. @@ -42,33 +41,32 @@ */ public class FlussRecordAsIcebergRecord extends FlussRowAsIcebergRecord { - // Lake table for iceberg will append three system columns: __bucket, __offset,__timestamp - private static final int LAKE_ICEBERG_SYSTEM_COLUMNS = SYSTEM_COLUMNS.size(); - private LogRecord logRecord; private final int bucket; + private final boolean isLegacy; - // the origin row fields in fluss, excluding the system columns in iceberg - private int originRowFieldCount; + // the count of user (business) columns; system columns (if any) start at this index + private final int businessFieldCount; public FlussRecordAsIcebergRecord( - int bucket, Types.StructType structType, RowType flussRowType) { + int bucket, Types.StructType structType, RowType flussRowType, boolean isLegacy) { super(structType, flussRowType); this.bucket = bucket; + this.isLegacy = isLegacy; + this.businessFieldCount = + isLegacy + ? structType.fields().size() - LEGACY_SYSTEM_COLUMNS.size() + : structType.fields().size(); } public void setFlussRecord(LogRecord logRecord) { this.logRecord = logRecord; this.internalRow = logRecord.getRow(); - this.originRowFieldCount = internalRow.getFieldCount(); - checkState( - originRowFieldCount == structType.fields().size() - LAKE_ICEBERG_SYSTEM_COLUMNS, - "The Iceberg table fields count must equals to LogRecord's fields count."); } @Override public Object getField(String name) { - if (SYSTEM_COLUMNS.containsKey(name)) { + if (isLegacy && LEGACY_SYSTEM_COLUMNS.containsKey(name)) { switch (name) { case BUCKET_COLUMN_NAME: return bucket; @@ -85,16 +83,14 @@ public Object getField(String name) { @Override public Object get(int pos) { - // firstly, for system columns - if (pos == originRowFieldCount) { - // bucket column - return bucket; - } else if (pos == originRowFieldCount + 1) { - // log offset column - return logRecord.logOffset(); - } else if (pos == originRowFieldCount + 2) { - // timestamp column - return toIcebergTimestampLtz(logRecord.timestamp()); + if (isLegacy) { + if (pos == businessFieldCount) { + return bucket; + } else if (pos == businessFieldCount + 1) { + return logRecord.logOffset(); + } else if (pos == businessFieldCount + 2) { + return toIcebergTimestampLtz(logRecord.timestamp()); + } } return super.get(pos); } diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergPartitionSpecValidator.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergPartitionSpecValidator.java index 732ea8de037..8f2bbb694a2 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergPartitionSpecValidator.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergPartitionSpecValidator.java @@ -20,6 +20,7 @@ import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.lake.iceberg.IcebergSchemaUtils; import org.apache.fluss.lake.iceberg.utils.IcebergPartitionSpecUtils; +import org.apache.fluss.lake.iceberg.utils.IcebergUtils; import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TableInfo; @@ -38,8 +39,13 @@ private IcebergPartitionSpecValidator() {} static void validate(Table icebergTable, TableInfo tableInfo) { TableDescriptor tableDescriptor = tableInfo.toTableDescriptor(); Schema icebergSchema = icebergTable.schema(); + // FIP-27: use a legacy expected schema when the physical table is legacy (has system cols) Schema expectedSchema = - IcebergSchemaUtils.createIcebergSchema(tableDescriptor, tableInfo.hasPrimaryKey()); + IcebergUtils.isLegacyTable(icebergSchema) + ? IcebergSchemaUtils.createLegacyIcebergSchema( + tableDescriptor, tableInfo.hasPrimaryKey()) + : IcebergSchemaUtils.createIcebergSchema( + tableDescriptor, tableInfo.hasPrimaryKey()); if (!IcebergSchemaUtils.compatibleWith(icebergSchema, expectedSchema)) { throw new InvalidTableException( String.format( diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/RecordWriter.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/RecordWriter.java index 238ef5be981..275f72a0988 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/RecordWriter.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/RecordWriter.java @@ -17,6 +17,7 @@ package org.apache.fluss.lake.iceberg.tiering; +import org.apache.fluss.lake.iceberg.utils.IcebergUtils; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.record.LogRecord; import org.apache.fluss.types.RowType; @@ -44,7 +45,10 @@ public RecordWriter( this.bucket = tableBucket.getBucket(); this.flussRecordAsIcebergRecord = new FlussRecordAsIcebergRecord( - tableBucket.getBucket(), icebergSchema.asStruct(), flussRowType); + tableBucket.getBucket(), + icebergSchema.asStruct(), + flussRowType, + IcebergUtils.isLegacyTable(icebergSchema)); } public abstract void write(LogRecord record) throws Exception; diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergConversions.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergConversions.java index 111df385ff9..70a15c58999 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergConversions.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergConversions.java @@ -61,7 +61,21 @@ public static PartitionKey toPartition( partitionKey.set(pos++, partition); } } - partitionKey.set(pos, bucket); + // Set the bucket value only when the partition spec has a trailing bucket field — either + // the legacy identity(__bucket) column or a bucket(userCol) transform. Unpartitioned clean + // tables (empty spec) and bucket-unaware partitioned tables (last field is an identity + // partition column) have no such field and must not set a bucket slot. + List fields = partitionSpec.fields(); + if (!fields.isEmpty()) { + PartitionField lastField = fields.get(fields.size() - 1); + Types.NestedField lastSourceField = schema.findField(lastField.sourceId()); + boolean lastIsLegacyBucket = + lastSourceField != null && lastSourceField.name().equals(BUCKET_COLUMN_NAME); + boolean lastIsBucketTransform = lastField.transform().toString().startsWith("bucket["); + if (lastIsLegacyBucket || lastIsBucketTransform) { + partitionKey.set(pos, bucket); + } + } return partitionKey; } @@ -85,7 +99,11 @@ public static Expression toFilterExpression( partition)); } } - expression = Expressions.and(expression, Expressions.equal(BUCKET_COLUMN_NAME, bucket)); + // FIP-27: legacy tables carry the __bucket column and are filtered per bucket. Clean + // tables have no __bucket column, so no bucket-level filter is applied. + if (table.schema().findField(BUCKET_COLUMN_NAME) != null) { + expression = Expressions.and(expression, Expressions.equal(BUCKET_COLUMN_NAME, bucket)); + } return expression; } diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergPartitionSpecUtils.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergPartitionSpecUtils.java index f9c7c0e0c61..0bb52d44d0c 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergPartitionSpecUtils.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergPartitionSpecUtils.java @@ -82,9 +82,13 @@ private static PartitionSpec createPartitionSpec( } if (bucketKeys.isEmpty()) { - // __offset and __timestamp are system data columns, but only __bucket is a - // partition field when the Fluss table has no bucket key. - builder.identity(BUCKET_COLUMN_NAME); + // FIP-27: a bucket-unaware table is partitioned by the __bucket system column only for + // legacy tables that still carry it. Clean tables have no __bucket column, so they are + // left unpartitioned (IcebergSplitPlanner treats an empty/partition-less spec as + // bucket-unaware). + if (icebergSchema.findField(BUCKET_COLUMN_NAME) != null) { + builder.identity(BUCKET_COLUMN_NAME); + } } else { builder.bucket(bucketKeys.get(0), bucketCount); } diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergUtils.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergUtils.java new file mode 100644 index 00000000000..4cb5aabb77c --- /dev/null +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergUtils.java @@ -0,0 +1,45 @@ +/* + * 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.iceberg.utils; + +import org.apache.iceberg.Schema; + +import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; + +/** + * Utility methods for Iceberg lake tables. + * + *

FIP-27: Newly created Iceberg lake tables ("clean" tables) contain only user columns. Legacy + * tables created before FIP-27 still carry the three trailing system columns (__bucket, __offset, + * __timestamp). This class provides detection logic to distinguish between the two layouts. + */ +public final class IcebergUtils { + + private IcebergUtils() {} + + /** + * Returns whether the given Iceberg table is a legacy table (has the three trailing system + * columns). + * + *

Detection: if the {@code __timestamp} system column exists in the physical schema, this is + * a legacy table. Clean tables have no system columns. + */ + public static boolean isLegacyTable(Schema icebergSchema) { + return icebergSchema.findField(TIMESTAMP_COLUMN_NAME) != null; + } +} diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java index 30f2b82f1c0..495bee71813 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java @@ -37,8 +37,6 @@ import org.apache.iceberg.PartitionField; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.RowLevelOperationMode; -import org.apache.iceberg.SortDirection; -import org.apache.iceberg.SortField; import org.apache.iceberg.SortOrder; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; @@ -160,13 +158,7 @@ void testCreatePrimaryKeyTable() { Arrays.asList( Types.NestedField.required(1, "id", Types.IntegerType.get()), Types.NestedField.optional( - 2, "name", Types.StringType.get(), "field name"), - Types.NestedField.required( - 3, BUCKET_COLUMN_NAME, Types.IntegerType.get()), - Types.NestedField.required( - 4, OFFSET_COLUMN_NAME, Types.LongType.get()), - Types.NestedField.required( - 5, TIMESTAMP_COLUMN_NAME, Types.TimestampType.withZone())), + 2, "name", Types.StringType.get(), "field name")), Collections.singleton(1)); assertThat(createdTable.schema().toString()).isEqualTo(expectIcebergSchema.toString()); @@ -222,13 +214,7 @@ void testCreatePartitionedPrimaryKeyTable() { Types.NestedField.optional( 4, "num_orders", Types.IntegerType.get()), Types.NestedField.required( - 5, "total_amount", Types.IntegerType.get()), - Types.NestedField.required( - 6, BUCKET_COLUMN_NAME, Types.IntegerType.get()), - Types.NestedField.required( - 7, OFFSET_COLUMN_NAME, Types.LongType.get()), - Types.NestedField.required( - 8, TIMESTAMP_COLUMN_NAME, Types.TimestampType.withZone())), + 5, "total_amount", Types.IntegerType.get())), identifierFieldIds); assertThat(createdTable.schema().toString()).isEqualTo(expectIcebergSchema.toString()); @@ -246,12 +232,8 @@ void testCreatePartitionedPrimaryKeyTable() { assertThat(partitionField2.transform().toString()).isEqualTo("bucket[10]"); assertThat(partitionField2.sourceId()).isEqualTo(2); - // Verify sort order - assertThat(createdTable.sortOrder().fields()).hasSize(1); - SortField sortField = createdTable.sortOrder().fields().get(0); - assertThat(sortField.sourceId()) - .isEqualTo(createdTable.schema().findField(OFFSET_COLUMN_NAME).fieldId()); - assertThat(sortField.direction()).isEqualTo(SortDirection.ASC); + // Verify sort order (FIP-27: clean tables are unsorted) + assertThat(createdTable.sortOrder().isUnsorted()).isTrue(); // Verify table properties assertThat(createdTable.properties()) @@ -325,29 +307,14 @@ void testCreateLogTable() { Types.NestedField.optional(1, "id", Types.LongType.get()), Types.NestedField.optional(2, "name", Types.StringType.get()), Types.NestedField.optional(3, "amount", Types.IntegerType.get()), - Types.NestedField.optional(4, "address", Types.StringType.get()), - Types.NestedField.required( - 5, BUCKET_COLUMN_NAME, Types.IntegerType.get()), - Types.NestedField.required( - 6, OFFSET_COLUMN_NAME, Types.LongType.get()), - Types.NestedField.required( - 7, TIMESTAMP_COLUMN_NAME, Types.TimestampType.withZone()))); + Types.NestedField.optional(4, "address", Types.StringType.get()))); - // Verify iceberg table schema + // Verify iceberg table schema (FIP-27: clean layout, no system columns) assertThat(createdTable.schema().toString()).isEqualTo(expectIcebergSchema.toString()); - // Verify partition field and transform - assertThat(createdTable.spec().fields()).hasSize(1); - PartitionField partitionField = createdTable.spec().fields().get(0); - assertThat(partitionField.name()).isEqualTo(BUCKET_COLUMN_NAME); - assertThat(partitionField.transform().toString()).isEqualTo("identity"); - - // Verify sort field and order - assertThat(createdTable.sortOrder().fields()).hasSize(1); - SortField sortField = createdTable.sortOrder().fields().get(0); - assertThat(sortField.sourceId()) - .isEqualTo(createdTable.schema().findField(OFFSET_COLUMN_NAME).fieldId()); - assertThat(sortField.direction()).isEqualTo(SortDirection.ASC); + // FIP-27: a clean bucket-unaware table is unpartitioned and unsorted. + assertThat(createdTable.spec().isUnpartitioned()).isTrue(); + assertThat(createdTable.sortOrder().isUnsorted()).isTrue(); } @Test @@ -384,33 +351,20 @@ void testCreatePartitionedLogTable() { Types.NestedField.optional(1, "id", Types.LongType.get()), Types.NestedField.optional(2, "name", Types.StringType.get()), Types.NestedField.optional(3, "amount", Types.IntegerType.get()), - Types.NestedField.optional(4, "order_type", Types.StringType.get()), - Types.NestedField.required( - 5, BUCKET_COLUMN_NAME, Types.IntegerType.get()), - Types.NestedField.required( - 6, OFFSET_COLUMN_NAME, Types.LongType.get()), - Types.NestedField.required( - 7, TIMESTAMP_COLUMN_NAME, Types.TimestampType.withZone()))); + Types.NestedField.optional( + 4, "order_type", Types.StringType.get()))); - // Verify iceberg table schema + // Verify iceberg table schema (FIP-27: clean layout, no system columns) assertThat(createdTable.schema().toString()).isEqualTo(expectIcebergSchema.toString()); - // Verify partition field and transform - assertThat(createdTable.spec().fields()).hasSize(2); + // Verify partition field and transform (FIP-27: only the partition key, no __bucket) + assertThat(createdTable.spec().fields()).hasSize(1); PartitionField firstPartitionField = createdTable.spec().fields().get(0); assertThat(firstPartitionField.name()).isEqualTo("order_type"); assertThat(firstPartitionField.transform().toString()).isEqualTo("identity"); - PartitionField secondPartitionField = createdTable.spec().fields().get(1); - assertThat(secondPartitionField.name()).isEqualTo(BUCKET_COLUMN_NAME); - assertThat(secondPartitionField.transform().toString()).isEqualTo("identity"); - - // Verify sort field and order - assertThat(createdTable.sortOrder().fields()).hasSize(1); - SortField sortField = createdTable.sortOrder().fields().get(0); - assertThat(sortField.sourceId()) - .isEqualTo(createdTable.schema().findField(OFFSET_COLUMN_NAME).fieldId()); - assertThat(sortField.direction()).isEqualTo(SortDirection.ASC); + // Verify sort order (FIP-27: clean tables are unsorted) + assertThat(createdTable.sortOrder().isUnsorted()).isTrue(); } @Test @@ -673,16 +627,7 @@ void testAlterTableAddColumnLastNullable() { table.schema().columns().stream() .map(Types.NestedField::name) .collect(Collectors.toList()); - assertThat(fieldNames) - .containsExactly( - "id", - "name", - "amount", - "address", - "new_col", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME); + assertThat(fieldNames).containsExactly("id", "name", "amount", "address", "new_col"); // Verify the new column's type and nullability Types.NestedField newCol = table.schema().findField("new_col"); @@ -860,15 +805,7 @@ void testAlterTableAddColumnWithComplexTypeTable() { table.schema().columns().stream() .map(Types.NestedField::name) .collect(Collectors.toList()); - assertThat(fieldNames) - .containsExactly( - "id", - "tags", - "metadata", - "new_col", - BUCKET_COLUMN_NAME, - OFFSET_COLUMN_NAME, - TIMESTAMP_COLUMN_NAME); + assertThat(fieldNames).containsExactly("id", "tags", "metadata", "new_col"); // Verify the new column Types.NestedField newCol = table.schema().findField("new_col"); @@ -939,7 +876,7 @@ void testAlterTableAddArrayColumn() { assertThat(listType.elementId()).isNotEqualTo(field.fieldId()); assertThat(table.schema().columns()) .extracting(Types.NestedField::name) - .containsSubsequence("new_arr", BUCKET_COLUMN_NAME); + .containsExactly("id", "name", "amount", "address", "new_arr"); } @Test diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkCatalogLakeTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkCatalogLakeTest.java index 51c25e813a1..aad6989910c 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkCatalogLakeTest.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkCatalogLakeTest.java @@ -41,7 +41,6 @@ import java.util.Map; import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED; -import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.SYSTEM_COLUMNS; import static org.assertj.core.api.Assertions.assertThat; /** Test class for {@link FlinkCatalog}. */ @@ -71,7 +70,7 @@ void testGetLakeTable() throws Exception { CatalogBaseTable lakeTable = catalog.getTable(new ObjectPath(DEFAULT_DB, "lake_table$lake")); Schema schema = lakeTable.getUnresolvedSchema(); - assertThat(schema.getColumns().size()).isEqualTo(3 + SYSTEM_COLUMNS.size()); + assertThat(schema.getColumns().size()).isEqualTo(3); assertThat(schema.getPrimaryKey().isPresent()).isTrue(); assertThat(schema.getPrimaryKey().get().getColumnNames()).isEqualTo(List.of("first")); } diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java index 5bd021b9ad9..e9d60e87902 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java @@ -338,8 +338,11 @@ protected void checkDataInIcebergAppendOnlyTable( InternalRow flussRow = flussRowIterator.next(); assertThat(actualRecord.get(0)).isEqualTo(flussRow.getInt(0)); assertThat(actualRecord.get(1)).isEqualTo(flussRow.getString(1).toString()); - // the idx 2 is __bucket, so use 3 - assertThat(actualRecord.get(3)).isEqualTo(startingOffset++); + // FIP-27: a clean table stores only user columns; only legacy tables carry the + // trailing __bucket/__offset/__timestamp columns (offset at idx 3). + if (actualRecord.struct().field(OFFSET_COLUMN_NAME) != null) { + assertThat(actualRecord.get(3)).isEqualTo(startingOffset++); + } } assertThat(flussRowIterator.hasNext()).isFalse(); } @@ -377,8 +380,11 @@ protected void checkDataInIcebergAppendOnlyPartitionedTable( assertThat(actualRecord.get(0)).isEqualTo(flussRow.getInt(0)); assertThat(actualRecord.get(1)).isEqualTo(flussRow.getString(1).toString()); assertThat(actualRecord.get(2)).isEqualTo(flussRow.getString(2).toString()); - // the idx 3 is __bucket, so use 4 - assertThat(actualRecord.get(4)).isEqualTo(startingOffset++); + // FIP-27: only legacy tables carry the trailing __bucket/__offset/__timestamp + // columns (offset at idx 4 for a 3-user-column partitioned table). + if (actualRecord.struct().field(OFFSET_COLUMN_NAME) != null) { + assertThat(actualRecord.get(4)).isEqualTo(startingOffset++); + } } assertThat(flussRowIterator.hasNext()).isFalse(); } @@ -402,7 +408,33 @@ private CloseableIterator getIcebergRows( // is log table, we want to compare __offset column // so sort data files by __offset according to the column stats List records = new ArrayList<>(); - int fieldId = table.schema().findField(OFFSET_COLUMN_NAME).fieldId(); + org.apache.iceberg.types.Types.NestedField offsetField = + table.schema().findField(OFFSET_COLUMN_NAME); + if (offsetField == null) { + // FIP-27: a clean table has no __offset column to sort by; read files directly. + table.refresh(); + TableScan cleanScan = filterByPartition(table.newScan(), partitionSpec); + cleanScan + .planFiles() + .iterator() + .forEachRemaining( + fileScanTask -> { + DataFile file = fileScanTask.file(); + Iterable iterable = + Parquet.read(table.io().newInputFile(file.location())) + .project(table.schema()) + .createReaderFunc( + fileSchema -> + GenericParquetReaders + .buildReader( + table.schema(), + fileSchema)) + .build(); + iterable.forEach(records::add); + }); + return CloseableIterator.withClose(records.iterator()); + } + int fieldId = offsetField.fieldId(); SortedSet files = new TreeSet<>( (f1, f2) -> { diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergSchemaEvolutionITCase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergSchemaEvolutionITCase.java index d4177a305f1..fe813d81857 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergSchemaEvolutionITCase.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergSchemaEvolutionITCase.java @@ -40,7 +40,6 @@ import java.util.stream.Collectors; import static org.apache.fluss.lake.iceberg.utils.IcebergConversions.toIceberg; -import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; import static org.apache.fluss.testutils.DataTestUtils.row; import static org.assertj.core.api.Assertions.assertThat; @@ -125,7 +124,8 @@ void testSchemaEvolutionLogTable() throws Exception { .map(Types.NestedField::name) .collect(Collectors.toList()); assertThat(fieldNames.indexOf("f_new")) - .isLessThan(fieldNames.indexOf(BUCKET_COLUMN_NAME)); + .isEqualTo( + fieldNames.size() - 1); // FIP-27: clean table appends new columns last // Post-ALTER rows: new rows carry f_new values; old rows must surface NULL // via Iceberg field-ID schema evolution. @@ -197,7 +197,8 @@ void testSchemaEvolutionPkTable() throws Exception { .map(Types.NestedField::name) .collect(Collectors.toList()); assertThat(fieldNames.indexOf("f_new")) - .isLessThan(fieldNames.indexOf(BUCKET_COLUMN_NAME)); + .isEqualTo( + fieldNames.size() - 1); // FIP-27: clean table appends new columns last // Post-ALTER upserts + snapshot trigger; verify all rows tier. List postAlterRows = Arrays.asList(row(4, "v4", 100), row(5, "v5", 200)); @@ -272,7 +273,8 @@ void testSchemaEvolutionLogTableWithComplexTypes() throws Exception { .map(Types.NestedField::name) .collect(Collectors.toList()); assertThat(fieldNames.indexOf("f_new")) - .isLessThan(fieldNames.indexOf(BUCKET_COLUMN_NAME)); + .isEqualTo( + fieldNames.size() - 1); // FIP-27: clean table appends new columns last assertThat(icebergTable.schema().findField("f_tags").type().isListType()).isTrue(); assertThat(icebergTable.schema().findField("f_meta").type().isMapType()).isTrue(); @@ -333,7 +335,8 @@ void testSchemaEvolutionLogTableAddComplexColumn() throws Exception { .map(Types.NestedField::name) .collect(Collectors.toList()); assertThat(fieldNames.indexOf("f_tags")) - .isLessThan(fieldNames.indexOf(BUCKET_COLUMN_NAME)); + .isEqualTo( + fieldNames.size() - 1); // FIP-27: clean table appends new columns last // Post-ALTER rows for a complex new column. Writing null exercises the // data-plane path through the new ARRAY field ID without forcing the test @@ -394,7 +397,8 @@ void testSchemaEvolutionPkTableWithComplexPreExisting() throws Exception { .map(Types.NestedField::name) .collect(Collectors.toList()); assertThat(fieldNames.indexOf("f_new")) - .isLessThan(fieldNames.indexOf(BUCKET_COLUMN_NAME)); + .isEqualTo( + fieldNames.size() - 1); // FIP-27: clean table appends new columns last // Post-ALTER upserts with f_new values + snapshot trigger. List postAlterRows =