[lake/paimon] Support clean and legacy Paimon lake table schemas

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java

@@ -68,8 +68,9 @@ public class PaimonLakeCatalog implements LakeCatalog {
         // We need __bucket system column to filter out the given bucket
         // for paimon bucket-unaware append only table.
         // It's not required for paimon bucket-aware table like primary key table
-        // and bucket-aware append only table, but we always add the system column
-        // for consistent behavior
+        // and bucket-aware append only table, but legacy tables always carry the system column
+        // for consistent behavior. Under FIP-27 these columns are no longer added to newly created
+        // (clean) tables; they only remain on legacy tables created before FIP-27.
         SYSTEM_COLUMNS.put(BUCKET_COLUMN_NAME, DataTypes.INT());
         SYSTEM_COLUMNS.put(OFFSET_COLUMN_NAME, DataTypes.BIGINT());
         SYSTEM_COLUMNS.put(TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP_LTZ_MILLIS());
@@ -133,12 +134,13 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Cont
                     currentPaimonSchema, toPaimonSchema(context.getCurrentTable()))) {
                 // if the paimon schema is same as current fluss schema, directly apply all the
                 // changes.
-                paimonSchemaChanges = toPaimonSchemaChanges(changesToApply);
+                paimonSchemaChanges = toPaimonSchemaChanges(table, changesToApply);
             } else if (isPaimonSchemaCompatible(
                     currentPaimonSchema, toPaimonSchema(context.getExpectedTable()))) {
                 // if the schema is same as applied fluss schema , skip adding columns.
                 paimonSchemaChanges =
                         toPaimonSchemaChanges(
+                                table,
                                 changesToApply.stream()
                                         .filter(
                                                 tableChange ->

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonRecordReader.java

@@ -46,6 +46,14 @@
 /** Record reader for paimon table. */
 public class PaimonRecordReader implements RecordReader {
 
+    /**
+     * Sentinel log offset / timestamp emitted for rows read from a clean lake table, which does not
+     * store the {@code __offset} / {@code __timestamp} system columns. A negative offset is
+     * interpreted downstream as "no valid offset" (snapshot phase), see {@code
+     * LakeRecordRecordEmitter}.
+     */
+    private static final long NO_SYSTEM_COLUMN_VALUE = -1L;
+
     protected PaimonRowAsFlussRecordIterator iterator;
     protected @Nullable int[][] project;
     protected RowType paimonRowType;
@@ -90,6 +98,13 @@ private ReadBuilder applyProject(
             ReadBuilder readBuilder, int[][] projects, RowType paimonFullRowType) {
         int[] projectIds = Arrays.stream(projects).mapToInt(project -> project[0]).toArray();
 
+        if (!hasSystemColumn(paimonFullRowType)) {
+            // Clean tables have no system columns to read, so project the business columns only.
+            return readBuilder.withProjection(projectIds);
+        }
+
+        // Legacy tables carry __offset/__timestamp, which the iterator needs to recover the log
+        // offset and timestamp of each record; append them to the projection.
         int offsetFieldPos = paimonFullRowType.getFieldIndex(OFFSET_COLUMN_NAME);
         int timestampFieldPos = paimonFullRowType.getFieldIndex(TIMESTAMP_COLUMN_NAME);
 
@@ -102,6 +117,15 @@ private ReadBuilder applyProject(
         return readBuilder.withProjection(paimonProject);
     }
 
+    /** A legacy table carries the three system columns, ending with {@code __timestamp}. */
+    private static boolean hasSystemColumn(RowType paimonRowType) {
+        return paimonRowType
+                .getFields()
+                .get(paimonRowType.getFieldCount() - 1)
+                .name()
+                .equals(TIMESTAMP_COLUMN_NAME);
+    }
+
     /** Iterator for paimon row as fluss record. */
     public static class PaimonRowAsFlussRecordIterator implements CloseableIterator<LogRecord> {
 
@@ -117,11 +141,25 @@ public PaimonRowAsFlussRecordIterator(
                 org.apache.paimon.utils.CloseableIterator<InternalRow> paimonRowIterator,
                 RowType paimonRowType) {
             this.paimonRowIterator = paimonRowIterator;
-            this.logOffsetColIndex = paimonRowType.getFieldIndex(OFFSET_COLUMN_NAME);
-            this.timestampColIndex = paimonRowType.getFieldIndex(TIMESTAMP_COLUMN_NAME);
 
-            int[] project = IntStream.range(0, paimonRowType.getFieldCount() - 2).toArray();
-            projectedRow = ProjectedRow.from(project);
+            int fieldCount = paimonRowType.getFieldCount();
+            if (!hasSystemColumn(paimonRowType)) {
+                // No system columns are read; all projected fields are business fields, and the
+                // log offset / timestamp are not available from the lake table.
+                this.logOffsetColIndex = -1;
+                this.timestampColIndex = -1;
+                projectedRow = ProjectedRow.from(IntStream.range(0, fieldCount).toArray());
+            } else {
+                // Legacy layout: applyProject appended exactly __offset and __timestamp (not
+                // __bucket) as the last two projected fields, so the business fields are all fields
+                // except those trailing two.
+                this.logOffsetColIndex = paimonRowType.getFieldIndex(OFFSET_COLUMN_NAME);
+                this.timestampColIndex = paimonRowType.getFieldIndex(TIMESTAMP_COLUMN_NAME);
+                int[] project = IntStream.range(0, fieldCount - 2).toArray();
+                projectedRow = ProjectedRow.from(project);
+            }
+            // The wrapped row is only ever accessed by index through projectedRow, which already
+            // drops the system columns, so no trailing-system-column trimming is needed here.
             paimonRowAsFlussRow = new PaimonRowAsFlussRow();
         }
 
@@ -143,8 +181,14 @@ public boolean hasNext() {
         public LogRecord next() {
             InternalRow paimonRow = paimonRowIterator.next();
             ChangeType changeType = toChangeType(paimonRow.getRowKind());
-            long offset = paimonRow.getLong(logOffsetColIndex);
-            long timestamp = paimonRow.getTimestamp(timestampColIndex, 6).getMillisecond();
+            long offset =
+                    logOffsetColIndex < 0
+                            ? NO_SYSTEM_COLUMN_VALUE
+                            : paimonRow.getLong(logOffsetColIndex);
+            long timestamp =
+                    timestampColIndex < 0
+                            ? NO_SYSTEM_COLUMN_VALUE
+                            : paimonRow.getTimestamp(timestampColIndex, 6).getMillisecond();
 
             return new GenericRecord(
                     offset,

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java

@@ -33,6 +33,7 @@
 /** To wrap Fluss {@link LogRecord} as paimon {@link InternalRow}. */
 public class FlussRecordAsPaimonRow extends FlussRowAsPaimonRow {
 
+    private final boolean paimonIncludingSystemColumns;
     private final int bucket;
     private LogRecord logRecord;
     private int originRowFieldCount;
@@ -41,10 +42,19 @@ public class FlussRecordAsPaimonRow extends FlussRowAsPaimonRow {
     private final int offsetFieldIndex;
     private final int timestampFieldIndex;
 
-    public FlussRecordAsPaimonRow(int bucket, RowType tableTowType) {
-        super(tableTowType);
+    public FlussRecordAsPaimonRow(int bucket, RowType tableRowType) {
+        this(bucket, tableRowType, false);
+    }
+
+    public FlussRecordAsPaimonRow(
+            int bucket, RowType tableRowType, boolean paimonIncludingSystemColumns) {
+        super(tableRowType);
         this.bucket = bucket;
-        this.businessFieldCount = tableRowType.getFieldCount() - SYSTEM_COLUMNS.size();
+        this.paimonIncludingSystemColumns = paimonIncludingSystemColumns;
+        this.businessFieldCount =
+                tableRowType.getFieldCount()
+                        - (paimonIncludingSystemColumns ? SYSTEM_COLUMNS.size() : 0);
+        // only valid when paimon includes the system columns
         this.bucketFieldIndex = businessFieldCount;
         this.offsetFieldIndex = businessFieldCount + 1;
         this.timestampFieldIndex = businessFieldCount + 2;
@@ -97,7 +107,7 @@ public boolean isNullAt(int pos) {
 
     @Override
     public int getInt(int pos) {
-        if (pos == bucketFieldIndex) {
+        if (paimonIncludingSystemColumns && pos == bucketFieldIndex) {
             // bucket system column
             return bucket;
         }
@@ -114,12 +124,14 @@ public int getInt(int pos) {
     @Override
     public long getLong(int pos) {
         checkState(logRecord != null, "setFlussRecord() must be called before accessing the row.");
-        if (pos == offsetFieldIndex) {
-            //  offset system column
-            return logRecord.logOffset();
-        } else if (pos == timestampFieldIndex) {
-            //  timestamp system column
-            return logRecord.timestamp();
+        if (paimonIncludingSystemColumns) {
+            if (pos == offsetFieldIndex) {
+                //  offset system column
+                return logRecord.logOffset();
+            } else if (pos == timestampFieldIndex) {
+                //  timestamp system column
+                return logRecord.timestamp();
+            }
         }
         if (pos >= originRowFieldCount) {
             throw new IllegalStateException(
@@ -135,7 +147,7 @@ public long getLong(int pos) {
     public Timestamp getTimestamp(int pos, int precision) {
         checkState(logRecord != null, "setFlussRecord() must be called before accessing the row.");
         // it's timestamp system column
-        if (pos == timestampFieldIndex) {
+        if (paimonIncludingSystemColumns && pos == timestampFieldIndex) {
             return Timestamp.fromEpochMillis(logRecord.timestamp());
         }
         if (pos >= originRowFieldCount) {

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java

@@ -39,6 +39,7 @@
 import java.util.Map;
 
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
+import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
 
 /** Implementation of {@link LakeWriter} for Paimon. */
 public class PaimonLakeWriter implements LakeWriter<PaimonWriteResult>, SupportsRecordBatchWrite {
@@ -58,21 +59,30 @@ public PaimonLakeWriter(
         List<String> partitionKeys = fileStoreTable.partitionKeys();
         RowType flussRowType = writerInitContext.tableInfo().getRowType();
 
+        // FIP-27: detect whether the target Paimon table is a clean table (only user columns) or a
+        // legacy table (carrying the three Fluss system columns). A legacy table is recognisable by
+        // the presence of the __timestamp system column. Writers emit system columns only for
+        // legacy tables.
+        boolean paimonIncludingSystemColumns =
+                fileStoreTable.rowType().getFieldIndex(TIMESTAMP_COLUMN_NAME) >= 0;
+
         this.recordWriter =
                 fileStoreTable.primaryKeys().isEmpty()
                         ? new AppendOnlyWriter(
                                 fileStoreTable,
                                 writerInitContext.tableBucket(),
                                 writerInitContext.partition(),
                                 partitionKeys,
-                                flussRowType)
+                                flussRowType,
+                                paimonIncludingSystemColumns)
                         : new MergeTreeWriter(
                                 fileStoreTable,
                                 writerInitContext.tableBucket(),
                                 writerInitContext.partition(),
                                 partitionKeys,
                                 flussRowType,
-                                writerInitContext.ioTmpDirs());
+                                writerInitContext.ioTmpDirs(),
+                                paimonIncludingSystemColumns);
     }
 
     @Override

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java

@@ -49,7 +49,8 @@ public RecordWriter(
             TableBucket tableBucket,
             @Nullable String partition,
             List<String> partitionKeys,
-            org.apache.fluss.types.RowType flussRowType) {
+            org.apache.fluss.types.RowType flussRowType,
+            boolean paimonIncludingSystemColumns) {
         this.tableWrite = tableWrite;
         this.tableRowType = tableRowType;
         this.bucket = tableBucket.getBucket();
@@ -62,7 +63,8 @@ public RecordWriter(
             this.partition = resolvePartition(partition, partitionKeys, flussRowType);
         }
         this.flussRecordAsPaimonRow =
-                new FlussRecordAsPaimonRow(tableBucket.getBucket(), tableRowType);
+                new FlussRecordAsPaimonRow(
+                        tableBucket.getBucket(), tableRowType, paimonIncludingSystemColumns);
     }
 
     public abstract void write(LogRecord record) throws Exception;

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyArrowBatchHelper.java

@@ -57,6 +57,7 @@ class AppendOnlyArrowBatchHelper implements AutoCloseable {
     private final TableWriteImpl<InternalRow> tableWrite;
     private final RowType tableRowType;
     private final int bucket;
+    private final boolean paimonIncludingSystemColumns;
 
     private static final Field BUCKET_FIELD =
             new Field(
@@ -88,11 +89,13 @@ class AppendOnlyArrowBatchHelper implements AutoCloseable {
             FileStoreTable fileStoreTable,
             TableWriteImpl<InternalRow> tableWrite,
             RowType tableRowType,
-            int bucket) {
+            int bucket,
+            boolean paimonIncludingSystemColumns) {
         this.fileStoreTable = fileStoreTable;
         this.tableWrite = tableWrite;
         this.tableRowType = tableRowType;
         this.bucket = bucket;
+        this.paimonIncludingSystemColumns = paimonIncludingSystemColumns;
     }
 
     /**
@@ -107,6 +110,16 @@ void writeArrowBatch(ArrowBatchData arrowBatchData, BinaryRow partition) throws
         }
 
         VectorSchemaRoot originalRoot = arrowBatchData.getVectorSchemaRoot();
+
+        if (!paimonIncludingSystemColumns) {
+            // Clean tables contain only user columns, so the incoming Arrow batch already matches
+            // the Paimon table schema. Write it directly without enriching system columns.
+            ArrowBundleRecords cleanRecords =
+                    new ArrowBundleRecords(originalRoot, tableRowType, false);
+            tableWrite.writeBundle(partition, writtenBucket, cleanRecords);
+            return;
+        }
+
         long baseOffset = arrowBatchData.getBaseLogOffset();
         long timestamp = arrowBatchData.getTimestamp();
         int rowCount = originalRoot.getRowCount();

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java

@@ -46,12 +46,15 @@ public class AppendOnlyWriter extends RecordWriter<InternalRow> {
      */
     @Nullable private AutoCloseable arrowBatchHelper;
 
+    private final boolean paimonIncludingSystemColumns;
+
     public AppendOnlyWriter(
             FileStoreTable fileStoreTable,
             TableBucket tableBucket,
             @Nullable String partition,
             List<String> partitionKeys,
-            RowType flussRowType) {
+            RowType flussRowType,
+            boolean paimonIncludingSystemColumns) {
         //noinspection unchecked
         super(
                 (TableWriteImpl<InternalRow>)
@@ -61,8 +64,10 @@ public AppendOnlyWriter(
                 tableBucket,
                 partition,
                 partitionKeys,
-                flussRowType);
+                flussRowType,
+                paimonIncludingSystemColumns);
         this.fileStoreTable = fileStoreTable;
+        this.paimonIncludingSystemColumns = paimonIncludingSystemColumns;
     }
 
     @Override
@@ -90,7 +95,11 @@ public void writeArrowBatch(ArrowBatchData arrowBatchData) throws Exception {
         if (arrowBatchHelper == null) {
             helper =
                     new AppendOnlyArrowBatchHelper(
-                            fileStoreTable, tableWrite, tableRowType, bucket);
+                            fileStoreTable,
+                            tableWrite,
+                            tableRowType,
+                            bucket,
+                            paimonIncludingSystemColumns);
             arrowBatchHelper = helper;
         } else {
             helper = (AppendOnlyArrowBatchHelper) arrowBatchHelper;

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java

@@ -49,8 +49,16 @@ public MergeTreeWriter(
             TableBucket tableBucket,
             @Nullable String partition,
             List<String> partitionKeys,
-            RowType flussRowType) {
-        this(fileStoreTable, tableBucket, partition, partitionKeys, flussRowType, (String[]) null);
+            RowType flussRowType,
+            boolean paimonIncludingSystemColumns) {
+        this(
+                fileStoreTable,
+                tableBucket,
+                partition,
+                partitionKeys,
+                flussRowType,
+                (String[]) null,
+                paimonIncludingSystemColumns);
     }
 
     public MergeTreeWriter(
@@ -59,14 +67,16 @@ public MergeTreeWriter(
             @Nullable String partition,
             List<String> partitionKeys,
             RowType flussRowType,
-            @Nullable String[] ioTmpDirs) {
+            @Nullable String[] ioTmpDirs,
+            boolean paimonIncludingSystemColumns) {
         this(
                 fileStoreTable,
                 createIOManager(ioTmpDirs),
                 tableBucket,
                 partition,
                 partitionKeys,
-                flussRowType);
+                flussRowType,
+                paimonIncludingSystemColumns);
     }
 
     MergeTreeWriter(
@@ -75,14 +85,16 @@ public MergeTreeWriter(
             TableBucket tableBucket,
             @Nullable String partition,
             List<String> partitionKeys,
-            RowType flussRowType) {
+            RowType flussRowType,
+            boolean paimonIncludingSystemColumns) {
         super(
                 createTableWrite(fileStoreTable, ioManager),
                 fileStoreTable.rowType(),
                 tableBucket,
                 partition,
                 partitionKeys,
-                flussRowType);
+                flussRowType,
+                paimonIncludingSystemColumns);
         this.rowKeyExtractor = fileStoreTable.createRowKeyExtractor();
         this.ioManager = ioManager;
     }