[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;
 
@@ -223,7 +225,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 +236,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 +256,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 +288,13 @@ boolean isIcebergSchemaCompatible(
         if (flussTableDescriptor == null) {
             return false;
         }
-        // Identifier fields don't affect the comparison.
-        Schema expectedSchema = IcebergSchemaUtils.createIcebergSchema(flussTableDescriptor, false);
+        // FIP-27: newly created tables are clean (no system columns). Legacy tables still carry the
+        // three trailing system columns. To handle re-enabling lake tiering on a legacy table, we
+        // build the expected schema to match what the physical table actually has.
+        Schema expectedSchema =
+                IcebergUtils.isLegacyTable(icebergSchema)
+                        ? IcebergSchemaUtils.createLegacyIcebergSchema(flussTableDescriptor, false)
+                        : IcebergSchemaUtils.createIcebergSchema(flussTableDescriptor, false);
         return IcebergSchemaUtils.compatibleWith(icebergSchema, expectedSchema);
     }
 
@@ -508,7 +528,11 @@ private void createDatabase(String databaseName) {
     }
 
     private SortOrder createSortOrder(Schema icebergSchema) {
-        // Sort by __offset system column for deterministic ordering
+        if (icebergSchema.findField(OFFSET_COLUMN_NAME) == null) {
+            // Clean tables (FIP-27) have no __offset system column; no sort order is needed.
+            return SortOrder.unsorted();
+        }
+        // Legacy tables: sort by __offset for deterministic ordering
         SortOrder.Builder builder = SortOrder.builderFor(icebergSchema);
         builder.asc(OFFSET_COLUMN_NAME);
         return builder.build();

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

@@ -38,27 +38,39 @@
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.SYSTEM_COLUMNS;
+import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.LEGACY_SYSTEM_COLUMNS;
+import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
 
 /** Adapter for Iceberg Record as fluss row. */
 public class IcebergRecordAsFlussRow implements InternalRow {
 
     private Record icebergRecord;
+    // cached business field count, recomputed when the record changes
+    private int businessFieldCount;
 
     public IcebergRecordAsFlussRow() {}
 
     public IcebergRecordAsFlussRow(Record icebergRecord) {
         this.icebergRecord = icebergRecord;
+        this.businessFieldCount = computeBusinessFieldCount(icebergRecord);
     }
 
     public IcebergRecordAsFlussRow replaceIcebergRecord(Record icebergRecord) {
         this.icebergRecord = icebergRecord;
+        this.businessFieldCount = computeBusinessFieldCount(icebergRecord);
         return this;
     }
 
+    private static int computeBusinessFieldCount(Record record) {
+        // A legacy table has __timestamp as the last column; a clean table has no system columns.
+        int total = record.struct().fields().size();
+        boolean isLegacy = record.struct().field(TIMESTAMP_COLUMN_NAME) != null;
+        return isLegacy ? total - LEGACY_SYSTEM_COLUMNS.size() : total;
+    }
+
     @Override
     public int getFieldCount() {
-        return icebergRecord.struct().fields().size() - SYSTEM_COLUMNS.size();
+        return businessFieldCount;
     }
 
     @Override

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

@@ -18,6 +18,7 @@
 
 package org.apache.fluss.lake.iceberg.source;
 
+import org.apache.fluss.lake.iceberg.utils.IcebergUtils;
 import org.apache.fluss.lake.source.RecordReader;
 import org.apache.fluss.record.ChangeType;
 import org.apache.fluss.record.GenericRecord;
@@ -53,35 +54,45 @@
  * org.apache.iceberg.Scan#ignoreResiduals()} for details.
  */
 public class IcebergRecordReader implements RecordReader {
+
+    /** Sentinel value emitted when the lake table has no per-record offset/timestamp. */
+    private static final long NO_SYSTEM_COLUMN_VALUE = -1L;
+
     protected IcebergRecordAsFlussRecordIterator iterator;
     protected @Nullable int[][] project;
     protected Types.StructType struct;
 
     public IcebergRecordReader(FileScanTask fileScanTask, Table table, @Nullable int[][] project) {
+        boolean isLegacy = IcebergUtils.isLegacyTable(table.schema());
         TableScan tableScan = table.newScan();
         if (project != null) {
-            tableScan = applyProject(tableScan, project);
+            tableScan = applyProject(tableScan, project, isLegacy);
         }
         IcebergGenericReader reader = new IcebergGenericReader(tableScan, true);
         struct = tableScan.schema().asStruct();
-        this.iterator = new IcebergRecordAsFlussRecordIterator(reader.open(fileScanTask), struct);
+        this.iterator =
+                new IcebergRecordAsFlussRecordIterator(reader.open(fileScanTask), struct, isLegacy);
     }
 
     @Override
     public CloseableIterator<LogRecord> read() throws IOException {
         return iterator;
     }
 
-    private TableScan applyProject(TableScan tableScan, int[][] projects) {
+    private TableScan applyProject(TableScan tableScan, int[][] projects, boolean isLegacy) {
         Types.StructType structType = tableScan.schema().asStruct();
         List<Types.NestedField> cols = new ArrayList<>(projects.length + 2);
 
         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));
+        if (isLegacy) {
+            // Legacy tables carry __offset and __timestamp; project them so the iterator can
+            // read the actual per-record offset and timestamp values.
+            cols.add(structType.field(OFFSET_COLUMN_NAME));
+            cols.add(structType.field(TIMESTAMP_COLUMN_NAME));
+        }
         return tableScan.project(new Schema(cols));
     }
 
@@ -97,13 +108,25 @@ public static class IcebergRecordAsFlussRecordIterator implements CloseableItera
         private final int timestampColIndex;
 
         public IcebergRecordAsFlussRecordIterator(
-                CloseableIterable<Record> icebergRecordIterator, Types.StructType struct) {
+                CloseableIterable<Record> icebergRecordIterator,
+                Types.StructType struct,
+                boolean isLegacy) {
             this.icebergRecordIterator = icebergRecordIterator.iterator();
-            this.logOffsetColIndex = struct.fields().indexOf(struct.field(OFFSET_COLUMN_NAME));
-            this.timestampColIndex = struct.fields().indexOf(struct.field(TIMESTAMP_COLUMN_NAME));
 
-            int[] project = IntStream.range(0, struct.fields().size() - 2).toArray();
-            projectedRow = ProjectedRow.from(project);
+            if (isLegacy) {
+                this.logOffsetColIndex = struct.fields().indexOf(struct.field(OFFSET_COLUMN_NAME));
+                this.timestampColIndex =
+                        struct.fields().indexOf(struct.field(TIMESTAMP_COLUMN_NAME));
+                // The last two projected columns are __offset and __timestamp; strip them from the
+                // business row.
+                int[] project = IntStream.range(0, struct.fields().size() - 2).toArray();
+                projectedRow = ProjectedRow.from(project);
+            } else {
+                this.logOffsetColIndex = -1;
+                this.timestampColIndex = -1;
+                projectedRow =
+                        ProjectedRow.from(IntStream.range(0, struct.fields().size()).toArray());
+            }
             icebergRecordAsFlussRow = new IcebergRecordAsFlussRow();
         }
 
@@ -124,12 +147,19 @@ public boolean hasNext() {
         @Override
         public LogRecord next() {
             Record icebergRecord = icebergRecordIterator.next();
-            long offset = icebergRecord.get(logOffsetColIndex, Long.class);
-            long timestamp =
-                    icebergRecord
-                            .get(timestampColIndex, OffsetDateTime.class)
-                            .toInstant()
-                            .toEpochMilli();
+            long offset;
+            long timestamp;
+            if (logOffsetColIndex >= 0) {
+                offset = icebergRecord.get(logOffsetColIndex, Long.class);
+                timestamp =
+                        icebergRecord
+                                .get(timestampColIndex, OffsetDateTime.class)
+                                .toInstant()
+                                .toEpochMilli();
+            } else {
+                offset = NO_SYSTEM_COLUMN_VALUE;
+                timestamp = NO_SYSTEM_COLUMN_VALUE;
+            }
 
             return new GenericRecord(
                     offset,