[SPARK-55948][SQL] Add DSv2 CDC connector API and analyzer resolution#54737
Closed
gengliangwang wants to merge 1 commit into
Closed
[SPARK-55948][SQL] Add DSv2 CDC connector API and analyzer resolution#54737gengliangwang wants to merge 1 commit into
gengliangwang wants to merge 1 commit into
Conversation
Introduce the foundational Change Data Capture (CDC) framework for DSv2: - Changelog interface: connector contract for exposing change data - ChangelogInfo/ChangelogRange: CDC query parameters (version/timestamp range, deduplication mode, compute updates) - TableCatalog.loadChangelog() and SUPPORT_CHANGELOG capability - ChangelogTable: DSv2 Table wrapper for Changelog - RelationChanges: unresolved logical node for CDC queries - Analyzer resolution: RelationChanges -> ChangelogTable -> DataSourceV2Relation - ChangelogInfoUtils: utility for parsing CDC options - TimeTravelSpec: extract shared resolveTimestampExpression method - Error conditions for CDC validation - InMemoryChangelogCatalog: test-only implementation with scan support - ChangelogInfoUtilsSuite: unit tests Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Member
Author
|
Merged into #54738 — combining the DSv2 API and SQL support into a single PR. |
gengliangwang
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
This is PR 1/3 of the CDC (Change Data Capture) framework for Spark. It introduces the foundational DSv2 connector API and analyzer resolution logic.
Connector API (new Java interfaces):
Changelog: The central connector interface for CDC. Connectors implement this to expose row-level change data with metadata columns (_change_type,_commit_version,_commit_timestamp). Connectors declare properties (containsCarryoverRows,containsIntermediateChanges,representsUpdateAsDeleteAndInsert) that tell Spark what post-processing is needed.ChangelogInfo: Encapsulates CDC query parameters — range, deduplication mode (NONE,DROP_CARRYOVERS,NET_CHANGES), andcomputeUpdatesflag.ChangelogRange: Sealed interface with three record types —VersionRange,TimestampRange, andUnbounded.TableCatalog.loadChangelog(Identifier, ChangelogInfo): Entry point for catalogs to provide aChangelog.TableCatalogCapability.SUPPORT_CHANGELOG: Catalog capability flag that gates CDC support.Analyzer resolution:
RelationChanges: Unresolved logical node representing a CDC query.ChangelogTable: DSv2Tablewrapper around aChangelog, declaringBATCH_READandMICRO_BATCH_READcapabilities.Analyzer: resolvesRelationChanges-> looks up catalog -> callsloadChangelog()-> wraps inChangelogTable-> createsDataSourceV2Relation(batch) orStreamingRelationV2(streaming).CheckAnalysis: ReportsTABLE_OR_VIEW_NOT_FOUNDfor unresolvedRelationChanges.CTESubstitution: Blocks CDC queries on CTE relations (mirrors time travel restriction).Utilities:
ChangelogInfoUtils: Parses DataFrame API options into aChangelogInfo.TimeTravelSpec.resolveTimestampExpression(): Extracted shared method for timestamp evaluation, deduplicating logic between time travel and CDC.Test infrastructure:
InMemoryChangelogCatalog: Test catalog with pre-populated change rows and a working scan pipeline. ReportscontainsCarryoverRows = false.Why are the changes needed?
Spark currently has no standardized framework for Change Data Capture (CDC) queries via DSv2 connectors. This PR establishes the connector API contract and analyzer resolution so that connectors can expose row-level change data in a uniform way, and Spark can handle post-processing (carry-over removal, update detection, net change computation) based on connector-declared properties.
Does this PR introduce any user-facing change?
No user-facing change in this PR alone. The connector API and analyzer resolution are internal building blocks. User-facing SQL syntax is added in PR 2 (#54738), and the DataFrame API in PR 3 (#54739).
How was this patch tested?
ChangelogInfoUtilsSuite— 18 unit tests covering version/timestamp range parsing, deduplication modes, bound inclusivity, timezone handling, and error cases.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.6)