[SPARK-55121][PYTHON][SS] Add DataStreamReader.name() to Classic PySpark #53898
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-55121 === This comment was automatically generated by GitHub Actions |
ericm-db
changed the title
ericm-db
changed the title
ericm-db
force-pushed
the
classic-datastream-reader-name
branch
from
02fa47f to
df74e5f
Compare
HyukjinKwon
approved these changes
Jan 21, 2026
Member
HyukjinKwon
left a comment
HyukjinKwon
left a comment
There was a problem hiding this comment.
I am fine with this but I think it would be better if @HeartSaVioR has some time to take a look.
anishshri-db
approved these changes
Jan 21, 2026
Contributor
anishshri-db
left a comment
anishshri-db
left a comment
There was a problem hiding this comment.
lgtm pending green CI
ueshin
reviewed
Jan 21, 2026
ueshin
approved these changes
Jan 21, 2026
python/pyspark/sql/tests/streaming/test_streaming_reader_name.py
Outdated
Show resolved
Hide resolved
python/pyspark/sql/tests/streaming/test_streaming_reader_name.py
Outdated
Show resolved
Hide resolved
Contributor
gaogaotiantian
left a comment
gaogaotiantian
left a comment
There was a problem hiding this comment.
Some final minor suggestions :)
python/pyspark/sql/tests/streaming/test_streaming_reader_name.py
Outdated
Show resolved
Hide resolved
python/pyspark/sql/tests/streaming/test_streaming_reader_name.py
Outdated
Show resolved
Hide resolved
### What changes were proposed in this pull request?
This PR adds the `name()` method to Classic PySpark's `DataStreamReader` class. This method allows users to specify a name for streaming sources, which is used in checkpoint metadata and enables stable checkpoint locations for source evolution.
Changes include:
- Add `name()` method to `DataStreamReader` in `python/pyspark/sql/streaming/readwriter.py`
- Add comprehensive test suite in `python/pyspark/sql/tests/streaming/test_streaming_reader_name.py`
- Update compatibility test to mark `name` as currently missing from Connect (until the Connect PR merges)
The method validates that the source_name contains only ASCII letters, digits, and underscores, raising `PySparkTypeError` or `PySparkValueError` for invalid inputs.
### Why are the changes needed?
This brings Classic PySpark to feature parity with the Scala/Java API for streaming source naming. The `name()` method is essential for:
1. Identifying sources in checkpoint metadata
2. Enabling stable checkpoint locations during source evolution
3. Providing consistency across Classic and Connect implementations
### Does this PR introduce _any_ user-facing change?
Yes. Users can now call `.name()` on DataStreamReader in Classic PySpark:
```python
spark.readStream.format("parquet").name("my_source").load("/path")
```
### How was this patch tested?
- Added comprehensive unit tests in `test_streaming_reader_name.py` covering:
- Valid name patterns (letters, digits, underscores)
- Invalid names (hyphens, spaces, dots, special characters, empty strings, None, wrong types)
- Method chaining
- Different data formats (parquet, json)
- Integration with streaming queries
- Updated compatibility tests to account for the current state where Classic has `name` but Connect doesn't yet
### Was this patch authored or co-authored using generative AI tooling?
Yes.
- Fix empty string validation to use VALUE_NOT_NON_EMPTY_STR error - Check type before checking emptiness to avoid confusing error messages - Combine all invalid name tests into single test with subTests - Use PySparkValueError instead of generic Exception for invalid names - Add more invalid name test cases (dollar, hash, exclamation)
- Remove redundant empty string validation check - empty and whitespace-only strings are now caught by the regex pattern validation - Remove unnecessary str() wrapper since type is already validated - Consolidate empty string test into invalid names test - Merge None and wrong type tests into single test method with multiple cases
ericm-db
force-pushed
the
classic-datastream-reader-name
branch
from
787bc26 to
f9df920
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 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 adds the
name()method to Classic PySpark'sDataStreamReaderclass. This method allows users to specify a name for streaming sources, which is used in checkpoint metadata and enables stable checkpoint locations for source evolution.Changes include:
name()method toDataStreamReaderinpython/pyspark/sql/streaming/readwriter.pypython/pyspark/sql/tests/streaming/test_streaming_reader_name.pynameas currently missing from Connect (until the Connect PR merges)The method validates that the source_name contains only ASCII letters, digits, and underscores, raising
PySparkTypeErrororPySparkValueErrorfor invalid inputs.Why are the changes needed?
This brings Classic PySpark to feature parity with the Scala/Java API for streaming source naming. The
name()method is essential for:Does this PR introduce any user-facing change?
Yes. Users can now call
.name()on DataStreamReader in Classic PySpark:How was this patch tested?
test_streaming_reader_name.pycovering:namebut Connect doesn't yetWas this patch authored or co-authored using generative AI tooling?
No.