You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/08/17 03:32:32 UTC
(fluss) branch main updated: [lake/paimon] Support clean and legacy Paimon lake table schemas (#3982)
This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new cc0228904 [lake/paimon] Support clean and legacy Paimon lake table schemas (#3982)
cc0228904 is described below
commit cc022890417931dc11cef81b574b17295718372a
Author: fhan <aa...@gmail.com>
AuthorDate: Mon Aug 17 11:32:27 2026 +0800
[lake/paimon] Support clean and legacy Paimon lake table schemas (#3982)
---------
Co-authored-by: luoyuxia <lu...@alumni.sjtu.edu.cn>
---
.../fluss/lake/paimon/PaimonLakeCatalog.java | 17 +-
.../paimon/lookup/PaimonLakeTableLookuper.java | 4 +-
.../lake/paimon/source/PaimonRecordReader.java | 55 +++--
.../paimon/tiering/FlussRecordAsPaimonRow.java | 36 ++-
.../lake/paimon/tiering/PaimonLakeWriter.java | 12 +-
.../fluss/lake/paimon/tiering/RecordWriter.java | 6 +-
.../tiering/append/AppendOnlyArrowBatchHelper.java | 15 +-
.../paimon/tiering/append/AppendOnlyWriter.java | 15 +-
.../paimon/tiering/mergetree/MergeTreeWriter.java | 24 +-
.../fluss/lake/paimon/utils/PaimonConversions.java | 52 +++--
.../lake/paimon/utils/PaimonRowAsFlussRow.java | 4 +-
.../lake/paimon/utils/PaimonTableValidation.java | 46 ++++
.../fluss/lake/paimon/utils/PaimonUtils.java | 40 ++++
.../lake/paimon/LakeEnabledTableCreateITCase.java | 259 +++++++--------------
.../fluss/lake/paimon/PaimonLakeCatalogTest.java | 20 +-
.../flink/FlinkUnionReadPrimaryKeyTableITCase.java | 51 +++-
.../lake/paimon/testutils/PaimonTestUtils.java | 53 +++++
.../paimon/tiering/FlussRecordAsPaimonRowTest.java | 74 ++++--
.../lake/paimon/tiering/PaimonTieringITCase.java | 66 +++++-
.../server/utils/TableDescriptorValidation.java | 1 +
.../lake/paimon/SparkLakePaimonCatalogTest.scala | 59 +----
21 files changed, 561 insertions(+), 348 deletions(-)
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java
index bd846cec2..3982e7474 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java
+++ b/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 class PaimonLakeCatalog implements LakeCatalog {
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 ->
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java
index 18afe2d98..9e04ce15e 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java
@@ -68,7 +68,7 @@ import java.util.List;
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 @@ public class PaimonLakeTableLookuper implements LakeTableLookuper {
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);
}
}
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonRecordReader.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonRecordReader.java
index 3af7467cf..dd2fca065 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonRecordReader.java
+++ b/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.io.IOException;
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 class PaimonRecordReader implements RecordReader {
@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 class PaimonRecordReader implements RecordReader {
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 class PaimonRecordReader implements RecordReader {
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 class PaimonRecordReader implements RecordReader {
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)));
}
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java
index bc0303010..78ddf3045 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java
@@ -25,7 +25,7 @@ import org.apache.paimon.data.Timestamp;
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;
@@ -33,6 +33,7 @@ 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 class FlussRecordAsPaimonRow extends FlussRowAsPaimonRow {
@Override
public int getInt(int pos) {
- if (pos == bucketFieldIndex) {
+ if (paimonIncludingSystemColumns && pos == bucketFieldIndex) {
// bucket system column
return bucket;
}
@@ -114,12 +124,14 @@ public class FlussRecordAsPaimonRow extends FlussRowAsPaimonRow {
@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 class FlussRecordAsPaimonRow extends FlussRowAsPaimonRow {
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) {
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java
index 2a38e388a..9b82403b6 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java
+++ b/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.ArrowRecordBatch;
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,6 +59,11 @@ public class PaimonLakeWriter implements LakeWriter<PaimonWriteResult>, Supports
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(
@@ -65,14 +71,16 @@ public class PaimonLakeWriter implements LakeWriter<PaimonWriteResult>, Supports
writerInitContext.tableBucket(),
writerInitContext.partition(),
partitionKeys,
- flussRowType)
+ flussRowType,
+ paimonIncludingSystemColumns)
: new MergeTreeWriter(
fileStoreTable,
writerInitContext.tableBucket(),
writerInitContext.partition(),
partitionKeys,
flussRowType,
- writerInitContext.ioTmpDirs());
+ writerInitContext.ioTmpDirs(),
+ paimonIncludingSystemColumns);
}
@Override
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java
index 173407bb9..2260d553b 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java
@@ -49,7 +49,8 @@ public abstract class RecordWriter<T> implements AutoCloseable {
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 abstract class RecordWriter<T> implements AutoCloseable {
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;
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyArrowBatchHelper.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyArrowBatchHelper.java
index 4fdb8bce7..a19c2628b 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyArrowBatchHelper.java
+++ b/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 @@ class AppendOnlyArrowBatchHelper implements AutoCloseable {
}
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();
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java
index 0caaed97c..23f61a331 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java
+++ b/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 class AppendOnlyWriter extends RecordWriter<InternalRow> {
tableBucket,
partition,
partitionKeys,
- flussRowType);
+ flussRowType,
+ paimonIncludingSystemColumns);
this.fileStoreTable = fileStoreTable;
+ this.paimonIncludingSystemColumns = paimonIncludingSystemColumns;
}
@Override
@@ -90,7 +95,11 @@ public class AppendOnlyWriter extends RecordWriter<InternalRow> {
if (arrowBatchHelper == null) {
helper =
new AppendOnlyArrowBatchHelper(
- fileStoreTable, tableWrite, tableRowType, bucket);
+ fileStoreTable,
+ tableWrite,
+ tableRowType,
+ bucket,
+ paimonIncludingSystemColumns);
arrowBatchHelper = helper;
} else {
helper = (AppendOnlyArrowBatchHelper) arrowBatchHelper;
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java
index b3e1ecdee..37aeef7af 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java
@@ -49,8 +49,16 @@ public class MergeTreeWriter extends RecordWriter<KeyValue> {
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 class MergeTreeWriter extends RecordWriter<KeyValue> {
@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 class MergeTreeWriter extends RecordWriter<KeyValue> {
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;
}
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
index becf8a2f0..cf3fc0348 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
@@ -37,6 +37,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.Table;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowKind;
@@ -50,7 +51,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.function.Function;
-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.utils.Preconditions.checkState;
/** Utils for conversion between Paimon and Fluss. */
@@ -172,7 +173,11 @@ public class PaimonConversions {
return partitionExtractor.apply(new FlussRowAsPaimonRow(partitionRow, paimonRowType));
}
- public static List<SchemaChange> toPaimonSchemaChanges(List<TableChange> tableChanges) {
+ public static List<SchemaChange> toPaimonSchemaChanges(
+ Table paimonTable, List<TableChange> tableChanges) {
+ // A legacy table (created before FIP-27) still carries the three trailing system columns,
+ // recognisable by the presence of the __timestamp column. A clean table has none of them.
+ boolean paimonIncludingSystemColumns = PaimonUtils.isLegacyTable(paimonTable.rowType());
List<SchemaChange> schemaChanges = new ArrayList<>(tableChanges.size());
for (TableChange tableChange : tableChanges) {
@@ -189,6 +194,13 @@ public class PaimonConversions {
} else if (tableChange instanceof TableChange.AddColumn) {
TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange;
+ if (LEGACY_SYSTEM_COLUMNS.containsKey(addColumn.getName())) {
+ throw new InvalidTableException(
+ "Column "
+ + addColumn.getName()
+ + " conflicts with a system column name of paimon table, please rename the column.");
+ }
+
if (!(addColumn.getPosition() instanceof TableChange.Last)) {
throw new UnsupportedOperationException(
"Only support to add column at last for paimon table.");
@@ -203,14 +215,27 @@ public class PaimonConversions {
org.apache.paimon.types.DataType paimonDataType =
flussDataType.accept(FlussDataTypeToPaimonDataType.INSTANCE);
- String firstSystemColumnName = SYSTEM_COLUMNS.keySet().iterator().next();
- schemaChanges.add(
- SchemaChange.addColumn(
- addColumn.getName(),
- paimonDataType,
- addColumn.getComment(),
- SchemaChange.Move.before(
- addColumn.getName(), firstSystemColumnName)));
+ if (paimonIncludingSystemColumns) {
+ // Legacy tables keep the three system columns as the last physical columns, so
+ // a new business column must be inserted right before the first system column.
+ String firstSystemColumnName = LEGACY_SYSTEM_COLUMNS.keySet().iterator().next();
+ schemaChanges.add(
+ SchemaChange.addColumn(
+ addColumn.getName(),
+ paimonDataType,
+ addColumn.getComment(),
+ SchemaChange.Move.before(
+ addColumn.getName(), firstSystemColumnName)));
+ } else {
+ // Clean tables have no trailing system columns, so a new business column is
+ // simply appended at the end.
+ schemaChanges.add(
+ SchemaChange.addColumn(
+ addColumn.getName(),
+ paimonDataType,
+ addColumn.getComment(),
+ SchemaChange.Move.last(addColumn.getName())));
+ }
} else {
throw new UnsupportedOperationException(
"Unsupported table change: " + tableChange.getClass());
@@ -252,7 +277,7 @@ public class PaimonConversions {
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 InvalidTableException(
"Column "
+ columnName
@@ -264,11 +289,6 @@ public class PaimonConversions {
column.getComment().orElse(null));
}
- // add system metadata columns to schema
- for (Map.Entry<String, DataType> systemColumn : SYSTEM_COLUMNS.entrySet()) {
- schemaBuilder.column(systemColumn.getKey(), systemColumn.getValue());
- }
-
// set pk
if (tableDescriptor.hasPrimaryKey()) {
schemaBuilder.primaryKey(
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonRowAsFlussRow.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonRowAsFlussRow.java
index fe956561a..5fa5a2fee 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonRowAsFlussRow.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonRowAsFlussRow.java
@@ -27,8 +27,6 @@ import org.apache.fluss.row.TimestampNtz;
import org.apache.paimon.data.Timestamp;
-import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS;
-
/** Adapter for paimon row as fluss row. */
public class PaimonRowAsFlussRow implements InternalRow {
@@ -47,7 +45,7 @@ public class PaimonRowAsFlussRow implements InternalRow {
@Override
public int getFieldCount() {
- return paimonRow.getFieldCount() - SYSTEM_COLUMNS.size();
+ return paimonRow.getFieldCount();
}
@Override
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonTableValidation.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonTableValidation.java
index 0b8f56052..c6254084c 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonTableValidation.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonTableValidation.java
@@ -32,6 +32,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
+import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS;
import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PAIMON_UNSETTABLE_OPTIONS;
import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY;
import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
@@ -40,6 +41,15 @@ import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
public class PaimonTableValidation {
public static boolean isPaimonSchemaCompatible(Schema existingSchema, Schema newSchema) {
+ // FIP-27: newly generated schemas are always clean (no system columns). When the existing
+ // table is a legacy table that still carries the three system columns (recognisable by the
+ // presence of the __timestamp column), re-enabling lake tiering must keep that physical
+ // layout. Enrich the clean new schema with the trailing system columns before comparison,
+ // so an existing legacy table is recognised as compatible and its layout is preserved.
+ if (PaimonUtils.isLegacyTable(existingSchema.rowType())) {
+ newSchema = appendSystemColumns(newSchema);
+ }
+
if (!equalPhysicalSchema(existingSchema, newSchema)) {
// Allow different precisions for __timestamp column for backward compatibility,
// old cluster will use precision 6, but new cluster will use precision 3,
@@ -50,6 +60,32 @@ public class PaimonTableValidation {
return true;
}
+ /**
+ * Returns a copy of {@code cleanSchema} with the three Fluss system columns appended as the
+ * last physical columns, so a clean schema generated by the current cluster can be compared
+ * against an existing legacy table. Partition keys, primary keys, options and comment are
+ * preserved. System-column field ids continue the existing id sequence to avoid collisions;
+ * they are irrelevant to {@link #equalPhysicalSchema}, which ignores field ids.
+ */
+ private static Schema appendSystemColumns(Schema cleanSchema) {
+ List<DataField> fields = new ArrayList<>(cleanSchema.fields());
+ int nextFieldId = 0;
+ for (DataField field : fields) {
+ nextFieldId = Math.max(nextFieldId, field.id() + 1);
+ }
+ for (Map.Entry<String, org.apache.paimon.types.DataType> systemColumn :
+ LEGACY_SYSTEM_COLUMNS.entrySet()) {
+ fields.add(
+ new DataField(nextFieldId++, systemColumn.getKey(), systemColumn.getValue()));
+ }
+ return new Schema(
+ fields,
+ cleanSchema.partitionKeys(),
+ cleanSchema.primaryKeys(),
+ cleanSchema.options(),
+ cleanSchema.comment());
+ }
+
/**
* Check if the {@code existingSchema} is compatible with {@code newSchema} by ignoring the
* precision difference of the system column {@code __timestamp}.
@@ -67,6 +103,16 @@ public class PaimonTableValidation {
public static boolean equalIgnoreSystemColumnTimestampPrecision(
Schema existingSchema, Schema newSchema) {
List<DataField> existingFields = new ArrayList<>(existingSchema.fields());
+ // The precision relaxation only applies to a legacy table's trailing __timestamp system
+ // column. A clean table (or any table whose last column is not __timestamp) has no such
+ // column, so there is nothing to relax and we compare the schemas directly.
+ if (existingFields.isEmpty()
+ || !existingFields
+ .get(existingFields.size() - 1)
+ .name()
+ .equals(TIMESTAMP_COLUMN_NAME)) {
+ return equalPhysicalSchema(existingSchema, newSchema);
+ }
DataField systemTimestampField = existingFields.get(existingFields.size() - 1);
if (systemTimestampField.name().equals(TIMESTAMP_COLUMN_NAME)
&& systemTimestampField
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonUtils.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonUtils.java
new file mode 100644
index 000000000..6b58f3f2b
--- /dev/null
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonUtils.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.paimon.utils;
+
+import org.apache.paimon.types.RowType;
+
+import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
+
+/** Common utilities for the Paimon lake integration. */
+public final class PaimonUtils {
+
+ private PaimonUtils() {}
+
+ /**
+ * Returns whether the physical Paimon table uses the legacy layout, i.e. it still carries the
+ * three trailing Fluss system columns ({@code __bucket}, {@code __offset}, {@code
+ * __timestamp}).
+ *
+ * <p>A legacy table is identified by the presence of the {@code __timestamp} system column;
+ * under FIP-27 newly created tables are clean and carry none of these columns.
+ */
+ public static boolean isLegacyTable(RowType tableRowType) {
+ return tableRowType.getFieldIndex(TIMESTAMP_COLUMN_NAME) >= 0;
+ }
+}
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
index f1d75d278..0a0c07d97 100644
--- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
@@ -44,7 +44,6 @@ import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.data.Timestamp;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.FileStoreTable;
@@ -74,6 +73,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.stream.Stream;
+import static org.apache.fluss.lake.paimon.testutils.PaimonTestUtils.adjustToLegacyV1Table;
import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PAIMON_UNSETTABLE_OPTIONS;
import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
@@ -172,19 +172,9 @@ class LakeEnabledTableCreateITCase {
RowType.of(
new DataType[] {
org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.STRING(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.STRING()
},
- new String[] {
- "log_c1",
- "log_c2",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
- }),
+ new String[] {"log_c1", "log_c2"}),
"log_c1,log_c2",
BUCKET_NUM);
@@ -210,19 +200,9 @@ class LakeEnabledTableCreateITCase {
RowType.of(
new DataType[] {
org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.STRING(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.STRING()
},
- new String[] {
- "log_c1",
- "log_c2",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
- }),
+ new String[] {"log_c1", "log_c2"}),
null,
BUCKET_NUM);
@@ -249,19 +229,9 @@ class LakeEnabledTableCreateITCase {
RowType.of(
new DataType[] {
org.apache.paimon.types.DataTypes.INT().notNull(),
- org.apache.paimon.types.DataTypes.STRING(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.STRING()
},
- new String[] {
- "pk_c1",
- "pk_c2",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
- }),
+ new String[] {"pk_c1", "pk_c2"}),
"pk_c1",
BUCKET_NUM);
@@ -292,20 +262,9 @@ class LakeEnabledTableCreateITCase {
new DataType[] {
org.apache.paimon.types.DataTypes.INT().notNull(),
org.apache.paimon.types.DataTypes.STRING(),
- org.apache.paimon.types.DataTypes.STRING().notNull(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.STRING().notNull()
},
- new String[] {
- "c1",
- "c2",
- "c3",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
- }),
+ new String[] {"c1", "c2", "c3"}),
"c1",
BUCKET_NUM);
}
@@ -359,32 +318,12 @@ class LakeEnabledTableCreateITCase {
org.apache.paimon.types.DataTypes.DATE(),
org.apache.paimon.types.DataTypes.TIME(),
org.apache.paimon.types.DataTypes.TIMESTAMP(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()
},
new String[] {
- "log_c1",
- "log_c2",
- "log_c3",
- "log_c4",
- "log_c5",
- "log_c6",
- "log_c7",
- "log_c8",
- "log_c9",
- "log_c10",
- "log_c11",
- "log_c12",
- "log_c13",
- "log_c14",
- "log_c15",
- "log_c16",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
+ "log_c1", "log_c2", "log_c3", "log_c4", "log_c5", "log_c6", "log_c7",
+ "log_c8", "log_c9", "log_c10", "log_c11", "log_c12", "log_c13",
+ "log_c14", "log_c15", "log_c16"
}),
null,
BUCKET_NUM);
@@ -599,10 +538,9 @@ class LakeEnabledTableCreateITCase {
.hasMessageContaining(
"The table `fluss`.`log_table_with_exist_lake_table` already exists in Paimon catalog, but the table schema is not compatible.")
.hasMessageContaining(
- "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE]")
+ "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT]")
.hasMessageContaining("options={bucket=-1")
- .hasMessageContaining(
- "new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE]")
+ .hasMessageContaining("new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT]")
.hasMessageContaining("options={bucket=3")
.hasMessageContaining("bucket-key=c1,c2")
.hasMessageEndingWith(
@@ -624,9 +562,9 @@ class LakeEnabledTableCreateITCase {
.hasMessageContaining(
"The table `fluss`.`log_table_with_exist_lake_table` already exists in Paimon catalog, but the table schema is not compatible.")
.hasMessageContaining(
- "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE]")
+ "Existing schema: UpdateSchema{fields=[`c1` STRING, `c2` INT]")
.hasMessageContaining(
- "new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `c3` STRING, `__bucket` INT, `__offset` BIGINT, `__timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE]")
+ "new schema: UpdateSchema{fields=[`c1` STRING, `c2` INT, `c3` STRING]")
.hasMessageEndingWith(
"Please first drop the table in Paimon catalog or use a new table name.");
@@ -781,19 +719,9 @@ class LakeEnabledTableCreateITCase {
RowType.of(
new DataType[] {
org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.STRING(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.STRING()
},
- new String[] {
- "log_c1",
- "log_c2",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
- }),
+ new String[] {"log_c1", "log_c2"}),
"log_c1,log_c2",
BUCKET_NUM);
@@ -888,19 +816,9 @@ class LakeEnabledTableCreateITCase {
RowType.of(
new DataType[] {
org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.STRING(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.STRING()
},
- new String[] {
- "c1",
- "c2",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
- }),
+ new String[] {"c1", "c2"}),
"c1,c2",
BUCKET_NUM);
@@ -919,19 +837,9 @@ class LakeEnabledTableCreateITCase {
RowType.of(
new DataType[] {
org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.STRING(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.STRING()
},
- new String[] {
- "c1",
- "c2",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
- }),
+ new String[] {"c1", "c2"}),
"c1,c2",
BUCKET_NUM);
@@ -966,19 +874,9 @@ class LakeEnabledTableCreateITCase {
RowType.of(
new DataType[] {
org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.STRING(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.STRING()
},
- new String[] {
- "c1",
- "c2",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
- }),
+ new String[] {"c1", "c2"}),
"c1,c2",
BUCKET_NUM);
@@ -1047,19 +945,9 @@ class LakeEnabledTableCreateITCase {
RowType.of(
new DataType[] {
org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.STRING(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.STRING()
},
- new String[] {
- "c1",
- "c2",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
- }),
+ new String[] {"c1", "c2"}),
"c1,c2",
BUCKET_NUM);
@@ -1100,19 +988,9 @@ class LakeEnabledTableCreateITCase {
RowType.of(
new DataType[] {
org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.STRING(),
- // for __bucket, __offset, __timestamp
- org.apache.paimon.types.DataTypes.INT(),
- org.apache.paimon.types.DataTypes.BIGINT(),
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()
+ org.apache.paimon.types.DataTypes.STRING()
},
- new String[] {
- "c1",
- "c2",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME
- }),
+ new String[] {"c1", "c2"}),
"c1,c2",
BUCKET_NUM);
@@ -1131,15 +1009,8 @@ class LakeEnabledTableCreateITCase {
paimonCatalog.getTable(Identifier.create(DATABASE, tablePath.getTableName()));
// Verify the new column c3 with comment was added to Paimon table
RowType alteredRowType = alteredPaimonTable.rowType();
- assertThat(alteredRowType.getFieldCount()).isEqualTo(6);
- assertThat(alteredRowType.getFieldNames())
- .containsExactly(
- "c1",
- "c2",
- "c3",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME);
+ assertThat(alteredRowType.getFieldCount()).isEqualTo(3);
+ assertThat(alteredRowType.getFieldNames()).containsExactly("c1", "c2", "c3");
// Verify c3 column has the correct type and comment
assertThat(alteredRowType.getField("c3").type())
.isEqualTo(org.apache.paimon.types.DataTypes.INT());
@@ -1159,12 +1030,21 @@ class LakeEnabledTableCreateITCase {
Identifier paimonIdentifier = Identifier.create(DATABASE, tablePath.getTableName());
- // alter to TIMESTAMP_WITH_LOCAL_TIME_ZONE to mock the legacy behavior
+ // FIP-27: a newly created table is clean (no system columns). To exercise the legacy
+ // compatibility path, first turn it into a legacy table by appending the three trailing
+ // system columns, using TIMESTAMP_WITH_LOCAL_TIME_ZONE for __timestamp to mock the
+ // precision-6 layout written by an old cluster.
paimonCatalog.alterTable(
paimonIdentifier,
- SchemaChange.updateColumnType(
- TIMESTAMP_COLUMN_NAME,
- org.apache.paimon.types.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()),
+ Arrays.asList(
+ SchemaChange.addColumn(
+ BUCKET_COLUMN_NAME, org.apache.paimon.types.DataTypes.INT()),
+ SchemaChange.addColumn(
+ OFFSET_COLUMN_NAME, org.apache.paimon.types.DataTypes.BIGINT()),
+ SchemaChange.addColumn(
+ TIMESTAMP_COLUMN_NAME,
+ org.apache.paimon.types.DataTypes
+ .TIMESTAMP_WITH_LOCAL_TIME_ZONE())),
false);
// disable data lake
@@ -1190,6 +1070,49 @@ class LakeEnabledTableCreateITCase {
.isTrue();
}
+ @Test
+ void testAddColumnForLegacyTableWithSystemColumns() throws Exception {
+ TablePath tablePath = TablePath.of(DATABASE, "legacy_add_column_table");
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("c1", DataTypes.INT())
+ .column("c2", DataTypes.STRING())
+ .build())
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .build();
+ admin.createTable(tablePath, tableDescriptor, false).get();
+
+ // turn the freshly created clean table into a legacy table carrying the three system
+ // columns
+ adjustToLegacyV1Table(tablePath, paimonCatalog);
+
+ // adding a business column to a legacy table must still work, and the new column must be
+ // inserted before the trailing system columns so the legacy physical layout is preserved.
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "c3",
+ DataTypes.INT(),
+ "c3 comment",
+ TableChange.ColumnPosition.last())),
+ false)
+ .get();
+
+ Identifier identifier = Identifier.create(DATABASE, tablePath.getTableName());
+ RowType rowType = paimonCatalog.getTable(identifier).rowType();
+ assertThat(rowType.getFieldNames())
+ .containsExactly(
+ "c1",
+ "c2",
+ "c3",
+ BUCKET_COLUMN_NAME,
+ OFFSET_COLUMN_NAME,
+ TIMESTAMP_COLUMN_NAME);
+ }
+
@Test
void testCreatePaimonDvTableWithNonStringPartitionColumn() throws Exception {
TablePath tablePath = TablePath.of(DATABASE, "invalid_dv_table");
@@ -1310,13 +1233,7 @@ class LakeEnabledTableCreateITCase {
BatchTableCommit commit = writeBuilder.newCommit()) {
for (int i = 0; i < 10; i++) {
- GenericRow row =
- GenericRow.of(
- i,
- BinaryString.fromString("row-" + i),
- 0,
- (long) i,
- Timestamp.fromEpochMillis(System.currentTimeMillis()));
+ GenericRow row = GenericRow.of(i, BinaryString.fromString("row-" + i));
write.write(row);
}
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java
index 1683533e5..1e67a92a2 100644
--- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java
@@ -160,15 +160,7 @@ class PaimonLakeCatalogTest {
Table table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier);
assertThat(table.rowType().getFieldNames())
- .containsSequence(
- "id",
- "name",
- "amount",
- "address",
- "new_col",
- "__bucket",
- "__offset",
- "__timestamp");
+ .containsSequence("id", "name", "amount", "address", "new_col");
}
@Test
@@ -205,15 +197,7 @@ class PaimonLakeCatalogTest {
assertThat(((FileStoreTable) table).schema().toSchema().comment()).isEqualTo("");
assertThat(table.options().get("fluss.key")).isEqualTo("value");
assertThat(table.rowType().getFieldNames())
- .containsSequence(
- "id",
- "name",
- "amount",
- "address",
- "is_direct_play",
- "__bucket",
- "__offset",
- "__timestamp");
+ .containsSequence("id", "name", "amount", "address", "is_direct_play");
}
@Test
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java
index 81fa9b050..9efa832e2 100644
--- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java
@@ -70,6 +70,7 @@ import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.as
import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertRowResultsIgnoreOrder;
import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectBatchRows;
import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectRowsWithTimeout;
+import static org.apache.fluss.lake.paimon.testutils.PaimonTestUtils.adjustToLegacyV1Table;
import static org.apache.fluss.testutils.DataTestUtils.row;
import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
import static org.assertj.core.api.Assertions.assertThat;
@@ -123,7 +124,9 @@ class FlinkUnionReadPrimaryKeyTableITCase extends FlinkUnionReadTestBase {
CollectionUtil.iteratorToList(tableResult.collect()).stream()
.map(
row -> {
- int userColumnCount = row.getArity() - 3;
+ // FIP-27: a clean lake table exposes only user columns via
+ // $lake, so there are no trailing system columns to strip.
+ int userColumnCount = row.getArity();
Object[] fields = new Object[userColumnCount];
for (int i = 0; i < userColumnCount; i++) {
fields[i] = row.getField(i);
@@ -277,7 +280,9 @@ class FlinkUnionReadPrimaryKeyTableITCase extends FlinkUnionReadTestBase {
.stream()
.map(
row -> {
- int columnCount = row.getArity() - 3;
+ // FIP-27: a clean lake table exposes only user columns via
+ // $lake, so there are no trailing system columns to strip.
+ int columnCount = row.getArity();
Object[] fields = new Object[columnCount];
for (int i = 0; i < columnCount; i++) {
fields[i] = row.getField(i);
@@ -546,6 +551,48 @@ class FlinkUnionReadPrimaryKeyTableITCase extends FlinkUnionReadTestBase {
.isEqualTo("[+I[0, v01, v02], +I[1, v11, v12], +I[2, v21, v22], +I[3, v31, v32]]");
}
+ @Test
+ void testUnionReadLegacyTable() throws Exception {
+ // FIP-27: union read (lake snapshot merged with fresh Fluss log) must still work for a
+ // legacy table that carries the three trailing system columns.
+ String tableName = "pk_table_union_read_legacy";
+ TablePath t1 = TablePath.of(DEFAULT_DB, tableName);
+ int bucketNum = 3;
+ long tableId = createSimplePkTable(t1, bucketNum, false, true);
+
+ // turn the freshly created clean table into a legacy table before any tiering happens
+ adjustToLegacyV1Table(t1, paimonCatalog);
+
+ // tier an initial version of every row into the lake
+ JobClient jobClient = buildTieringJob(execEnv);
+ List<InternalRow> rows = new ArrayList<>();
+ Map<TableBucket, Long> bucketLogEndOffset = new HashMap<>();
+ for (int i = 0; i < bucketNum; i++) {
+ rows.add(
+ GenericRow.of(
+ i, BinaryString.fromString("v1_1"), BinaryString.fromString("v1_2")));
+ bucketLogEndOffset.put(new TableBucket(tableId, i), 1L);
+ }
+ writeRows(t1, rows, false);
+ waitUntilBucketsSynced(bucketLogEndOffset.keySet());
+ assertReplicaStatus(bucketLogEndOffset);
+
+ // stop tiering, then overwrite every row so the fresh values only exist in the Fluss log
+ jobClient.cancel().get();
+ rows.clear();
+ for (int i = 0; i < bucketNum; i++) {
+ rows.add(
+ GenericRow.of(
+ i, BinaryString.fromString("v2_1"), BinaryString.fromString("v2_2")));
+ }
+ writeRows(t1, rows, false);
+
+ // union read must merge the lake snapshot with the fresh log and return the latest values
+ List<String> result = toSortedRows(batchTEnv.executeSql("select * from " + tableName));
+ assertThat(result.toString())
+ .isEqualTo("[+I[0, v2_1, v2_2], +I[1, v2_1, v2_2], +I[2, v2_1, v2_2]]");
+ }
+
@ParameterizedTest
@ValueSource(booleans = {false, true})
void testUnionReadInStreamMode(Boolean isPartitioned) throws Exception {
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/PaimonTestUtils.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/PaimonTestUtils.java
new file mode 100644
index 000000000..590ff4886
--- /dev/null
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/PaimonTestUtils.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.paimon.testutils;
+
+import org.apache.fluss.metadata.TablePath;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.types.DataTypes;
+
+import java.util.Arrays;
+
+import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
+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;
+
+/** The utils for paimon testing. */
+public class PaimonTestUtils {
+
+ /**
+ * Adjusts the Paimon table to a legacy table by appending the three trailing Fluss system
+ * columns ({@code __bucket}, {@code __offset}, {@code __timestamp}). This simulates a table
+ * created before FIP-27, when the lake table always carried these system columns, so tests can
+ * verify that legacy tables remain readable and writable.
+ */
+ public static void adjustToLegacyV1Table(TablePath tablePath, Catalog paimonCatalog)
+ throws Exception {
+ paimonCatalog.alterTable(
+ toPaimon(tablePath),
+ Arrays.asList(
+ SchemaChange.addColumn(BUCKET_COLUMN_NAME, DataTypes.INT()),
+ SchemaChange.addColumn(OFFSET_COLUMN_NAME, DataTypes.BIGINT()),
+ SchemaChange.addColumn(
+ TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP_LTZ_MILLIS())),
+ false);
+ }
+}
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java
index e085b694b..891ce12c5 100644
--- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java
@@ -49,6 +49,38 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link FlussRecordAsPaimonRow}. */
class FlussRecordAsPaimonRowTest {
+ @Test
+ void testCleanTableWritesNoSystemColumns() {
+ // A clean table (FIP-27 default) contains only user columns; the writer must not emit any
+ // of the three system columns.
+ int tableBucket = 3;
+ RowType tableRowType =
+ RowType.of(
+ new org.apache.paimon.types.IntType(),
+ new org.apache.paimon.types.VarCharType());
+
+ FlussRecordAsPaimonRow flussRecordAsPaimonRow =
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, false);
+
+ long logOffset = 7L;
+ long timeStamp = System.currentTimeMillis();
+ GenericRow genericRow = new GenericRow(2);
+ genericRow.setField(0, 1);
+ genericRow.setField(1, BinaryString.fromString("v1"));
+ LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
+ flussRecordAsPaimonRow.setFlussRecord(logRecord);
+
+ // exposes exactly the two business columns, no trailing system columns
+ assertThat(flussRecordAsPaimonRow.getFieldCount()).isEqualTo(2);
+ assertThat(flussRecordAsPaimonRow.getInt(0)).isEqualTo(1);
+ assertThat(flussRecordAsPaimonRow.getString(1).toString()).isEqualTo("v1");
+
+ // the no-arg-layout constructor defaults to the clean layout
+ FlussRecordAsPaimonRow defaultRow = new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ defaultRow.setFlussRecord(logRecord);
+ assertThat(defaultRow.getFieldCount()).isEqualTo(2);
+ }
+
@Test
void testLogTableRecordAllTypes() {
// Construct a FlussRecordAsPaimonRow instance
@@ -75,7 +107,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(14);
@@ -143,7 +175,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.BigIntType(),
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -185,7 +217,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 10;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(2);
@@ -224,7 +256,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 5;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -259,7 +291,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -313,7 +345,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(7);
@@ -386,7 +418,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -420,7 +452,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -454,7 +486,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -483,7 +515,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -508,7 +540,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -537,7 +569,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -564,7 +596,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 7L;
long timeStamp = System.currentTimeMillis();
@@ -595,7 +627,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 7L;
long timeStamp = System.currentTimeMillis();
@@ -690,7 +722,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(8);
@@ -918,7 +950,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, nestedMapRowType);
+ new FlussRecordAsPaimonRow(tableBucket, nestedMapRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow nestedMapGenericRow = new GenericRow(1);
@@ -971,7 +1003,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.BigIntType(),
new org.apache.paimon.types.LocalZonedTimestampType(3));
- FlussRecordAsPaimonRow flussRow = new FlussRecordAsPaimonRow(tableBucket, rowType);
+ FlussRecordAsPaimonRow flussRow = new FlussRecordAsPaimonRow(tableBucket, rowType, true);
GenericRow genericRow = new GenericRow(1);
genericRow.setField(0, new GenericMap(mapData));
LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
@@ -997,7 +1029,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -1023,7 +1055,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -1058,7 +1090,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.LocalZonedTimestampType(3));
FlussRecordAsPaimonRow flussRecordAsPaimonRow =
- new FlussRecordAsPaimonRow(tableBucket, tableRowType);
+ new FlussRecordAsPaimonRow(tableBucket, tableRowType, true);
long logOffset = 0;
long timeStamp = System.currentTimeMillis();
GenericRow genericRow = new GenericRow(1);
@@ -1082,7 +1114,7 @@ class FlussRecordAsPaimonRowTest {
new org.apache.paimon.types.IntType(),
new org.apache.paimon.types.BigIntType(),
new org.apache.paimon.types.LocalZonedTimestampType(3));
- FlussRecordAsPaimonRow row = new FlussRecordAsPaimonRow(0, rowType);
+ FlussRecordAsPaimonRow row = new FlussRecordAsPaimonRow(0, rowType, true);
assertThatThrownBy(row::getRowKind)
.isInstanceOf(IllegalStateException.class)
.hasMessageContaining(expectedMsg);
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java
index 3d4da4fe5..6d035aec0 100644
--- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java
@@ -67,6 +67,7 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
+import static org.apache.fluss.lake.paimon.testutils.PaimonTestUtils.adjustToLegacyV1Table;
import static org.apache.fluss.testutils.DataTestUtils.row;
import static org.assertj.core.api.Assertions.assertThat;
@@ -188,6 +189,37 @@ class PaimonTieringITCase extends FlinkPaimonTieringTestBase {
}
}
+ @Test
+ void testTieringForLegacyTable() throws Exception {
+ // FIP-27: verify that tiering still works for a legacy table that carries the three
+ // trailing
+ // system columns (__bucket, __offset, __timestamp).
+ TablePath t = TablePath.of(DEFAULT_DB, "legacyLogTable");
+ long tId = createLogTable(t);
+
+ // turn the freshly created clean table into a legacy table before writing/tiering
+ adjustToLegacyV1Table(t, paimonCatalog);
+
+ TableBucket tBucket = new TableBucket(tId, 0);
+ List<InternalRow> flussRows = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ List<InternalRow> rows = Arrays.asList(row(1, "v1"), row(2, "v2"), row(3, "v3"));
+ flussRows.addAll(rows);
+ writeRows(t, rows, true);
+ }
+
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ assertReplicaStatus(tBucket, 30);
+ assertThat(getLeaderReplica(tBucket).getLogTablet().getLakeMaxTimestamp())
+ .isGreaterThan(-1);
+ // the legacy table keeps the system columns, and the tiered __offset must be correct
+ checkDataInPaimonLegacyAppendOnlyTable(t, flussRows, 0);
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
private static Stream<Arguments> tieringAllTypesWriteArgs() {
return Stream.of(Arguments.of(true), Arguments.of(false));
}
@@ -504,7 +536,28 @@ class PaimonTieringITCase extends FlinkPaimonTieringTestBase {
InternalRow flussRow = flussRowIterator.next();
assertThat(row.getInt(0)).isEqualTo(flussRow.getInt(0));
assertThat(row.getString(1).toString()).isEqualTo(flussRow.getString(1).toString());
- // system columns are always the last three: __bucket, __offset, __timestamp
+ // FIP-27: a clean lake table only stores user columns, so the Paimon row has exactly
+ // the same field count as the expected Fluss row (no trailing
+ // __bucket/__offset/__timestamp columns).
+ assertThat(row.getFieldCount()).isEqualTo(flussRow.getFieldCount());
+ }
+ assertThat(flussRowIterator.hasNext()).isFalse();
+ }
+
+ private void checkDataInPaimonLegacyAppendOnlyTable(
+ TablePath tablePath, List<InternalRow> expectedRows, long startingOffset)
+ throws Exception {
+ Iterator<org.apache.paimon.data.InternalRow> paimonRowIterator =
+ getPaimonRowCloseableIterator(tablePath);
+ Iterator<InternalRow> flussRowIterator = expectedRows.iterator();
+ while (paimonRowIterator.hasNext()) {
+ org.apache.paimon.data.InternalRow row = paimonRowIterator.next();
+ InternalRow flussRow = flussRowIterator.next();
+ assertThat(row.getInt(0)).isEqualTo(flussRow.getInt(0));
+ assertThat(row.getString(1).toString()).isEqualTo(flussRow.getString(1).toString());
+ // A legacy table keeps the two business columns plus the three trailing system columns
+ // __bucket/__offset/__timestamp, so the tiered offset is stored at fieldCount - 2.
+ assertThat(row.getFieldCount()).isEqualTo(flussRow.getFieldCount() + 3);
int offsetIndex = row.getFieldCount() - 2;
assertThat(row.getLong(offsetIndex)).isEqualTo(startingOffset++);
}
@@ -526,8 +579,10 @@ class PaimonTieringITCase extends FlinkPaimonTieringTestBase {
assertThat(row.getInt(0)).isEqualTo(flussRow.getInt(0));
assertThat(row.getString(1).toString()).isEqualTo(flussRow.getString(1).toString());
assertThat(row.getString(2).toString()).isEqualTo(flussRow.getString(2).toString());
- // the idx 3 is __bucket, so use 4
- assertThat(row.getLong(4)).isEqualTo(startingOffset++);
+ // FIP-27: a clean lake table only stores user columns, so the Paimon row has exactly
+ // the same field count as the expected Fluss row (no trailing
+ // __bucket/__offset/__timestamp columns).
+ assertThat(row.getFieldCount()).isEqualTo(flussRow.getFieldCount());
}
assertThat(flussRowIterator.hasNext()).isFalse();
}
@@ -594,9 +649,8 @@ class PaimonTieringITCase extends FlinkPaimonTieringTestBase {
FileStoreTable paimonTable = (FileStoreTable) paimonCatalog.getTable(tableIdentifier);
List<String> fieldNames = paimonTable.rowType().getFieldNames();
- // Should have exact fields in order: a, b, c3, __bucket, __offset, __timestamp
- assertThat(fieldNames)
- .containsExactly("a", "b", "c3", "__bucket", "__offset", "__timestamp");
+ // FIP-27: a clean lake table only stores user columns, in order: a, b, c3.
+ assertThat(fieldNames).containsExactly("a", "b", "c3");
// 9. Verify both schema evolution and data correctness
// For initial rows (before ADD COLUMN), c3 should be NULL
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
index e8ba9714d..e898420e1 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
@@ -137,6 +137,7 @@ public class TableDescriptorValidation {
/** Validates the schema after altering table columns. */
@Internal
public static void validateAlterTableSchema(TableInfo table, Schema newSchema) {
+ checkSystemColumns(newSchema.getRowType());
if (table.getTableConfig()
.getMergeEngineType()
.map(MergeEngineType.AGGREGATION::equals)
diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/paimon/SparkLakePaimonCatalogTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/paimon/SparkLakePaimonCatalogTest.scala
index fe5d3eb5d..c08fe5aa7 100644
--- a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/paimon/SparkLakePaimonCatalogTest.scala
+++ b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/paimon/SparkLakePaimonCatalogTest.scala
@@ -21,7 +21,6 @@ import org.apache.fluss.config.{ConfigOptions, Configuration, FlussConfigUtils}
import org.apache.fluss.exception.FlussRuntimeException
import org.apache.fluss.lake.paimon.utils.PaimonConversions
import org.apache.fluss.metadata._
-import org.apache.fluss.metadata.TableDescriptor.{BUCKET_COLUMN_NAME, OFFSET_COLUMN_NAME, TIMESTAMP_COLUMN_NAME}
import org.apache.fluss.server.utils.LakeStorageUtils
import org.apache.fluss.spark.SparkCatalogTest
import org.apache.fluss.spark.SparkConnectorOptions.{BUCKET_KEY, BUCKET_NUMBER, PRIMARY_KEY}
@@ -154,12 +153,9 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest {
.of(
Array.apply(
org.apache.paimon.types.DataTypes.INT,
- org.apache.paimon.types.DataTypes.STRING,
- org.apache.paimon.types.DataTypes.INT,
- org.apache.paimon.types.DataTypes.BIGINT,
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS
+ org.apache.paimon.types.DataTypes.STRING
),
- Array.apply("id", "name", BUCKET_COLUMN_NAME, OFFSET_COLUMN_NAME, TIMESTAMP_COLUMN_NAME)
+ Array.apply("id", "name")
),
"id",
2
@@ -187,12 +183,9 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest {
.of(
Array.apply(
org.apache.paimon.types.DataTypes.INT,
- org.apache.paimon.types.DataTypes.STRING,
- org.apache.paimon.types.DataTypes.INT,
- org.apache.paimon.types.DataTypes.BIGINT,
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS
+ org.apache.paimon.types.DataTypes.STRING
),
- Array.apply("id", "name", BUCKET_COLUMN_NAME, OFFSET_COLUMN_NAME, TIMESTAMP_COLUMN_NAME)
+ Array.apply("id", "name")
),
null,
2
@@ -222,18 +215,9 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest {
Array.apply(
org.apache.paimon.types.DataTypes.INT.notNull(),
org.apache.paimon.types.DataTypes.STRING,
- org.apache.paimon.types.DataTypes.STRING.notNull(),
- org.apache.paimon.types.DataTypes.INT,
- org.apache.paimon.types.DataTypes.BIGINT,
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS
+ org.apache.paimon.types.DataTypes.STRING.notNull()
),
- Array.apply(
- "id",
- "name",
- "pk1",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME)
+ Array.apply("id", "name", "pk1")
),
"id",
2
@@ -265,19 +249,9 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest {
org.apache.paimon.types.DataTypes.INT.notNull(),
org.apache.paimon.types.DataTypes.STRING,
org.apache.paimon.types.DataTypes.STRING.notNull(),
- org.apache.paimon.types.DataTypes.STRING.notNull(),
- org.apache.paimon.types.DataTypes.INT,
- org.apache.paimon.types.DataTypes.BIGINT,
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS
+ org.apache.paimon.types.DataTypes.STRING.notNull()
),
- Array.apply(
- "id",
- "name",
- "pk1",
- "pt1",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME)
+ Array.apply("id", "name", "pk1", "pt1")
),
"id",
2
@@ -328,10 +302,7 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest {
),
org.apache.paimon.types.DataTypes.MAP(
org.apache.paimon.types.DataTypes.STRING.notNull(),
- org.apache.paimon.types.DataTypes.INT),
- org.apache.paimon.types.DataTypes.INT,
- org.apache.paimon.types.DataTypes.BIGINT,
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS
+ org.apache.paimon.types.DataTypes.INT)
),
Array.apply(
"c1",
@@ -349,10 +320,7 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest {
"c13",
"c14",
"c15",
- "c16",
- BUCKET_COLUMN_NAME,
- OFFSET_COLUMN_NAME,
- TIMESTAMP_COLUMN_NAME)
+ "c16")
),
null,
2
@@ -402,12 +370,9 @@ class SparkLakePaimonCatalogTest extends SparkCatalogTest {
.of(
Array.apply(
org.apache.paimon.types.DataTypes.INT,
- org.apache.paimon.types.DataTypes.STRING,
- org.apache.paimon.types.DataTypes.INT,
- org.apache.paimon.types.DataTypes.BIGINT,
- org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS
+ org.apache.paimon.types.DataTypes.STRING
),
- Array.apply("id", "name", BUCKET_COLUMN_NAME, OFFSET_COLUMN_NAME, TIMESTAMP_COLUMN_NAME)
+ Array.apply("id", "name")
),
null,
2