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)) + } }