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..9861a63b9d7 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; @@ -116,6 +118,8 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co createTable( tablePath, tableBuilder, + tableDescriptor, + isPkTable, icebergSchema, partitionSpec, sortOrder, @@ -127,6 +131,8 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co createTable( tablePath, tableBuilder, + tableDescriptor, + isPkTable, icebergSchema, partitionSpec, sortOrder, @@ -223,7 +229,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 +240,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 +260,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 +292,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); } @@ -285,6 +309,8 @@ private TableIdentifier toIcebergTableIdentifier(TablePath tablePath) { private void createTable( TablePath tablePath, Catalog.TableBuilder tableBuilder, + TableDescriptor tableDescriptor, + boolean isPkTable, Schema newIcebergSchema, PartitionSpec expectedSpec, SortOrder expectedSortOrder, @@ -297,6 +323,22 @@ private void createTable( TableIdentifier icebergId = toIcebergTableIdentifier(tablePath); Table existingTable = icebergCatalog.loadTable(icebergId); Schema existingSchema = existingTable.schema(); + + // FIP-27: re-enabling tiering on a legacy table (created before FIP-27) hits this path. + // The expectations above were built for the clean layout, so rebuild the expected + // schema/spec/sort-order in the legacy layout to match what the physical table actually + // has; otherwise the compatibility checks below would spuriously reject the legacy + // table. + boolean existingIsLegacy = IcebergUtils.isLegacyTable(existingSchema); + if (existingIsLegacy) { + newIcebergSchema = + IcebergSchemaUtils.createLegacyIcebergSchema(tableDescriptor, isPkTable); + expectedSpec = + IcebergPartitionSpecUtils.createPartitionSpec( + tableDescriptor, newIcebergSchema); + expectedSortOrder = createSortOrder(newIcebergSchema); + } + if (!isIcebergSchemaCompatibleWithSchema(existingSchema, newIcebergSchema)) { throw new TableAlreadyExistException( String.format( @@ -323,17 +365,23 @@ private void createTable( existingSchema, expectedSortOrder, newIcebergSchema)) { + // Legacy tables expect ASC(__offset); clean tables (no __offset) expect unsorted(). + String sortOrderGuidance = + existingIsLegacy + ? String.format( + "or with ASC(%s) set explicitly, ", OFFSET_COLUMN_NAME) + : ""; throw new TableAlreadyExistException( String.format( "The table %s already exists in Iceberg catalog, but the sort order is not compatible. " + "Existing sort order: %s, new sort order: %s. " + "Please first drop the table in Iceberg catalog, or pre-create the " - + "Iceberg table without a custom sort order, or with ASC(%s) set explicitly, " + + "Iceberg table without a custom sort order, %s" + "or use a new table name.", tablePath, existingTable.sortOrder(), expectedSortOrder, - OFFSET_COLUMN_NAME), + sortOrderGuidance), e); } @@ -508,7 +556,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..cc884585df4 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 @@ -28,6 +28,7 @@ import org.apache.fluss.utils.BytesUtils; import org.apache.iceberg.data.Record; +import org.apache.iceberg.types.Types; import java.math.BigDecimal; import java.nio.ByteBuffer; @@ -38,27 +39,51 @@ import java.util.List; import java.util.Map; -import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.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; /** 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) { + // Subtract the system columns that are actually present in this record's struct rather than + // a fixed count: a projected legacy record may carry only a subset (e.g. __offset and + // __timestamp but not __bucket), and a clean record carries none. + Types.StructType struct = record.struct(); + int systemColumns = 0; + if (struct.field(BUCKET_COLUMN_NAME) != null) { + systemColumns++; + } + if (struct.field(OFFSET_COLUMN_NAME) != null) { + systemColumns++; + } + if (struct.field(TIMESTAMP_COLUMN_NAME) != null) { + systemColumns++; + } + return struct.fields().size() - systemColumns; + } + @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..fe1da1144bb 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,7 +28,7 @@ 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; @@ -42,33 +42,40 @@ */ 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(); + // Guard against a schema/record mismatch surfacing later as a bare + // ArrayIndexOutOfBoundsException inside the Iceberg appender. The Fluss row carries only + // business columns; system columns (if any) are supplied by this wrapper. checkState( - originRowFieldCount == structType.fields().size() - LAKE_ICEBERG_SYSTEM_COLUMNS, - "The Iceberg table fields count must equals to LogRecord's fields count."); + internalRow.getFieldCount() == businessFieldCount, + "The Fluss record's field count (%s) must equal the business field count (%s).", + internalRow.getFieldCount(), + businessFieldCount); } @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 +92,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/IcebergLakeWriter.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeWriter.java index 93f60ed4809..f4351cc87c7 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeWriter.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeWriter.java @@ -22,6 +22,7 @@ import org.apache.fluss.lake.iceberg.tiering.writer.AppendOnlyTaskWriter; import org.apache.fluss.lake.iceberg.tiering.writer.DeltaTaskWriter; import org.apache.fluss.lake.iceberg.tiering.writer.TaskWriterFactory; +import org.apache.fluss.lake.iceberg.utils.IcebergUtils; import org.apache.fluss.lake.writer.LakeWriter; import org.apache.fluss.lake.writer.WriterInitContext; import org.apache.fluss.metadata.TablePath; @@ -73,7 +74,16 @@ public IcebergLakeWriter( // Create a record writer this.recordWriter = createRecordWriter(writerInitContext); - if (writerInitContext.tableInfo().getTableConfig().isDataLakeAutoCompaction()) { + // FIP-27: auto compaction relies on the per-bucket __bucket predicate to give each writer + // an + // exclusive file set. Clean tables have no __bucket column, so different buckets' + // compaction + // jobs would plan overlapping files and race on rewrite/commit. Only schedule compaction + // for + // legacy tables until a bucket-scoped ownership model for clean tables exists. + boolean isLegacyTable = IcebergUtils.isLegacyTable(icebergTable.schema()); + if (isLegacyTable + && writerInitContext.tableInfo().getTableConfig().isDataLakeAutoCompaction()) { this.compactionExecutor = Executors.newSingleThreadExecutor( new ExecutorThreadFactory( 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..91b83f17ad9 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 @@ -49,10 +49,18 @@ public static TableIdentifier toIceberg(TablePath tablePath) { return TableIdentifier.of(tablePath.getDatabaseName(), tablePath.getTableName()); } + @Nullable public static PartitionKey toPartition( Table table, @Nullable String partitionName, int bucket) { PartitionSpec partitionSpec = table.spec(); Schema schema = table.schema(); + // FIP-27: an unpartitioned spec (clean bucket-unaware table) has no partition fields. + // Returning null lets the writer take Iceberg's canonical unpartitioned path + // (data/file.parquet); a non-null empty PartitionKey would route through the partitioned + // path and can produce a malformed data//file.parquet location. + if (partitionSpec.isUnpartitioned()) { + return null; + } PartitionKey partitionKey = new PartitionKey(partitionSpec, schema); int pos = 0; if (partitionName != null) { @@ -61,7 +69,19 @@ 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. 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(); + 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 +105,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..6ec9361c055 --- /dev/null +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergUtils.java @@ -0,0 +1,105 @@ +/* + * 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.fluss.lake.iceberg.IcebergSchemaUtils; + +import org.apache.iceberg.Schema; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * 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, i.e. one that carries the three + * system columns ({@code __bucket}, {@code __offset}, {@code __timestamp}) as its trailing + * columns. + * + *

Detection requires all three system columns to be present, as the last columns, in + * canonical order and with matching types. This is deliberately strict: a user table that + * merely happens to contain a {@code __timestamp} (or one/two of the system names) must be + * treated as a clean table and never be misinterpreted as legacy — otherwise onboarding such a + * table to Fluss would corrupt it by enabling system-column enrichment. + * + *

    + *
  • none of the system columns present → clean (returns {@code false}); + *
  • all three present as the trailing columns in canonical order and type → legacy; + *
  • a partial / out-of-order / type-incompatible match → rejected with {@link + * IllegalStateException}, since such a physical layout cannot be safely handled either + * way. + *
+ */ + public static boolean isLegacyTable(Schema icebergSchema) { + Map systemColumns = IcebergSchemaUtils.LEGACY_SYSTEM_COLUMNS; + List fields = icebergSchema.columns(); + + int presentCount = 0; + List presentNames = new ArrayList<>(); + for (Types.NestedField field : fields) { + if (systemColumns.containsKey(field.name())) { + presentCount++; + presentNames.add(field.name()); + } + } + + // No system columns → clean table. + if (presentCount == 0) { + return false; + } + + // Some but not all system columns → this is not a Fluss-managed layout. Treat a table that + // does not carry the complete set as clean rather than legacy, so a user table that happens + // to reuse a single system-column name is not misdetected. + if (presentCount < systemColumns.size()) { + return false; + } + + // All three names are present: they must be the trailing columns, in canonical order, with + // matching types. Anything else is an ambiguous layout we cannot safely process. + List expectedOrder = new ArrayList<>(systemColumns.keySet()); + int systemStart = fields.size() - systemColumns.size(); + for (int i = 0; i < systemColumns.size(); i++) { + Types.NestedField field = fields.get(systemStart + i); + String expectedName = expectedOrder.get(i); + Type expectedType = systemColumns.get(expectedName); + if (!field.name().equals(expectedName) || !field.type().equals(expectedType)) { + throw new IllegalStateException( + "The Iceberg table carries the reserved system column names " + + presentNames + + " but not as the trailing " + + expectedOrder + + " columns in canonical order and type. Such a layout is not a " + + "valid Fluss lake table and cannot be tiered."); + } + } + return true; + } +} 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..868a6bfd47b 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 @@ -23,6 +23,7 @@ import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.exception.TableAlreadyExistException; import org.apache.fluss.exception.TableNotExistException; +import org.apache.fluss.lake.iceberg.testutils.IcebergTestUtils; import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableChange; @@ -37,8 +38,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 +159,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 +215,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 +233,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 +308,103 @@ 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 + void testEnableLakeTableWithLegacySystemTimestampColumn() { + // FIP-27: a legacy table (created before FIP-27, carrying the three trailing system columns + // with an identity(__bucket) partition and asc(__offset) sort order) must survive + // disable-then-re-enable tiering. Re-enabling goes through createTable() on the existing + // physical table; the clean-layout expectations are rebuilt in the legacy layout before the + // compatibility checks, so this must not throw and must preserve the legacy layout. + String database = "test_db"; + String tableName = "legacy_log_table"; + TablePath tablePath = TablePath.of(database, tableName); + + TableDescriptor td = + TableDescriptor.builder().schema(FLUSS_SCHEMA).distributedBy(3).build(); + + // Create the table clean, then turn it into a legacy table (simulating a pre-FIP-27 table). + flussIcebergCatalog.createTable(tablePath, td, new TestingLakeCatalogContext()); + IcebergTestUtils.adjustToLegacyV1Table( + flussIcebergCatalog.getIcebergCatalog(), TableIdentifier.of(database, tableName)); + + // Sanity: the table is now legacy (has __timestamp, identity(__bucket), asc(__offset)). + Table legacy = + flussIcebergCatalog + .getIcebergCatalog() + .loadTable(TableIdentifier.of(database, tableName)); + assertThat(legacy.schema().findField(TIMESTAMP_COLUMN_NAME)).isNotNull(); + assertThat(legacy.spec().isUnpartitioned()).isFalse(); + assertThat(legacy.sortOrder().isUnsorted()).isFalse(); + + // Re-enabling tiering (createTable on the existing legacy table) must be accepted. + flussIcebergCatalog.createTable(tablePath, td, new TestingLakeCatalogContext()); + + // The legacy physical layout must be preserved. + Table reloaded = + flussIcebergCatalog + .getIcebergCatalog() + .loadTable(TableIdentifier.of(database, tableName)); + assertThat(reloaded.schema().findField(TIMESTAMP_COLUMN_NAME)).isNotNull(); + assertThat(reloaded.spec().isUnpartitioned()).isFalse(); + assertThat(reloaded.sortOrder().isUnsorted()).isFalse(); + } + + @Test + void testAddColumnForLegacyTableWithSystemColumns() { + // FIP-27: adding a business column to a legacy table must insert it before the trailing + // system columns, so the __bucket/__offset/__timestamp columns stay last. (Clean tables + // append the new column last; that path is covered by testAlterTableAddColumnLastNullable.) + String database = "test_db"; + String tableName = "legacy_add_col_table"; + TablePath tablePath = TablePath.of(database, tableName); + + // Create clean, then convert to legacy layout. + flussIcebergCatalog.createTable( + tablePath, + TableDescriptor.builder().schema(FLUSS_SCHEMA).distributedBy(3).build(), + new TestingLakeCatalogContext()); + IcebergTestUtils.adjustToLegacyV1Table( + flussIcebergCatalog.getIcebergCatalog(), TableIdentifier.of(database, tableName)); + + List changes = + Collections.singletonList( + TableChange.addColumn( + "new_col", + DataTypes.INT(), + "new_col comment", + TableChange.ColumnPosition.last())); + flussIcebergCatalog.alterTable( + tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA, changes)); + + Table table = + flussIcebergCatalog + .getIcebergCatalog() + .loadTable(TableIdentifier.of(database, tableName)); + List fieldNames = + table.schema().columns().stream() + .map(Types.NestedField::name) + .collect(Collectors.toList()); + // The new column goes before the trailing system columns. + assertThat(fieldNames) + .containsExactly( + "id", + "name", + "amount", + "address", + "new_col", + BUCKET_COLUMN_NAME, + OFFSET_COLUMN_NAME, + TIMESTAMP_COLUMN_NAME); } @Test @@ -384,33 +441,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 +717,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 +895,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 +966,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 @@ -1600,8 +1627,10 @@ void testCreateTableFailsWithIncompatibleSortOrder() { tablePath, td, new TestingLakeCatalogContext())) .isInstanceOf(TableAlreadyExistException.class) .hasMessageContaining("sort order is not compatible") - .hasMessageContaining("ASC(__offset)") - .hasMessageContaining("pre-create the Iceberg table without a custom sort order"); + .hasMessageContaining("pre-create the Iceberg table without a custom sort order") + // FIP-27: this is a clean table (no __offset), so the guidance must not tell the + // user to pre-create with ASC(__offset) — that only applies to legacy tables. + .hasMessageNotContaining("ASC(__offset)"); } /** 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/maintenance/IcebergRewriteITCase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java index 4b6d9de71a8..355a0adcdf7 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java @@ -32,7 +32,6 @@ import org.junit.jupiter.api.Test; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.Comparator; import java.util.Iterator; @@ -69,7 +68,11 @@ protected static void beforeAll() { } @Test - void testPkTableCompaction() throws Exception { + void testPkTableCompactionDisabledForCleanTable() throws Exception { + // FIP-27: clean-schema tables have no __bucket column, so per-bucket compaction can no + // longer scope an exclusive file set. Auto compaction is therefore disabled for clean + // tables; verify the files written here are NOT merged. Legacy-table compaction remains + // covered by IcebergRewriteTest. JobClient jobClient = buildTieringJob(execEnv); try { TablePath t1 = TablePath.of(DEFAULT_DB, "pk_table_1"); @@ -85,19 +88,17 @@ void testPkTableCompaction() throws Exception { writeIcebergTableRecords(t1, t1Bucket, 2, false, rows); flussRows.addAll(rows); - // add pos-delete - rows = Arrays.asList(row(3, "v1"), row(3, "v2")); - writeIcebergTableRecords(t1, t1Bucket, 5, false, rows); - // one UPDATE_BEFORE and one UPDATE_AFTER - checkFileStatusInIcebergTable(t1, 3, true); - flussRows.add(rows.get(1)); + rows = Collections.singletonList(row(3, "v1")); + writeIcebergTableRecords(t1, t1Bucket, 3, false, rows); + flussRows.addAll(rows); - // trigger compaction rows = Collections.singletonList(row(4, "v1")); - writeIcebergTableRecords(t1, t1Bucket, 6, false, rows); - checkFileStatusInIcebergTable(t1, 2, false); + writeIcebergTableRecords(t1, t1Bucket, 4, false, rows); flussRows.addAll(rows); + // Would have triggered compaction on a legacy table; a clean table must not compact. + checkNoCompactionInIcebergTable(t1, 4); + // Data must still be complete and correct even without compaction. checkRecords(getIcebergRecords(t1), flussRows); } finally { jobClient.cancel().get(); @@ -126,45 +127,9 @@ private void checkRecords(List actualRows, List expectedRow } @Test - void testPkTableCompactionWithConflict() throws Exception { - JobClient jobClient = buildTieringJob(execEnv); - try { - TablePath t1 = TablePath.of(DEFAULT_DB, "pk_table_2"); - long t1Id = createPkTable(t1, 1, true, pkSchema); - TableBucket t1Bucket = new TableBucket(t1Id, 0); - List flussRows = new ArrayList<>(); - - List rows = Collections.singletonList(row(1, "v1")); - flussRows.addAll(writeIcebergTableRecords(t1, t1Bucket, 1, false, rows)); - checkFileStatusInIcebergTable(t1, 1, false); - - rows = Collections.singletonList(row(2, "v1")); - flussRows.addAll(writeIcebergTableRecords(t1, t1Bucket, 2, false, rows)); - - rows = Collections.singletonList(row(3, "v1")); - flussRows.addAll(writeIcebergTableRecords(t1, t1Bucket, 3, false, rows)); - - // add pos-delete and trigger compaction - rows = Arrays.asList(row(4, "v1"), row(4, "v2")); - flussRows.add(writeIcebergTableRecords(t1, t1Bucket, 6, false, rows).get(1)); - // rewritten files should fail to commit due to conflict, add check here - checkRecords(getIcebergRecords(t1), flussRows); - // 4 data file and 1 delete file - checkFileStatusInIcebergTable(t1, 4, true); - - // previous compaction conflicts won't prevent further compaction, and check iceberg - // records - rows = Collections.singletonList(row(5, "v1")); - flussRows.addAll(writeIcebergTableRecords(t1, t1Bucket, 7, false, rows)); - checkRecords(getIcebergRecords(t1), flussRows); - checkFileStatusInIcebergTable(t1, 2, false); - } finally { - jobClient.cancel().get(); - } - } - - @Test - void testLogTableCompaction() throws Exception { + void testLogTableCompactionDisabledForCleanTable() throws Exception { + // FIP-27: a clean bucket-aware log table still has no __bucket column; auto compaction is + // disabled for clean tables, so the small files written here must not be merged. JobClient jobClient = buildTieringJob(execEnv); try { TablePath t1 = TablePath.of(DEFAULT_DB, "log_table"); @@ -184,18 +149,14 @@ void testLogTableCompaction() throws Exception { flussRows.addAll( writeIcebergTableRecords( t1, t1Bucket, ++i, true, Collections.singletonList(row(1, "v1")))); - checkFileStatusInIcebergTable(t1, 3, false); - // Write should trigger compaction now since the current data file count is greater or - // equal MIN_FILES_TO_COMPACT flussRows.addAll( writeIcebergTableRecords( t1, t1Bucket, ++i, true, Collections.singletonList(row(1, "v1")))); - // Should only have two files now, one file it for newly written, one file is for target - // compacted file - checkFileStatusInIcebergTable(t1, 2, false); - // check data in iceberg to make sure compaction won't lose data or duplicate data + // FIP-27: a clean table is not auto-compacted, so the files must not be merged. + checkNoCompactionInIcebergTable(t1, 4); + // check data in iceberg to make sure no data is lost or duplicated without compaction checkRecords(getIcebergRecords(t1), flussRows); } finally { jobClient.cancel().get(); diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRowTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRowTest.java index e5e142dfb91..b94a5ac141b 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRowTest.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRowTest.java @@ -124,6 +124,44 @@ void testGetFieldCount() { assertThat(icebergRecordAsFlussRow.getFieldCount()).isEqualTo(19); } + @Test + void testGetFieldCountCleanRecord() { + // A clean record (FIP-27) has no system columns; the business field count is the full size. + Schema cleanSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "name", Types.StringType.get()), + optional(3, "age", Types.IntegerType.get())); + Record cleanRecord = GenericRecord.create(cleanSchema); + cleanRecord.setField("id", 1L); + cleanRecord.setField("name", "John"); + cleanRecord.setField("age", 30); + + icebergRecordAsFlussRow.replaceIcebergRecord(cleanRecord); + + assertThat(icebergRecordAsFlussRow.getFieldCount()).isEqualTo(3); + } + + @Test + void testGetFieldCountLegacyProjectedRecord() { + // A projected legacy record carries only the system columns that were projected. The reader + // projects __offset and __timestamp (but not __bucket), so a single projected business + // column yields [id, __offset, __timestamp]; the business field count must be 1, not 0. + Schema projectedSchema = + new Schema( + required(1, "id", Types.LongType.get()), + required(23, "__offset", Types.LongType.get()), + required(24, "__timestamp", Types.TimestampType.withZone())); + Record projectedRecord = GenericRecord.create(projectedSchema); + projectedRecord.setField("id", 1L); + projectedRecord.setField("__offset", 100L); + projectedRecord.setField("__timestamp", OffsetDateTime.now(ZoneOffset.UTC)); + + icebergRecordAsFlussRow.replaceIcebergRecord(projectedRecord); + + assertThat(icebergRecordAsFlussRow.getFieldCount()).isEqualTo(1); + } + @Test void testIsNullAt() { record.setField("id", 1L); 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..84c75aa64d8 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 @@ -69,6 +69,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.Comparator; import java.util.HashMap; import java.util.Iterator; import java.util.List; @@ -338,7 +339,31 @@ 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 + // FIP-27: a clean table stores only user columns and carries none of the trailing + // __bucket/__offset/__timestamp system columns. + assertThat(actualRecord.struct().field(OFFSET_COLUMN_NAME)).isNull(); + } + assertThat(flussRowIterator.hasNext()).isFalse(); + } + } + + /** + * Legacy counterpart of {@link #checkDataInIcebergAppendOnlyTable}: a legacy table carries the + * trailing {@code __bucket}/{@code __offset}/{@code __timestamp} system columns, so this + * asserts the {@code __offset} column is present and advances per row (offset at idx 3 for a + * 2-user- column table). + */ + protected void checkDataInIcebergLegacyAppendOnlyTable( + TablePath tablePath, List expectedRows, long startingOffset) + throws Exception { + try (CloseableIterator records = getIcebergRows(tablePath)) { + Iterator flussRowIterator = expectedRows.iterator(); + while (records.hasNext()) { + Record actualRecord = records.next(); + InternalRow flussRow = flussRowIterator.next(); + assertThat(actualRecord.get(0)).isEqualTo(flussRow.getInt(0)); + assertThat(actualRecord.get(1)).isEqualTo(flussRow.getString(1).toString()); + assertThat(actualRecord.struct().field(OFFSET_COLUMN_NAME)).isNotNull(); assertThat(actualRecord.get(3)).isEqualTo(startingOffset++); } assertThat(flussRowIterator.hasNext()).isFalse(); @@ -363,6 +388,38 @@ protected void checkFileStatusInIcebergTable( assertThat(count).isEqualTo(expectedFileCount); } + /** + * Waits until the iceberg table has at least {@code minFileCount} data files and asserts the + * count then stays there (i.e. compaction does not reduce it). Used to verify that clean-schema + * tables (FIP-27) are intentionally NOT auto-compacted: their per-bucket ownership boundary + * (the {@code __bucket} predicate) no longer exists, so compaction is disabled for them. + */ + protected void checkNoCompactionInIcebergTable(TablePath tablePath, int minFileCount) + throws Exception { + org.apache.iceberg.Table table = icebergCatalog.loadTable(toIceberg(tablePath)); + // Wait until the writes have produced at least minFileCount files. + waitUntil( + () -> planDataFileCount(table) >= minFileCount, + Duration.ofMinutes(1), + "Expected at least " + minFileCount + " data files to be tiered."); + // Give any (erroneously scheduled) compaction a chance to run, then assert nothing merged. + Thread.sleep(2000L); + assertThat(planDataFileCount(table)).isGreaterThanOrEqualTo(minFileCount); + } + + private static int planDataFileCount(org.apache.iceberg.Table table) { + table.refresh(); + int count = 0; + try (CloseableIterable tasks = table.newScan().planFiles()) { + for (FileScanTask ignored : tasks) { + count++; + } + } catch (IOException e) { + throw new RuntimeException("Failed to plan files for table " + table.name(), e); + } + return count; + } + protected void checkDataInIcebergAppendOnlyPartitionedTable( TablePath tablePath, Map partitionSpec, @@ -377,7 +434,33 @@ 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 + // FIP-27: a clean partitioned table carries no trailing system columns. + assertThat(actualRecord.struct().field(OFFSET_COLUMN_NAME)).isNull(); + } + assertThat(flussRowIterator.hasNext()).isFalse(); + } + } + + /** + * Legacy counterpart of {@link #checkDataInIcebergAppendOnlyPartitionedTable}: asserts the + * trailing {@code __offset} column is present and advances per row (offset at idx 4 for a 3- + * user-column partitioned table). + */ + protected void checkDataInIcebergLegacyAppendOnlyPartitionedTable( + TablePath tablePath, + Map partitionSpec, + List expectedRows, + long startingOffset) + throws Exception { + try (CloseableIterator records = getIcebergRows(tablePath, partitionSpec)) { + Iterator flussRowIterator = expectedRows.iterator(); + while (records.hasNext()) { + Record actualRecord = records.next(); + InternalRow flussRow = flussRowIterator.next(); + 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()); + assertThat(actualRecord.struct().field(OFFSET_COLUMN_NAME)).isNotNull(); assertThat(actualRecord.get(4)).isEqualTo(startingOffset++); } assertThat(flussRowIterator.hasNext()).isFalse(); @@ -402,7 +485,34 @@ 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. Iterate files in a + // deterministic order (by data-file location) so the row order is stable regardless + // of manifest ordering, which FastAppend may change across tiering commits. + table.refresh(); + TableScan cleanScan = filterByPartition(table.newScan(), partitionSpec); + List cleanFiles = new ArrayList<>(); + cleanScan + .planFiles() + .iterator() + .forEachRemaining(fileScanTask -> cleanFiles.add(fileScanTask.file())); + cleanFiles.sort(Comparator.comparing(f -> f.location())); + for (DataFile file : cleanFiles) { + 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/testutils/IcebergTestUtils.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/IcebergTestUtils.java new file mode 100644 index 00000000000..5ef1fd41573 --- /dev/null +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/IcebergTestUtils.java @@ -0,0 +1,66 @@ +/* + * 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.testutils; + +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.types.Types; + +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; + +/** Utils for iceberg testing. */ +public class IcebergTestUtils { + + /** + * Adjusts a clean Iceberg table into a legacy one by appending the three trailing Fluss system + * columns ({@code __bucket}, {@code __offset}, {@code __timestamp}), then restoring the legacy + * physical layout of an {@code identity(__bucket)} partition spec and an {@code ASC(__offset)} + * sort order. This simulates a table created before FIP-27, when the lake table always carried + * these system columns, so tests can verify legacy tables remain readable and writable. + * + *

Mirrors Paimon's {@code PaimonTestUtils.adjustToLegacyV1Table}, but additionally restores + * the Iceberg-specific partition spec and sort order that a legacy table carries. + */ + public static void adjustToLegacyV1Table(Catalog icebergCatalog, TableIdentifier tableId) { + Table table = icebergCatalog.loadTable(tableId); + + // 1. Append the three trailing system columns. They must be REQUIRED to faithfully match a + // real pre-FIP-27 legacy table (created via IcebergSchemaUtils, which adds them required). + // addRequiredColumn is allowed here because the table is still empty right after creation. + table.updateSchema() + .allowIncompatibleChanges() + .addRequiredColumn(BUCKET_COLUMN_NAME, Types.IntegerType.get()) + .addRequiredColumn(OFFSET_COLUMN_NAME, Types.LongType.get()) + .addRequiredColumn(TIMESTAMP_COLUMN_NAME, Types.TimestampType.withZone()) + .commit(); + + // 2. Restore the legacy identity(__bucket) partition spec. + table.refresh(); + table.updateSpec().addField(BUCKET_COLUMN_NAME).commit(); + + // 3. Restore the legacy ASC(__offset) sort order. + table.refresh(); + Schema schema = table.schema(); + table.replaceSortOrder().asc(schema.findField(OFFSET_COLUMN_NAME).name()).commit(); + table.refresh(); + } +} 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..c8af3a490eb 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 @@ -18,6 +18,7 @@ package org.apache.fluss.lake.iceberg.tiering; import org.apache.fluss.lake.iceberg.testutils.FlinkIcebergTieringTestBase; +import org.apache.fluss.lake.iceberg.testutils.IcebergTestUtils; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableChange; @@ -41,6 +42,8 @@ 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.metadata.TableDescriptor.OFFSET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; import static org.apache.fluss.testutils.DataTestUtils.row; import static org.assertj.core.api.Assertions.assertThat; @@ -125,7 +128,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. @@ -151,6 +155,61 @@ void testSchemaEvolutionLogTable() throws Exception { } } + @Test + void testSchemaEvolutionLegacyLogTable() throws Exception { + // FIP-27: a legacy table keeps the three trailing system columns, so a newly added business + // column must be inserted BEFORE the first system column (the moveBefore branch), not + // appended last. This complements testSchemaEvolutionLogTable, which covers the clean + // append-last path. + TablePath tablePath = TablePath.of(DEFAULT_DB, "schemaEvoLegacyLogTable"); + long tableId = createLogTable(tablePath, 1, false, INITIAL_LOG_SCHEMA); + TableBucket bucket = new TableBucket(tableId, 0); + + List initialRows = Arrays.asList(row(1, "v1"), row(2, "v2"), row(3, "v3")); + writeRows(tablePath, initialRows, true); + + JobClient jobClient = buildTieringJob(execEnv); + try { + assertReplicaStatus(bucket, 3); + // Initial tiered data is still clean at this point. + checkDataInIcebergAppendOnlyTable(tablePath, initialRows, 0); + + // Convert the tiered table into a legacy table (system columns + identity(__bucket) + + // asc(__offset)), simulating a table created before FIP-27. + IcebergTestUtils.adjustToLegacyV1Table(icebergCatalog, toIceberg(tablePath)); + + // ALTER TABLE ADD COLUMN via Admin API. + admin.alterTable( + tablePath, + Collections.singletonList( + TableChange.addColumn( + "f_new", + DataTypes.STRING(), + null, + TableChange.ColumnPosition.last())), + false) + .get(); + + org.apache.iceberg.Table icebergTable = icebergCatalog.loadTable(toIceberg(tablePath)); + icebergTable.refresh(); + List fieldNames = + icebergTable.schema().columns().stream() + .map(Types.NestedField::name) + .collect(Collectors.toList()); + // The new column must sit before the three trailing system columns. + assertThat(fieldNames) + .containsExactly( + "f_int", + "f_str", + "f_new", + BUCKET_COLUMN_NAME, + OFFSET_COLUMN_NAME, + TIMESTAMP_COLUMN_NAME); + } finally { + jobClient.cancel().get(); + } + } + @Test void testSchemaEvolutionPkTable() throws Exception { TablePath tablePath = TablePath.of(DEFAULT_DB, "schemaEvoPkTable"); @@ -197,7 +256,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 +332,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 +394,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 +456,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 = diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringITCase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringITCase.java index 0e1c80eea71..4d52ab187cb 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringITCase.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringITCase.java @@ -20,6 +20,7 @@ import org.apache.fluss.config.AutoPartitionTimeUnit; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.lake.iceberg.testutils.FlinkIcebergTieringTestBase; +import org.apache.fluss.lake.iceberg.testutils.IcebergTestUtils; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableDescriptor; @@ -54,6 +55,7 @@ import java.util.List; import java.util.Map; +import static org.apache.fluss.lake.iceberg.utils.IcebergConversions.toIceberg; import static org.apache.fluss.testutils.DataTestUtils.row; import static org.assertj.core.api.Assertions.assertThat; @@ -268,6 +270,35 @@ void testTiering() throws Exception { } } + @Test + void testTieringForLegacyTable() throws Exception { + // FIP-27: verify tiering still works for a legacy table that carries the three trailing + // system columns (__bucket, __offset, __timestamp). This covers the legacy writer path, + // whereas testTiering covers the clean writer path. + TablePath t = TablePath.of(DEFAULT_DB, "legacyLogTable"); + long tId = createLogTable(t, 1, false, logSchema); + + // Turn the freshly created clean table into a legacy table before writing/tiering. + IcebergTestUtils.adjustToLegacyV1Table(icebergCatalog, toIceberg(t)); + + 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); + // The legacy table keeps the system columns, and the tiered __offset must be correct. + checkDataInIcebergLegacyAppendOnlyTable(t, flussRows, 0); + } finally { + jobClient.cancel().get(); + } + } + private void checkDataInIcebergPrimaryKeyTable( TablePath tablePath, List expectedRows) throws Exception { Iterator actualIterator = getIcebergRecords(tablePath).iterator(); diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/utils/IcebergConversionsTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/utils/IcebergConversionsTest.java index 512024807cd..5915e042585 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/utils/IcebergConversionsTest.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/utils/IcebergConversionsTest.java @@ -60,6 +60,30 @@ void testToPartition(@TempDir File tempWarehouseDir) { assertThat(partitionKey.toPath()).isEqualTo("country=china/region=region1/__bucket=2"); } + @Test + void testToPartitionReturnsNullForUnpartitionedTable(@TempDir File tempWarehouseDir) { + Catalog catalog = getIcebergCatalog(tempWarehouseDir); + + // FIP-27: a clean bucket-unaware table is unpartitioned. toPartition must return null so + // the + // writer uses Iceberg's canonical unpartitioned path instead of a non-null empty key. + TablePath tablePath = TablePath.of("default", "fluss_clean_unpartitioned_table"); + Schema schema = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "name", Types.StringType.get())); + TableIdentifier tableIdentifier = toIceberg(tablePath); + try { + ((SupportsNamespaces) catalog).createNamespace(tableIdentifier.namespace()); + } catch (AlreadyExistsException ignore) { + // ignore + } + catalog.createTable(tableIdentifier, schema, PartitionSpec.unpartitioned()); + Table table = catalog.loadTable(tableIdentifier); + + assertThat(IcebergConversions.toPartition(table, null, 1)).isNull(); + } + private Catalog getIcebergCatalog(File tempWarehouseDir) { Configuration configuration = new Configuration(); configuration.setString("warehouse", tempWarehouseDir.toURI().toString()); diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/utils/IcebergUtilsTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/utils/IcebergUtilsTest.java new file mode 100644 index 00000000000..c9eb5ad0c7c --- /dev/null +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/utils/IcebergUtilsTest.java @@ -0,0 +1,115 @@ +/* + * 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 org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; + +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.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** UT for {@link IcebergUtils#isLegacyTable(Schema)}. */ +class IcebergUtilsTest { + + private static Types.NestedField bucketField(int id) { + return required(id, BUCKET_COLUMN_NAME, Types.IntegerType.get()); + } + + private static Types.NestedField offsetField(int id) { + return required(id, OFFSET_COLUMN_NAME, Types.LongType.get()); + } + + private static Types.NestedField timestampField(int id) { + return required(id, TIMESTAMP_COLUMN_NAME, Types.TimestampType.withZone()); + } + + @Test + void testCleanTableHasNoSystemColumns() { + Schema clean = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "name", Types.StringType.get())); + assertThat(IcebergUtils.isLegacyTable(clean)).isFalse(); + } + + @Test + void testLegacyTableWithAllThreeTrailingSystemColumns() { + Schema legacy = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "name", Types.StringType.get()), + bucketField(3), + offsetField(4), + timestampField(5)); + assertThat(IcebergUtils.isLegacyTable(legacy)).isTrue(); + } + + @Test + void testUserTableWithOnlyTimestampNameIsClean() { + // A pre-existing user table that merely reuses the __timestamp name (but is not a full + // Fluss + // legacy layout) must be treated as clean, not misdetected as legacy. + Schema oneMatch = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, TIMESTAMP_COLUMN_NAME, Types.TimestampType.withZone())); + assertThat(IcebergUtils.isLegacyTable(oneMatch)).isFalse(); + } + + @Test + void testUserTableWithTwoSystemColumnNamesIsClean() { + Schema twoMatch = + new Schema( + required(1, "id", Types.LongType.get()), offsetField(2), timestampField(3)); + assertThat(IcebergUtils.isLegacyTable(twoMatch)).isFalse(); + } + + @Test + void testAllThreePresentButNotTrailingIsRejected() { + // System columns present but a user column follows them → ambiguous, must be rejected. + Schema notTrailing = + new Schema( + required(1, "id", Types.LongType.get()), + bucketField(2), + offsetField(3), + timestampField(4), + optional(5, "extra", Types.StringType.get())); + assertThatThrownBy(() -> IcebergUtils.isLegacyTable(notTrailing)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("trailing"); + } + + @Test + void testAllThreePresentButOutOfOrderIsRejected() { + Schema outOfOrder = + new Schema( + required(1, "id", Types.LongType.get()), + offsetField(2), + bucketField(3), + timestampField(4)); + assertThatThrownBy(() -> IcebergUtils.isLegacyTable(outOfOrder)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("canonical order"); + } +}