[tkaymak]

Part of #36841.

When one unbounded source PCollection feeds two branches of a streaming query, for example a Flatten of a PCollection with itself, Spark does not deduplicate streaming relations and plans one scan per branch. Both scans open a reader for the same split in the same micro batch, so they shared one BeamReaderCache entry: the second task either took over the reader in flight or invalidated it. With bounded input the end of stream row was then emitted again on every batch and the query never went idle. The shared entry exists since #39971, the repeated end of stream row since #40766.

The second scan now gets a detached reader, restored from the start mark of the batch, never cached and writing no marks or done markers. A sibling is recognised by the query run id, so a restarted query in the same JVM still restores from the durable mark. The detached reader inherits whether the end of stream row was already emitted. A failed mark write now also drops the cached reader.

Sibling scans still read independently. A batch cut by maxBatchDurationMillis can differ per sibling on a live source, tracked in #40786. Bounded inputs that drain within the deadline are exact.

PAssertStreamingTest.testFlattenSameSource flattens Create with itself and asserts every element twice. It hangs without the fix and passed 3 of 3 runs with it. Full :runners:spark:4:test and :runners:spark:3:test pass locally.

[tkaymak]

R: @Abacn

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

[Abacn]

Thanks, this LGTM