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 dfa2c0ca4f [core] Preserve sequence-group deletes in projected reads
(#10135)
dfa2c0ca4f is described below
commit dfa2c0ca4f7609b18785f216b3aa6db6875ab9aa
Author: Ran Tao <[email protected]>
AuthorDate: Thu Sep 24 10:58:30 2026 +0800
[core] Preserve sequence-group deletes in projected reads (#10135)
---
.../compact/PartialUpdateMergeFunction.java | 9 +-
.../compact/PartialUpdateMergeFunctionTest.java | 145 +++++++++++++++++++++
.../paimon/spark/sql/DeleteFromTableTestBase.scala | 47 +++++++
3 files changed, 199 insertions(+), 2 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
index 00bc83b152..b97d7a29da 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -621,8 +621,13 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
LinkedHashSet<DataField> extraFields = new LinkedHashSet<>();
List<String> readFieldNames = readType.getFieldNames();
- for (DataField readField : readType.getFields()) {
- int index = rowType.getFieldIndex(readField.name());
+ LinkedHashSet<Integer> requiredFields =
+ readFieldNames.stream()
+ .map(rowType::getFieldIndex)
+
.collect(Collectors.toCollection(LinkedHashSet::new));
+ // These groups determine whether the whole row exists, even for
an empty projection.
+
sequenceGroupPartialDelete.stream().sorted().forEach(requiredFields::add);
+ for (int index : requiredFields) {
Supplier<FieldsComparator> comparatorSupplier =
fieldSeqComparators.get(index);
if (comparatorSupplier == null) {
continue;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
index 9de9d22957..3c7eca4e72 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
@@ -26,10 +26,13 @@ import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import static org.apache.paimon.CoreOptions.FIELDS_DEFAULT_AGG_FUNC;
import static
org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
@@ -424,6 +427,122 @@ public class PartialUpdateMergeFunctionTest {
validate(func, 1, 1, 1, 2, 2);
}
+ @ParameterizedTest
+ @ValueSource(strings = {"", "id", "v1", "seq1", "v2", "seq2", "extra",
"v2,id"})
+ public void testSequenceGroupDeleteWithProjection(String selectedFields) {
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("v1", DataTypes.INT())
+ .field("seq1", DataTypes.INT())
+ .field("v2", DataTypes.INT())
+ .field("seq2", DataTypes.INT())
+ .field("extra", DataTypes.INT())
+ .build();
+ Options options = new Options();
+ options.set("fields.seq1.sequence-group", "v1");
+ options.set("fields.seq2.sequence-group", "v2");
+ options.set("partial-update.remove-record-on-sequence-group", "seq2");
+ MergeFunctionFactory<KeyValue> factory =
+ PartialUpdateMergeFunction.factory(options, rowType,
ImmutableList.of("id"));
+ // An empty selection models COUNT(*); other selections cover each
field and reordering.
+ RowType readType =
+ factory.adjustReadType(
+ rowType.project(
+ selectedFields.isEmpty()
+ ? new String[0]
+ : selectedFields.split(",")));
+ MergeFunction<KeyValue> func = factory.create(readType);
+ ProjectedRow projection = ProjectedRow.from(readType, rowType);
+ GenericRow inserted = GenericRow.of(1, 10, 1, 20, 1, 100);
+
+ // A newer or equal seq2 deletes the whole row, even if seq2 was not
selected.
+ assertProjectedDelete(
+ func,
+ projection,
+ inserted,
+ GenericRow.of(1, null, null, null, 2, null),
+ RowKind.DELETE);
+ assertProjectedDelete(
+ func,
+ projection,
+ inserted,
+ GenericRow.of(1, null, null, null, 1, null),
+ RowKind.DELETE);
+ // An older seq2 must not delete the row.
+ assertProjectedDelete(
+ func,
+ projection,
+ inserted,
+ GenericRow.of(1, null, null, null, 0, null),
+ RowKind.INSERT);
+ // A null seq2 skips the group and must not delete the row.
+ assertProjectedDelete(
+ func,
+ projection,
+ inserted,
+ GenericRow.of(1, null, null, null, null, null),
+ RowKind.INSERT);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", "id", "v1", "seq1", "v2", "seq2", "subSeq2",
"v2,id"})
+ public void testMultiSequenceFieldsDeleteWithProjection(String
selectedFields) {
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("v1", DataTypes.INT())
+ .field("seq1", DataTypes.INT())
+ .field("v2", DataTypes.INT())
+ .field("seq2", DataTypes.INT())
+ .field("subSeq2", DataTypes.INT())
+ .build();
+ Options options = new Options();
+ options.set("fields.seq1.sequence-group", "v1");
+ options.set("fields.seq2,subSeq2.sequence-group", "v2");
+ options.set("partial-update.remove-record-on-sequence-group", "seq2");
+ MergeFunctionFactory<KeyValue> factory =
+ PartialUpdateMergeFunction.factory(options, rowType,
ImmutableList.of("id"));
+ RowType readType =
+ factory.adjustReadType(
+ rowType.project(
+ selectedFields.isEmpty()
+ ? new String[0]
+ : selectedFields.split(",")));
+ MergeFunction<KeyValue> func = factory.create(readType);
+ ProjectedRow projection = ProjectedRow.from(readType, rowType);
+ GenericRow inserted = GenericRow.of(1, 10, 1, 20, 1, 1);
+
+ // seq2 stays equal: subSeq2 must be read to recognize the newer
delete.
+ assertProjectedDelete(
+ func,
+ projection,
+ inserted,
+ GenericRow.of(1, null, null, null, 1, 2),
+ RowKind.DELETE);
+ // Equal composite sequences also allow deletion.
+ assertProjectedDelete(
+ func,
+ projection,
+ inserted,
+ GenericRow.of(1, null, null, null, 1, 1),
+ RowKind.DELETE);
+ // The older subSeq2 must prevent deletion, despite the equal seq2.
+ assertProjectedDelete(
+ func,
+ projection,
+ inserted,
+ GenericRow.of(1, null, null, null, 1, 0),
+ RowKind.INSERT);
+ // A fully null sequence group must not delete the row.
+ assertProjectedDelete(
+ func,
+ projection,
+ inserted,
+ GenericRow.of(1, null, null, null, null, null),
+ RowKind.INSERT);
+ }
+
@Test
public void testMultiSequenceFieldsAdjustProjectionProject() {
Options options = new Options();
@@ -1035,6 +1154,32 @@ public class PartialUpdateMergeFunctionTest {
validate(func, 1, 2, 2, null);
}
+ private void assertProjectedDelete(
+ MergeFunction<KeyValue> function,
+ ProjectedRow projection,
+ GenericRow inserted,
+ GenericRow deleted,
+ RowKind expectedKind) {
+ function.reset();
+ function.add(
+ new KeyValue()
+ .replace(
+ GenericRow.of(1),
+ sequence++,
+ RowKind.INSERT,
+ projection.replaceRow(inserted)));
+ function.add(
+ new KeyValue()
+ .replace(
+ GenericRow.of(1),
+ sequence++,
+ RowKind.DELETE,
+ projection.replaceRow(deleted)));
+ assertThat(function.getResult().valueKind())
+ .as("delete record %s", deleted)
+ .isEqualTo(expectedKind);
+ }
+
private void add(MergeFunction<KeyValue> function, Integer... f) {
add(function, RowKind.INSERT, f);
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
index 300c6b7b2d..fcbc0390c5 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
@@ -20,8 +20,10 @@ package org.apache.paimon.spark.sql
import org.apache.paimon.{CoreOptions, Snapshot}
import org.apache.paimon.CoreOptions.MergeEngine
+import org.apache.paimon.data.GenericRow
import org.apache.paimon.spark.PaimonSparkTestBase
import org.apache.paimon.spark.catalyst.analysis.Delete
+import org.apache.paimon.types.RowKind
import org.apache.spark.sql.Row
import org.assertj.core.api.Assertions.{assertThat, assertThatThrownBy}
@@ -614,4 +616,49 @@ abstract class DeleteFromTableTestBase extends
PaimonSparkTestBase {
}
}
}
+
+ test("Paimon Delete: sequence-group delete records with projected reads") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, v1 INT, seq1 INT, v2 INT, seq2 INT)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'id',
+ | 'bucket' = '1',
+ | 'merge-engine' = 'partial-update',
+ | 'fields.seq1.sequence-group' = 'v1',
+ | 'fields.seq2.sequence-group' = 'v2',
+ | 'partial-update.remove-record-on-sequence-group' = 'seq2',
+ | 'write-only' = 'true')
+ |""".stripMargin)
+
+ // mock two streams updating independent field groups for the same primary
keys.
+ spark.sql("INSERT INTO T VALUES (1, 10, 1, NULL, NULL), (2, 20, 1, NULL,
NULL)")
+ spark.sql("INSERT INTO T VALUES (1, NULL, NULL, 100, 1), (2, NULL, NULL,
200, 1)")
+ checkAnswer(spark.sql("SELECT * FROM T"), Seq(Row(1, 10, 1, 100, 1),
Row(2, 20, 1, 200, 1)))
+
+ // A newer seq2 triggers whole-row deletion; seq1=NULL skips the first
group.
+ // Use the Paimon write API because SQL DELETE rewrites files for this
configuration.
+ val builder = loadTable("T").newBatchWriteBuilder()
+ val write = builder.newWrite()
+ val commit = builder.newCommit()
+ try {
+ val delete = GenericRow.of(1, null, null, null, 2)
+ delete.setRowKind(RowKind.DELETE)
+ write.write(delete)
+ commit.commit(write.prepareCommit())
+ } finally {
+ write.close()
+ commit.close()
+ }
+
+ // Keep both streams' insert files and the delete file separate for
merging during reads.
+ checkAnswer(spark.sql("SELECT COUNT(*) FROM `T$files`"), Row(3L))
+
+ // Reading all columns retains seq2, so merging removes id=1 and combines
both groups for id=2.
+ checkAnswer(spark.sql("SELECT * FROM T"), Row(2, 20, 1, 200, 1))
+
+ // seq2 controls row existence even when only the key, another group, or
no columns are read.
+ checkAnswer(spark.sql("SELECT COUNT(*) FROM T"), Row(1L))
+ checkAnswer(spark.sql("SELECT id FROM T"), Row(2))
+ checkAnswer(spark.sql("SELECT v1 FROM T"), Row(20))
+ }
}
