feat(scan): let a scan request bound the splits it decodes at once - #11127

Summary

• Tracking Issue: Let a scan limit how many splits it decodes at once #11124

A scan decodes 4 × available_parallelism() splits at once. In Spark, every task in an executor runs its own scan sized to the whole machine, and each split holds the decoded chunks of every column it overlaps. On wide, nested files that is GiBs of native memory per executor (#11123).

This adds a per-scan limit on how many splits are decoded at once.

Result: a full scan of a 180 MB list-layout file, peak live native bytes:

Changes

• vortex-scan / vortex-layout: ScanRequest::split_concurrency, honoured by both layout data sources.
  • It is a cap, separate from the per-worker with_concurrency. A later with_concurrency call (as the DataFusion opener makes) leaves it in place, and concurrency() keeps its meaning.
  • Per-worker scaling saturates before the cap is applied, so it can't overflow.
• JNI: ScanOptions.splitConcurrency().
  • A capped scan drives its Arrow stream directly, not through the runtime's result queue, which can hold one extra finished batch per core. The Arrow conversion done ahead of Java is limited by the same value. Uncapped scans keep the existing queue.
• Spark: the vortex.splitConcurrency read option, for V1 and V2 reads, documented in docs/user-guide/spark.md.
• It doesn't limit I/O: segment reads for later splits may still be coalesced ahead of them. The docs say so.

Tests

• Rust: a scan of 64 one-row splits decodes exactly 1, and exactly 2, at once with the cap set, against 48 by default. A later with_concurrency keeps the cap. Concurrency scaling saturates instead of overflowing.
• JNI: ScanOptions validation, and a capped scan reads every row. Paused-consumer regressions with two background workers.
• Spark: option parsing, validation and serialization, plus V1 and V2 reads with the option set.

On the fork (quantcast#64, same change), the Rust tests for vortex-layout, vortex-scan, vortex-jni and vortex-file passed, along with the JNI, Spark 3.5 and Spark 4.0 suites.

This branch ports that change onto develop. The fork's Spark V1/V2 read test lived in a test class that develop doesn't have, so it moved into VortexV1FallbackTest. Only git diff --check has been run on the port.

API Changes

• New public field ScanRequest::split_concurrency. Code that builds ScanRequest with ..Default::default() is unaffected.
• New ScanOptions.splitConcurrency() builder option in JNI, and a new vortex.splitConcurrency Spark read option. Unset keeps today's behaviour.

AI assistance

Disclosed per the AI policy: AI tools helped write and review this change. Claude Code prepared this upstream port, issue and PR description at my request. I am the responsible contributor.

🤖 Generated with Claude Code