[core] Preserve sequence-group deletes in projected reads

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java

@@ -621,8 +621,13 @@ public RowType adjustReadType(RowType readType) {
 
             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;

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<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 void testInitRowWithNullableFieldOnDelete() {
         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);
     }

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