cosmow35 opened a new pull request, #16941: URL: https://github.com/apache/iceberg/pull/16941
## What `SyncSparkMicroBatchPlanner.latestOffset` now short-circuits an unbounded `ReadLimit` (`ReadLimit.allAvailable()`) instead of walking every snapshot in `(committed, head]`. Applied to Spark 3.5, 4.0, and 4.1. Fixes #16940. ## Why `Trigger.AvailableNow` computes its target offset once per run via `latestOffset(committedOffset, ReadLimit.allAvailable())`. The sync planner served that by walking every snapshot in the committed→head range — reading each snapshot's manifests and iterating every `FileScanTask` on the driver. For an unbounded limit none of that per-file work bounds the result: the end offset is just the latest valid snapshot and its added-file count. So the traversal is wasted — `O(snapshots-in-gap × files)`, single-threaded. A scheduled `Trigger.AvailableNow` consumer of a high-commit-cadence table is always ~one trigger interval behind, so the gap (and thus this cost) grows run over run and can diverge until the job exceeds its budget. `AsyncSparkMicroBatchPlanner` already short-circuits `ReadAllAvailable` by advancing only the snapshot chain (metadata, skipping rewrite/delete snapshots); the default (sync) planner did not. ## What it does For `ReadLimit.allAvailable()`, `latestOffset` delegates to a small helper that: - returns the pre-computed `Trigger.AvailableNow` cap if set, else - advances only the snapshot chain (metadata; `nextValidSnapshot`, preserving `streaming-skip-overwrite-snapshots` / `streaming-skip-delete-snapshots`) to the latest valid snapshot and returns `(snapshotId, addedFilesCount, false)`, - preserving the existing no-new-data → `null` contract. This mirrors `AsyncSparkMicroBatchPlanner#latestOffset`. The bounded (rate-limited / `CompositeReadLimit`) path is unchanged. ## Equivalence / safety Perf change with no intended behavior change for the unbounded path: - The returned offset is identical to what the full walk produced — the walk's per-file accumulators only feed the rate-limit checks, which never fire for an unbounded limit, so the result depends only on the latest valid snapshot. - `scanAllFiles` is hardcoded `false` here, matching the async planner. Every `StreamingOffset` construction in the Spark source currently passes `false`, so this is equivalent today; glad to revisit if that invariant changes. ## Tests Added a parameterized `TestStructuredStreamingRead3` case (`testLatestOffsetForUnboundedLimitAdvancesToHeadSnapshot`) asserting `latestOffset(initialOffset, ReadLimit.allAvailable())` advances to the head snapshot's offset. It runs under both `async = false` and `async = true`, pinning sync↔async parity. Verified locally for Spark 3.5 (Scala 2.12), 4.0 and 4.1 (Scala 2.13): `spotlessApply`, `compileTestJava`, `checkstyleMain`/`checkstyleTest`, and the new test all pass. -- 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]
