From 282667b24374a3a1c13912cd1495c6866c8267d4 Mon Sep 17 00:00:00 2001 From: Ran Tao Date: Wed, 23 Sep 2026 11:34:12 +0800 Subject: [PATCH] [core] Preserve sequence-group deletes in projected reads Retain all ordering fields required by sequence groups that can delete whole rows, even when those fields are absent from the query projection. Add tests for empty, key-only and unrelated-field projections, including sequence groups with multiple ordering fields. --- .../compact/PartialUpdateMergeFunction.java | 9 +- .../PartialUpdateMergeFunctionTest.java | 145 ++++++++++++++++++ .../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 00bc83b152e6..b97d7a29dac1 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 RowType adjustReadType(RowType readType) { LinkedHashSet extraFields = new LinkedHashSet<>(); List readFieldNames = readType.getFieldNames(); - for (DataField readField : readType.getFields()) { - int index = rowType.getFieldIndex(readField.name()); + LinkedHashSet 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 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 9de9d2295731..3c7eca4e7281 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.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 void testAdjustProjectionSequenceFieldsProject() { 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 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 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 factory = + PartialUpdateMergeFunction.factory(options, rowType, ImmutableList.of("id")); + RowType readType = + factory.adjustReadType( + rowType.project( + selectedFields.isEmpty() + ? new String[0] + : selectedFields.split(","))); + MergeFunction 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 void testInitRowWithNullableFieldOnDelete() { validate(func, 1, 2, 2, null); } + private void assertProjectedDelete( + MergeFunction 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 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 300c6b7b2d9c..fcbc0390c530 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)) + } }