Repository navigation
Spark: Short-circuit SyncSparkMicroBatchPlanner.latestOffset for unbounded limits - #18494
binarybrain-009 wants to merge 3 commits into
Conversation
…unded limits Fixes apache#16940 Co-authored-by: cosmow35 <166751777+cosmow35@users.noreply.github.com>
|
@manuzhang could you take a look when you have a chance? You've worked on |
| latestStreamingOffset = | ||
| new StreamingOffset( | ||
| lastValidSnapshot.snapshotId(), | ||
| MicroBatchUtils.addedFilesCount(table(), lastValidSnapshot), |
There was a problem hiding this comment.
Does addedFilesCount always match the end position? MicroBatches includes existing files when computing manifest offsets. For manifests [added=1, existing=1] and [added=1], the added files are at positions 0 and 2, so returning 2 skips the second file. Could we add a test with merged manifests?
There was a problem hiding this comment.
Good catch, thanks! It doesn't: a merged manifest's existing entries take positions too, so the file after it was skipped. The short-circuit now ends at the start of the snapshot's last manifest plus that manifest's added files, which is where the full loop ends, and the Trigger.AvailableNow cap gets the same position instead of being returned as is. Added latestOffsetForUnboundedLimitIncludesFilesAfterMergedManifest, which fails without the change.
The async planner makes the same assumption (its unbounded latestOffset and the positions it queues), and with this layout its planFiles keeps waiting for an end offset that never arrives, so the test runs only for the sync planner. planFiles also uses addedFilesCount as the end of intermediate snapshots. Both predate this PR, so I'd rather fix them in a follow-up if that works for you.
…tOffset Existing entries in merged manifests take positions too, so a snapshot's added-file count can end before its last added file and the files after a merged manifest were skipped. End at the position the per-file loop reaches, for both the latest valid snapshot and the Trigger.AvailableNow cap.
With an unbounded read limit (
Trigger.AvailableNoworReadLimit.allAvailable()),SyncSparkMicroBatchPlanner.latestOffsetreads every manifest between the start offset and the head although no limit can stop it, andplanFilesthen reads the same range again. ForReadAllAvailableit now walks only the snapshot chain and indexes the manifests of the latest valid snapshot, or of the Trigger.AvailableNow cap, to find the end position the per-file loop would reach. That position is not the added-file count, because existing entries in merged manifests take positions too. Refreshes the stale #16941 and covers Spark 3.5 to 4.2.Test plan: the end offset reaches the head snapshot, and files listed after a merged manifest are still planned (sync planner only; the async planner has the same gap issue, left for a follow-up).
Fixes #16940
AI Disclosure