ericm-db opened a new pull request, #59526:
URL: https://github.com/apache/spark/pull/59526
### What changes were proposed in this pull request?
Preserve `DataStreamReader.name(...).table(...)` source names in the Python
and Scala Spark Connect clients and server:
- Add optional `Read.NamedTable.source_name = 3` and regenerate the Python
protobuf bindings.
- Copy the reader's source name into named-table reads in both clients.
- Validate the wire-provided name and carry it into analysis with
`NamedStreamingRelation`.
- Leave batch reads, unnamed table reads, and the existing named `load()`
path unchanged.
Classic already propagates table source names, so it needs no production
change. A shared Python regression covers both Classic and Connect. CDC
`changes()` is outside this patch's scope.
### Why are the changes needed?
Connect currently serializes named and unnamed streaming table reads
identically. The server cannot recover the missing name for source-evolution
validation or checkpoint source IDs.
For example, with an existing parquet table and source evolution disabled:
```python
spark.conf.set("spark.sql.streaming.queryEvolution.enableSourceEvolution",
"false")
spark.readStream.name("orders_source").table("db.orders").schema
```
Classic rejects the explicit name with
`STREAMING_QUERY_EVOLUTION_ERROR.SOURCE_NAMING_NOT_SUPPORTED`. Connect loses
the name instead. With source evolution enabled, the table read must retain
`orders_source` in its checkpoint rather than becoming an unnamed source.
### Does this PR introduce _any_ user-facing change?
Yes. Spark Connect streaming table reads now honor the existing experimental
`.name()` API, including its validation and feature-gate errors, matching
Classic.
The protocol addition is optional and wire-compatible. Both a fixed client
and a fixed server are required: an older server accepts but ignores the
unknown field, so upgrading only the client is insufficient. No new public
method is added.
### How was this patch tested?
Passed locally:
- All 85 tests in `pyspark.sql.tests.connect.test_connect_plan`, including
the new named/unnamed table serialization regression.
- A descriptor comparison proving the only semantic protobuf change is
optional `NamedTable.source_name = 3`; an old-schema round trip accepts and
preserves the unknown field but cannot apply its name.
- Repository protobuf generation, focused Ruff lint/format checks, and Scala
formatter validation for all three changed Connect modules.
- `git diff --check`.
Added, but not run locally:
- Scala Connect client serialization coverage in `ClientStreamingQuerySuite`.
- Server planner coverage for named/unnamed, streaming/batch, invalid and
empty names in `SparkConnectPlannerSuite`.
- A shared Classic/Connect Python integration test using a one-row parquet
table, an `availableNow` trigger, a fresh checkpoint, bounded termination and
cleanup. It checks the V2 offset log contains `orders_source:` and verifies the
disabled-feature error.
The JVM-backed suites require building the changed server/client. No
existing worktree's build outputs were reused and no broad shared build was
started. This PR remains draft pending that validation.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex CLI 0.156.0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
