Add tests: SPIP matrix coverage, combined post-processing, composite rowId

gengliangwang · gengliangwang · commit e2f65a20f986 · 2026-05-01T00:41:06.000Z
- ChangelogEndToEndSuite: 3 new e2e tests covering missing SPIP matrix cells
  ((true, false) DELETE, (true, true) without computeUpdates), and combined
  netChanges + carry-over removal post-processing path.
- ResolveChangelogTableStreamingPostProcessingSuite: assert grouping
  attribute on netChanges-alone test; add composite rowId plan-shape test.
- Remove stale "deduplicationMode=netChanges is rejected on streaming" test
  (netChanges is now supported on streaming) and the now-unused
  AnalysisException import.

Co-authored-by: Isaac
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogEndToEndSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogEndToEndSuite.scala
@@ -874,4 +874,149 @@ class ChangelogEndToEndSuite extends SharedSparkSession {
       }
     }
   }
+
+  test("streaming netChanges emits DELETE for pre-existing row deleted in range") {
+    // Exercises the SPIP `(true, false)` cell: existedBefore = true (first event is a
+    // delete or update_preimage), existsAfter = false (last event is a delete).
+    val id = recreateWithRowVersion()
+    catalog.setChangelogProperties(id, ChangelogProperties(
+      containsIntermediateChanges = true,
+      rowIdNames = Seq("id"),
+      rowVersionName = Some("row_commit_version")))
+
+    catalog.addChangeRows(id, Seq(
+      // v1: pre-existing Alice gets updated to Bob.
+      ppRow(1L, "Alice", 1L, CHANGE_TYPE_UPDATE_PREIMAGE,  1L, 1000000L),
+      ppRow(1L, "Bob",   1L, CHANGE_TYPE_UPDATE_POSTIMAGE, 1L, 1000000L),
+      // v2: Bob deleted -- the v1 preimage is the first event and the v2 delete is
+      // the last event for row identity 1 across the entire range.
+      ppRow(1L, "Bob",   1L, CHANGE_TYPE_DELETE, 2L, 2000000L),
+      // v3: insert Carol -- gives the watermark something to advance past, so row
+      // identity 1's timer fires before end-of-input.
+      ppRow(2L, "Carol", 3L, CHANGE_TYPE_INSERT, 3L, 3000000L)))
+
+    withSQLConf(rocksDbProviderConf) {
+      val q = spark.readStream
+        .option("startingVersion", "1")
+        .option("deduplicationMode", "netChanges")
+        .changes(fullTableName)
+        .select("id", "data", "_change_type")
+        .writeStream
+        .format("memory")
+        .queryName("cdc_stream_netchanges_delete")
+        .outputMode("append")
+        .start()
+      try {
+        q.processAllAvailable()
+        checkAnswer(
+          spark.sql("SELECT * FROM cdc_stream_netchanges_delete"),
+          Seq(
+            // (true, false): emit a single DELETE carrying the *first* event's data
+            // (the preimage), per the batch contract.
+            Row(1L, "Alice", CHANGE_TYPE_DELETE),
+            Row(2L, "Carol", CHANGE_TYPE_INSERT)))
+      } finally {
+        q.stop()
+      }
+    }
+  }
+
+  test("streaming netChanges without computeUpdates keeps persisting rows as DELETE+INSERT") {
+    // Exercises the SPIP `(true, true)` cell with computeUpdates = false: the pair is
+    // emitted as DELETE + INSERT rather than relabeled as
+    // update_preimage + update_postimage.
+    val id = recreateWithRowVersion()
+    catalog.setChangelogProperties(id, ChangelogProperties(
+      containsIntermediateChanges = true,
+      rowIdNames = Seq("id"),
+      rowVersionName = Some("row_commit_version")))
+
+    catalog.addChangeRows(id, Seq(
+      // v1: pre-existing Alice updated to Bob.
+      ppRow(1L, "Alice", 1L, CHANGE_TYPE_UPDATE_PREIMAGE,  1L, 1000000L),
+      ppRow(1L, "Bob",   1L, CHANGE_TYPE_UPDATE_POSTIMAGE, 1L, 1000000L),
+      // v2: Bob updated to Robert -- row identity 1 spans (preimage Alice) ..
+      // (postimage Robert).
+      ppRow(1L, "Bob",    1L, CHANGE_TYPE_UPDATE_PREIMAGE,  2L, 2000000L),
+      ppRow(1L, "Robert", 2L, CHANGE_TYPE_UPDATE_POSTIMAGE, 2L, 2000000L)))
+
+    withSQLConf(rocksDbProviderConf) {
+      val q = spark.readStream
+        .option("startingVersion", "1")
+        .option("deduplicationMode", "netChanges")
+        // computeUpdates defaults to false.
+        .changes(fullTableName)
+        .select("id", "data", "_change_type")
+        .writeStream
+        .format("memory")
+        .queryName("cdc_stream_netchanges_no_compute_updates")
+        .outputMode("append")
+        .start()
+      try {
+        q.processAllAvailable()
+        checkAnswer(
+          spark.sql("SELECT * FROM cdc_stream_netchanges_no_compute_updates"),
+          Seq(
+            Row(1L, "Alice",  CHANGE_TYPE_DELETE),
+            Row(1L, "Robert", CHANGE_TYPE_INSERT)))
+      } finally {
+        q.stop()
+      }
+    }
+  }
+
+  test("streaming netChanges + carry-over removal: combined post-processing") {
+    // Validates the design point that the row-level rewrite and the netChanges rewrite
+    // share a single EventTimeWatermark on `_commit_timestamp` and produce the
+    // expected combined result. Carry-over CoW pairs are dropped before the netChanges
+    // collapse runs, so the final emission only reflects real content changes.
+    val id = recreateWithRowVersion()
+    catalog.setChangelogProperties(id, ChangelogProperties(
+      containsCarryoverRows = true,
+      containsIntermediateChanges = true,
+      rowIdNames = Seq("id"),
+      rowVersionName = Some("row_commit_version")))
+
+    catalog.addChangeRows(id, Seq(
+      // v1: insert Alice, Bob (rcv=1).
+      ppRow(1L, "Alice", 1L, CHANGE_TYPE_INSERT, 1L, 1000000L),
+      ppRow(2L, "Bob",   1L, CHANGE_TYPE_INSERT, 1L, 1000000L),
+      // v2: real delete Alice + carry-over for Bob (rcv unchanged means CoW rewrite,
+      // no content change).
+      ppRow(1L, "Alice", 1L, CHANGE_TYPE_DELETE, 2L, 2000000L),
+      ppRow(2L, "Bob",   1L, CHANGE_TYPE_DELETE, 2L, 2000000L),
+      ppRow(2L, "Bob",   1L, CHANGE_TYPE_INSERT, 2L, 2000000L),
+      // v3: insert Carol -- advances the watermark past v2 so timers for row
+      // identities 1 and 2 fire and the netChanges output is emitted.
+      ppRow(3L, "Carol", 3L, CHANGE_TYPE_INSERT, 3L, 3000000L)))
+
+    withSQLConf(rocksDbProviderConf) {
+      val q = spark.readStream
+        .option("startingVersion", "1")
+        .option("deduplicationMode", "netChanges")
+        .changes(fullTableName)
+        .select("id", "data", "_change_type")
+        .writeStream
+        .format("memory")
+        .queryName("cdc_stream_netchanges_with_carryover")
+        .outputMode("append")
+        .start()
+      try {
+        q.processAllAvailable()
+        // After carry-over removal: Alice has v1 INSERT + v2 DELETE; Bob has only
+        // v1 INSERT (the v2 CoW pair was dropped); Carol has v3 INSERT.
+        // After netChanges:
+        //   Alice -- (false, false) -> no output
+        //   Bob   -- (false, true)  -> emit INSERT
+        //   Carol -- (false, true)  -> emit INSERT
+        checkAnswer(
+          spark.sql("SELECT * FROM cdc_stream_netchanges_with_carryover"),
+          Seq(
+            Row(2L, "Bob",   CHANGE_TYPE_INSERT),
+            Row(3L, "Carol", CHANGE_TYPE_INSERT)))
+      } finally {
+        q.stop()
+      }
+    }
+  }
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/ResolveChangelogTableStreamingPostProcessingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/ResolveChangelogTableStreamingPostProcessingSuite.scala
@@ -21,7 +21,7 @@ import java.util.Collections
 
 import org.scalatest.BeforeAndAfterEach
 
