[lake/iceberg] Support clean and legacy Iceberg lake table schemas

fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java

@@ -20,10 +20,12 @@
 import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.InvalidAlterTableException;
+import org.apache.fluss.exception.InvalidTableException;
 import org.apache.fluss.exception.TableAlreadyExistException;
 import org.apache.fluss.exception.TableNotExistException;
 import org.apache.fluss.lake.iceberg.utils.IcebergCatalogUtils;
 import org.apache.fluss.lake.iceberg.utils.IcebergPartitionSpecUtils;
+import org.apache.fluss.lake.iceberg.utils.IcebergUtils;
 import org.apache.fluss.lake.lakestorage.LakeCatalog;
 import org.apache.fluss.metadata.TableChange;
 import org.apache.fluss.metadata.TableDescriptor;
@@ -60,7 +62,7 @@
 import java.util.Map;
 import java.util.Set;
 
-import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.SYSTEM_COLUMNS;
+import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.LEGACY_SYSTEM_COLUMNS;
 import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
 import static org.apache.fluss.utils.Preconditions.checkArgument;
 
@@ -116,6 +118,8 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co
             createTable(
                     tablePath,
                     tableBuilder,
+                    tableDescriptor,
+                    isPkTable,
                     icebergSchema,
                     partitionSpec,
                     sortOrder,
@@ -127,6 +131,8 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co
                 createTable(
                         tablePath,
                         tableBuilder,
+                        tableDescriptor,
+                        isPkTable,
                         icebergSchema,
                         partitionSpec,
                         sortOrder,
@@ -223,7 +229,8 @@ private void applySchemaChanges(Table table, List<TableChange> schemaChanges, Co
         }
 
         UpdateSchema updateSchema = table.updateSchema();
-        String firstSystemColumnName = SYSTEM_COLUMNS.keySet().iterator().next();
+        boolean isLegacyTable = IcebergUtils.isLegacyTable(currentIcebergSchema);
+        String firstSystemColumnName = LEGACY_SYSTEM_COLUMNS.keySet().iterator().next();
         boolean hasChanges = false;
 
         for (TableChange tableChange : schemaChanges) {
@@ -233,6 +240,13 @@ private void applySchemaChanges(Table table, List<TableChange> schemaChanges, Co
                 }
                 TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange;
 
+                if (LEGACY_SYSTEM_COLUMNS.containsKey(addColumn.getName())) {
+                    throw new InvalidTableException(
+                            "Column '"
+                                    + addColumn.getName()
+                                    + "' conflicts with a reserved system column name.");
+                }
+
                 if (!(addColumn.getPosition() instanceof TableChange.Last)) {
                     throw new UnsupportedOperationException(
                             "Only support to add column at last for iceberg table.");
@@ -246,7 +260,12 @@ private void applySchemaChanges(Table table, List<TableChange> schemaChanges, Co
 
                 Type icebergType = flussDataType.accept(new FlussDataTypeToIcebergDataType());
                 updateSchema.addColumn(addColumn.getName(), icebergType, addColumn.getComment());
-                updateSchema.moveBefore(addColumn.getName(), firstSystemColumnName);
+                if (isLegacyTable) {
+                    // Legacy tables keep the three system columns as the trailing columns, so a new
+                    // business column must be inserted right before the first system column. Clean
+                    // tables have no system columns, so the new column is simply appended last.
+                    updateSchema.moveBefore(addColumn.getName(), firstSystemColumnName);
+                }
                 hasChanges = true;
             } else {
                 throw new UnsupportedOperationException(
@@ -273,8 +292,13 @@ boolean isIcebergSchemaCompatible(
         if (flussTableDescriptor == null) {
             return false;
         }
-        // Identifier fields don't affect the comparison.
-        Schema expectedSchema = IcebergSchemaUtils.createIcebergSchema(flussTableDescriptor, false);
+        // FIP-27: newly created tables are clean (no system columns). Legacy tables still carry the
+        // three trailing system columns. To handle re-enabling lake tiering on a legacy table, we
+        // build the expected schema to match what the physical table actually has.
+        Schema expectedSchema =
+                IcebergUtils.isLegacyTable(icebergSchema)
+                        ? IcebergSchemaUtils.createLegacyIcebergSchema(flussTableDescriptor, false)
+                        : IcebergSchemaUtils.createIcebergSchema(flussTableDescriptor, false);
         return IcebergSchemaUtils.compatibleWith(icebergSchema, expectedSchema);
     }
 
@@ -285,6 +309,8 @@ private TableIdentifier toIcebergTableIdentifier(TablePath tablePath) {
     private void createTable(
             TablePath tablePath,
             Catalog.TableBuilder tableBuilder,
+            TableDescriptor tableDescriptor,
+            boolean isPkTable,
             Schema newIcebergSchema,
             PartitionSpec expectedSpec,
             SortOrder expectedSortOrder,
@@ -297,6 +323,22 @@ private void createTable(
             TableIdentifier icebergId = toIcebergTableIdentifier(tablePath);
             Table existingTable = icebergCatalog.loadTable(icebergId);
             Schema existingSchema = existingTable.schema();
+
+            // FIP-27: re-enabling tiering on a legacy table (created before FIP-27) hits this path.
+            // The expectations above were built for the clean layout, so rebuild the expected
+            // schema/spec/sort-order in the legacy layout to match what the physical table actually
+            // has; otherwise the compatibility checks below would spuriously reject the legacy
+            // table.
+            boolean existingIsLegacy = IcebergUtils.isLegacyTable(existingSchema);
+            if (existingIsLegacy) {
+                newIcebergSchema =
+                        IcebergSchemaUtils.createLegacyIcebergSchema(tableDescriptor, isPkTable);
+                expectedSpec =
+                        IcebergPartitionSpecUtils.createPartitionSpec(
+                                tableDescriptor, newIcebergSchema);
+                expectedSortOrder = createSortOrder(newIcebergSchema);
+            }
+
             if (!isIcebergSchemaCompatibleWithSchema(existingSchema, newIcebergSchema)) {
                 throw new TableAlreadyExistException(
                         String.format(
@@ -323,17 +365,23 @@ private void createTable(
                     existingSchema,
                     expectedSortOrder,
                     newIcebergSchema)) {
+                // Legacy tables expect ASC(__offset); clean tables (no __offset) expect unsorted().
+                String sortOrderGuidance =
+                        existingIsLegacy
+                                ? String.format(
+                                        "or with ASC(%s) set explicitly, ", OFFSET_COLUMN_NAME)
+                                : "";
                 throw new TableAlreadyExistException(
                         String.format(
                                 "The table %s already exists in Iceberg catalog, but the sort order is not compatible. "
                                         + "Existing sort order: %s, new sort order: %s. "
                                         + "Please first drop the table in Iceberg catalog, or pre-create the "
-                                        + "Iceberg table without a custom sort order, or with ASC(%s) set explicitly, "
+                                        + "Iceberg table without a custom sort order, %s"
                                         + "or use a new table name.",
                                 tablePath,
                                 existingTable.sortOrder(),
                                 expectedSortOrder,
-                                OFFSET_COLUMN_NAME),
+                                sortOrderGuidance),
                         e);
             }
 
@@ -508,7 +556,11 @@ private void createDatabase(String databaseName) {
     }
 
     private SortOrder createSortOrder(Schema icebergSchema) {
-        // Sort by __offset system column for deterministic ordering
+        if (icebergSchema.findField(OFFSET_COLUMN_NAME) == null) {
+            // Clean tables (FIP-27) have no __offset system column; no sort order is needed.
+            return SortOrder.unsorted();
+        }
+        // Legacy tables: sort by __offset for deterministic ordering
         SortOrder.Builder builder = SortOrder.builderFor(icebergSchema);
         builder.asc(OFFSET_COLUMN_NAME);
         return builder.build();

fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergSchemaUtils.java

@@ -41,34 +41,58 @@
 @Internal
 public final class IcebergSchemaUtils {
 
-    /** The Iceberg-only system columns appended to every Fluss-managed Iceberg table. */
-    public static final Map<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 static Schema createIcebergSchema(
             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) {

fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteDataFiles.java

@@ -106,9 +106,13 @@ private List<CombinedScanTask> planRewriteFileGroups(long snapshotId) throws IOE
 
         // then, pack the fileScanTasks into compaction units which contains compactable
         // fileScanTasks, after compaction, we want to it still keep order by __offset column,
-        // so, let's first sort by __offset column
-        int offsetFieldId = table.schema().findField(OFFSET_COLUMN_NAME).fieldId();
-        fileScanTasks.sort(sortFileScanTask(offsetFieldId));
+        // so, let's first sort by __offset column. FIP-27: a clean table has no __offset column,
+        // so this ordering is skipped (files are packed as-is).
+        org.apache.iceberg.types.Types.NestedField offsetField =
+                table.schema().findField(OFFSET_COLUMN_NAME);
+        if (offsetField != null) {
+            fileScanTasks.sort(sortFileScanTask(offsetField.fieldId()));
+        }
 
         // do package now
         BinPacking.ListPacker<FileScanTask> packer =

fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRow.java

@@ -28,6 +28,7 @@
 import org.apache.fluss.utils.BytesUtils;
 
 import org.apache.iceberg.data.Record;
+import org.apache.iceberg.types.Types;
 
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
@@ -38,27 +39,51 @@
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.SYSTEM_COLUMNS;
+import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
+import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
+import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
 
 /** Adapter for Iceberg Record as fluss row. */
 public class IcebergRecordAsFlussRow implements InternalRow {
 
     private Record icebergRecord;
+    // cached business field count, recomputed when the record changes
+    private int businessFieldCount;
 
     public IcebergRecordAsFlussRow() {}
 
     public IcebergRecordAsFlussRow(Record icebergRecord) {
         this.icebergRecord = icebergRecord;
+        this.businessFieldCount = computeBusinessFieldCount(icebergRecord);
     }
 
     public IcebergRecordAsFlussRow replaceIcebergRecord(Record icebergRecord) {
         this.icebergRecord = icebergRecord;
+        this.businessFieldCount = computeBusinessFieldCount(icebergRecord);
         return this;
     }
 
+    private static int computeBusinessFieldCount(Record record) {
+        // Subtract the system columns that are actually present in this record's struct rather than
+        // a fixed count: a projected legacy record may carry only a subset (e.g. __offset and
+        // __timestamp but not __bucket), and a clean record carries none.
+        Types.StructType struct = record.struct();
+        int systemColumns = 0;
+        if (struct.field(BUCKET_COLUMN_NAME) != null) {
+            systemColumns++;
+        }
+        if (struct.field(OFFSET_COLUMN_NAME) != null) {
+            systemColumns++;
+        }
+        if (struct.field(TIMESTAMP_COLUMN_NAME) != null) {
+            systemColumns++;
+        }
+        return struct.fields().size() - systemColumns;
+    }
+
     @Override
     public int getFieldCount() {
-        return icebergRecord.struct().fields().size() - SYSTEM_COLUMNS.size();
+        return businessFieldCount;
     }
 
     @Override