ericm-db opened a new pull request, #58213:
URL: https://github.com/apache/spark/pull/58213
### 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)
--
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]
