This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 022ad03332 [core] Fix incorrect sequenceNumber in manifest after
row-tracking compaction (#7409)
022ad03332 is described below
commit 022ad03332bb3842b5c76d3732e91c2ce01fce79
Author: Juntao Zhang <[email protected]>
AuthorDate: Fri Jun 19 21:24:35 2026 +0800
[core] Fix incorrect sequenceNumber in manifest after row-tracking
compaction (#7409)
---
.../io/RowDataFileSequenceNumberTracker.java | 102 +++++++++++++++++++++
.../org/apache/paimon/io/RowDataFileWriter.java | 12 ++-
.../operation/commit/RowTrackingCommitUtils.java | 13 ++-
.../paimon/spark/sql/RowTrackingTestBase.scala | 31 +++++++
4 files changed, 152 insertions(+), 6 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileSequenceNumberTracker.java
b/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileSequenceNumberTracker.java
new file mode 100644
index 0000000000..0372c7f885
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileSequenceNumberTracker.java
@@ -0,0 +1,102 @@
+/*
+ * 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.paimon.io;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.operation.commit.RowTrackingCommitUtils;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.LongCounter;
+
+import java.util.function.Supplier;
+
+/**
+ * Tracks sequence number range for rows written to a data file.
+ *
+ * <p>Two modes of operation:
+ *
+ * <ul>
+ * <li><b>Non-row-tracking mode</b> (no {@code _SEQUENCE_NUMBER} field in
schema): sequence
+ * numbers are generated monotonically by a {@link LongCounter}. min =
counter - rowCount, max
+ * = counter - 1.
+ * <li><b>Row-tracking mode</b> (schema contains {@code _SEQUENCE_NUMBER}):
sequence numbers are
+ * extracted from each row. min/max are computed from actual row values.
Null sequence numbers
+ * are tracked and cause max() to return 0, which signals {@link
+ * RowTrackingCommitUtils#assignSnapshotId} to assign the current
snapshot ID.
+ * </ul>
+ */
+public class RowDataFileSequenceNumberTracker {
+ private final LongCounter seqNumCounter;
+ private final Supplier<Long> recordCountSupplier;
+ private final int seqNumberFieldIndex;
+ private long minSeqNumber;
+ private long maxSeqNumber;
+ private boolean hasNullSeqNumber;
+
+ public RowDataFileSequenceNumberTracker(
+ RowType writeSchema,
+ Supplier<LongCounter> seqNumCounterSupplier,
+ Supplier<Long> recordCountSupplier) {
+ this.seqNumCounter = seqNumCounterSupplier.get();
+ this.recordCountSupplier = recordCountSupplier;
+ this.seqNumberFieldIndex =
writeSchema.getFieldIndex(SpecialFields.SEQUENCE_NUMBER.name());
+ this.minSeqNumber = Long.MAX_VALUE;
+ this.maxSeqNumber = Long.MIN_VALUE;
+ this.hasNullSeqNumber = false;
+ }
+
+ /** Returns the minimum sequence number for the file. */
+ public long min() {
+ if (seqNumberFieldIndex == -1) {
+ return seqNumCounter.getValue() - recordCountSupplier.get();
+ }
+ // minSeqNumber stays at Long.MAX_VALUE when all records have null
sequence numbers.
+ // Returning 0 triggers RowTrackingCommitUtils.assignSnapshotId() to
use snapshot ID.
+ return minSeqNumber == Long.MAX_VALUE ? 0 : minSeqNumber;
+ }
+
+ /** Returns the maximum sequence number for the file. */
+ public long max() {
+ if (seqNumberFieldIndex == -1) {
+ return seqNumCounter.getValue() - 1;
+ }
+ // When hasNullSeqNumber is true, some records have null sequence
numbers.
+ // Returning 0 triggers RowTrackingCommitUtils.assignSnapshotId() to
use snapshot ID for
+ // max.
+ return hasNullSeqNumber ? 0 : maxSeqNumber;
+ }
+
+ /**
+ * Processes one row: increments the sequence counter, in row-tracking
mode, extracts and tracks
+ * the sequence number from the row.
+ */
+ public void update(InternalRow row) {
+ seqNumCounter.add(1L);
+
+ if (seqNumberFieldIndex != -1 && !row.isNullAt(seqNumberFieldIndex)) {
+ // If sequence number field exists, extract min/max from row data
+ long seqNum = row.getLong(seqNumberFieldIndex);
+ minSeqNumber = Math.min(minSeqNumber, seqNum);
+ maxSeqNumber = Math.max(maxSeqNumber, seqNum);
+ } else if (seqNumberFieldIndex != -1) {
+ // Manifest will calculate the correct max based on snapshot id
+ hasNullSeqNumber = true;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java
index 7f8715ab08..aa3b203588 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java
@@ -46,12 +46,12 @@ import static
org.apache.paimon.io.DataFilePathFactory.dataFileToFileIndexPath;
public class RowDataFileWriter extends
StatsCollectingSingleFileWriter<InternalRow, DataFileMeta> {
private final long schemaId;
- private final LongCounter seqNumCounter;
private final boolean isExternalPath;
private final SimpleStatsConverter statsArraySerializer;
@Nullable private final DataFileIndexWriter dataFileIndexWriter;
private final FileSource fileSource;
@Nullable private final List<String> writeCols;
+ private final RowDataFileSequenceNumberTracker sequenceNumberTracker;
public RowDataFileWriter(
FileIO fileIO,
@@ -68,7 +68,6 @@ public class RowDataFileWriter extends
StatsCollectingSingleFileWriter<InternalR
@Nullable List<String> writeCols) {
super(fileIO, context, path, Function.identity(), writeSchema,
asyncFileWrite);
this.schemaId = schemaId;
- this.seqNumCounter = seqNumCounterSupplier.get();
this.isExternalPath = isExternalPath;
this.statsArraySerializer = new SimpleStatsConverter(writeSchema,
statsDenseStore);
this.dataFileIndexWriter =
@@ -76,6 +75,9 @@ public class RowDataFileWriter extends
StatsCollectingSingleFileWriter<InternalR
fileIO, dataFileToFileIndexPath(path), writeSchema,
fileIndexOptions);
this.fileSource = fileSource;
this.writeCols = writeCols;
+ this.sequenceNumberTracker =
+ new RowDataFileSequenceNumberTracker(
+ writeSchema, seqNumCounterSupplier,
super::recordCount);
}
@Override
@@ -85,7 +87,7 @@ public class RowDataFileWriter extends
StatsCollectingSingleFileWriter<InternalR
if (dataFileIndexWriter != null) {
dataFileIndexWriter.write(row);
}
- seqNumCounter.add(1L);
+ sequenceNumberTracker.update(row);
}
@Override
@@ -111,8 +113,8 @@ public class RowDataFileWriter extends
StatsCollectingSingleFileWriter<InternalR
fileSize,
recordCount(),
statsPair.getRight(),
- seqNumCounter.getValue() - super.recordCount(),
- seqNumCounter.getValue() - 1,
+ sequenceNumberTracker.min(),
+ sequenceNumberTracker.max(),
schemaId,
indexResult.independentIndexFile() == null
? Collections.emptyList()
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowTrackingCommitUtils.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowTrackingCommitUtils.java
index ad8533f894..f6e0b660ce 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowTrackingCommitUtils.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowTrackingCommitUtils.java
@@ -50,9 +50,20 @@ public class RowTrackingCommitUtils {
private static void assignSnapshotId(
long snapshotId, List<ManifestEntry> deltaFiles,
List<ManifestEntry> snapshotAssigned) {
for (ManifestEntry entry : deltaFiles) {
- if (entry.file().minSequenceNumber() == 0L) {
+ long minSeqNumber = entry.file().minSequenceNumber();
+ long maxSeqNumber = entry.file().maxSequenceNumber();
+ if (minSeqNumber == 0L) {
+ // Case 1: New file (e.g., from INSERT)
+ // All records in this file get the current snapshot ID as
sequence number
snapshotAssigned.add(entry.assignSequenceNumber(snapshotId,
snapshotId));
+ } else if (maxSeqNumber == 0L) {
+ // Case 2: File with some modified records
+ // - min: Preserve original sequence number (from unmodified
records)
+ // - max: Assign current snapshot ID
+ snapshotAssigned.add(entry.assignSequenceNumber(minSeqNumber,
snapshotId));
} else {
+ // Case 3: Pure compact file (no modified records)
+ // Preserve original min/max sequence numbers from source files
snapshotAssigned.add(entry);
}
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
index 1da318e282..5efcaf79b8 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
@@ -303,6 +303,37 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
sql("SELECT *, _ROW_ID, _SEQUENCE_NUMBER FROM t ORDER BY id"),
Seq(Row(1, 1, 0, 1), Row(2, 2, 1, 2), Row(3, 3, 2, 3))
)
+
+ sql("INSERT INTO t VALUES (4, '4')")
+ sql("INSERT INTO t VALUES (5, '5')")
+ // snapshot 7: should merge files with sequence numbers [1, 6]
+ sql("CALL sys.compact(table => 't')")
+ checkAnswer(
+ sql("SELECT min_sequence_number, max_sequence_number FROM `t$files`"),
+ Seq(Row(1, 6))
+ )
+ // snapshot 8: Updated record has null sequence number
+ sql("UPDATE t SET data = 22 WHERE id = 2")
+
+ // snapshot 9 ~ 10: add new file, and set sequence number to null
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS data FROM
range(6, 8)")
+ sql("UPDATE t SET data = 67 WHERE _SEQUENCE_NUMBER = 9")
+ checkAnswer(
+ sql(
+ "SELECT min_sequence_number, max_sequence_number FROM `t$files`
order by min_sequence_number"),
+ Seq(Row(1, 8), Row(10, 10))
+ )
+ checkAnswer(
+ sql("SELECT *, _ROW_ID, _SEQUENCE_NUMBER FROM t ORDER BY id"),
+ Seq(
+ Row(1, 1, 0, 1),
+ Row(2, 22, 1, 8),
+ Row(3, 3, 2, 3),
+ Row(4, 4, 3, 5),
+ Row(5, 5, 4, 6),
+ Row(6, 67, 5, 10),
+ Row(7, 67, 6, 10))
+ )
}
}
