feat: add parallel json log reads for snapshot construction (#3246) - #3254
feat: add parallel json log reads for snapshot construction (#3246)#3254andrei-ionescu wants to merge 6 commits into
Conversation
cbf528c to
542eb2b
Compare
|
@chiinlquah, @scottsand-db, @dengsh12, @scovich, @DrakeLin, @rliao147 Could you have a look on this PR? Greatly appreciated 🙏 |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #3254 +/- ##
==========================================
+ Coverage 90.32% 90.33% +0.01%
==========================================
Files 250 250
Lines 89038 89525 +487
Branches 89038 89525 +487
==========================================
+ Hits 80422 80875 +453
- Misses 5682 5713 +31
- Partials 2934 2937 +3 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Benchmark results: ✅ PassSummary: 🚀 0 · ✅ 8 · ☑️ 6 · 🚧 1 · ❌ 0 Per-benchmark results (15 rows)
Legend: 🚀 ≥1.15x faster · ✅ faster or unchanged · ☑️ ≤1.03x slower · 🚧 1.03x-1.15x slower · ❌ ≥1.15x slower |
542eb2b to
fe86e71
Compare
|
I did some changes to fix the failing actions. Fast turnaround. @chiinlquah, @scottsand-db, @dengsh12, @scovich, @DrakeLin, @rliao147 can you approve the actions/workflows run again? Greatly appreciated 🙏 |
d8c5fc1 to
21c3656
Compare
|
@chiinlquah, @scottsand-db, @dengsh12, @scovich, @DrakeLin, @rliao147 could you approve the execution of the Github actions? I did some changes trying to fix any failing actions. Also, can someone be assigned for review if is not too much? Thank you very much 🙏 |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
AI Review (draft - human review required)
Show review
This adds an opt-in parallel JSON log read path. The ordering design is sound: chunks are contiguous slices of the input file list, each chunk runs the existing read_json_files_impl (which preserves per-file order), and flat_map over the in-order (receiver, handle) pairs drains chunks strictly in order, so the global yield order matches the serial path. The protocol reviewer confirmed log-replay ordering and last-write-wins reconciliation are unaffected, and the panic-to-JoinFailure handling correctly avoids a silent truncated commit stream. Two issues in the concurrency plumbing should be fixed before merge.
Blocking issues
Blocker1 - default-engine/src/json.rs:183 (spawn) and :212 (drain)
Chunk tasks are spawned with tokio::spawn and referenced only by their JoinHandle. Dropping a JoinHandle detaches the task, it does not abort it. When the consumer cancels (CancellableStreamIterator drops the outer stream on a cancelled token) or stops early (for example try_collect short-circuits on the first Err), the receivers and handles are dropped but the spawned tasks keep issuing object-store GETs (up to buffered(per_chunk_buffer) in flight) until their next tx.send() observes the closed channel. This contradicts the cancellation contract documented on stream_future_to_cancellable_iter that dropping the stream releases in-flight buffered work.
Raised by: maintainer-codex-reviewer, maintainer-claude-reviewer
Suggested fix: hold the handles in a tokio::task::JoinSet, or wrap each in an abort-on-drop handle (for example tokio_util::task::AbortOnDropHandle), so dropping the stream aborts outstanding tasks. Still await each handle on the normal path to preserve JoinFailure detection. Add a test that drops the iterator mid-stream and asserts the tasks stop.
Blocker2 - default-engine/src/json.rs:167,172
channel_cap = per_chunk_buffer.saturating_mul(4) and each chunk pipeline also keeps about per_chunk_buffer reads in flight, so aggregate peak is roughly 4 * buffer_size + buffer_size = about 5x buffer_size batches. Because chunks run concurrently but are consumed strictly in chunk order, every not-yet-drained chunk fills its channel while the consumer is still on chunk 0, so this is steady state rather than a rare worst case. It breaks the memory bound documented on with_buffer_size.
Raised by: maintainer-claude-reviewer
Suggested fix: drop the x4 multiplier (channel_cap = per_chunk_buffer) or use a small fixed cap, and update the with_buffer_size doc to describe behavior under parallelism.
Non-blocking notes
Nit1 - default-engine/src/lib.rs:298
The builder doc says Some(n) "splits the file list into n chunks", but the actual count can be lower: num_chunks is clamped to files.len() and chunk_size uses div_ceil, so 8 files with n=5 yields 4 chunks, and fewer files than n yields at most files.len() chunks.
Raised by: docs-reviewer
Suggested fix: describe n as an upper bound, for example "splits the file list into up to n chunks (fewer when there are fewer files than n)".
Nit2 - default-engine/src/json.rs:183
The handler holds task_executor to inject the async runtime, but the parallel path calls tokio::spawn directly, coupling it to the ambient Tokio runtime. Under the default TokioBackgroundExecutor (a current-thread runtime) chunk tasks multiplex on one thread and the speedup does not materialize, which is why the benchmark had to switch to TokioMultiThreadExecutor. The public with_parallel_chunks knob therefore has no effect on the default engine.
Raised by: architecture-reviewer, maintainer-codex-reviewer, maintainer-claude-reviewer
Suggested fix: spawn through the executor abstraction (extending TaskExecutor::spawn to return a joinable handle, which is what powers JoinFailure detection), or document that with_parallel_chunks requires a multi-thread executor.
Nit3 - default-engine/src/json.rs:197
Negative and concurrency paths are only exercised with a single file, which is a single chunk. An error originating in a non-first chunk, the consumer-drop/backpressure path (tx.send().await.is_err()), and non-divisible chunk boundaries are untested; value-asserting tests only use exact multiples.
Raised by: test-coverage-reviewer
Suggested fix: add a test where a later chunk errors after earlier chunks produced data (assert prior data arrives before the error), a test that drops the iterator mid-stream, and ragged chunk cases such as 7 files with 3 chunks.
Summary
The parallelization preserves the ordering guarantees log replay depends on and hardens against silent commit truncation, so the feature is functionally correct on the happy path. Before merge, tie spawned task lifetime to the stream so cancellation and early drop stop background I/O, and rein in the channel sizing so the parallel path respects the documented buffer_size memory bound. The executor bypass and test gaps are worth addressing but are not blocking.
Automated review - workflow run
| let predicate = predicate.clone(); | ||
|
|
||
| let handle = tokio::spawn(async move { | ||
| let result = read_json_files_impl( |
There was a problem hiding this comment.
Nit2 The handler holds task_executor to inject the async runtime, but the parallel path calls tokio::spawn directly, coupling it to the ambient Tokio runtime. Under the default TokioBackgroundExecutor (current-thread runtime) chunk tasks multiplex on one thread and the speedup does not materialize, which is why the benchmark switched to TokioMultiThreadExecutor; the public with_parallel_chunks knob has no effect on the default engine. Raised by: architecture-reviewer, maintainer-codex-reviewer, maintainer-claude-reviewer. Suggested fix: spawn through the executor abstraction (extending TaskExecutor::spawn to return a joinable handle), or document that with_parallel_chunks requires a multi-thread executor.
| Ok(batch_stream) => { | ||
| let mut batch_stream = std::pin::pin!(batch_stream); | ||
| while let Some(batch) = batch_stream.next().await { | ||
| if tx.send(batch).await.is_err() { |
There was a problem hiding this comment.
Nit3 Negative and concurrency paths are only exercised with a single file (single chunk). An error originating in a non-first chunk, the consumer-drop/backpressure path (tx.send().await.is_err()), and non-divisible chunk boundaries are untested; value-asserting tests use only exact multiples. Raised by: test-coverage-reviewer. Suggested fix: add a test where a later chunk errors after earlier chunks produced data, a test that drops the iterator mid-stream, and ragged chunk cases such as 7 files with 3 chunks.
8dbd959 to
b52c222
Compare
|
I tried to address the feedback from |
b52c222 to
114236f
Compare
|
Once more please @chiinlquah, @scottsand-db, @dengsh12, @scovich, @DrakeLin, @rliao147? I feel bad to ask you so many times for this PR. 🙏 |
|
@andrei-ionescu No worries : ) Thanks for the contribution!! Will try to get to review the PR soon as well |
|
Thank you @chiinlquah 😄 |
…#3246) ## What changes are proposed in this pull request? Opt-in parallel JSON log reads for snapshot construction. `None` (default) keeps the existing serial path. `Some(n)` splits the ordered file list into `n` chunks, parses them concurrently as tokio tasks, and concatenates results in proper order. Here are some performance results: | Run | Serial | Parallel | Speedup | |-----|---------|----------|---------| | 1 | 96.6 ms | 40.0 ms | 2.42x | | 2 | 95.9 ms | 39.7 ms | 2.42x | | 3 | 96.5 ms | 39.6 ms | 2.44x | | 4 | 94.6 ms | 39.1 ms | 2.42x | This is **2.42x faster** (saved ~59% of time spent)! ## How was this change tested? - `test_read_json_files_parallel_ordering` with 10k files, assert row order - `test_drain_chunk_panicked_task_is_join_failure_not_eof` checking panic yields `JoinFailure`, not EOF - `cargo nextest run -p delta_kernel_default_engine --lib --all-features` - `cargo bench -p delta_kernel --bench metadata_bench -- create_snapshot`
114236f to
9245f77
Compare
|
Rebased and ready for what's next... @chiinlquah, @scottsand-db, @dengsh12, @scovich, @DrakeLin, @rliao147? |
2 participants
What changes are proposed in this pull request?
Opt-in parallel JSON log reads for snapshot construction.
None(default) keeps the existing serial path.Some(n)splits the ordered file list intonchunks, parses them concurrently as tokio tasks, and concatenates results in proper order.Here are some performance results:
Important
This is 2.41x faster (saved ~59% of time spent)!
This metrics were extracted from a similar output of
cargo bench -p delta_kernel --bench metadata_bench -- create_snapshot:What does it fix?
Closes #3246
How was this change tested?
test_read_json_files_parallel_orderingwith 10k files, assert correct ordertest_drain_chunk_panicked_task_is_join_failure_not_eofchecking panic yieldsJoinFailure, not EOFcargo +nightly fmt --checkcargo testcargo nextest run -p delta_kernel_default_engine --lib --all-featurescargo bench -p delta_kernel --bench metadata_bench -- create_snapshot