[SPARK-54867][SS] Introduce NamedStreamingRelation wrapper for source identification during analysis #53639
+257
−0
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
… identification during analysis ### What changes were proposed in this pull request? This PR introduces infrastructure for tracking and propagating source identifying names through query analysis for streaming queries. It adds: 1. **StreamingSourceIdentifyingName** - A sealed trait hierarchy representing the naming state of streaming sources: - `UserProvided(name)` - Explicitly set via `.name()` API - `FlowAssigned(name)` - Assigned by external flow systems (e.g., SDP) - `Unassigned` - No name assigned yet (to be auto-generated) 2. **NamedStreamingRelation** - A transparent wrapper node that: - Carries source identifying names through the analyzer phase - Extends `UnaryNode` for transparent interaction with analyzer rules - Stays unresolved until explicitly unwrapped by a future `NameStreamingSources` analyzer rule - Provides `withUserProvidedName()` to attach user-specified names 3. **NAMED_STREAMING_RELATION** tree pattern for efficient pattern matching ### Why are the changes needed? Streaming sources need stable, predictable names for: - **Checkpoint location stability** - Schema evolution and offset tracking require consistent source identification - **Schema lookup at specific offsets** - Analysis-time operations need to reference sources by name - **Flow integration** - SDP and similar systems need per-source metadata paths - **User control** - Allow users to explicitly name sources via the `.name()` API By introducing this wrapper during analysis (rather than at execution planning), we enable these capabilities while maintaining a clean separation between parsing, analysis, and execution phases. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? New unit tests in `NamedStreamingRelationSuite` covering: - Source name state transitions (Unassigned → UserProvided) - Output delegation to child plan - Tree pattern registration - Resolved state behavior - String representation ### Was this patch authored or co-authored using generative AI tooling? No.
ericm-db
force-pushed
the
named-streaming-relation
branch
from
d7588b7 to
89b259a
Compare
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 infrastructure for tracking and propagating source identifying names through query analysis for streaming queries. It adds:
StreamingSourceIdentifyingName - A sealed trait hierarchy representing the naming state of streaming sources:
UserProvided(name)- Explicitly set via.name()APIFlowAssigned(name)- Assigned by external flow systems (e.g., SDP)Unassigned- No name assigned yet (to be auto-generated)NamedStreamingRelation - A transparent wrapper node that:
UnaryNodefor transparent interaction with analyzer rulesNameStreamingSourcesanalyzer rulewithUserProvidedName()to attach user-specified namesNAMED_STREAMING_RELATION tree pattern for efficient pattern matching
Why are the changes needed?
Streaming sources need stable, predictable names for:
By introducing this wrapper during analysis (rather than at execution planning), we enable these capabilities while maintaining a clean separation between parsing, analysis, and execution phases.
Does this PR introduce any user-facing change?
No.
How was this patch tested?
New unit tests in
NamedStreamingRelationSuitecovering:Was this patch authored or co-authored using generative AI tooling?
No.