[SPARK-56686][FOLLOWUP][SQL] Mark CDC streaming rewrite via attribute metadata

### What changes were proposed in this pull request?

Follow-up to #55636 addressing post-merge review comments from zikangh:

1. **Deduplicate `isCarryoverPair`.** The carry-over predicate (`_del_cnt = 1 AND _ins_cnt = 1 AND _rv_cnt = 2 AND _min_rv = _max_rv`) was duplicated between the batch path's `addCarryOverPairFilter` and the streaming path's inline filter. Extracted a shared `buildCarryOverPairPredicate` helper and call it from both.

2. **Mark the streaming row-level rewrite via attribute metadata rather than helper column name.** `UnsupportedOperationChecker` previously detected the rewrite by string-matching the `__spark_cdc_events` aggregate alias name. Switched to a metadata marker (`ResolveChangelogTable.streamingPostProcessingMarker`) attached to the alias's output attribute -- mirroring the existing `EventTimeWatermark.delayKey` and `SessionWindow.marker` patterns. The marker travels with the attribute through optimization.

3. **Expand streaming E2E coverage.** New tests in `ChangelogEndToEndSuite`:
   - composite rowId carry-over removal,
   - composite rowId update detection (different tuples kept raw),
   - carry-over + update detection across multiple commits,
   - DELETE-all-rows and UPDATE-all-rows fixtures,
   - append-only workload pass-through,
   - no-op UPDATE labeled as update (rcv differs on pre/post),
   - large carry-over removal (9 carry-over pairs + 1 real delete).

### Why are the changes needed?

zikangh raised these on the merged PR. Bundled together so they can be reviewed and shipped as one follow-up.

### Does this PR introduce _any_ user-facing change?

