[SPARK-60270][CONNECT] Preserve streaming source names for table reads

[ericm-db]

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 wire-provided names and wrap every streaming table read in NamedStreamingRelation, matching Classic. Unnamed tables retain Unassigned so source-evolution enforcement can reject them.
• Leave batch reads and the existing named load() path unchanged. Unnamed streaming table reads still work when source evolution is disabled.

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:

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. With source evolution enabled, unnamed table reads also fail with the intended analysis error instead of bypassing source-naming enforcement.

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, including unnamed-source enforcement through NameStreamingSources.
• 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:, verifies unnamed reads are rejected when enforcement is enabled, and covers both named/unnamed behavior when it is disabled.

The JVM-backed suites require building the changed server/client and remain unverified locally. No existing worktree's build outputs were reused and no broad shared build was started.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Codex CLI 0.156.0

[funrollloops]

It's unfortunate this isn't alongside is_streaming, but this is the best option now.