[lake/hudi] Use a clean lake table schema for Hudi lake storage

fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiArrayAsFlussArray.java

@@ -149,7 +149,7 @@ public InternalMap getMap(int pos) {
     @Override
     public InternalRow getRow(int pos, int numFields) {
         RowData nestedRow = hudiArray.getRow(pos, numFields);
-        return nestedRow == null ? null : new HudiRowAsFlussRow(nestedRow, false);
+        return nestedRow == null ? null : new HudiRowAsFlussRow(nestedRow);
     }
 
     @Override

fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java

@@ -55,10 +55,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
-import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS;
 import static org.apache.fluss.lake.hudi.utils.HudiConversions.toChangeType;
-import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
-import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
 
 /** Record reader for Hudi tables. */
 public class HudiRecordReader implements RecordReader {
@@ -126,10 +123,7 @@ public HudiRecordReader(
                     unifiedHudiTableReader.readFileSlice(fileSlice);
             this.iterator =
                     new HudiRecordAsFlussRecordIterator(
-                            hudiRecordIterator,
-                            requiredSchema,
-                            metadataFieldCount,
-                            SYSTEM_COLUMNS.size());
+                            hudiRecordIterator, requiredSchema, metadataFieldCount);
         }
     }
 
@@ -164,22 +158,16 @@ private static int[] selectedFields(
             return IntStream.range(0, schema.getFields().size()).toArray();
         }
 
+        // FIP-27: Hudi lake tables contain only user columns, so the selected fields are the Hudi
+        // metadata columns followed by the projected business columns.
         int[] hudiMetadataFields = IntStream.range(0, metadataFieldCount).toArray();
         int[] projectedDataFields =
                 Arrays.stream(project)
                         .filter(projectPath -> projectPath.length > 0)
                         .mapToInt(projectPath -> projectPath[0] + metadataFieldCount)
                         .toArray();
-        int[] systemFields =
-                SYSTEM_COLUMNS.keySet().stream()
-                        .mapToInt(systemColumn -> requiredFieldPosition(schema, systemColumn))
-                        .toArray();
 
-        return IntStream.concat(
-                        IntStream.concat(
-                                IntStream.of(hudiMetadataFields),
-                                IntStream.of(projectedDataFields)),
-                        IntStream.of(systemFields))
+        return IntStream.concat(IntStream.of(hudiMetadataFields), IntStream.of(projectedDataFields))
                 .toArray();
     }
 
@@ -216,42 +204,29 @@ private static DataType producedDataType(RowType rowType, int[] selectedFields)
                 .bridgedTo(RowData.class);
     }
 
