You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/08/24 01:27:34 UTC
(fluss) branch main updated: [lake/iceberg] Support clean and legacy Iceberg lake table schemas (#4019)
This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new ce854ae9a [lake/iceberg] Support clean and legacy Iceberg lake table schemas (#4019)
ce854ae9a is described below
commit ce854ae9a1ee6a5559411d40aeec57a371cf668a
Author: fhan <aa...@gmail.com>
AuthorDate: Mon Aug 24 09:27:29 2026 +0800
[lake/iceberg] Support clean and legacy Iceberg lake table schemas (#4019)
---
.../fluss/lake/iceberg/IcebergLakeCatalog.java | 74 +++++--
.../fluss/lake/iceberg/IcebergSchemaUtils.java | 46 ++++-
.../maintenance/IcebergRewriteDataFiles.java | 10 +-
.../iceberg/source/IcebergRecordAsFlussRow.java | 29 ++-
.../lake/iceberg/source/IcebergRecordReader.java | 49 ++---
.../lake/iceberg/source/IcebergSplitPlanner.java | 68 +++----
.../tiering/FlussRecordAsIcebergRecord.java | 47 +++--
.../lake/iceberg/tiering/IcebergLakeWriter.java | 23 ++-
.../tiering/IcebergPartitionSpecValidator.java | 8 +-
.../fluss/lake/iceberg/tiering/RecordWriter.java | 6 +-
.../lake/iceberg/utils/IcebergConversions.java | 23 ++-
.../iceberg/utils/IcebergPartitionSpecUtils.java | 30 ++-
.../fluss/lake/iceberg/utils/IcebergUtils.java | 52 +++++
.../fluss/lake/iceberg/IcebergLakeCatalogTest.java | 197 +++++++++++--------
.../lake/iceberg/flink/FlinkCatalogLakeTest.java | 3 +-
.../iceberg/maintenance/IcebergRewriteITCase.java | 216 ---------------------
.../source/IcebergRecordAsFlussRowTest.java | 38 ++++
.../testutils/FlinkIcebergTieringTestBase.java | 122 +++---------
.../lake/iceberg/testutils/IcebergTestUtils.java | 87 +++++++++
.../tiering/IcebergSchemaEvolutionITCase.java | 75 ++++++-
.../lake/iceberg/tiering/IcebergTieringITCase.java | 35 +++-
.../lake/iceberg/utils/IcebergConversionsTest.java | 24 +++
.../fluss/lake/iceberg/utils/IcebergUtilsTest.java | 86 ++++++++
.../datalake-formats/iceberg.md | 4 +
24 files changed, 829 insertions(+), 523 deletions(-)
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 7db0a1957..70886c9c8 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 @@ package org.apache.fluss.lake.iceberg;
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.List;
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 class IcebergLakeCatalog implements LakeCatalog {
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<String, String> expectedProperties = buildTableProperties(tableDescriptor, isPkTable);
tableBuilder.withProperties(expectedProperties);
tableBuilder.withPartitionSpec(partitionSpec);
@@ -116,6 +120,8 @@ public class IcebergLakeCatalog implements LakeCatalog {
createTable(
tablePath,
tableBuilder,
+ tableDescriptor,
+ isPkTable,
icebergSchema,
partitionSpec,
sortOrder,
@@ -127,6 +133,8 @@ public class IcebergLakeCatalog implements LakeCatalog {
createTable(
tablePath,
tableBuilder,
+ tableDescriptor,
+ isPkTable,
icebergSchema,
partitionSpec,
sortOrder,
@@ -223,7 +231,8 @@ public class IcebergLakeCatalog implements LakeCatalog {
}
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 @@ public class IcebergLakeCatalog implements LakeCatalog {
}
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 @@ public class IcebergLakeCatalog implements LakeCatalog {
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 @@ public class IcebergLakeCatalog implements LakeCatalog {
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 @@ public class IcebergLakeCatalog implements LakeCatalog {
private void createTable(
TablePath tablePath,
Catalog.TableBuilder tableBuilder,
+ TableDescriptor tableDescriptor,
+ boolean isPkTable,
Schema newIcebergSchema,
PartitionSpec expectedSpec,
SortOrder expectedSortOrder,
@@ -297,6 +325,23 @@ public class IcebergLakeCatalog implements LakeCatalog {
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 @@ public class IcebergLakeCatalog implements LakeCatalog {
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 @@ public class IcebergLakeCatalog implements LakeCatalog {
}
}
- 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<String, String> buildTableProperties(
TableDescriptor tableDescriptor, boolean isPkTable) {
Map<String, String> 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 7559625f9..21ffd3217 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 @@ import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
@Internal
public final class IcebergSchemaUtils {
- /** The Iceberg-only system columns appended to every Fluss-managed Iceberg table. */
- public static final Map<String, Type> 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<String, Type> LEGACY_SYSTEM_COLUMNS;
static {
LinkedHashMap<String, Type> 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.
+ *
+ * <p>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.
+ *
+ * <p>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<Types.NestedField> 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 final class IcebergSchemaUtils {
fields.add(field);
}
- for (Map.Entry<String, Type> systemColumn : SYSTEM_COLUMNS.entrySet()) {
- fields.add(
- Types.NestedField.required(
- fieldId++, systemColumn.getKey(), systemColumn.getValue()));
+ if (includeSystemCols) {
+ for (Map.Entry<String, Type> 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 570c05603..96c1e44b8 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 @@ public class IcebergRewriteDataFiles {
// 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<FileScanTask> 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 a618df40e..cc884585d 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.row.TimestampNtz;
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.time.OffsetDateTime;
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 f654fb5d0..cd0f87bbd 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 org.apache.iceberg.types.Types;
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 @@ import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
* 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 class IcebergRecordReader implements RecordReader {
private TableScan applyProject(TableScan tableScan, int[][] projects) {
Types.StructType structType = tableScan.schema().asStruct();
- List<Types.NestedField> cols = new ArrayList<>(projects.length + 2);
-
+ List<Types.NestedField> 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 class IcebergRecordReader implements RecordReader {
private final ProjectedRow projectedRow;
private final IcebergRecordAsFlussRow icebergRecordAsFlussRow;
- private final int logOffsetColIndex;
- private final int timestampColIndex;
-
public IcebergRecordAsFlussRecordIterator(
CloseableIterable<Record> 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 class IcebergRecordReader implements RecordReader {
@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 a397a5af0..34c021756 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 @@ package org.apache.fluss.lake.iceberg.source;
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.Collectors;
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<IcebergSplit> {
@@ -143,47 +142,44 @@ public class IcebergSplitPlanner implements Planner<IcebergSplit> {
}
private Function<FileScanTask, Integer> createBucketExtractor(Table table) {
- PartitionSpec partitionSpec = table.spec();
- List<PartitionField> 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<PartitionField> 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<FileScanTask, List<String>> createPartitionExtractor(Table table) {
- PartitionSpec partitionSpec = table.spec();
- List<PartitionField> partitionFields = partitionSpec.fields();
+ List<PartitionField> 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<Integer> 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<Integer> 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 60f400575..fe1da1144 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.Instant;
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 @@ import static org.apache.fluss.utils.Preconditions.checkState;
*/
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 class FlussRecordAsIcebergRecord extends FlussRowAsIcebergRecord {
@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 93f60ed48..646bdddc4 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 class IcebergLakeWriter implements LakeWriter<IcebergWriteResult> {
// 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 732ea8de0..8f2bbb694 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 @@ package org.apache.fluss.lake.iceberg.tiering;
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 @@ final class 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 238ef5be9..275f72a09 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 abstract class RecordWriter implements AutoCloseable {
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 111df385f..1279cf6a1 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 class IcebergConversions {
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 class IcebergConversions {
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<PartitionField> 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 class IcebergConversions {
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 f9c7c0e0c..a2e4c2a99 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.annotation.Internal;
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 @@ public final class IcebergPartitionSpecUtils {
}
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.
+ *
+ * <p>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 000000000..7e7ddb4ff
--- /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.
+ *
+ * <p>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}).
+ *
+ * <p>Detection requires <b>all three</b> 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 30f2b82f1..868a6bfd4 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.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.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.FileFormat;
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 @@ class IcebergLakeCatalogTest {
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 @@ class IcebergLakeCatalogTest {
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 @@ class IcebergLakeCatalogTest {
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 @@ class IcebergLakeCatalogTest {
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<TableChange> 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<String> 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 @@ class IcebergLakeCatalogTest {
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 @@ class IcebergLakeCatalogTest {
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 @@ class IcebergLakeCatalogTest {
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 @@ class IcebergLakeCatalogTest {
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 @@ class IcebergLakeCatalogTest {
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 51c25e813..aad698991 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.List;
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 @@ class FlinkCatalogLakeTest extends FlinkIcebergTieringTestBase {
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 4b6d9de71..000000000
--- 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<InternalRow> flussRows = new ArrayList<>();
-
- List<InternalRow> 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<Record> actualRows, List<InternalRow> expectedRows) {
- // check records size
- assertThat(actualRows.size()).isEqualTo(expectedRows.size());
-
- // check records content
- Iterator<Record> actualIterator =
- actualRows.stream()
- .sorted(Comparator.comparingInt((Record r) -> (int) r.get(0)))
- .iterator();
- Iterator<InternalRow> 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<InternalRow> flussRows = new ArrayList<>();
-
- List<InternalRow> 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<InternalRow> 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<InternalRow> writeIcebergTableRecords(
- TablePath tablePath,
- TableBucket tableBucket,
- long expectedLogEndOffset,
- boolean append,
- List<InternalRow> 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 e5e142dfb..b94a5ac14 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 @@ class IcebergRecordAsFlussRowTest {
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 5bd021b9a..8c8206181 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.fluss.server.zk.data.lake.LakeTable;
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 org.junit.jupiter.api.extension.RegisterExtension;
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 @@ public class FlinkIcebergTieringTestBase {
}
protected void checkDataInIcebergAppendOnlyTable(
- TablePath tablePath, List<InternalRow> expectedRows, long startingOffset)
- throws Exception {
+ TablePath tablePath, List<InternalRow> expectedRows) throws Exception {
+ List<List<Object>> actualRows = new ArrayList<>();
try (CloseableIterator<Record> records = getIcebergRows(tablePath)) {
- Iterator<InternalRow> 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.<Object>asList(actualRecord.get(0), actualRecord.get(1)));
}
- assertThat(flussRowIterator.hasNext()).isFalse();
}
+
+ List<List<Object>> expectedRecords = new ArrayList<>();
+ for (InternalRow expectedRow : expectedRows) {
+ expectedRecords.add(
+ Arrays.<Object>asList(
+ expectedRow.getInt(0), expectedRow.getString(1).toString()));
+ }
+ assertThat(actualRows).containsExactlyInAnyOrderElementsOf(expectedRecords);
}
protected void checkFileStatusInIcebergTable(
@@ -364,24 +356,27 @@ public class FlinkIcebergTieringTestBase {
}
protected void checkDataInIcebergAppendOnlyPartitionedTable(
- TablePath tablePath,
- Map<String, String> partitionSpec,
- List<InternalRow> expectedRows,
- long startingOffset)
+ TablePath tablePath, Map<String, String> partitionSpec, List<InternalRow> expectedRows)
throws Exception {
+ List<List<Object>> actualRows = new ArrayList<>();
try (CloseableIterator<Record> records = getIcebergRows(tablePath, partitionSpec)) {
- Iterator<InternalRow> 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.<Object>asList(
+ actualRecord.get(0), actualRecord.get(1), actualRecord.get(2)));
}
- assertThat(flussRowIterator.hasNext()).isFalse();
}
+
+ List<List<Object>> expectedRecords = new ArrayList<>();
+ for (InternalRow expectedRow : expectedRows) {
+ expectedRecords.add(
+ Arrays.<Object>asList(
+ expectedRow.getInt(0),
+ expectedRow.getString(1).toString(),
+ expectedRow.getString(2).toString()));
+ }
+ assertThat(actualRows).containsExactlyInAnyOrderElementsOf(expectedRecords);
}
private CloseableIterator<Record> getIcebergRows(TablePath tablePath) {
@@ -392,59 +387,9 @@ public class FlinkIcebergTieringTestBase {
private CloseableIterator<Record> getIcebergRows(
TablePath tablePath, Map<String, String> 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<Record> records = new ArrayList<>();
- int fieldId = table.schema().findField(OFFSET_COLUMN_NAME).fieldId();
- SortedSet<DataFile> 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<Record> 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 @@ public class FlinkIcebergTieringTestBase {
return scanBuilder;
}
- private TableScan filterByPartition(TableScan tableScan, Map<String, String> partitionSpec) {
- for (Map.Entry<String, String> 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<TableBucket, Long> 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 000000000..0c5dadf28
--- /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.
+ *
+ * <p>The partition spec is preserved in a spec-aware way, matching how a real pre-FIP-27 table
+ * looked:
+ *
+ * <ul>
+ * <li>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);
+ * <li>a bucket-unaware table gains an {@code identity(__bucket)} partition, which is how such
+ * tables physically carried the bucket before FIP-27.
+ * </ul>
+ *
+ * <p>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 d4177a305..bafcf7526 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 java.util.stream.Collectors;
import static org.apache.fluss.lake.iceberg.utils.IcebergConversions.toIceberg;
import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
+import static org.apache.fluss.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 @@ class IcebergSchemaEvolutionITCase extends FlinkIcebergTieringTestBase {
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 @@ class IcebergSchemaEvolutionITCase extends FlinkIcebergTieringTestBase {
.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 @@ class IcebergSchemaEvolutionITCase extends FlinkIcebergTieringTestBase {
}
}
+ @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<InternalRow> 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<String> 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 @@ class IcebergSchemaEvolutionITCase extends FlinkIcebergTieringTestBase {
.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<InternalRow> postAlterRows = Arrays.asList(row(4, "v4", 100), row(5, "v5", 200));
@@ -272,7 +332,8 @@ class IcebergSchemaEvolutionITCase extends FlinkIcebergTieringTestBase {
.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 @@ class IcebergSchemaEvolutionITCase extends FlinkIcebergTieringTestBase {
.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 @@ class IcebergSchemaEvolutionITCase extends FlinkIcebergTieringTestBase {
.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<InternalRow> 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 0e1c80eea..b83b764c7 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 @@ package org.apache.fluss.lake.iceberg.tiering;
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.Iterator;
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 @@ class IcebergTieringITCase extends FlinkIcebergTieringTestBase {
}
}
+ @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<InternalRow> flussRows = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ List<InternalRow> 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<InternalRow> expectedRows) throws Exception {
Iterator<Record> actualIterator = getIcebergRecords(tablePath).iterator();
@@ -353,7 +383,7 @@ class IcebergTieringITCase extends FlinkIcebergTieringTestBase {
assertReplicaStatus(t2Bucket, 30);
// check data in iceberg
- checkDataInIcebergAppendOnlyTable(t2, flussRows, 0);
+ checkDataInIcebergAppendOnlyTable(t2, flussRows);
}
private void testPartitionedTableTiering() throws Exception {
@@ -385,8 +415,7 @@ class IcebergTieringITCase extends FlinkIcebergTieringTestBase {
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 512024807..5915e0425 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 @@ class IcebergConversionsTest {
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 000000000..b687a02c4
--- /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 5a9cccb70..55a50e058 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"