fix(scanner): honor batch_readahead to bound v2 scan decode concurrency#7632
Conversation
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
|
|
||
| // Bound the decode fan-out by `batch_readahead`. | ||
| read_options = read_options.with_threading_mode( | ||
| FilteredReadThreadingMode::OnePartitionMultipleThreads(self.batch_readahead), |
There was a problem hiding this comment.
Do we need to make sure self.batch_readahead is not equal to 0?
| /// This controls how decode work is parallelized. For the default single-partition | ||
| /// scan, the parameter of [`FilteredReadThreadingMode::OnePartitionMultipleThreads`] | ||
| /// bounds how many batch-decode tasks are buffered in flight (via `try_buffered`). | ||
| pub fn with_threading_mode(mut self, threading_mode: FilteredReadThreadingMode) -> Self { |
There was a problem hiding this comment.
Do we allow OnePartitionMultipleThreads(0) / MultiplePartitions(0) or any check the input parameter?
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
💤 Files with no reviewable changes (1)
📝 WalkthroughWalkthrough
ChangesBatch readahead behavior
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant Scanner
participant FilteredReadOptions
participant FilteredReadExec
Scanner->>FilteredReadOptions: set decode threading from batchReadahead
FilteredReadOptions->>FilteredReadExec: provide threading mode
FilteredReadExec->>FilteredReadExec: reject zero parallelism
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
java/src/main/java/org/lance/ipc/ScanOptions.java (1)
105-105: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winJavadoc doesn't document the new
batchReadahead > 0constraint.The constructor validates
batchReadahead > 0(lines 137-138), but the@param batchReadaheadJavadoc (line 105),getBatchReadahead()Javadoc (lines 270-274), andBuilder.batchReadahead()Javadoc (lines 559-564) still only say "Number of batches to read ahead." / "the number of batches to read ahead." with no mention of the positive-value requirement, so callers have no way to discover this constraint without hitting the runtime exception.As per coding guidelines, "Copy Rust documentation about defaults, constraints, and invariants into Javadoc for binding classes."
📝 Suggested doc update
/** * Set the batch readahead. * * `@param` batchReadahead Number of batches to read ahead. + * Must be greater than 0. * `@return` Builder instance for method chaining. */ public Builder batchReadahead(int batchReadahead) {Also applies to: 271-277, 559-568
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@java/src/main/java/org/lance/ipc/ScanOptions.java` at line 105, Update the Javadocs for the ScanOptions constructor’s batchReadahead parameter, getBatchReadahead(), and Builder.batchReadahead() to explicitly state that batchReadahead must be greater than zero, preserving the existing description while documenting the validation constraint consistently.Source: Coding guidelines
java/src/test/java/org/lance/ScannerTest.java (1)
419-455: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winAdd a
batchReadahead(0)rejection case
ScannerTestonly covers the low-readahead happy path. Add anassertThrows(IllegalArgumentException.class, () -> new ScanOptions.Builder().batchReadahead(0).build())case so the zero-value validation stays covered here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@java/src/test/java/org/lance/ScannerTest.java` around lines 419 - 455, Add a validation test in ScannerTest, alongside testDatasetScannerBatchReadahead, that asserts new ScanOptions.Builder().batchReadahead(0).build() throws IllegalArgumentException using assertThrows.
🧹 Nitpick comments (1)
python/python/lance/dataset.py (1)
6184-6193: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
batch_readaheaddocstring lacks an example.The updated docstring documents the new constraint but has no usage example or links to related methods.
As per coding guidelines, "All public APIs must have documentation with examples, and documentation should link to relevant structs and methods" for
**/*.{rs,md,py}files.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/python/lance/dataset.py` around lines 6184 - 6193, Add a usage example and relevant cross-references to the public `batch_readahead` docstring, showing a valid positive value and linking to related scanner configuration methods or APIs, while preserving the documented greater-than-zero constraint.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 1490-1499: Validate the thread count before accepting it in
`with_threading_mode` (or centrally in `FilteredReadExec::try_new`), rejecting
`OnePartitionMultipleThreads(0)` and `MultiplePartitions(0)` with
`Error::invalid_input_source`. Ensure the guard applies to all public
construction paths so `get_stream` never calls `try_buffered` with zero.
---
Outside diff comments:
In `@java/src/main/java/org/lance/ipc/ScanOptions.java`:
- Line 105: Update the Javadocs for the ScanOptions constructor’s batchReadahead
parameter, getBatchReadahead(), and Builder.batchReadahead() to explicitly state
that batchReadahead must be greater than zero, preserving the existing
description while documenting the validation constraint consistently.
In `@java/src/test/java/org/lance/ScannerTest.java`:
- Around line 419-455: Add a validation test in ScannerTest, alongside
testDatasetScannerBatchReadahead, that asserts new
ScanOptions.Builder().batchReadahead(0).build() throws IllegalArgumentException
using assertThrows.
---
Nitpick comments:
In `@python/python/lance/dataset.py`:
- Around line 6184-6193: Add a usage example and relevant cross-references to
the public `batch_readahead` docstring, showing a valid positive value and
linking to related scanner configuration methods or APIs, while preserving the
documented greater-than-zero constraint.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 5fa73b89-be82-47ef-a13c-41a77af2c5d8
📒 Files selected for processing (5)
java/src/main/java/org/lance/ipc/ScanOptions.javajava/src/test/java/org/lance/ScannerTest.javapython/python/lance/dataset.pyrust/lance/src/dataset/scanner.rsrust/lance/src/io/exec/filtered_read.rs
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
rust/lance/src/io/exec/filtered_read.rs (2)
1486-1497: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDocument both threading modes with a runnable example.
This public builder documents only
OnePartitionMultipleThreadsand has no example. Add ano_runexample using the current signature and explain or link both enum variants.As per coding guidelines, all public Rust APIs must have documentation with examples and links to relevant structs and methods.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 1486 - 1497, Expand the documentation for FilteredReadExec::with_threading_mode to describe and link both FilteredReadThreadingMode variants, including their respective behavior. Add a runnable `no_run` Rust example using the current builder signature and showing threading mode configuration, with links to relevant types and methods.Source: Coding guidelines
3933-3939: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winParameterize the mode cases with
rstest.These cases differ only by input. Replace the loop with named
#[case::...]parameters and one test body.As per coding guidelines, Rust tests whose cases differ only by inputs must use
rstestwith readable case names.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 3933 - 3939, The test currently loops over two threading modes that differ only by input; refactor it to use an rstest parameterized test with readable named cases such as #[case::...] for each FilteredReadThreadingMode value. Keep a single test body that receives the mode parameter and uses it to build FilteredReadOptions and invoke FilteredReadExec::try_new.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 3939-3946: Update the test around FilteredReadExec::try_new to
retain the returned error instead of converting it directly to a string, then
match it as Error::InvalidInput and assert that the contained message includes
"must be greater than 0"; ensure other variants fail the test.
---
Nitpick comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 1486-1497: Expand the documentation for
FilteredReadExec::with_threading_mode to describe and link both
FilteredReadThreadingMode variants, including their respective behavior. Add a
runnable `no_run` Rust example using the current builder signature and showing
threading mode configuration, with links to relevant types and methods.
- Around line 3933-3939: The test currently loops over two threading modes that
differ only by input; refactor it to use an rstest parameterized test with
readable named cases such as #[case::...] for each FilteredReadThreadingMode
value. Keep a single test body that receives the mode parameter and uses it to
build FilteredReadOptions and invoke FilteredReadExec::try_new.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: af689aef-007a-413d-bbdd-19c122f50a17
📒 Files selected for processing (1)
rust/lance/src/io/exec/filtered_read.rs
|
HI @yanghua Thanks a lot for your review. All comments are addressed. PTAL ~ |
2 participants
Background & Motivation
Under concurrent-scan workloads like Spark, a single large worker (Executor) typically runs many Tasks in parallel (in Lance's read path, 1 Task = 1 fragment = 1 scan). The problem: each Task independently sizes its own scan concurrency from the core count of the current Executor / host.
Concretely in Lance, the decode concurrency of the v2 read path
FilteredReadExec(plan nameLanceRead) — thetry_buffered(num_threads)— is unconditionally set toget_num_compute_intensive_cpus()=num_cpus − 2.Why It Was Deprecated, and the Gap It Left Behind
Scanner.batch_readaheadis markedIgnored in v2 and newer format— a deliberate decision, not an oversight. In v1 it controlled two things: prefetch depth and decode concurrency. v2 replaced prefetch with a byte-budget model (io_buffer_size+fragment_readahead), so its prefetch role became meaningless and was rightly dropped.The gap: v2 split the decode-concurrency role into
FilteredReadExec's threading mode (try_buffered(num_threads)), hard-coded toget_num_compute_intensive_cpus(). The only override today is the process-wideLANCE_CPU_THREADSenv var — far too coarse, since it governs every compute-intensive path at once (vector/KNN search, index building,take, update/merge-insert, …), not just scan decode concurrency. With no per-scan knob, there's no way to rein in the over-parallelization above.What This Change Does & Its Impact
Have
new_filtered_readpassScanner.batch_readaheadthrough asFilteredReadThreadingMode::OnePartitionMultipleThreads(batch_readahead), reattachingbatch_readaheadto the decode-concurrency dimension that v2 had left without a knob.Summary by CodeRabbit
Bug Fixes
batch_readahead/batchReadaheadvalues (0 or less) are now rejected consistently across Java, Python, and Rust.batch_readaheadmust be greater than 0.batch_readaheadto control decoding parallelism more reliably.Tests
batch_readaheadtests to validate deterministic scanned content in addition to row counts.