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 7db0a1957a..70886c9c83 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; @@ -107,7 +109,9 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co PartitionSpec partitionSpec = IcebergPartitionSpecUtils.createPartitionSpec(tableDescriptor, icebergSchema); - SortOrder sortOrder = createSortOrder(icebergSchema); + // FIP-27: newly created tables are clean and carry no __offset column, so they need no + // sort order. (Legacy tables keep ASC(__offset), but those are never created here.) + SortOrder sortOrder = SortOrder.unsorted(); Map expectedProperties = buildTableProperties(tableDescriptor, isPkTable); tableBuilder.withProperties(expectedProperties); tableBuilder.withPartitionSpec(partitionSpec); @@ -116,6 +120,8 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co createTable( tablePath, tableBuilder, + tableDescriptor, + isPkTable, icebergSchema, partitionSpec, sortOrder, @@ -127,6 +133,8 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co createTable( tablePath, tableBuilder, + tableDescriptor, + isPkTable, icebergSchema, partitionSpec, sortOrder, @@ -223,7 +231,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 +242,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 +262,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 +294,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 +311,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 +325,23 @@ 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 = + SortOrder.builderFor(newIcebergSchema).asc(OFFSET_COLUMN_NAME).build(); + } + if (!isIcebergSchemaCompatibleWithSchema(existingSchema, newIcebergSchema)) { throw new TableAlreadyExistException( String.format( @@ -323,17 +368,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); } @@ -507,13 +558,6 @@ private void createDatabase(String databaseName) { } } - private SortOrder createSortOrder(Schema icebergSchema) { - // Sort by __offset system column for deterministic ordering - SortOrder.Builder builder = SortOrder.builderFor(icebergSchema); - builder.asc(OFFSET_COLUMN_NAME); - return builder.build(); - } - private Map buildTableProperties( TableDescriptor tableDescriptor, boolean isPkTable) { Map icebergProperties = new HashMap<>(); 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 7559625f91..21ffd32173 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 570c056032..96c1e44b82 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 a618df40e3..cc884585df 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 f654fb5d04..cd0f87bbd8 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; @@ -37,13 +38,11 @@ import javax.annotation.Nullable; import java.io.IOException; -import java.time.OffsetDateTime; import java.util.ArrayList; import java.util.List; import java.util.stream.IntStream; -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.lake.iceberg.IcebergSchemaUtils.LEGACY_SYSTEM_COLUMNS; /** * Iceberg record reader. The filter is applied during the plan phase of IcebergSplitPlanner, so the @@ -53,6 +52,10 @@ * 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; @@ -74,14 +77,14 @@ public CloseableIterator read() throws IOException { private TableScan applyProject(TableScan tableScan, int[][] projects) { Types.StructType structType = tableScan.schema().asStruct(); - List cols = new ArrayList<>(projects.length + 2); - + List cols = new ArrayList<>(projects.length); + // The projected column ids reference the user (business) columns; the log offset / + // timestamp + // are not read from the lake table (a clean table has none, and for a legacy table we no + // longer read them), so no system column needs to be projected. for (int[] project : projects) { cols.add(structType.fields().get(project[0])); } - - cols.add(structType.field(OFFSET_COLUMN_NAME)); - cols.add(structType.field(TIMESTAMP_COLUMN_NAME)); return tableScan.project(new Schema(cols)); } @@ -93,17 +96,18 @@ public static class IcebergRecordAsFlussRecordIterator implements CloseableItera private final ProjectedRow projectedRow; private final IcebergRecordAsFlussRow icebergRecordAsFlussRow; - private final int logOffsetColIndex; - private final int timestampColIndex; - public IcebergRecordAsFlussRecordIterator( CloseableIterable icebergRecordIterator, Types.StructType struct) { 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); + // A legacy table read without projection still exposes its three trailing system + // columns; trim them so only the business columns are emitted. A clean table (or any + // projected read) has no system columns to trim. + int businessFieldCount = + IcebergUtils.isLegacyTable(new Schema(struct.fields())) + ? struct.fields().size() - LEGACY_SYSTEM_COLUMNS.size() + : struct.fields().size(); + projectedRow = ProjectedRow.from(IntStream.range(0, businessFieldCount).toArray()); icebergRecordAsFlussRow = new IcebergRecordAsFlussRow(); } @@ -124,16 +128,13 @@ 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(); - + // The lake table does not carry a per-record log offset / timestamp (a clean table has + // no system columns, and for a legacy table we no longer read them), so a sentinel -1 + // is + // emitted, consistent with the Paimon reader and the LakeRecordRecordEmitter contract. return new GenericRecord( - offset, - timestamp, + NO_SYSTEM_COLUMN_VALUE, + NO_SYSTEM_COLUMN_VALUE, ChangeType.INSERT, projectedRow.replaceRow( icebergRecordAsFlussRow.replaceIcebergRecord(icebergRecord))); 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 a397a5af00..34c0217569 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 @@ -21,12 +21,12 @@ import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.config.Configuration; import org.apache.fluss.lake.iceberg.utils.IcebergCatalogUtils; +import org.apache.fluss.lake.iceberg.utils.IcebergPartitionSpecUtils; import org.apache.fluss.lake.source.Planner; import org.apache.fluss.metadata.TablePath; import org.apache.iceberg.FileScanTask; import org.apache.iceberg.PartitionField; -import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Table; import org.apache.iceberg.TableScan; import org.apache.iceberg.catalog.Catalog; @@ -49,7 +49,6 @@ import java.util.stream.IntStream; import static org.apache.fluss.lake.iceberg.utils.IcebergConversions.toIceberg; -import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; /** Iceberg split planner. */ public class IcebergSplitPlanner implements Planner { @@ -143,47 +142,44 @@ private Set union(Set left, Set right) { } 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); - - 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 + List partitionFields = table.spec().fields(); + if (partitionFields.isEmpty()) { return task -> -1; - } else { - int bucketFieldIndex = partitionFields.size() - 1; - return task -> task.file().partition().get(bucketFieldIndex, Integer.class); } + + // Only a bucket(bucketKey) transform carries a physical bucket value in the partition + // tuple. + // A legacy identity(__bucket) partition or a bucket-unaware user partition has none. + int bucketFieldIndex = partitionFields.size() - 1; + if (!IcebergPartitionSpecUtils.isBucketTransform(partitionFields.get(bucketFieldIndex))) { + return task -> -1; + } + return task -> task.file().partition().get(bucketFieldIndex, Integer.class); } private Function> createPartitionExtractor(Table table) { - PartitionSpec partitionSpec = table.spec(); - List partitionFields = partitionSpec.fields(); + List partitionFields = table.spec().fields(); + if (partitionFields.isEmpty()) { + return task -> Collections.emptyList(); + } + + // The trailing field is the Fluss bucket (legacy identity(__bucket) or a bucket(bucketKey) + // transform); everything before it is the Fluss partition columns. + PartitionField lastField = partitionFields.get(partitionFields.size() - 1); + int partitionColCount = + IcebergPartitionSpecUtils.isFlussBucketField(table.schema(), lastField) + ? partitionFields.size() - 1 + : partitionFields.size(); - // 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 (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 60f400575d..fe1da1144b 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 93f60ed480..646bdddc45 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 @@ -17,11 +17,13 @@ package org.apache.fluss.lake.iceberg.tiering; +import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.lake.iceberg.maintenance.IcebergRewriteDataFiles; import org.apache.fluss.lake.iceberg.maintenance.RewriteDataFileResult; 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,13 +75,32 @@ 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 autoCompaction = + writerInitContext.tableInfo().getTableConfig().isDataLakeAutoCompaction(); + boolean isLegacyTable = IcebergUtils.isLegacyTable(icebergTable.schema()); + if (autoCompaction && isLegacyTable) { this.compactionExecutor = Executors.newSingleThreadExecutor( new ExecutorThreadFactory( "iceberg-compact-" + writerInitContext.tableBucket())); scheduleCompaction(writerInitContext); } else { + if (autoCompaction) { + // A clean-schema table (FIP-27) cannot be safely compacted by the tiering service + // yet, so the enabled auto-compaction is a no-op. Warn instead of silently ignoring + // it, so operators do not assume small-file maintenance is happening. + LOG.warn( + "Table {} has '{}' enabled, but tiering-managed compaction is not supported " + + "for clean-schema Iceberg tables and will not run. Please use " + + "external Iceberg compaction for this table.", + writerInitContext.tablePath(), + ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key()); + } this.compactionExecutor = null; } } 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 732ea8de03..8f2bbb694a 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 238ef5be98..275f72a098 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 111df385ff..1279cf6a11 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,14 @@ public static PartitionKey toPartition( partitionKey.set(pos++, partition); } } - partitionKey.set(pos, bucket); + // Set the bucket value only when the trailing partition field is the Fluss bucket field + // (legacy identity(__bucket) or a bucket(bucketKey) transform). Bucket-unaware partitioned + // tables (a trailing identity partition column) have no such field. + List fields = partitionSpec.fields(); + PartitionField lastField = fields.get(fields.size() - 1); + if (IcebergPartitionSpecUtils.isFlussBucketField(schema, lastField)) { + partitionKey.set(pos, bucket); + } return partitionKey; } @@ -85,7 +100,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 f9c7c0e0c6..a2e4c2a990 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 @@ -21,6 +21,7 @@ import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.metadata.TableDescriptor; +import org.apache.iceberg.PartitionField; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; @@ -82,12 +83,35 @@ 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); } return builder.build(); } + + /** Returns whether the partition field uses an Iceberg bucket transform. */ + public static boolean isBucketTransform(PartitionField partitionField) { + return partitionField.transform().toString().startsWith("bucket["); + } + + /** + * Returns whether the partition field is the trailing physical bucket field maintained by + * Fluss. + * + *

