[SPARK-58938][SS] Add Real-Time Mode test coverage for streaming source evolution

[ericm-db]

What changes were proposed in this pull request?

This is a test-only PR. It adds coverage for the previously untested intersection of two existing features: streaming source naming / source evolution (spark.sql.streaming.queryEvolution.enableSourceEvolution) and Real-Time Mode (RealTimeTrigger).

The two features were developed independently and their startup paths meet in MicroBatchExecution: source evolution makes sourceIdMap name-keyed and forces offset log v2 (OffsetMap), while RTM writes its end offsets at the end of a batch keyed by the physical scan's SparkDataStream rather than by source id, which StreamProgress.toOffsetMap must then resolve back to a source name. That resolution step was not exercised by any existing suite — the evolution suites use non-RTM triggers, and the RTM suites never enable source evolution or name a source.

Two suites are added:

• RealTimeModeSourceEvolutionSuite (sql/core), using LowLatencyMemoryStream. Names are attached by wrapping the stream's plan in a NamedStreamingRelation, following StreamingSourceEvolutionSuite, since .name() lives on DataStreamReader and a memory stream is not reachable through spark.readStream. Path assertions read StreamingDataSourceV2Relation.metadataPath off the plan, because a memory stream never materializes sources/ on disk. Covers:

  • a named source getting a name-keyed checkpoint path (sources/<name>) and a v2 name-keyed OffsetMap under RTM, with the enforcement flag persisted in the offset metadata;
  • a union of two named sources getting separate name-keyed paths;
  • a restart that reorders the two named sources, asserting each resumes from its own offset rather than the other's;
  • a restart that adds a named source, asserting the pre-existing source keeps its committed offset;
  • duplicate source names rejected;
  • a positional-path baseline with evolution off.

• KafkaRealTimeModeSourceEvolutionSuite (connector/kafka-0-10-sql), exercising the same behaviors through the .name() API on spark.readStream.format("kafka"), plus unnamed-source rejection at load() under enforcement.

Why are the changes needed?

Source evolution and Real-Time Mode were built independently and nothing exercised them together, leaving the source-name resolution on RTM's end-of-batch offset path untested. These tests pin down that the name-keyed checkpoint bookkeeping — metadata paths, offset-map keys, and offsets surviving reorder and add-source across restarts — holds on the RTM offset path, and that the naming enforcement rules (unnamed and duplicate sources) fire under RTM.

Does this PR introduce any user-facing change?

No. Tests only; no production code changes.

How was this patch tested?

New suites, run locally against master:

• build/sbt "sql/testOnly org.apache.spark.sql.streaming.RealTimeModeSourceEvolutionSuite" — 6 tests, all pass.
• build/sbt "sql-kafka-0-10/testOnly org.apache.spark.sql.kafka010.KafkaRealTimeModeSourceEvolutionSuite" — 5 tests, all pass.

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

Generated-by: Claude Code (Claude Opus 4.8)