DuckDB V2 scan: cut chunks on file boundaries, default it for TPC-DS and ClickBench

bench-orchestrator/bench_orchestrator/runner/executor.py

@@ -17,6 +17,12 @@
 
 console = Console()
 
+# The benchmarks each engine runs on the planner-driven V2 Vortex scan.
+SCAN_V2_BENCHMARKS: dict[Engine, frozenset[Benchmark]] = {
+    Engine.DATAFUSION: frozenset({Benchmark.TPCH}),
+    Engine.DUCKDB: frozenset({Benchmark.TPCH, Benchmark.TPCDS, Benchmark.CLICKBENCH}),
+}
+
 
 @final
 class BenchmarkExecutor:
@@ -56,6 +62,8 @@ def build_command(
             cmd.extend(["--formats", ",".join(fmt.value for fmt in formats)])
         if self.backend == Engine.DUCKDB:
             cmd.append("--delete-duckdb-database")
+        if benchmark in SCAN_V2_BENCHMARKS.get(self.backend, frozenset()):
+            cmd.extend(["--scan-v2", "true"])
 
         if query is not None:
             cmd.extend(["--queries", str(query)])

bench-orchestrator/tests/test_executor.py

@@ -5,6 +5,7 @@
 import textwrap
 from pathlib import Path
 
+import pytest
 from bench_orchestrator.config import Benchmark, Engine, Format
 from bench_orchestrator.runner.executor import BenchmarkExecutor
 
@@ -33,6 +34,29 @@ def test_build_command_adds_duckdb_cleanup_flag() -> None:
     assert cmd[cmd.index("--opt") + 1] == "scale-factor=1.0"
 
 
+@pytest.mark.parametrize(
+    ("engine", "benchmark", "expected"),
+    [
+        (Engine.DUCKDB, Benchmark.TPCDS, True),
+        (Engine.DUCKDB, Benchmark.CLICKBENCH, True),
+        (Engine.DUCKDB, Benchmark.TPCH, True),
+        (Engine.DUCKDB, Benchmark.FINEWEB, False),
+        (Engine.DATAFUSION, Benchmark.TPCH, True),
+        (Engine.DATAFUSION, Benchmark.CLICKBENCH, False),
+        (Engine.LANCE, Benchmark.TPCH, False),
+    ],
+)
+def test_build_command_turns_on_scan_v2(engine: Engine, benchmark: Benchmark, expected: bool) -> None:
+    executor = BenchmarkExecutor(Path("/tmp/bench"), engine)
+
+    cmd = executor.build_command(benchmark=benchmark, formats=[Format.VORTEX])
+
+    if expected:
+        assert cmd[cmd.index("--scan-v2") + 1] == "true"
+    else:
+        assert "--scan-v2" not in cmd
+
+
 def test_build_command_serializes_vortex_spatial_native_format() -> None:
     executor = BenchmarkExecutor(Path("/tmp/duckdb-bench"), Engine.DUCKDB)
 

benchmarks/datafusion-bench/src/main.rs

@@ -120,9 +120,9 @@ struct Args {
     #[arg(long, value_delimiter = ',', value_parser = value_parser!(Format))]
     formats: Vec<Format>,
 
-    /// Use the planner-driven V2 Vortex scan (defaults to true for TPC-H).
-    #[arg(long, action = clap::ArgAction::Set)]
-    scan_v2: Option<bool>,
+    /// Use the planner-driven V2 Vortex scan.
+    #[arg(long, default_value_t = false, action = clap::ArgAction::Set)]
+    scan_v2: bool,
 
     #[arg(long = "opt", value_delimiter = ',', value_parser = value_parser!(Opt))]
     options: Vec<Opt>,
@@ -136,10 +136,7 @@ async fn main() -> anyhow::Result<()> {
     vortex_bench::configure_allocator();
     let args = Args::parse();
     let opts = Opts::from(args.options);
-    let scan_v2 = args.scan_v2.unwrap_or_else(|| {
-        matches!(args.benchmark, BenchmarkArg::TpcH)
-            && !std::env::var("VORTEX_SCAN_V2").is_ok_and(|v| v == "0")
-    });
+    let scan_v2 = args.scan_v2;
 
     set_join_set_tracer(get_static_tracer())?;
     setup_logging_and_tracing(args.verbose, args.tracing)?;

benchmarks/duckdb-bench/src/main.rs

@@ -103,9 +103,9 @@ struct Args {
     )]
     reuse: bool,
 
-    /// Use the planner-driven V2 Vortex scan (defaults to true for TPC-H).
-    #[arg(long, action = clap::ArgAction::Set)]
-    scan_v2: Option<bool>,
+    /// Use the planner-driven V2 Vortex scan.
+    #[arg(long, default_value_t = false, action = clap::ArgAction::Set)]
+    scan_v2: bool,
 }
 
 #[global_allocator]
@@ -116,11 +116,9 @@ fn main() -> anyhow::Result<()> {
     let args = Args::parse();
     let opts = Opts::from(args.options);
 
-    let scan_v2 = args.scan_v2.unwrap_or_else(|| {
-        matches!(args.benchmark, BenchmarkArg::TpcH)
-            && !std::env::var("VORTEX_SCAN_V2").is_ok_and(|v| v == "0")
-    });
-    vortex_duckdb::set_scan_v2(scan_v2);
+    if args.scan_v2 {
+        vortex_duckdb::set_scan_v2(true);
+    }
 
     setup_logging_and_tracing(args.verbose, args.tracing)?;
 

vortex-duckdb/src/e2e_test/vortex_scan_test.rs

@@ -1410,6 +1410,10 @@ fn query_i64_column(path: &str, query: &str) -> Result<Vec<i64>> {
 #[case::expression("SELECT a + b FROM ? WHERE a >= 50000 ORDER BY 1")]
 #[case::full_scan("SELECT a FROM ? ORDER BY a")]
 #[case::empty("SELECT a FROM ? WHERE b > 5000")]
+#[case::top_n("SELECT a FROM ? ORDER BY b DESC, a LIMIT 7")]
+#[case::filtered_top_n("SELECT a FROM ? WHERE b < 500 ORDER BY a DESC LIMIT 10")]
+// DuckDB fetches the columns beyond the sort key by row id, as in ClickBench q23.
+#[case::late_materialized_top_n("SELECT * FROM ? WHERE b < 700 ORDER BY a DESC LIMIT 5")]
 fn test_vortex_scan_v2_matches_v1(#[case] query: &str) -> Result<()> {
     // Many row blocks, so the file is cut into several V2 splits.
     let rows = 100_000i64;

vortex-duckdb/src/file_reader.rs

@@ -12,6 +12,7 @@ use url::Url;
 use vortex::array::ArrayRef;
 use vortex::array::VortexSessionExecute as _;
 use vortex::array::arrays::struct_::StructArrayExt as _;
+use vortex::array::scalar_fn::fns::dynamic::DynamicComparison;
 use vortex::dtype::FieldPath;
 use vortex::error::VortexExpect;
 use vortex::error::VortexResult;
@@ -28,6 +29,8 @@ use vortex::io::runtime::BlockingRuntime as _;
 use vortex::io::std_file::StdFileSystem;
 use vortex::layout::LayoutReaderRef;
 use vortex::layout::scan::scan_builder::ScanBuilder;
+use vortex::layout::scan::scan_builder::referenced_field_masks;
+use vortex::layout::scan::split_by::SplitBy;
 use vortex::layout::scan::v2::ScanBuilder as V2ScanBuilder;
 use vortex::layout::scan::v2::ScanFile;
 use vortex::mask::Mask;
@@ -216,7 +219,18 @@ pub fn reader_initialize(file: &mut OpenFileReader, global: &GlobalState) -> Vor
     // global lock.
     if global.scan_v2 {
         let state = file.scan_state();
-        let scan = FileScan::try_new(v2_builder(state, global))?;
+        let dynamic = global
+            .filter
+            .filter
+            .as_ref()
+            .map(|filter| filter.contains::<DynamicComparison>())
+            .transpose()?
+            .unwrap_or(false);
+        let scan = FileScan::try_new(
+            v2_builder(state, global),
+            &v2_boundaries(state, global)?,
+            dynamic,
+        )?;
         if state.v2.set(scan).is_err() {
             vortex_panic!("file initialized twice");
         }
@@ -257,6 +271,16 @@ fn v2_builder(state: &FileScanState, global: &GlobalState) -> V2ScanBuilder {
         .with_row_range(rows)
 }
 
+/// The rows the file's chunks of the columns the scan reads end at, the boundaries V1 splits on.
+fn v2_boundaries(state: &FileScanState, global: &GlobalState) -> VortexResult<Vec<u64>> {
+    let field_mask = referenced_field_masks(&global.projection, global.filter.filter.as_ref())?;
+    SplitBy::default().splits(
+        state.reader.as_ref(),
+        &(0..state.reader.row_count()),
+        &field_mask,
+    )
+}
+
 /// What a thread's split yields next.
 enum SplitItem {
     /// The split has no more rows.
@@ -301,13 +325,13 @@ fn next_split_item(local: &mut LocalState) -> VortexResult<SplitItem> {
 /// the global lock. It only moves references between the file and the local
 /// state, and reader_scan drops the retired ones.
 pub fn reader_try_initialize_scan(file: &mut OpenFileReader, local: &mut LocalState) -> bool {
-    // A V2 scan's threads take chunks from the file's scan until it has spawned
-    // them all, so the file has work for another thread until then.
+    // A V2 scan's threads take chunks from the file's queue until it is empty,
+    // so the file has work for one more thread per queued chunk at most.
     let v2 = file
         .scan_state
         .as_ref()
         .and_then(|state| state.v2.get())
-        .map(|scan| (!scan.is_exhausted()).then_some(Split::V2));
+        .map(|scan| scan.claim().then_some(Split::V2));
     let Some(split) = v2.unwrap_or_else(|| file.splits.pop()) else {
         // DuckDB releases the file after this call, under the global lock.
         // Give the reference of the file to this thread so it drops outside.