Repository navigation
feat(scan): let a scan request bound the splits it decodes at once - #64
Conversation
ScanRequest gains split_concurrency: the most splits of each partition decoded at once, in all. Both layout data sources pass it to ScanBuilder::with_some_max_splits_in_flight, a cap kept apart from the per-worker concurrency. JNI takes it as ScanOptions.splitConcurrency, and also bounds the Arrow conversion it runs ahead of Java by it; Spark takes it as the vortex.splitConcurrency read option. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Christopher Wirt <[email protected]>
Drive capped Arrow streams directly instead of feeding the runtime result queue, which can retain one extra completed batch per core. Preserve the existing queue for uncapped scans. Add paused-consumer regressions with two background workers. Saturate per-worker concurrency scaling before applying the absolute cap and add an overflow regression. Clarify that split concurrency reduces decoded memory rather than bounding total native memory or I/O read-ahead. Signed-off-by: Christopher Wirt <[email protected]>
512a010 to
31ae92c
Compare
gitRasheed
left a comment
gitRasheed
left a comment
There was a problem hiding this comment.
Looks good overall. The bound applies where decoding happens, and callers that don't set it see no change.
One thing worth fixing before merge: the JNI lookahead cap doesn't bound what sits ahead of Java. block_on_stream_thread_safe (vortex-io/src/runtime/current.rs:75) sends results into a channel sized to available_parallelism, so with splitConcurrency=1 on 12 cores up to 12 converted batches, plus one blocked send, can still queue while Java is slow. The ScanOptions Javadoc promises otherwise. Either carry the bound into that channel or narrow the Javadoc. A test with several batches and a paused consumer would catch it; the Java and Spark tests only check that every row arrives.
Minor: with a filter and a limit, execute_stream buffers at two stages (repeated_scan.rs:283 and :290), so it's about 2n split arrays. The builder doc already says so, but the Spark docs row reads like a hard cap.
Signed-off-by: Christopher Wirt <[email protected]>
|
Thanks, both points are fair. The queue issue is addressed in 31ae92c: when a split cap is set, JNI drives the stream directly instead of feeding the per-core result channel. Uncapped scans keep the existing path. That commit also adds paused-consumer regressions for caps of 1 and 2, with background workers running. I've pushed 11e695f to make the other caveat explicit in the Spark docs and Java option Javadocs. With both a filter and a limit, the mapping stage can retain another trimmed array per split, so the option reduces concurrent decode/conversion memory but isn't a cap on total native memory or I/O read-ahead. The Java docs still said it bounded native memory; that wording is gone too. This follow-up only changes docs; I haven't rerun the test suites. |
|
Rust's For our Spark workloads, we should start with |
gitRasheed
left a comment
gitRasheed
left a comment
There was a problem hiding this comment.
Both fixes look right. Capped scans now skip the per-core result channel, and the paused-consumer tests cover caps of 1 and 2. Agreed on starting at vortex.splitConcurrency=4 with spark.task.cpus=1 and measuring; defaulting to 4 x TaskContext.cpus() later makes sense.
Closes #63. Helps with #61, but doesn't fix it.
Problem
4 × available_parallelism()splits at once.Change
An absolute bound on splits decoded at once, settable per scan:
ScanRequest::split_concurrency→ScanBuilder::with_some_max_splits_in_flightScanOptions.splitConcurrency()vortex.splitConcurrencyread optionwith_concurrency. A laterwith_concurrency(as the DataFusion opener calls) leaves it in place, andconcurrency()keeps its meaning.buffered(worker_count)) by the same value.Effect
A full scan of a 180 MB list-layout file, peak live native bytes:
Tests
with_concurrencykeeps the bound.ScanOptionsvalidation, and a scan with the bound reads every row.🤖 Generated with Claude Code