[core] Row lineage core support. #5935

leaves12138 · 2025-07-22T08:43:13Z

Purpose

For paimon append table (bucket = -1), support new mod, row with unique row id.

This pull request introduce a new kind of table, which based on unaware bucket append table. It writes external two column in file (_row_id BIGINT, _snapshot_version BIGINT), it follows the following rules:

If the file was firstly written. The _row_id column and _snapshot_version column is set to null. It will be allocated by committer.
If one row moved from one file to another for any reason, it should copy the _row_id value from read, to the target file.
If one row moved from one file to another for any reason, it changed , than _snapshot_version column should be set to null, otherwise, copied from the source.
The read from file which enable row_lineage, will set _row_id column and _snapshot_version column from file or metadata. (If the file target column is null, then fetch it from metadata)
Only FileKind equals APPEND (not COMPACT) file, will be assign firstRowId, the firstRowId is a long, which stored in DataFileMeta, the _row_id for the first time is set to firstRowId + offset in file.

RELATED:
https://cwiki.apache.org/confluence/pages/resumedraft.action?draftId=373886632&draftShareId=6a8aca3b-913f-4123-83e4-9106f2825736&

Tests

API and Format

Documentation

JingsongLi · 2025-07-23T11:06:20Z

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1904,6 +1904,11 @@ public InlineElement getDescription() {
                                    + "respectively. When not configured, it will automatically determine the algorithm based on the number of columns "
                                    + "in 'sink.clustering.by-columns'. 'order' is used for 1 column, 'zorder' for less than 5 columns, "
                                    + "and 'hilbert' for 5 or more columns.");
+    public static final ConfigOption<Boolean> APPEND_ROW_LINEAGE_ENABLED =
+            key("append.row.lineage.enabled")


row-tracking.enabled

You should check the table should be append table in SchemaValidation.

row-tracking.enabled

You should check the table should be append table in SchemaValidation.

OK

JingsongLi · 2025-07-23T11:08:07Z

paimon-api/src/main/java/org/apache/paimon/table/SpecialFields.java

+    public static final DataField ROW_ID =
+            new DataField(Integer.MAX_VALUE - 5, "_row_id", DataTypes.BIGINT());
+
+    public static final DataField SNAPSHOT_VERSION =


I prefer using SEQUENCE_NUMBER.

OK, then we should make SEQUENCE_NUMBER nullable

JingsongLi · 2025-07-23T11:08:27Z

paimon-api/src/main/java/org/apache/paimon/table/SpecialFields.java

@@ -87,6 +87,12 @@ public class SpecialFields {
            new DataField(
                    Integer.MAX_VALUE - 4, "rowkind", new VarCharType(VarCharType.MAX_LENGTH));

+    public static final DataField ROW_ID =
+            new DataField(Integer.MAX_VALUE - 5, "_row_id", DataTypes.BIGINT());


JingsongLi · 2025-07-23T11:09:11Z

paimon-core/src/main/java/org/apache/paimon/append/AppendCompactTask.java

@@ -52,6 +54,10 @@ public class AppendCompactTask {
    public AppendCompactTask(BinaryRow partition, List<DataFileMeta> files) {
        Preconditions.checkArgument(files != null);
        this.partition = partition;
+        files.sort(


Why add this?

I want to keep the row id in one file in order, this will may make "add new column function" more convenient.

JingsongLi · 2025-07-24T03:20:06Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

-                            new DataField(17, "_EXTERNAL_PATH", newStringType(true))));
+                            new DataField(17, "_EXTERNAL_PATH", newStringType(true)),
+                            new DataField(18, "_ROW_ID", new BigIntType(true)),
+                            new DataField(19, "_SEQUENCE_NUMBER", new BigIntType(true))));


Use min max sequence number.

JingsongLi · 2025-07-24T03:21:51Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