No. Internal refactor (#1, #2) and additional test coverage (#3). The behavior of streaming CDC reads is unchanged.

### How was this patch tested?

All 157 tests pass across the four CDC suites:
- `ChangelogResolutionSuite`
- `ResolveChangelogTablePostProcessingSuite`
- `ResolveChangelogTableStreamingPostProcessingSuite`
- `ChangelogEndToEndSuite`

Also confirmed:
- `UnsupportedOperationsSuite` (216 tests) still passes after the marker-based detection switch.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (claude-opus-4-7)

Closes #55653 from gengliangwang/streamingCDC-followup-zikangh.

Authored-by: Gengliang Wang <gengliang@apache.org>
Signed-off-by: Gengliang Wang <gengliang@apache.org>
(cherry picked from commit 84d9c84)
Signed-off-by: Gengliang Wang <gengliang@apache.org>

Author: gengliangwang

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveChangelogTable.scala

@@ -90,6 +90,17 @@ object ResolveChangelogTable extends Rule[LogicalPlan] {
     val all: Set[String] = Set(DelCnt, InsCnt, MinRv, MaxRv, RvCnt, Events)
   }
 
+  /**
+   * Metadata-key marker placed on the `__spark_cdc_events` aggregate's output attribute
+   * by [[addStreamingRowLevelPostProcessing]]. Downstream rules
+   * (`UnsupportedOperationChecker`'s CDC-specific output-mode check) detect the
+   * streaming row-level rewrite by looking for this marker rather than by string-matching
+   * the helper column's name -- mirroring the existing `EventTimeWatermark.delayKey` and
+   * `SessionWindow.marker` patterns and surviving any optimization that might relabel
+   * or rewrite the alias.
+   */
+  final val streamingPostProcessingMarker = "spark.cdc.streamingPostProcessing"
+
   /**
    * Reserved (`__spark_cdc_*`) column names used internally by net-change
    * computation; connectors must not emit columns with these names.
@@ -385,23 +396,22 @@ object ResolveChangelogTable extends Rule[LogicalPlan] {
     // both inside the struct and as top-level grouping outputs; the top-level duplicates
     // are dropped via `unrequiredChildIndex` below.
     val structOfAllCols = CreateStruct(watermarked.output)
+    // Attach a metadata marker to the `__spark_cdc_events` alias so downstream rules
+    // can detect the streaming row-level rewrite by metadata rather than by helper
+    // column name (mirrors `SessionWindow.marker` / `EventTimeWatermark.delayKey`).
+    val eventsMetadata = new MetadataBuilder()
+      .putBoolean(streamingPostProcessingMarker, true)
+      .build()
     val eventsAlias = Alias(
-      new CollectList(structOfAllCols).toAggregateExpression(), HelperColumn.Events)()
+      new CollectList(structOfAllCols).toAggregateExpression(), HelperColumn.Events)(
+      explicitMetadata = Some(eventsMetadata))
 
     val aggregateExprs: Seq[NamedExpression] =
       groupingNamedExprs ++ Seq(delCntAlias, insCntAlias) ++ rvAliases :+ eventsAlias
     val aggregated = Aggregate(groupingExprs, aggregateExprs, watermarked)
 
     val filtered: LogicalPlan = if (requiresCarryOverRemoval) {
-      val delCnt = getAttribute(aggregated, HelperColumn.DelCnt)
-      val insCnt = getAttribute(aggregated, HelperColumn.InsCnt)
-      val minRv = getAttribute(aggregated, HelperColumn.MinRv)
-      val maxRv = getAttribute(aggregated, HelperColumn.MaxRv)
-      val rvCnt = getAttribute(aggregated, HelperColumn.RvCnt)
-      val isCarryoverPair = And(
-        And(EqualTo(delCnt, Literal(1L)), EqualTo(insCnt, Literal(1L))),
-        And(EqualTo(rvCnt, Literal(2L)), EqualTo(minRv, maxRv)))
-      Filter(Not(isCarryoverPair), aggregated)
+      Filter(Not(buildCarryOverPairPredicate(aggregated)), aggregated)
     } else aggregated
 
     // Inline the struct array back into rows. Drop the events column (consumed by Inline)
@@ -531,23 +541,33 @@ object ResolveChangelogTable extends Rule[LogicalPlan] {
   }
 
   /**
-   * Adds a Filter node that drops rows belonging to a CoW carry-over pair.
-   * A pair is a carry-over iff
-   * `_del_cnt = 1 AND _ins_cnt = 1 AND _rv_cnt = 2 AND _min_rv = _max_rv`.
-   * The `_rv_cnt = 2` clause guards against a NULL rowVersion silently matching
+   * Builds the carry-over pair predicate against the helper attributes exposed by
+   * `input`: a pair is a CoW carry-over iff
+   * `_del_cnt = 1 AND _ins_cnt = 1 AND _rv_cnt = 2 AND _min_rv = _max_rv`. The
+   * `_rv_cnt = 2` clause guards against a NULL rowVersion silently matching
    * `_min_rv = _max_rv` (Spark's min/max skip NULLs).
+   *
+   * Used by both the batch path (`addCarryOverPairFilter` over a Window child) and the
+   * streaming path (in `addStreamingRowLevelPostProcessing` over an Aggregate child).
+   * The helper-attribute layout is the same in both cases.
    */
-  private def addCarryOverPairFilter(input: LogicalPlan): LogicalPlan = {
+  private def buildCarryOverPairPredicate(input: LogicalPlan): Expression = {
     val delCnt = getAttribute(input, HelperColumn.DelCnt)
     val insCnt = getAttribute(input, HelperColumn.InsCnt)
     val minRv = getAttribute(input, HelperColumn.MinRv)
     val maxRv = getAttribute(input, HelperColumn.MaxRv)
     val rvCnt = getAttribute(input, HelperColumn.RvCnt)
-
-    val isCarryoverPair = And(
+    And(
       And(EqualTo(delCnt, Literal(1L)), EqualTo(insCnt, Literal(1L))),
       And(EqualTo(rvCnt, Literal(2L)), EqualTo(minRv, maxRv)))
-    Filter(Not(isCarryoverPair), input)
+  }
+
+  /**
+   * Adds a Filter node that drops rows belonging to a CoW carry-over pair, using the
+   * shared `buildCarryOverPairPredicate`.
+   */
+  private def addCarryOverPairFilter(input: LogicalPlan): LogicalPlan = {
+    Filter(Not(buildCarryOverPairPredicate(input)), input)
   }
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala

@@ -284,9 +284,10 @@ object UnsupportedOperationChecker extends Logging {
     // [[ResolveChangelogTable]] are designed and validated only for Append output mode.
     // Two markers can identify a CDC-rewritten plan:
     //   - The row-level rewrite (`addStreamingRowLevelPostProcessing`) injects a
-    //     streaming Aggregate buffering input rows into the helper column
-    //     `ResolveChangelogTable.HelperColumn.Events` ("__spark_cdc_events") before
-    //     re-emitting them via `Generate(Inline(...))`.
+    //     streaming Aggregate whose `__spark_cdc_events` alias's output attribute
+    //     carries the metadata marker
+    //     `ResolveChangelogTable.streamingPostProcessingMarker` (mirrors
+    //     `SessionWindow.marker` / `EventTimeWatermark.delayKey`).
     //   - The netChanges rewrite (`addStreamingNetChangeComputation`) injects a
     //     `TransformWithState` driven by `CdcNetChangesStatefulProcessor`.
     // Under Update or Complete the Aggregate / TransformWithState would re-emit
@@ -296,9 +297,10 @@ object UnsupportedOperationChecker extends Logging {
     // netChanges-only marker is needed here primarily to catch Update mode.) Reject
     // those modes at analysis time with a clear error rather than silently producing
     // a misleading change feed.
-    val containsCdcEventsAggregate = aggregates.exists(a => a.aggregateExpressions.exists {
+    val containsCdcRowLevelRewrite = aggregates.exists(a => a.aggregateExpressions.exists {
       case ne: NamedExpression if ne.resolved =>
-        ne.name == ResolveChangelogTable.HelperColumn.Events
+        ne.metadata.contains(ResolveChangelogTable.streamingPostProcessingMarker) &&
+          ne.metadata.getBoolean(ResolveChangelogTable.streamingPostProcessingMarker)
       case _ => false
     })
     val containsCdcNetChangesProcessor = plan.exists {
@@ -307,7 +309,7 @@ object UnsupportedOperationChecker extends Logging {
       case _ => false
     }
     if (outputMode != InternalOutputModes.Append &&
-        (containsCdcEventsAggregate || containsCdcNetChangesProcessor)) {
+        (containsCdcRowLevelRewrite || containsCdcNetChangesProcessor)) {
       throw QueryCompilationErrors.unsupportedOutputModeForStreamingOperationError(
         outputMode, "Change Data Capture (CDC) streaming reads with post-processing")
     }