Default duckdb-bench to the V2 scan for TPC-DS and ClickBench (#11047)

Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>

Author: joseph-isaacs

benchmarks/duckdb-bench/src/main.rs

@@ -103,7 +103,7 @@ struct Args {
     )]
     reuse: bool,
 
-    /// Use the planner-driven V2 Vortex scan (defaults to true for TPC-H).
+    /// Use the planner-driven V2 Vortex scan (defaults to true for TPC-H, TPC-DS and ClickBench).
     #[arg(long, action = clap::ArgAction::Set)]
     scan_v2: Option<bool>,
 }
@@ -117,8 +117,10 @@ fn main() -> anyhow::Result<()> {
     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")
+        matches!(
+            args.benchmark,
+            BenchmarkArg::TpcH | BenchmarkArg::TpcDS | BenchmarkArg::ClickBench
+        ) && !std::env::var("VORTEX_SCAN_V2").is_ok_and(|v| v == "0")
     });
     vortex_duckdb::set_scan_v2(scan_v2);
 

vortex-duckdb/src/exporter/mod.rs

@@ -422,6 +422,26 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_set_validity_values_with_bit_buffer_offset() {
+        let logical_type = LogicalType::new(DUCKDB_TYPE::DUCKDB_TYPE_BIGINT);
+        let mut vector = Vector::with_capacity(&logical_type, 100);
+
+        let bits = (0..80).map(|i| i % 3 == 0).collect::<Vec<_>>();
+        let bit_buffer = BitBuffer::from(bits.as_slice()).slice(2..80);
+        assert_eq!(bit_buffer.offset(), 2);
+        let mask = Mask::from(bit_buffer);
+
+        let all_null = unsafe { vector.set_validity_zero_copy(&mask, 5, 60, None) };
+
+        assert!(!all_null);
+
+        let validity = unsafe { vector.validity_bitslice_mut(60).unwrap() };
+        for i in 0..60 {
+            assert_eq!(validity[i], bits[i + 2 + 5]);
+        }
+    }
+
     #[test]
     fn test_set_validity_values_64bit_alignment() {
         let logical_type = LogicalType::new(DUCKDB_TYPE::DUCKDB_TYPE_BIGINT);

vortex-duckdb/src/exporter/vector.rs

@@ -60,11 +60,11 @@ impl VectorRef {
             unsafe { self.set_validity_data(u64_offset, len, validity_data) };
         } else {
             debug_assert!(zero_copy.is_none() || !offset.is_multiple_of(64));
-            let source = values.bit_buffer().inner().as_slice();
+            let bits = values.bit_buffer();
             copy_from_slice(
                 unsafe { self.ensure_validity_slice(len) },
-                source,
-                offset,
+                bits.inner().as_slice(),
+                bits.offset() + offset,
                 len,
             );
         }

vortex-duckdb/src/file_reader.rs

@@ -11,7 +11,10 @@ use object_store::registry::ObjectStoreRegistry;
 use url::Url;
 use vortex::array::ArrayRef;
 use vortex::array::VortexSessionExecute as _;
+use vortex::array::arrays::Chunked;
+use vortex::array::arrays::chunked::ChunkedArrayExt as _;
 use vortex::array::arrays::struct_::StructArrayExt as _;
+use vortex::array::scalar_fn::fns::dynamic::DynamicExprUpdates;
 use vortex::dtype::FieldPath;
 use vortex::error::VortexExpect;
 use vortex::error::VortexResult;
@@ -221,7 +224,16 @@ 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 splits = ScanBuilder::new(SESSION.clone(), Arc::clone(&state.reader))
+            .with_projection(global.projection.clone())
+            .with_some_filter(global.filter.filter.clone())
+            .full_file_splits()?;
+        let dynamic = global
+            .filter
+            .filter
+            .as_ref()
+            .is_some_and(|filter| DynamicExprUpdates::new(filter).is_some());
+        let scan = FileScan::try_new(v2_builder(state, global), &splits, dynamic)?;
         if state.v2.set(scan).is_err() {
             vortex_panic!("file initialized twice");
         }
@@ -284,8 +296,23 @@ fn next_split_item(local: &mut LocalState) -> VortexResult<SplitItem> {
                 .v2
                 .get()
                 .vortex_expect("V2 split of a file without a V2 scan");
-            let Some(array) = scan.next_array(&mut local.run)? else {
-                return Ok(SplitItem::Done);
+            let array = match local.pending.pop() {
+                Some(array) => array,
+                None => {
+                    let Some(array) = scan.next_array(&mut local.run)? else {
+                        return Ok(SplitItem::Done);
+                    };
+                    // A V2 batch can join chunks of several layout chunks. Canonicalizing it
+                    // would decode their dictionaries, so export each chunk on its own, as a V1
+                    // scan yields them.
+                    match array.as_opt::<Chunked>() {
+                        Some(chunked) => {
+                            local.pending.extend(chunked.chunks().into_iter().rev());
+                            local.pending.pop().unwrap_or(array)
+                        }
+                        None => array,
+                    }
+                }
             };
             local.split = Some(Split::V2);
             Ok(if array.is_empty() {