@@ -83,7 +83,9 @@ public class DataFileMeta {
                                    16,
                                    "_VALUE_STATS_COLS",
                                    DataTypes.ARRAY(DataTypes.STRING().notNull())),
-                            new DataField(17, "_EXTERNAL_PATH", newStringType(true))));
+                            new DataField(17, "_EXTERNAL_PATH", newStringType(true)),
+                            new DataField(18, "_ROW_ID", new BigIntType(true)),


_FIRST_ROW_ID

JingsongLi · 2025-07-25T07:59:04Z

paimon-api/src/main/java/org/apache/paimon/table/SpecialFields.java

@@ -74,7 +74,7 @@ public class SpecialFields {
    public static final int KEY_FIELD_ID_START = SYSTEM_FIELD_ID_START;

    public static final DataField SEQUENCE_NUMBER =
-            new DataField(Integer.MAX_VALUE - 1, "_SEQUENCE_NUMBER", DataTypes.BIGINT().notNull());


Just add nullable to append table.

JingsongLi · 2025-07-28T02:20:51Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

-    private final long minSequenceNumber;
-    private final long maxSequenceNumber;
+    // As for row-lineage table, this will be reassigned while committing
+    private long minSequenceNumber;


You should use copy instead making it variable.

JingsongLi · 2025-07-28T02:21:10Z

paimon-core/src/main/java/org/apache/paimon/append/AppendCompactTask.java

@@ -74,6 +78,13 @@ public CommitMessage doCompact(FileStoreTable table, BaseAppendFileStoreWrite wr
        Preconditions.checkArgument(
                dvEnabled || compactBefore.size() > 1,
                "AppendOnlyCompactionTask need more than one file input.");
+        if (table.coreOptions().rowTrackingEnabled()) {


Remove this? Let's finish compact in next PR.

JingsongLi · 2025-07-28T02:25:02Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

@@ -124,6 +126,8 @@ public class DataFileMeta {
    /** external path of file, if it is null, it is in the default warehouse path. */
    private final @Nullable String externalPath;

+    private @Nullable Long rowIdStart;


_FIRST_ROW_ID or rowIdStart?

JingsongLi · 2025-07-28T02:30:25Z

paimon-common/src/main/java/org/apache/paimon/utils/LongCounter.java


 /** An counter that sums up {@code long} values. */
-public class LongCounter implements Serializable {
+public class LongCounter implements SequenceNumberCounter {


You don't need to change this class. This class just have 50 lines, we don't need to reuse codes for it.

JingsongLi · 2025-07-28T02:31:38Z

paimon-common/src/main/java/org/apache/paimon/utils/MinSequenceCounter.java

+import org.apache.paimon.data.InternalRow;
+
+/** Sequence number counter only generate min sequence. */
+public class MinSequenceCounter implements SequenceNumberCounter {


Remove it, it just for compact.

JingsongLi · 2025-07-28T02:38:13Z

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

+        // assign row id for new files
+        long start = startRowId;
+        for (ManifestEntry entry : deltaFiles) {
+            if (entry.file().fileSource().orElse(FileSource.COMPACT).equals(FileSource.APPEND)) {


assert must have fileSource.

JingsongLi · 2025-07-28T02:39:50Z

paimon-core/src/main/java/org/apache/paimon/rowlineage/PartialMappingRow.java

+import org.apache.paimon.types.RowKind;
+
+/** Row with row lineage inject in. */
+public class PartialMappingRow implements InternalRow {


RowWithLineage

JingsongLi · 2025-07-28T02:41:43Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileRecordReader.java

@@ -81,6 +102,29 @@ public FileRecordIterator<InternalRow> readBatch() throws IOException {
                final ProjectedRow projectedRow = ProjectedRow.from(indexMapping);
                iterator = iterator.transform(projectedRow::replaceRow);
            }
+
+            if (rowLineageEnabled && !metaColumnIndex.isEmpty()) {
+                GenericRow genericRow = new GenericRow(metaColumnIndex.size());


JingsongLi · 2025-07-28T02:46:02Z

paimon-core/src/main/java/org/apache/paimon/rowlineage/PartialMappingRow.java

+
+    @Override
+    public boolean isNullAt(int pos) {
+        for (int i = 0; i < mappings.length; i++) {


Remove this loop.

JingsongLi · 2025-07-28T02:47:03Z

paimon-core/src/main/java/org/apache/paimon/table/system/RowLineageTable.java

+/** A {@link Table} for reading row id of table. */
+public class RowLineageTable implements DataTable, ReadonlyTable {
+
+    public static final String WITH_METADATA = "row_lineage";


ROW_LINEAGE

JingsongLi · 2025-07-28T07:24:36Z

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1911,6 +1911,11 @@ public InlineElement getDescription() {
                                    + "respectively. When not configured, it will automatically determine the algorithm based on the number of columns "
                                    + "in 'sink.clustering.by-columns'. 'order' is used for 1 column, 'zorder' for less than 5 columns, "
                                    + "and 'hilbert' for 5 or more columns.");
+    public static final ConfigOption<Boolean> ROW_TRACKING_ENABLED =


Keep a line above

JingsongLi · 2025-07-28T07:25:54Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

@@ -124,6 +126,8 @@ public class DataFileMeta {
    /** external path of file, if it is null, it is in the default warehouse path. */
    private final @Nullable String externalPath;

+    private @Nullable Long firstRowId;


JingsongLi · 2025-07-28T07:27:07Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

+        this.firstRowId = firstRowId;
+    }
+
+    public void setFirstRowId(@Nullable Long firstRowId) {


assignFirstRowId

JingsongLi · 2025-07-28T07:27:24Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

+                firstRowId);
+    }
+
+    public DataFileMeta copyWithMaxSequenceNumber(long maxSequenceNumber) {


assignSequenceNumber(min, max)

JingsongLi · 2025-07-28T07:27:46Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileRecordReader.java

+            if (rowLineageEnabled && !metaColumnIndex.isEmpty()) {
+                GenericRow lineageRow = new GenericRow(metaColumnIndex.size());
+
+                int[] fallbackToMetaRowLineageMappings = new int[tableRowType.getFieldCount()];


fallbackToLineageMappings

JingsongLi · 2025-07-28T07:30:49Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileRecordReader.java

+            boolean rowLineageEnabled,
+            @Nullable Long firstRowId,
+            long snapshotId,
+            Map<String, Integer> metaColumnIndex)


systemFields

JingsongLi · 2025-07-28T07:30:52Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileRecordReader.java

+    private final boolean rowLineageEnabled;
+    @Nullable private final Long firstRowId;
+    private final long snapshotId;
+    private final Map<String, Integer> metaColumnIndex;


systemFields

JingsongLi · 2025-07-28T07:32:49Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileRecordReader.java

    private final FileRecordReader<InternalRow> reader;
    @Nullable private final int[] indexMapping;
    @Nullable private final PartitionInfo partitionInfo;
    @Nullable private final CastFieldGetter[] castMapping;
+    private final boolean rowLineageEnabled;
+    @Nullable private final Long firstRowId;
+    private final long snapshotId;


maxSequenceNumber

JingsongLi · 2025-07-28T07:33:28Z

...src/test/java/org/apache/paimon/manifest/ManifestCommittableSerializerCompatibilityTest.java

@@ -76,7 +76,8 @@ public void testCompatibilityToV4CommitV7() throws IOException {
                        new byte[] {1, 2, 4},
                        FileSource.COMPACT,
                        Arrays.asList("field1", "field2", "field3"),
-                        "hdfs://localhost:9000/path/to/file");
+                        "hdfs://localhost:9000/path/to/file",
+                        null);
        List<DataFileMeta> dataFiles = Collections.singletonList(dataFile);

        LinkedHashMap<String, DeletionVectorMeta> dvMetas = new LinkedHashMap<>();


JingsongLi · 2025-07-28T07:33:48Z

paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java

@@ -426,7 +427,10 @@ private static FunctionWithIOException<DataInputView, DataFileMeta> getFileMetaS
        } else if (version == 3 || version == 4) {
            DataFileMeta10LegacySerializer serializer = new DataFileMeta10LegacySerializer();
            return serializer::deserialize;
-        } else if (version >= 5) {
+        } else if (version == 5 || version == 6) {
+            DataFileMeta12LegacySerializer serializer = new DataFileMeta12LegacySerializer();


Add test too.

JingsongLi · 2025-07-28T09:36:03Z

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

@@ -136,7 +140,8 @@ public static DataFileMeta forAppend(
            @Nullable byte[] embeddedIndex,
            @Nullable FileSource fileSource,
            @Nullable List<String> valueStatsCols,
-            @Nullable String externalPath) {
+            @Nullable String externalPath,
+            @Nullable Long rowIdStart) {


JingsongLi · 2025-07-28T09:40:47Z

paimon-core/src/main/java/org/apache/paimon/utils/FormatReaderMapping.java

@@ -165,8 +173,13 @@ public FormatReaderMapping build(

            // extract the whole data fields in logic.
            List<DataField> allDataFields = fieldsExtractor.apply(dataSchema);
+            if (CoreOptions.fromMap(dataSchema.options()).rowTrackingEnabled()) {


pass rowTrackingEnabled to this class.

JingsongLi

+1

leaves12138 force-pushed the first_row_id branch from 3e9a745 to a1af33a Compare July 22, 2025 08:45

leaves12138 changed the title ~~[WIP] [core] Row lineage core support.~~ [core] Row lineage core support. Jul 22, 2025

JingsongLi reviewed Jul 23, 2025

View reviewed changes

leaves12138 requested a review from JingsongLi July 24, 2025 03:07

JingsongLi reviewed Jul 24, 2025

View reviewed changes

leaves12138 force-pushed the first_row_id branch from 813df40 to 2eec9cc Compare July 25, 2025 06:27

JingsongLi reviewed Jul 25, 2025

View reviewed changes

JingsongLi requested changes Jul 28, 2025

View reviewed changes

leaves12138 requested a review from JingsongLi July 28, 2025 05:19

[core] Row lineage core support.

7b0f148

leaves12138 force-pushed the first_row_id branch from 6abe0be to 7b0f148 Compare July 28, 2025 05:49

leaves12138 added 7 commits July 28, 2025 14:36

Fix comment

7d63b27

Fix comment

ea9c775

Fix comment

ce61c29

Fix comment

374ba51

Fix comment

1ebf083

Fix comment

43ae655

Fix comment

a26d841

JingsongLi reviewed Jul 28, 2025

View reviewed changes

leaves12138 added 4 commits July 28, 2025 16:19

Fix comment

40f4c41

Fix comment

8c54bfd

Fix comment

a1b3bf5

Fix comment

2eaaaa3

leaves12138 requested a review from JingsongLi July 28, 2025 08:54

JingsongLi reviewed Jul 28, 2025

View reviewed changes

Fix comment

2b24f15

JingsongLi reviewed Jul 28, 2025

View reviewed changes

leaves12138 added 2 commits July 28, 2025 17:44

Fix comment

8dbb2a3

Fix comment

fb2da82

JingsongLi approved these changes Jul 28, 2025

View reviewed changes

JingsongLi merged commit 020ed14 into apache:master Jul 28, 2025
22 of 23 checks passed

leaves12138 mentioned this pull request Jul 30, 2025

[core] Introduce Row Id for append table #5795

Closed

[core] Row lineage core support. #5935

[core] Row lineage core support. #5935

Uh oh!

Conversation

leaves12138 commented Jul 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Tests

API and Format

Documentation

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

leaves12138 commented Jul 22, 2025 •

edited

Loading