For bucket-aware tables, this is {@code bucket(bucketKey)}. For legacy bucket-unaware + * tables, this is {@code identity(__bucket)}. + */ + public static boolean isFlussBucketField(Schema icebergSchema, PartitionField partitionField) { + return isBucketTransform(partitionField) + || (partitionField.transform().isIdentity() + && BUCKET_COLUMN_NAME.equals( + icebergSchema.findColumnName(partitionField.sourceId()))); + } } 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 0000000000..7e7ddb4ffc --- /dev/null +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergUtils.java @@ -0,0 +1,52 @@ +/* + * 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; + +/** + * 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}). + * + *

Detection requires all three system columns to be present. This guards against + * misdetecting a user table that merely reuses one of the system-column names (e.g. a table + * with only a {@code __timestamp} column that is being onboarded to Fluss): such a table has + * fewer than three system columns and is therefore treated as clean. + */ + public static boolean isLegacyTable(Schema icebergSchema) { + for (String systemColumn : IcebergSchemaUtils.LEGACY_SYSTEM_COLUMNS.keySet()) { + if (icebergSchema.findField(systemColumn) == null) { + return false; + } + } + 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 30f2b82f1c..868a6bfd47 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 51c25e813a..aad6989910 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 deleted file mode 100644 index 4b6d9de71a..0000000000 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java +++ /dev/null @@ -1,216 +0,0 @@ -/* - * 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.maintenance; - -import org.apache.fluss.lake.iceberg.testutils.FlinkIcebergTieringTestBase; -import org.apache.fluss.metadata.Schema; -import org.apache.fluss.metadata.TableBucket; -import org.apache.fluss.metadata.TablePath; -import org.apache.fluss.row.InternalRow; -import org.apache.fluss.types.DataTypes; - -import org.apache.flink.core.execution.JobClient; -import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; -import org.apache.iceberg.data.Record; -import org.junit.jupiter.api.BeforeAll; -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; -import java.util.List; - -import static org.apache.fluss.testutils.DataTestUtils.row; -import static org.assertj.core.api.Assertions.assertThat; - -/** Integration test for Iceberg compaction. */ -class IcebergRewriteITCase extends FlinkIcebergTieringTestBase { - protected static final String DEFAULT_DB = "fluss"; - - private static StreamExecutionEnvironment execEnv; - - private static final Schema pkSchema = - Schema.newBuilder() - .column("f_int", DataTypes.INT()) - .column("f_string", DataTypes.STRING()) - .primaryKey("f_int") - .build(); - - private static final Schema logSchema = - Schema.newBuilder() - .column("f_int", DataTypes.INT()) - .column("f_string", DataTypes.STRING()) - .build(); - - @BeforeAll - protected static void beforeAll() { - FlinkIcebergTieringTestBase.beforeAll(); - execEnv = StreamExecutionEnvironment.getExecutionEnvironment(); - execEnv.setParallelism(2); - execEnv.enableCheckpointing(1000); - } - - @Test - void testPkTableCompaction() throws Exception { - JobClient jobClient = buildTieringJob(execEnv); - try { - TablePath t1 = TablePath.of(DEFAULT_DB, "pk_table_1"); - long t1Id = createPkTable(t1, 1, true, pkSchema); - TableBucket t1Bucket = new TableBucket(t1Id, 0); - List flussRows = new ArrayList<>(); - - List rows = Collections.singletonList(row(1, "v1")); - writeIcebergTableRecords(t1, t1Bucket, 1, false, rows); - flussRows.addAll(rows); - - rows = Collections.singletonList(row(2, "v1")); - 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)); - - // trigger compaction - rows = Collections.singletonList(row(4, "v1")); - writeIcebergTableRecords(t1, t1Bucket, 6, false, rows); - checkFileStatusInIcebergTable(t1, 2, false); - flussRows.addAll(rows); - - checkRecords(getIcebergRecords(t1), flussRows); - } finally { - jobClient.cancel().get(); - } - } - - private void checkRecords(List actualRows, List expectedRows) { - // check records size - assertThat(actualRows.size()).isEqualTo(expectedRows.size()); - - // check records content - Iterator actualIterator = - actualRows.stream() - .sorted(Comparator.comparingInt((Record r) -> (int) r.get(0))) - .iterator(); - Iterator expectedIterator = - expectedRows.stream().sorted(Comparator.comparingInt(r -> r.getInt(0))).iterator(); - while (actualIterator.hasNext() && expectedIterator.hasNext()) { - Record record = actualIterator.next(); - InternalRow row = expectedIterator.next(); - assertThat(record.get(0)).isEqualTo(row.getInt(0)); - assertThat(record.get(1)).isEqualTo(row.getString(1).toString()); - } - assertThat(actualIterator.hasNext()).isFalse(); - assertThat(expectedIterator.hasNext()).isFalse(); - } - - @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 { - JobClient jobClient = buildTieringJob(execEnv); - try { - TablePath t1 = TablePath.of(DEFAULT_DB, "log_table"); - long t1Id = createLogTable(t1, 1, true, logSchema); - TableBucket t1Bucket = new TableBucket(t1Id, 0); - - int i = 0; - List flussRows = new ArrayList<>(); - flussRows.addAll( - writeIcebergTableRecords( - t1, t1Bucket, ++i, true, Collections.singletonList(row(1, "v1")))); - - flussRows.addAll( - writeIcebergTableRecords( - t1, t1Bucket, ++i, true, Collections.singletonList(row(1, "v1")))); - - 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 - checkRecords(getIcebergRecords(t1), flussRows); - } finally { - jobClient.cancel().get(); - } - } - - private List writeIcebergTableRecords( - TablePath tablePath, - TableBucket tableBucket, - long expectedLogEndOffset, - boolean append, - List rows) - throws Exception { - writeRows(tablePath, rows, append); - assertReplicaStatus(tableBucket, expectedLogEndOffset); - return rows; - } -} 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 e5e142dfb9..b94a5ac141 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 5bd021b9ad..8c8206181d 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 @@ -43,18 +43,14 @@ import org.apache.flink.api.common.RuntimeExecutionMode; import org.apache.flink.core.execution.JobClient; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; -import org.apache.iceberg.DataFile; import org.apache.iceberg.FileScanTask; import org.apache.iceberg.Snapshot; -import org.apache.iceberg.TableScan; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.data.IcebergGenerics; import org.apache.iceberg.data.Record; -import org.apache.iceberg.data.parquet.GenericParquetReaders; import org.apache.iceberg.hadoop.HadoopCatalog; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.CloseableIterator; -import org.apache.iceberg.parquet.Parquet; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; @@ -62,25 +58,19 @@ import java.io.Closeable; import java.io.IOException; -import java.nio.ByteBuffer; -import java.nio.ByteOrder; import java.nio.file.Files; import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; -import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.SortedSet; -import java.util.TreeSet; import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL; import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY; import static org.apache.fluss.lake.iceberg.utils.IcebergConversions.toIceberg; -import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; import static org.apache.fluss.testutils.DataTestUtils.row; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil; @@ -329,20 +319,22 @@ protected List getIcebergRecords(TablePath tablePath) throws IOException } protected void checkDataInIcebergAppendOnlyTable( - TablePath tablePath, List expectedRows, long startingOffset) - throws Exception { + TablePath tablePath, List expectedRows) throws Exception { + List> actualRows = new ArrayList<>(); 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()); - // the idx 2 is __bucket, so use 3 - assertThat(actualRecord.get(3)).isEqualTo(startingOffset++); + actualRows.add(Arrays.asList(actualRecord.get(0), actualRecord.get(1))); } - assertThat(flussRowIterator.hasNext()).isFalse(); } + + List> expectedRecords = new ArrayList<>(); + for (InternalRow expectedRow : expectedRows) { + expectedRecords.add( + Arrays.asList( + expectedRow.getInt(0), expectedRow.getString(1).toString())); + } + assertThat(actualRows).containsExactlyInAnyOrderElementsOf(expectedRecords); } protected void checkFileStatusInIcebergTable( @@ -364,24 +356,27 @@ protected void checkFileStatusInIcebergTable( } protected void checkDataInIcebergAppendOnlyPartitionedTable( - TablePath tablePath, - Map partitionSpec, - List expectedRows, - long startingOffset) + TablePath tablePath, Map partitionSpec, List expectedRows) throws Exception { + List> actualRows = new ArrayList<>(); 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()); - // the idx 3 is __bucket, so use 4 - assertThat(actualRecord.get(4)).isEqualTo(startingOffset++); + actualRows.add( + Arrays.asList( + actualRecord.get(0), actualRecord.get(1), actualRecord.get(2))); } - assertThat(flussRowIterator.hasNext()).isFalse(); } + + List> expectedRecords = new ArrayList<>(); + for (InternalRow expectedRow : expectedRows) { + expectedRecords.add( + Arrays.asList( + expectedRow.getInt(0), + expectedRow.getString(1).toString(), + expectedRow.getString(2).toString())); + } + assertThat(actualRows).containsExactlyInAnyOrderElementsOf(expectedRecords); } private CloseableIterator getIcebergRows(TablePath tablePath) { @@ -392,59 +387,9 @@ private CloseableIterator getIcebergRows(TablePath tablePath) { private CloseableIterator getIcebergRows( TablePath tablePath, Map partitionSpec) { org.apache.iceberg.Table table = icebergCatalog.loadTable(toIceberg(tablePath)); - // is primary key, we don't care about records order, - // use iceberg read api directly - if (!table.schema().identifierFieldIds().isEmpty()) { - IcebergGenerics.ScanBuilder scanBuilder = - filterByPartition(IcebergGenerics.read(table), partitionSpec); - return scanBuilder.build().iterator(); - } else { - // 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(); - SortedSet files = - new TreeSet<>( - (f1, f2) -> { - ByteBuffer buffer1 = - (ByteBuffer) - f1.lowerBounds() - .get(fieldId) - .order(ByteOrder.LITTLE_ENDIAN) - .rewind(); - long offset1 = buffer1.getLong(); - ByteBuffer buffer2 = - (ByteBuffer) - f2.lowerBounds() - .get(fieldId) - .order(ByteOrder.LITTLE_ENDIAN) - .rewind(); - long offset2 = buffer2.getLong(); - return Long.compare(offset1, offset2); - }); - - table.refresh(); - TableScan tableScan = filterByPartition(table.newScan(), partitionSpec); - tableScan - .includeColumnStats() - .planFiles() - .iterator() - .forEachRemaining(fileScanTask -> files.add(fileScanTask.file())); - - for (DataFile file : files) { - 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()); - } + IcebergGenerics.ScanBuilder scanBuilder = + filterByPartition(IcebergGenerics.read(table), partitionSpec); + return scanBuilder.build().iterator(); } private IcebergGenerics.ScanBuilder filterByPartition( @@ -457,15 +402,6 @@ private IcebergGenerics.ScanBuilder filterByPartition( return scanBuilder; } - private TableScan filterByPartition(TableScan tableScan, Map partitionSpec) { - for (Map.Entry partitionKeyAndValue : partitionSpec.entrySet()) { - String partitionCol = partitionKeyAndValue.getKey(); - String partitionValue = partitionKeyAndValue.getValue(); - tableScan = tableScan.filter(equal(partitionCol, partitionValue)); - } - return tableScan; - } - protected void checkFlussOffsetsInSnapshot( TablePath tablePath, Map expectedOffsets) throws Exception { org.apache.iceberg.Table table = icebergCatalog.loadTable(toIceberg(tablePath)); 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 0000000000..0c5dadf284 --- /dev/null +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/IcebergTestUtils.java @@ -0,0 +1,87 @@ +/* + * 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.fluss.lake.iceberg.utils.IcebergPartitionSpecUtils; + +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}) and restoring the + * Iceberg-specific legacy physical layout ({@code ASC(__offset)} sort order, plus an {@code + * identity(__bucket)} partition for bucket-unaware tables). 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. + * + *

The partition spec is preserved in a spec-aware way, matching how a real pre-FIP-27 table + * looked: + * + *

    + *
  • a bucket-aware table keeps its {@code bucket(bucketKey)} transform on the user key; no + * {@code identity(__bucket)} is added (adding it would create a second, invalid bucket + * field that {@code IcebergPartitionSpecValidator} rejects); + *
  • a bucket-unaware table gains an {@code identity(__bucket)} partition, which is how such + * tables physically carried the bucket before FIP-27. + *
+ * + *

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); + + // A bucket-aware table already partitions by bucket(bucketKey); a bucket-unaware table has + // no bucket transform and needs identity(__bucket) added below. + boolean bucketAware = + table.spec().fields().stream() + .anyMatch(IcebergPartitionSpecUtils::isBucketTransform); + + // 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. For a bucket-unaware table, restore the legacy identity(__bucket) partition. A + // bucket-aware table keeps its existing bucket(bucketKey) transform untouched. + if (!bucketAware) { + table.refresh(); + table.updateSpec().addField(BUCKET_COLUMN_NAME).commit(); + } + + // 3. Restore the legacy ASC(__offset) sort order. + table.refresh(); + table.replaceSortOrder().asc(OFFSET_COLUMN_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 d4177a305f..bafcf7526a 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; @@ -94,7 +97,7 @@ void testSchemaEvolutionLogTable() throws Exception { try { // Wait until initial data is tiered assertReplicaStatus(bucket, 3); - checkDataInIcebergAppendOnlyTable(tablePath, initialRows, 0); + checkDataInIcebergAppendOnlyTable(tablePath, initialRows); // Verify initial Iceberg schema has no extra columns org.apache.iceberg.Table icebergTable = icebergCatalog.loadTable(toIceberg(tablePath)); @@ -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); + + // 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 0e1c80eea7..b83b764c71 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,34 @@ 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); + checkDataInIcebergAppendOnlyTable(t, flussRows); + } finally { + jobClient.cancel().get(); + } + } + private void checkDataInIcebergPrimaryKeyTable( TablePath tablePath, List expectedRows) throws Exception { Iterator actualIterator = getIcebergRecords(tablePath).iterator(); @@ -353,7 +383,7 @@ private void testLogTableTiering() throws Exception { assertReplicaStatus(t2Bucket, 30); // check data in iceberg - checkDataInIcebergAppendOnlyTable(t2, flussRows, 0); + checkDataInIcebergAppendOnlyTable(t2, flussRows); } private void testPartitionedTableTiering() throws Exception { @@ -385,8 +415,7 @@ private void testPartitionedTableTiering() throws Exception { checkDataInIcebergAppendOnlyPartitionedTable( partitionedTablePath, Collections.singletonMap(partitionCol, partitionName), - writtenRowsByPartition.get(partitionName), - 0); + writtenRowsByPartition.get(partitionName)); } checkFlussOffsetsInSnapshot(partitionedTablePath, expectedOffsets); 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 512024807c..5915e04258 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 0000000000..b687a02c49 --- /dev/null +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/utils/IcebergUtilsTest.java @@ -0,0 +1,86 @@ +/* + * 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; + +/** 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(); + } +} diff --git a/website/docs/streaming-lakehouse/datalake-formats/iceberg.md b/website/docs/streaming-lakehouse/datalake-formats/iceberg.md index 5a9cccb703..55a50e058e 100644 --- a/website/docs/streaming-lakehouse/datalake-formats/iceberg.md +++ b/website/docs/streaming-lakehouse/datalake-formats/iceberg.md @@ -558,6 +558,10 @@ When integrating with Iceberg, Fluss automatically converts between Fluss data t The table option `table.datalake.auto-compaction` (disabled by default) provides per-table control over automatic compaction. When enabled for a specific table, compaction is automatically triggered during write operations to that table by the tiering service. +:::note +Tiering-managed auto compaction is currently only supported for legacy Iceberg lake tables (tables that still carry the `__bucket`/`__offset`/`__timestamp` system columns). Newly created ("clean") Iceberg lake tables contain only user columns and have no `__bucket` column to scope per-bucket compaction, so `table.datalake.auto-compaction` is a no-op for them (a warning is logged by the tiering service). Use external Iceberg compaction to maintain small files for such tables. +::: + #### Configuration ```sql title="Flink SQL"