[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)

[uros-b]

The test asserts each source's offset advanced after run 2 (offset != offsetsAfterFirstRun(name)) but cannot detect an offset-swap regression: both input1 and input2 each receive exactly one record in run 1, so their committed LowLatencyMemoryStreamOffset values are identical ({"0":1}). If the system incorrectly assigned alpha's checkpoint slot to beta's object (or vice versa), both sources would still restart from the same byte position and the advance assertion would still pass. The stated goal is "each source resumes from its own offset rather than the other's", but that claim is unverifiable when both starting offsets collide. Fix: give input1 and input2 different record counts in run 1 (e.g., AddData(input1, 1, 2) and AddData(input2, 3)) so their offsets diverge, then after run 2 cross-check that each name maps to the expected post-resume offset.