[SPARK-55953][SQL] Compute net changes in ResolveChangelogTable for batch CDC reads by SanJSp · Pull Request #55583 · apache/spark

SanJSp · 2026-04-28T12:55:08Z

What changes were proposed in this pull request?

This PR adds the netChanges deduplication mode to the ResolveChangelogTable analyzer rule (SPARK-55668 / #55508 ). When a CDC read sets deduplicationMode = 'netChanges', intermediate changes per row identity are collapsed into a single net effect, per the SPIPDeduplication Semantics in B.8.

Net change collapse rules (per SPIP)

Change Sequence	Net Result
INSERT → DELETE	No row (cancels out)
INSERT → UPDATE(s)	Final insert only
UPDATE(s)	First pre-image + final post-image
UPDATE(s) → DELETE	delete with original values

Implementation: 2x2 matrix on `(existedBefore, existsAfter)`

The four SPIP rules map onto a 2x2 matrix that the implementation evaluates per rowId partition:

existedBefore is true iff the partition's first event is delete or update_preimage.
existsAfter is true iff the partition's last event is insert or update_postimage.

These two booleans are sufficient to reproduce the SPIP rules above, because the SPIP only cares about whether the row existed at the boundaries of the version range — never about the intermediate events.

existedBefore	existsAfter	output
false	false	(cancel)
false	true	insert
true	false	delete
true	true	update_preimage + update_postimage

If computeUpdates = false, the update_preimage + update_postimage pair is emitted as delete + insert instead.

Pipeline: Window (per-rowId aggregates: row number, row count, first/last _change_type) → Filter (keep first and/or last row per partition) → Project (relabel _change_type, drop helper columns).

Why are the changes needed?

This completes the net-change post-processing capability of the DSv2 CDC API per the SPIP. Without it, connectors that surface intermediate changes cannot expose a deduplicated change feed to users via the standard CDC API.

Does this PR introduce any user-facing change?

Yes. Requesting deduplicationMode = 'netChanges' on a CDC read now produces a deduplicated change stream. Previously the same request was rejected up-front.

How was this patch tested?

Added ResolveChangelogTableNetChangesSuite — a trait + 2 concrete suite classes (...WithComputeUpdatesSuite, ...WithoutComputeUpdatesSuite) running the same 16-test body under both modes (32 invocations total). Coverage:

4 single-event tests (lone insert/delete across various range shapes).
9 matrix tests covering all (first_change_type, last_change_type) cells.
1 range-narrowing test (events outside the requested version range are not seen).
2 multi-rowId tests (independent partitions, mixed mode-dependent cells).

Removed 2 obsolete tests in ResolveChangelogTablePostProcessingSuite that asserted the previous "not supported" rejection.

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

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

johanl-db

Only minor comments

johanl-db · 2026-04-28T15:07:07Z

+ computeUpdates: Boolean): LogicalPlan = {
+ val windowedPlan = addNetChangesWindow(plan, cl)
+ val filteredAndRelabeledPlan =
+ removeIntermediateChangelogEntriesAndRelabelChangeTypes(windowedPlan, computeUpdates)


nit:
Suggested change
removeIntermediateChangelogEntriesAndRelabelChangeTypes(windowedPlan, computeUpdates)
removeIntermediateChanges(windowedPlan, computeUpdates)

I'd prefer to keep the longer name. It makes both responsibilities of the function explicit (filter + relabel)

gengliangwang

Net-change implementation looks correct and the per-cell tests are thorough. Two notes:

Dead code after the rejection path was removed: cdcNetChangesNotYetSupported at sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:3872-3876 and the corresponding INVALID_CDC_OPTION.NET_CHANGES_NOT_YET_SUPPORTED entry at common/utils/src/main/resources/error/error-conditions.json:3295-3299 are now unreferenced — the only call site in evaluateRequirements is gone and the two ..._is_rejected tests in ResolveChangelogTablePostProcessingSuite were deleted in this PR. Consider removing both in this PR (or as a quick follow-up) so the error catalog stays accurate.
One inline comment below on test coverage of the combined post-processing pipeline.

gengliangwang · 2026-04-28T17:53:07Z

+ cat.setChangelogProperties(ident, ChangelogProperties(
+ containsIntermediateChanges = true,
+ containsCarryoverRows = false,
+ representsUpdateAsDeleteAndInsert = false,


The trait pins representsUpdateAsDeleteAndInsert = false, which keeps addRowLevelPostProcessing (update detection) out of the pipeline. As a result, the chained path where update detection's relabel produces update_preimage/update_postimage rows that then feed injectNetChangeComputation is not exercised end-to-end. Consider at least one variant with representsUpdateAsDeleteAndInsert = true so the integration of the two passes is covered.

Done — see the new WithUpdateDetectionSuite (16 tests × representsUpdateAsDeleteAndInsert = true, computeUpdates = true).

dongjoon-hyun

Please hold on this CDC PR because the master branch was broken at CDC commit and we are recovering it.

SanJSp · 2026-04-29T12:54:22Z

@gengliangwang regarding your top level comment:
I've removed the dead code. The new WithUpdateDetectionSuite actually caught a bug, so well spotted 👍
Claude explanation of bug:

injectNetChangeComputation was calling V2ExpressionUtils.resolveRefs(cl.rowId(), plan) on the Project(Window(...)) produced by addRowLevelPostProcessing, and V2 resolution only works on a bare DataSourceV2Relation. Fixed by hoisting rowId resolution to the top of apply against the bare relation; the resolved Catalyst attributes flow through any wrapping operators by ExprId, so addNetChangesWindow doesn't re-resolve.

Feel free to give it another check, thanks in advance 🙏

SanJSp · 2026-04-29T12:55:28Z

As Gengliang has resolved the issues, this PR should no longer be blocked @dongjoon-hyun ?
See commit 83af083

- ResolveChangelogTable: move the hoisted V2 rowId resolution inside the `if (req.requiresNetChanges)` branch. The hoist was needed to resolve against the bare DataSourceV2Relation (V2ExpressionUtils.resolveRefs doesn't work on the wrapped Project/Window plan), but it should not run unconditionally: connectors that report all of containsCarryoverRows / containsIntermediateChanges / representsUpdateAsDeleteAndInsert as false are allowed by the Changelog contract to inherit the default rowId() impl, which throws. Gating preserves the previous "only call when needed" behavior while keeping the V2-resolution fix. - pom.xml: revert to upstream/master. The diff was stale-rebase noise (maven 3.9.15->3.9.14, scala-maven-plugin 4.9.10->4.9.9, commons-codec 1.22.0->1.21.0, guava 33.6.0->33.5.0, etc.), unrelated to this PR.

gengliangwang · 2026-04-30T02:46:44Z

Tests passed in https://github.com/gengliangwang/spark/actions/runs/25135903704.
Merging to master

### What changes were proposed in this pull request? This PR implements row-level CDC post-processing (carry-over removal and update detection) for DSv2 streaming reads. Previously, streaming `changes()` rejected any post-processing with a blanket `INVALID_CDC_OPTION.STREAMING_POST_PROCESSING_NOT_SUPPORTED` error. The batch path (added in #55508 and #55583) uses a Catalyst `Window` keyed by `(rowId, _commit_version)`, which `UnsupportedOperationChecker` rejects on streaming queries (`NON_TIME_WINDOW_NOT_SUPPORTED_IN_STREAMING`). The streaming rewrite in `ResolveChangelogTable` now expresses the same logic with streaming-allowed primitives: ``` EventTimeWatermark(_commit_timestamp, 0s) -> Aggregate keyed by (rowId..., _commit_version, _commit_timestamp) (count_if delete/insert, [min/max/count rowVersion,] collect_list(struct(*))) -> [Filter on the carry-over predicate] -> Generate(Inline(events)) -> [Project relabeling _change_type for delete+insert pairs] -> Project dropping __spark_cdc_* helpers ``` Including `_commit_timestamp` in the grouping keys is required to satisfy the Append-mode streaming aggregation contract (the watermark attribute must appear among the grouping expressions). By CDC convention all rows in a single commit share `_commit_timestamp`, so this is a no-op semantically relative to the batch `(rowId, _commit_version)` grouping. `deduplicationMode = netChanges` is still rejected -- net change computation partitions by `rowId` alone and reasons over the entire requested range, which is fundamentally cross-batch. The existing error class `INVALID_CDC_OPTION.STREAMING_POST_PROCESSING_NOT_SUPPORTED` is replaced with the more specific `INVALID_CDC_OPTION.STREAMING_NET_CHANGES_NOT_SUPPORTED`, which now names the offending option and points users at the supported streaming alternatives. Doc updates: - `Changelog.java` clarifies that all rows of a single `_commit_version` must share `_commit_timestamp`, and that streaming reads expect non-decreasing `_commit_timestamp` across micro-batches. - `Changelog.java` notes that `containsIntermediateChanges()` is range-scoped, hence the streaming limitation for `netChanges`. - `DataStreamReader.changes()` Scaladoc lists the `netChanges` streaming limitation. ### Why are the changes needed? Without this PR, any streaming CDC read against a connector that emits CoW carry-over pairs (`containsCarryoverRows = true`) or represents updates as raw delete+insert (`representsUpdateAsDeleteAndInsert = true`) raises an analysis error, forcing users to fall back to batch reads. The batch-only restriction is unnecessary for these passes -- they don't need cross-version state -- and it surprises users since the same options work on batch reads. ### Does this PR introduce _any_ user-facing change? Yes. - Streaming `spark.readStream.changes(...)` now supports `computeUpdates = true` and `deduplicationMode = dropCarryovers`. Previously these threw `INVALID_CDC_OPTION.STREAMING_POST_PROCESSING_NOT_SUPPORTED`. - The error class `INVALID_CDC_OPTION.STREAMING_POST_PROCESSING_NOT_SUPPORTED` is renamed to `INVALID_CDC_OPTION.STREAMING_NET_CHANGES_NOT_SUPPORTED` with a more specific message. The new error fires only for `deduplicationMode = netChanges` on streaming reads. - `DataStreamReader.changes()` Scaladoc is updated accordingly. - `Changelog.java` Scaladoc clarifies the `_commit_timestamp` contract for streaming. ### How was this patch tested? 86 tests across 4 CDC suites (all passing): - `ResolveChangelogTableStreamingPostProcessingSuite` (new, 5 tests) -- plan-shape assertions covering carry-over only, update detection only, both fused, and the no-rewrite pass-through cases. Verifies the `EventTimeWatermark` + `Aggregate` + `Generate(Inline)` rewrite shape. - `ChangelogResolutionSuite` -- the two existing streaming throw-tests are flipped to plan-shape assertions; a new test covers the `netChanges` streaming throw. - `ResolveChangelogTablePostProcessingSuite` -- the existing streaming throw test is updated to cover the `netChanges`-only case. - `ChangelogEndToEndSuite` -- three new streaming end-to-end tests using `InMemoryChangelogCatalog`: carry-over removal drops CoW pairs, update detection relabels delete+insert as update, and `netChanges` throws. Also confirmed `UnsupportedOperationsSuite` (216 tests) still passes -- the rewritten plan does not contain `Window` or any other streaming-rejected operator. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (claude-opus-4-7) Closes#55636 from gengliangwang/streamingCDC. Authored-by: Gengliang Wang <gengliang@apache.org> Signed-off-by: Gengliang Wang <gengliang@apache.org>

…reads tests ### What changes were proposed in this pull request? Add 5 tests verifying CACHE TABLE impact on reads for DSv2 tables. Write operations ignore the cache, so these tests verify how reads behave when a cached table is mutated by session SQL or external catalog API calls. Tests run in both classic and Connect modes. `DSv2CacheTableReadTests` extends `DSv2ExternalMutationTestBase`, following the same pattern as `DSv2TempViewWithStoredPlanTests` (PR #55571) and `DSv2RepeatedTableAccessTests` (PR #55583). The tests cover five scenarios from the design doc: - **Scenario 1 (external write)**: Cache pins the read, external write invisible until `REFRESH TABLE`. - **Scenario 2 (session write and more external changes)**: Session write rebuilds cache, subsequent external write invisible until `REFRESH TABLE`. - **Scenario 3 (external schema changes)**: Cache pinned at original schema, external ADD COLUMN invisible until `REFRESH TABLE`. - **Scenario 4 (session schema changes and more external changes)**: Session ALTER rebuilds cache with new schema, subsequent external write invisible until `REFRESH TABLE`. - **Scenario 5 (external drop and recreate table)**: Query sees the new empty table. #### New files - **`DSv2CacheTableReadTests`**: Shared trait containing all 5 tests, using `session.sql(...)` with `.collect()` calls (harmless in classic mode, required for Connect). #### Modified files - **`DataSourceV2DataFrameSuite`**: Mixes in `DSv2CacheTableReadTests` alongside existing traits (classic runner, `testPrefix = ""`). - **`DataSourceV2DataFrameConnectSuite`**: Mixes in `DSv2CacheTableReadTests` alongside existing traits (Connect runner, `testPrefix = "[connect] "`). - **`DSv2ExternalMutationTestBase`**: Updated Scaladoc to reference all three consumer traits. - **`DSv2TempViewWithStoredPlanTests`**: Removed stale comment about inherited vals. ### Why are the changes needed? These tests document and lock down the expected CACHE TABLE behavior with DSv2 tables: cached reads are pinned against external mutations, and `REFRESH TABLE` invalidates the cache. This prevents regressions if cache invalidation logic changes. ### Does this PR introduce _any_ user-facing change? No. This PR is test-only. ### How was this patch tested? 5 new tests run in both classic and Connect modes (10 total): Classic (175 total, all pass): ``` build/sbt 'sql/testOnly *DataSourceV2DataFrameSuite' ``` Connect (35 total, all pass): ``` build/sbt 'connect/testOnly *DataSourceV2DataFrameConnectSuite' ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (claude-opus-4-6) Closes#55536 from longvu-db/spark-dsv2-cache-scenario-5. Authored-by: Thang Long Vu <long.vu@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

…reads tests ### What changes were proposed in this pull request? Add 5 tests verifying CACHE TABLE impact on reads for DSv2 tables. Write operations ignore the cache, so these tests verify how reads behave when a cached table is mutated by session SQL or external catalog API calls. Tests run in both classic and Connect modes. `DSv2CacheTableReadTests` extends `DSv2ExternalMutationTestBase`, following the same pattern as `DSv2TempViewWithStoredPlanTests` (PR #55571) and `DSv2RepeatedTableAccessTests` (PR #55583). The tests cover five scenarios from the design doc: - **Scenario 1 (external write)**: Cache pins the read, external write invisible until `REFRESH TABLE`. - **Scenario 2 (session write and more external changes)**: Session write rebuilds cache, subsequent external write invisible until `REFRESH TABLE`. - **Scenario 3 (external schema changes)**: Cache pinned at original schema, external ADD COLUMN invisible until `REFRESH TABLE`. - **Scenario 4 (session schema changes and more external changes)**: Session ALTER rebuilds cache with new schema, subsequent external write invisible until `REFRESH TABLE`. - **Scenario 5 (external drop and recreate table)**: Query sees the new empty table. #### New files - **`DSv2CacheTableReadTests`**: Shared trait containing all 5 tests, using `session.sql(...)` with `.collect()` calls (harmless in classic mode, required for Connect). #### Modified files - **`DataSourceV2DataFrameSuite`**: Mixes in `DSv2CacheTableReadTests` alongside existing traits (classic runner, `testPrefix = ""`). - **`DataSourceV2DataFrameConnectSuite`**: Mixes in `DSv2CacheTableReadTests` alongside existing traits (Connect runner, `testPrefix = "[connect] "`). - **`DSv2ExternalMutationTestBase`**: Updated Scaladoc to reference all three consumer traits. - **`DSv2TempViewWithStoredPlanTests`**: Removed stale comment about inherited vals. ### Why are the changes needed? These tests document and lock down the expected CACHE TABLE behavior with DSv2 tables: cached reads are pinned against external mutations, and `REFRESH TABLE` invalidates the cache. This prevents regressions if cache invalidation logic changes. ### Does this PR introduce _any_ user-facing change? No. This PR is test-only. ### How was this patch tested? 5 new tests run in both classic and Connect modes (10 total): Classic (175 total, all pass): ``` build/sbt 'sql/testOnly *DataSourceV2DataFrameSuite' ``` Connect (35 total, all pass): ``` build/sbt 'connect/testOnly *DataSourceV2DataFrameConnectSuite' ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (claude-opus-4-6) Closes#55536 from longvu-db/spark-dsv2-cache-scenario-5. Authored-by: Thang Long Vu <long.vu@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit 5a47a30) Signed-off-by: Wenchen Fan <wenchen@databricks.com>

johanl-db approved these changes Apr 28, 2026
View reviewed changes

gengliangwang reviewed Apr 28, 2026
View reviewed changes

dongjoon-hyun requested changes Apr 28, 2026
View reviewed changes

SanJSp requested review from dongjoon-hyun and gengliangwang April 29, 2026 12:54

SanJSpand others added 4 commits April 29, 2026 21:30

First working netChanges implementation
63796ff

Simplified test suite, refactored netChanges, javadocs
746f37c

PR feedback
23b2c42

gengliangwang force-pushed the SPARK-55953-net-changes branch from 5d71af6 to 312495bCompare April 29, 2026 21:31

gengliangwang approved these changes Apr 30, 2026
View reviewed changes

gengliangwang closed this in 5ac03acApr 30, 2026

gengliangwang mentioned this pull request Apr 30, 2026
[SPARK-56686][SQL] Support streaming row-level CDC post-processing #55636
Closed

longvu-db mentioned this pull request May 27, 2026
[SPARK-54022][SPARK-56617][SQL][TESTS] Add DSv2 CACHE TABLE impact on reads tests #55536
Closed

Navigation Menu

Uh oh!

[SPARK-55953][SQL] Compute net changes in ResolveChangelogTable for batch CDC reads - #55583

[SPARK-55953][SQL] Compute net changes in ResolveChangelogTable for batch CDC reads#55583
SanJSp wants to merge 4 commits into
apache:masterfrom
SanJSp:SPARK-55953-net-changes

SanJSp commented Apr 28, 2026•
edited
Loading

Uh oh!

johanl-db left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

johanl-dbApr 28, 2026

Uh oh!

SanJSpApr 29, 2026

Uh oh!

Uh oh!

Uh oh!

gengliangwang left a comment

Uh oh!

gengliangwangApr 28, 2026

Uh oh!

SanJSpApr 29, 2026

Uh oh!

dongjoon-hyun left a comment

Uh oh!

SanJSp commented Apr 29, 2026•
edited
Loading

Uh oh!

SanJSp commented Apr 29, 2026•
edited
Loading

Uh oh!

gengliangwang commented Apr 30, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

	removeIntermediateChangelogEntriesAndRelabelChangeTypes(windowedPlan, computeUpdates)
	removeIntermediateChanges(windowedPlan, computeUpdates)

Uh oh!

Conversation

SanJSp commented Apr 28, 2026• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Net change collapse rules (per SPIP)

Implementation: 2x2 matrix on (existedBefore, existsAfter)

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

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

Uh oh!

johanl-db left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

johanl-dbApr 28, 2026

Choose a reason for hiding this comment

Uh oh!

SanJSpApr 29, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

gengliangwang left a comment

Choose a reason for hiding this comment

Uh oh!

gengliangwangApr 28, 2026

Choose a reason for hiding this comment

Uh oh!

SanJSpApr 29, 2026

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

SanJSp commented Apr 29, 2026• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SanJSp commented Apr 29, 2026• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gengliangwang commented Apr 30, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

SanJSp commented Apr 28, 2026•
edited
Loading

Implementation: 2x2 matrix on `(existedBefore, existsAfter)`

SanJSp commented Apr 29, 2026•
edited
Loading

SanJSp commented Apr 29, 2026•
edited
Loading