-    private static int requiredFieldPosition(Schema schema, String fieldName) {
-        Schema.Field field = schema.getField(fieldName);
-        if (field == null) {
-            throw new IllegalArgumentException(
-                    String.format(
-                            "Required Hudi system column '%s' does not exist in Hudi schema.",
-                            fieldName));
-        }
-        return field.pos();
-    }
-
     /** Iterator for Hudi {@link RowData} as Fluss {@link LogRecord}. */
     public static class HudiRecordAsFlussRecordIterator implements CloseableIterator<LogRecord> {
 
+        /** Sentinel offset / timestamp emitted for rows read from a lake table. */
+        private static final long NO_SYSTEM_COLUMN_VALUE = -1L;
+
         private final ClosableIterator<RowData> hudiRecordIterator;
         private final ProjectedRow projectedRow;
         private final HudiRowAsFlussRow hudiRowAsFlussRow;
-        private final int logOffsetColIndex;
-        private final int timestampColIndex;
 
         private boolean closed;
 
         public HudiRecordAsFlussRecordIterator(
                 ClosableIterator<RowData> hudiRecordIterator,
                 Schema schema,
-                int metadataFieldCount,
-                int systemFieldCount) {
+                int metadataFieldCount) {
             this.hudiRecordIterator = hudiRecordIterator;
-            this.logOffsetColIndex = requiredFieldPosition(schema, OFFSET_COLUMN_NAME);
-            this.timestampColIndex = requiredFieldPosition(schema, TIMESTAMP_COLUMN_NAME);
             this.hudiRowAsFlussRow = new HudiRowAsFlussRow();
+            // FIP-27: the physical schema is metadata columns followed by user columns only; strip
+            // the Hudi metadata columns so only the business columns are emitted.
             this.projectedRow =
                     ProjectedRow.from(
-                            IntStream.range(
-                                            metadataFieldCount,
-                                            schema.getFields().size() - systemFieldCount)
+                            IntStream.range(metadataFieldCount, schema.getFields().size())
                                     .toArray());
         }
 
@@ -278,12 +253,12 @@ public boolean hasNext() {
         public LogRecord next() {
             RowData rowData = hudiRecordIterator.next();
             ChangeType changeType = toChangeType(rowData.getRowKind());
-            long offset = rowData.getLong(logOffsetColIndex);
-            long timestamp = rowData.getTimestamp(timestampColIndex, 6).getMillisecond();
-
+            // The lake table does not carry a per-record log offset / timestamp meaningful to
+            // downstream consumers, so a sentinel -1 is emitted, consistent with the Paimon and
+            // Iceberg readers and the LakeRecordRecordEmitter contract.
             return new GenericRecord(
-                    offset,
-                    timestamp,
+                    NO_SYSTEM_COLUMN_VALUE,
+                    NO_SYSTEM_COLUMN_VALUE,
                     changeType,
                     projectedRow.replaceRow(hudiRowAsFlussRow.replaceRow(rowData)));
         }

fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRow.java

@@ -31,29 +31,15 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.TimestampData;
 
-import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS;
-
 /** Wraps a Hudi/Flink {@link RowData} as a Fluss {@link InternalRow}. */
 public class HudiRowAsFlussRow implements InternalRow {
 
     private RowData rowData;
-    private final boolean stripSystemColumns;
 
-    public HudiRowAsFlussRow() {
-        this(true);
-    }
+    public HudiRowAsFlussRow() {}
 
     public HudiRowAsFlussRow(RowData rowData) {
-        this(rowData, true);
-    }
-
-    HudiRowAsFlussRow(RowData rowData, boolean stripSystemColumns) {
         this.rowData = rowData;
-        this.stripSystemColumns = stripSystemColumns;
-    }
-
-    private HudiRowAsFlussRow(boolean stripSystemColumns) {
-        this.stripSystemColumns = stripSystemColumns;
     }
 
     public HudiRowAsFlussRow replaceRow(RowData rowData) {
@@ -63,7 +49,9 @@ public HudiRowAsFlussRow replaceRow(RowData rowData) {
 
     @Override
     public int getFieldCount() {
-        return stripSystemColumns ? rowData.getArity() - SYSTEM_COLUMNS.size() : rowData.getArity();
+        // FIP-27: Hudi lake tables contain only user columns; the arity is the business field
+        // count.
+        return rowData.getArity();
     }
 
     @Override
@@ -170,6 +158,6 @@ public InternalMap getMap(int pos) {
     @Override
     public InternalRow getRow(int pos, int numFields) {
         RowData value = rowData.getRow(pos, numFields);
-        return value == null ? null : new HudiRowAsFlussRow(value, false);
+        return value == null ? null : new HudiRowAsFlussRow(value);
     }
 }

fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSortedRecordReader.java

@@ -54,7 +54,6 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
-import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS;
 import static org.apache.fluss.utils.Preconditions.checkState;
 
 /** Sorted Hudi record reader for primary key table union read. */
@@ -167,9 +166,6 @@ private static RecordKeyInfo resolveRecordKeyInfo(HudiTableInfo hudiTableInfo) {
         int userFieldPosition = 0;
         for (Schema.UnresolvedColumn column :
                 hudiTableInfo.getHudiTable().getUnresolvedSchema().getColumns()) {
-            if (SYSTEM_COLUMNS.containsKey(column.getName())) {
-                continue;
-            }
             DataType dataType = getDataType(column);
             dataTypesByName.put(column.getName(), dataType);
             userFieldPositionsByName.put(column.getName(), userFieldPosition++);

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

@@ -61,8 +61,7 @@ public RecordWriter(
             HudiWriteTableInfo hudiTableInfo,
             CkpMetadata ckpMetadata) {
         this.bucketNum = writerInitContext.tableBucket().getBucket();
-        this.flussRecordAsHudiRecord =
-                new FlussRecordAsHudiRow(bucketNum, hudiTableInfo.getRowType());
+        this.flussRecordAsHudiRecord = new FlussRecordAsHudiRow(hudiTableInfo.getRowType());
         this.hudiTableInfo = hudiTableInfo;
         this.ckpMetadata = ckpMetadata;
         this.recordWriteBuffer =

fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRow.java

@@ -19,72 +19,37 @@
 
 import org.apache.fluss.record.LogRecord;
 
-import org.apache.flink.table.data.TimestampData;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.types.RowKind;
 
-import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS;
 import static org.apache.fluss.lake.hudi.utils.HudiConversions.toRowKind;
 import static org.apache.fluss.utils.Preconditions.checkState;
 
-/** Wraps a Fluss {@link LogRecord} as a Hudi/Flink row with Fluss system columns. */
+/** Wraps a Fluss {@link LogRecord} as a Hudi/Flink row. */
 public class FlussRecordAsHudiRow extends FlussRowAsHudiRow {
 
-    private final int bucket;
+    // FIP-27: Hudi lake tables contain only user columns; no Fluss system columns are written.
+    private final int fieldCount;
 
     private LogRecord logRecord;
-    private int originRowFieldCount;
 
-    public FlussRecordAsHudiRow(int bucket, RowType rowType) {
+    public FlussRecordAsHudiRow(RowType rowType) {
         super(rowType);
-        this.bucket = bucket;
+        this.fieldCount = rowType.getFieldCount();
     }
 
     public void setFlussRecord(LogRecord logRecord) {
         this.logRecord = logRecord;
         this.internalRow = logRecord.getRow();
-        this.originRowFieldCount = internalRow.getFieldCount();
         checkState(
-                originRowFieldCount == rowType.getFieldCount() - SYSTEM_COLUMNS.size(),
-                "Hudi table field count must equal Fluss LogRecord field count plus system fields.");
+                internalRow.getFieldCount() == fieldCount,
+                "The Fluss record's field count (%s) must equal the Hudi table field count (%s).",
+                internalRow.getFieldCount(),
+                fieldCount);
     }
 
     @Override
     public RowKind getRowKind() {
         return toRowKind(logRecord.getChangeType());
     }
-
-    @Override
-    public boolean isNullAt(int pos) {
-        if (pos < originRowFieldCount) {
-            return super.isNullAt(pos);
-        }
-        return false;
-    }
-
-    @Override
-    public int getInt(int pos) {
-        if (pos == originRowFieldCount) {
-            return bucket;
-        }
-        return super.getInt(pos);
-    }
-
-    @Override
-    public long getLong(int pos) {
-        if (pos == originRowFieldCount + 1) {
-            return logRecord.logOffset();
-        } else if (pos == originRowFieldCount + 2) {
-            return logRecord.timestamp();
-        }
-        return super.getLong(pos);
-    }
-
-    @Override
-    public TimestampData getTimestamp(int pos, int precision) {
-        if (pos == originRowFieldCount + 2) {
-            return TimestampData.fromEpochMillis(logRecord.timestamp());
-        }
-        return super.getTimestamp(pos, precision);
-    }
 }

fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java

@@ -33,7 +33,6 @@
 import org.apache.flink.table.catalog.ResolvedSchema;
 import org.apache.flink.table.catalog.UniqueConstraint;
 import org.apache.flink.table.factories.FactoryUtil;
-import org.apache.flink.table.types.DataType;
 import org.apache.flink.types.RowKind;
 import org.apache.hudi.common.model.HoodieTableType;
 import org.apache.hudi.configuration.FlinkOptions;
@@ -112,7 +111,8 @@ public static ResolvedSchema convertToFlinkResolvedSchema(
 
         List<Column> columns = new ArrayList<>();
 
-        // Add regular columns
+        // FIP-27: Hudi lake tables contain only user columns; the Fluss system columns
+        // (__bucket/__offset/__timestamp) are not written to the physical schema.
         for (org.apache.fluss.metadata.Schema.Column column :
                 tableDescriptor.getSchema().getColumns()) {
             String columnName = column.getName();
@@ -133,11 +133,6 @@ public static ResolvedSchema convertToFlinkResolvedSchema(
             columns.add(Column.physical(columnName, column.getDataType().accept(converter)));
         }
 
-        // add system metadata columns to schema
-        for (Map.Entry<String, DataType> systemColumn : SYSTEM_COLUMNS.entrySet()) {
-            columns.add(Column.physical(systemColumn.getKey(), systemColumn.getValue()));
-        }
-
         UniqueConstraint constraint = null;
         // Set primary key if this is a PK table
         if (isPkTable && tableDescriptor.hasPrimaryKey()) {

fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeCatalogTest.java

@@ -562,12 +562,10 @@ void testHudiMetadataColumnPrefixConflictThrowsException() {
 
     private org.apache.flink.table.api.Schema buildExpectedHudiSchema(
             DataType idType, String primaryKeyName) {
+        // FIP-27: newly created tables are clean and carry only user columns.
         return org.apache.flink.table.api.Schema.newBuilder()
                 .column("id", idType)
                 .column("name", org.apache.flink.table.api.DataTypes.STRING())
-                .column("__bucket", org.apache.flink.table.api.DataTypes.INT())
-                .column("__offset", org.apache.flink.table.api.DataTypes.BIGINT())
-                .column("__timestamp", org.apache.flink.table.api.DataTypes.TIMESTAMP(6))
                 .primaryKeyNamed(primaryKeyName, "id")
                 .build();
     }