[SPARK-54907][SS] Introduce NameStreamingSources analyzer rule for streaming source evolution #53684
+465
−1
Conversation
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
JIRA Issue Information=== Task SPARK-54907 === This comment was automatically generated by GitHub Actions |
ericm-db
changed the title
ericm-db
changed the title
ericm-db
changed the title
ericm-db
force-pushed
the
SPARK-54684-streaming-source-naming
branch
2 times, most recently
from
2515f71 to
7dd09f9
Compare
…reaming source evolution This PR introduces the `NameStreamingSources` analyzer rule and supporting infrastructure to enable streaming source evolution. This allows streaming queries to add, remove, or reorder sources without losing state by assigning stable names to sources. Key changes: - Added `HasStreamingSourceIdentifyingName` trait for uniform name propagation - Updated `StreamingRelationV2` to support source identifying names - Created `NameStreamingSources` analyzer rule to propagate names from `NamedStreamingRelation` wrappers - Added `spark.sql.streaming.queryEvolution.enableStreamingSourceEvolution` config flag - Added error handling for unnamed sources when enforcement is enabled Currently, streaming sources are identified by their position in the query plan (sources/0, sources/1, etc.). This makes it impossible to add, remove, or reorder sources without breaking checkpoint compatibility. By assigning stable names to sources, we enable: 1. **Source evolution**: Add/remove/reorder sources without losing state 2. **Stable checkpoint locations**: sources/<name> instead of sources/0, sources/1 3. **Better debugging**: Named sources are easier to identify and debug No. The infrastructure is in place but the user-facing `.name()` DataFrame API is not yet exposed. The analyzer rule handles existing `NamedStreamingRelation` nodes that may be created internally. - Added comprehensive unit tests in `NameStreamingSourcesSuite` (15 test cases) - Tests cover name propagation, enforcement checks, error messages, and edge cases - Tests verify behavior with UserProvided, FlowAssigned, and Unassigned names No.
ericm-db
force-pushed
the
SPARK-54684-streaming-source-naming
branch
from
7dd09f9 to
22baf62
Compare
Contributor
Author
|
@dtenedor PTAL |
dtenedor
approved these changes
Jan 9, 2026
Contributor
dtenedor
left a comment
dtenedor
left a comment
There was a problem hiding this comment.
We talked offline about a couple small mismatches between config names in error messages, etc. LGTM now.
Contributor
|
LGTM, merging to master. |
Yicong-Huang
pushed a commit
to Yicong-Huang/spark
that referenced
this pull request
Jan 9, 2026
…reaming source evolution ## What changes were proposed in this pull request? This PR introduces the `NameStreamingSources` analyzer rule and supporting infrastructure to enable streaming source evolution. This allows streaming queries to add, remove, or reorder sources without losing state by assigning stable names to sources. Key changes: - Added `HasStreamingSourceIdentifyingName` trait for uniform name propagation - Updated `StreamingRelationV2` to support source identifying names - Created `NameStreamingSources` analyzer rule to propagate names from `NamedStreamingRelation` wrappers - Added `spark.sql.streaming.queryEvolution.enableStreamingSourceEvolution` config flag - Added error handling for unnamed sources when enforcement is enabled ## Why are the changes needed? Currently, streaming sources are identified by their position in the query plan (sources/0, sources/1, etc.). This makes it impossible to add, remove, or reorder sources without breaking checkpoint compatibility. By assigning stable names to sources, we enable: 1. **Source evolution**: Add/remove/reorder sources without losing state 2. **Stable checkpoint locations**: sources/<name> instead of sources/0, sources/1 3. **Better debugging**: Named sources are easier to identify and debug ## Does this PR introduce _any_ user-facing change? No. The infrastructure is in place but the user-facing `.name()` DataFrame API is not yet exposed. The analyzer rule handles existing `NamedStreamingRelation` nodes that may be created internally. ## How was this patch tested? - Added comprehensive unit tests in `NameStreamingSourcesSuite` (15 test cases) - Tests cover name propagation, enforcement checks, error messages, and edge cases - Tests verify behavior with UserProvided, FlowAssigned, and Unassigned names ## Was this patch authored or co-authored using generative AI tooling? No. Closes apache#53684 from ericm-db/SPARK-54684-streaming-source-naming. Lead-authored-by: Eric Marnadi <132308037+ericm-db@users.noreply.github.com> Co-authored-by: ericm-db <eric.marnadi@databricks.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
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 PR introduces the
NameStreamingSourcesanalyzer rule and supporting infrastructure to enable streaming source evolution. This allows streaming queries to add, remove, or reorder sources without losing state by assigning stable names to sources.Key changes:
HasStreamingSourceIdentifyingNametrait for uniform name propagationStreamingRelationV2to support source identifying namesNameStreamingSourcesanalyzer rule to propagate names fromNamedStreamingRelationwrappersspark.sql.streaming.queryEvolution.enableStreamingSourceEvolutionconfig flagWhy are the changes needed?
Currently, streaming sources are identified by their position in the query plan (sources/0, sources/1, etc.). This makes it impossible to add, remove, or reorder sources without breaking checkpoint compatibility. By assigning stable names to sources, we enable:
Does this PR introduce any user-facing change?
No. The infrastructure is in place but the user-facing
.name()DataFrame API is not yet exposed. The analyzer rule handles existingNamedStreamingRelationnodes that may be created internally.How was this patch tested?
NameStreamingSourcesSuite(15 test cases)Was this patch authored or co-authored using generative AI tooling?
No.