[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

@@ -62,17 +62,19 @@ public class PaimonLakeCatalog implements LakeCatalog {
 
     private static final Logger LOG = LoggerFactory.getLogger(PaimonLakeCatalog.class);
     private static final String PAIMON_PATH_KEY = "paimon.path";
-    public static final LinkedHashMap<String, DataType> SYSTEM_COLUMNS = new LinkedHashMap<>();
+    public static final LinkedHashMap<String, DataType> LEGACY_SYSTEM_COLUMNS =
+            new LinkedHashMap<>();
 
     static {
         // 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
-        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());
+        // 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.
+        LEGACY_SYSTEM_COLUMNS.put(BUCKET_COLUMN_NAME, DataTypes.INT());
+        LEGACY_SYSTEM_COLUMNS.put(OFFSET_COLUMN_NAME, DataTypes.BIGINT());
+        LEGACY_SYSTEM_COLUMNS.put(TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP_LTZ_MILLIS());
     }
 
     private final Catalog paimonCatalog;
@@ -133,12 +135,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/lookup/PaimonLakeTableLookuper.java

@@ -68,7 +68,7 @@
 import java.util.Set;
 
 import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2;
-import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS;
+import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonPartition;
 import static org.apache.fluss.utils.Preconditions.checkArgument;
@@ -280,7 +280,7 @@ private static int[] businessFieldProjection(FileStoreTable fileStoreTable) {
         List<DataField> fields = fileStoreTable.schema().logicalRowType().getFields();
         List<Integer> projectedFields = new ArrayList<>();
         for (int i = 0; i < fields.size(); i++) {
-            if (!SYSTEM_COLUMNS.containsKey(fields.get(i).name())) {
+            if (!LEGACY_SYSTEM_COLUMNS.containsKey(fields.get(i).name())) {
                 projectedFields.add(i);
             }
         }

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

@@ -19,6 +19,7 @@
 package org.apache.fluss.lake.paimon.source;
 
 import org.apache.fluss.lake.paimon.utils.PaimonRowAsFlussRow;
+import org.apache.fluss.lake.paimon.utils.PaimonUtils;
 import org.apache.fluss.lake.source.RecordReader;
 import org.apache.fluss.record.ChangeType;
 import org.apache.fluss.record.GenericRecord;
@@ -39,13 +40,20 @@
 import java.util.Arrays;
 import java.util.stream.IntStream;
 
+import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toChangeType;
-import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
-import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
 
 /** Record reader for paimon table. */
 public class PaimonRecordReader implements RecordReader {
 
+    /**
+     * Sentinel log offset / timestamp emitted for rows read from a lake table. The lake table does
+     * not carry a per-record log offset (a clean table has no system columns, and for a legacy
+     * table we no longer read them), so a negative value is emitted and 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;
@@ -57,9 +65,8 @@ public PaimonRecordReader(
             @Nullable Predicate predicate)
             throws IOException {
         ReadBuilder readBuilder = fileStoreTable.newReadBuilder();
-        RowType paimonFullRowType = fileStoreTable.rowType();
         if (project != null) {
-            readBuilder = applyProject(readBuilder, project, paimonFullRowType);
+            readBuilder = applyProject(readBuilder, project);
         }
 
         if (predicate != null) {
@@ -86,20 +93,11 @@ public CloseableIterator<LogRecord> read() throws IOException {
         return iterator;
     }
 
-    private ReadBuilder applyProject(
-            ReadBuilder readBuilder, int[][] projects, RowType paimonFullRowType) {
+    private ReadBuilder applyProject(ReadBuilder readBuilder, int[][] projects) {
+        // The projected column ids reference the user (business) columns; the log offset /
+        // timestamp are not read from the lake table, so no system column needs to be projected.
         int[] projectIds = Arrays.stream(projects).mapToInt(project -> project[0]).toArray();
-
-        int offsetFieldPos = paimonFullRowType.getFieldIndex(OFFSET_COLUMN_NAME);
-        int timestampFieldPos = paimonFullRowType.getFieldIndex(TIMESTAMP_COLUMN_NAME);
-
-        int[] paimonProject =
-                IntStream.concat(
-                                IntStream.of(projectIds),
-                                IntStream.of(offsetFieldPos, timestampFieldPos))
-                        .toArray();
-
-        return readBuilder.withProjection(paimonProject);
+        return readBuilder.withProjection(projectIds);
     }
 
     /** Iterator for paimon row as fluss record. */
@@ -110,18 +108,20 @@ public static class PaimonRowAsFlussRecordIterator implements CloseableIterator<
         private final ProjectedRow projectedRow;
         private final PaimonRowAsFlussRow paimonRowAsFlussRow;
 
-        private final int logOffsetColIndex;
-        private final int timestampColIndex;
-
         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);
+            // 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 fieldCount = paimonRowType.getFieldCount();
+            int businessFieldCount =
+                    PaimonUtils.isLegacyTable(paimonRowType)
+                            ? fieldCount - LEGACY_SYSTEM_COLUMNS.size()
+                            : fieldCount;
+            projectedRow = ProjectedRow.from(IntStream.range(0, businessFieldCount).toArray());
             paimonRowAsFlussRow = new PaimonRowAsFlussRow();
         }
 
@@ -143,12 +143,9 @@ 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();
-
             return new GenericRecord(
-                    offset,
-                    timestamp,
+                    NO_SYSTEM_COLUMN_VALUE,
+                    NO_SYSTEM_COLUMN_VALUE,
                     changeType,
                     projectedRow.replaceRow(paimonRowAsFlussRow.replaceRow(paimonRow)));
         }

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

@@ -25,14 +25,15 @@
 import org.apache.paimon.types.RowKind;
 import org.apache.paimon.types.RowType;
 
-import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS;
+import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toRowKind;
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
 import static org.apache.fluss.utils.Preconditions.checkState;
 
 /** 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 ? LEGACY_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

@@ -21,6 +21,7 @@
 import org.apache.fluss.lake.batch.RecordBatch;
 import org.apache.fluss.lake.paimon.tiering.append.AppendOnlyWriter;
 import org.apache.fluss.lake.paimon.tiering.mergetree.MergeTreeWriter;
+import org.apache.fluss.lake.paimon.utils.PaimonUtils;
 import org.apache.fluss.lake.writer.LakeWriter;
 import org.apache.fluss.lake.writer.SupportsRecordBatchWrite;
 import org.apache.fluss.lake.writer.WriterInitContext;
@@ -58,21 +59,28 @@ 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). Writers emit system columns only
+        // for legacy tables.
+        boolean paimonIncludingSystemColumns = PaimonUtils.isLegacyTable(fileStoreTable.rowType());
+
         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;