-import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest}
+import org.apache.spark.sql.{DataFrame, QueryTest}
 import org.apache.spark.sql.catalyst.expressions.Inline
 import org.apache.spark.sql.catalyst.plans.logical.{
   Aggregate, EventTimeWatermark, Filter, Generate, LogicalPlan, Project, TransformWithState}
@@ -236,6 +236,38 @@ class ResolveChangelogTableStreamingPostProcessingSuite
     assertHelperColumnsRemoved(analyzed)
   }
 
+  test("netChanges with composite rowId groups by all helper columns") {
+    // Recreate with a two-column rowId so we exercise the rowIdColumn(idx) helper
+    // for idx > 0. The single-rowId test asserts the size-1 case; this guards
+    // against a regression that hard-codes a single helper attribute.
+    val cat = catalog
+    val ident = identifier
+    cat.dropTable(ident)
+    cat.createTable(
+      ident,
+      Array(
+        Column.create("ns", LongType, false),
+        Column.create("id", LongType, false),
+        Column.create("row_commit_version", LongType, false)),
+      Array.empty[Transform],
+      Collections.emptyMap[String, String]())
+    cat.setChangelogProperties(ident, ChangelogProperties(
+      containsIntermediateChanges = true,
+      rowIdNames = Seq("ns", "id"),
+      rowVersionName = Some("row_commit_version")))
+
+    val analyzed = streamingDf(
+      "deduplicationMode" -> "netChanges").queryExecution.analyzed
+    assertWatermarkOnCommitTimestamp(analyzed)
+    val tws = analyzed.collect { case t: TransformWithState => t }
+    assert(tws.size == 1, s"Expected one TransformWithState. Plan:\n$analyzed")
+    val groupingNames = tws.head.groupingAttributes.map(_.name)
+    assert(groupingNames == Seq("__spark_cdc_rowid_0", "__spark_cdc_rowid_1"),
+      s"Expected grouping by [__spark_cdc_rowid_0, __spark_cdc_rowid_1]; got $groupingNames. " +
+        s"Plan:\n$analyzed")
+    assertHelperColumnsRemoved(analyzed)
+  }
+
   test("netChanges + carry-over removal share a single watermark") {
     catalog.setChangelogProperties(identifier, ChangelogProperties(
       containsCarryoverRows = true,
@@ -276,20 +308,4 @@ class ResolveChangelogTableStreamingPostProcessingSuite
     assertNoStreamingPostProcessing(analyzed)
   }
 
-  // ===========================================================================
-  // Net change computation rejection
-  // ===========================================================================
-
-  test("deduplicationMode=netChanges is rejected on streaming") {
-    catalog.setChangelogProperties(identifier, ChangelogProperties(
-      containsIntermediateChanges = true,
-      rowIdNames = Seq("id"),
-      rowVersionName = Some("row_commit_version")))
-
-    val e = intercept[AnalysisException] {
-      streamingDf("deduplicationMode" -> "netChanges").queryExecution.analyzed
-    }
-    assert(e.getCondition == "INVALID_CDC_OPTION.STREAMING_NET_CHANGES_NOT_SUPPORTED",
-      s"Unexpected error: ${e.getMessage}")
-  }
 }
