Tuning and fixes and stuff

Signed-off-by: Adam Gutglick <adam@spiraldb.com>

Author: AdamGS

vortex-array/src/arrow/executor/list.rs

@@ -2,8 +2,6 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::sync::Arc;
-use std::sync::atomic::AtomicU64;
-use std::sync::atomic::Ordering;
 
 use arrow_array::ArrayRef as ArrowArrayRef;
 use arrow_array::GenericListArray;
@@ -19,6 +17,7 @@ use crate::ArrayRef;
 use crate::Canonical;
 use crate::DynArray;
 use crate::ExecutionCtx;
+use crate::arrays::Chunked;
 use crate::arrays::List;
 use crate::arrays::ListArray;
 use crate::arrays::ListView;
@@ -33,31 +32,6 @@ use crate::dtype::NativePType;
 use crate::dtype::Nullability;
 use crate::vtable::ValidityHelper;
 
-static LIST_TO_LIST_COUNT: AtomicU64 = AtomicU64::new(0);
-static LIST_VIEW_ZCTL_COUNT: AtomicU64 = AtomicU64::new(0);
-static EXECUTE_LIST_VIEW_COUNT: AtomicU64 = AtomicU64::new(0);
-
-#[derive(Debug, Clone, Copy, Default)]
-pub struct ListArrowPathCounters {
-    pub list_to_list: u64,
-    pub list_view_zctl: u64,
-    pub execute_list_view: u64,
-}
-
-pub(crate) fn reset_list_arrow_path_counters() {
-    LIST_TO_LIST_COUNT.store(0, Ordering::Relaxed);
-    LIST_VIEW_ZCTL_COUNT.store(0, Ordering::Relaxed);
-    EXECUTE_LIST_VIEW_COUNT.store(0, Ordering::Relaxed);
-}
-
-pub(crate) fn list_arrow_path_counters() -> ListArrowPathCounters {
-    ListArrowPathCounters {
-        list_to_list: LIST_TO_LIST_COUNT.load(Ordering::Relaxed),
-        list_view_zctl: LIST_VIEW_ZCTL_COUNT.load(Ordering::Relaxed),
-        execute_list_view: EXECUTE_LIST_VIEW_COUNT.load(Ordering::Relaxed),
-    }
-}
-
 /// Convert a Vortex array into an Arrow GenericBinaryArray.
 pub(super) fn to_arrow_list<O: OffsetSizeTrait + NativePType>(
     array: ArrayRef,
@@ -66,14 +40,12 @@ pub(super) fn to_arrow_list<O: OffsetSizeTrait + NativePType>(
 ) -> VortexResult<ArrowArrayRef> {
     // If the Vortex array is already in List format, we can directly convert it.
     if let Some(array) = array.as_opt::<List>() {
-        LIST_TO_LIST_COUNT.fetch_add(1, Ordering::Relaxed);
         return list_to_list::<O>(array, elements_field, ctx);
     }
 
     // If the Vortex array is a ListViewArray, rebuild to ZCTL if needed and convert.
     let array = match array.try_into::<ListView>() {
         Ok(array) => {
-            LIST_VIEW_ZCTL_COUNT.fetch_add(1, Ordering::Relaxed);
             let zctl = if array.is_zero_copy_to_list() {
                 array
             } else {
@@ -84,18 +56,34 @@ pub(super) fn to_arrow_list<O: OffsetSizeTrait + NativePType>(
         Err(a) => a,
     };
 
+    // Handle ChunkedArray by converting each chunk individually.
+    // This preserves the fast list_to_list path for inner ListArray chunks
+    // instead of falling through to the expensive execute::<ListViewArray> path.
+    if let Some(chunked) = array.as_opt::<Chunked>() {
+        let mut arrow_chunks: Vec<ArrowArrayRef> = Vec::with_capacity(chunked.nchunks());
+        for chunk in chunked.chunks() {
+            arrow_chunks.push(to_arrow_list::<O>(chunk.clone(), elements_field, ctx)?);
+        }
+        if arrow_chunks.len() == 1 {
+            return Ok(arrow_chunks
+                .into_iter()
+                .next()
+                .vortex_expect("known length"));
+        }
+        let refs: Vec<&dyn arrow_array::Array> = arrow_chunks.iter().map(|a| a.as_ref()).collect();
+        return Ok(arrow_select::concat::concat(&refs)?);
+    }
+
     // Otherwise, we execute the array to become a ListViewArray, then rebuild to ZCTL.
     // Note: arrow_cast::cast supports ListView → List (apache/arrow-rs#8735), but it
     // unconditionally uses take. Our rebuild uses a heuristic that picks list-by-list
     // for large lists, which avoids materializing a large index buffer.
-    EXECUTE_LIST_VIEW_COUNT.fetch_add(1, Ordering::Relaxed);
     let list_view = array.execute::<ListViewArray>(ctx)?;
     let zctl = if list_view.is_zero_copy_to_list() {
         list_view
     } else {
         list_view.rebuild(ListViewRebuildMode::MakeZeroCopyToList)?
     };
-    LIST_VIEW_ZCTL_COUNT.fetch_add(1, Ordering::Relaxed);
     list_view_zctl::<O>(zctl, elements_field, ctx)
 }
 

vortex-array/src/arrow/executor/mod.rs

@@ -25,9 +25,6 @@ use arrow_schema::Field;
 use arrow_schema::FieldRef;
 use arrow_schema::Schema;
 use itertools::Itertools;
-pub(crate) use list::ListArrowPathCounters;
-pub(crate) use list::list_arrow_path_counters;
-pub(crate) use list::reset_list_arrow_path_counters;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_ensure;

vortex-array/src/arrow/mod.rs

@@ -15,9 +15,6 @@ mod null_buffer;
 mod record_batch;
 
 pub use datum::*;
-pub(crate) use executor::ListArrowPathCounters;
-pub(crate) use executor::list_arrow_path_counters;
-pub(crate) use executor::reset_list_arrow_path_counters;
 pub use executor::*;
 pub use iter::*;
 

vortex-array/src/debug.rs

@@ -41,7 +41,6 @@ pub mod builtins;
 mod canonical;
 mod columnar;
 pub mod compute;
-pub mod debug;
 pub mod display;
 pub mod dtype;
 mod executor;

vortex-array/src/lib.rs

@@ -3,8 +3,6 @@
 
 //! Generic benchmark runner infrastructure to reduce boilerplate across engine-specific benchmarks.
 
-use std::env;
-use std::fmt;
 use std::fs::File;
 use std::future::Future;
 use std::io::Write;
@@ -14,11 +12,7 @@ use std::time::Duration;
 use std::time::Instant;
 
 use indicatif::ProgressBar;
-use vortex::array::debug::arrow_path_counters;
-use vortex::array::debug::reset_arrow_path_counters;
 use vortex::error::vortex_panic;
-use vortex::scan::debug::counters as scan_debug_counters;
-use vortex::scan::debug::reset_counters as reset_scan_debug_counters;
 
 use crate::Benchmark;
 use crate::BenchmarkDataset;
@@ -51,47 +45,6 @@ use crate::measurements::QueryMeasurement;
 use crate::memory::BenchmarkMemoryTracker;
 use crate::url_scheme_to_storage;
 
-#[derive(Debug, Clone, Copy, Default)]
-struct DebugCounterSnapshot {
-    list_to_list: u64,
-    list_view_zctl: u64,
-    execute_list_view: u64,
-    legacy_stream_fallbacks: u64,
-}
-
-impl fmt::Display for DebugCounterSnapshot {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(
-            f,
-            "list_to_list={}, list_view_zctl={}, execute_list_view={}, legacy_stream_fallbacks={}",
-            self.list_to_list,
-            self.list_view_zctl,
-            self.execute_list_view,
-            self.legacy_stream_fallbacks
-        )
-    }
-}
-
-fn debug_counters_enabled() -> bool {
-    env::var_os("VORTEX_BENCH_DEBUG_COUNTERS").is_some()
-}
-
-fn reset_debug_counters() {
-    reset_arrow_path_counters();
-    reset_scan_debug_counters();
-}
-
-fn snapshot_debug_counters() -> DebugCounterSnapshot {
-    let arrow = arrow_path_counters();
-    let scan = scan_debug_counters();
-    DebugCounterSnapshot {
-        list_to_list: arrow.list_to_list,
-        list_view_zctl: arrow.list_view_zctl,
-        execute_list_view: arrow.execute_list_view,
-        legacy_stream_fallbacks: scan.legacy_stream_fallbacks,
-    }
-}
-
 /// Results from a benchmark run.
 pub struct BenchmarkResults {
     pub query_measurements: Vec<QueryMeasurement>,
@@ -177,9 +130,6 @@ impl SqlBenchmarkRunner {
         let mut row_count = None;
 
         for _ in 0..iterations {
-            if debug_counters_enabled() {
-                reset_debug_counters();
-            }
             let start = Instant::now();
             let (timing, result) = f();
             let elapsed = timing.unwrap_or_else(|| start.elapsed());
@@ -188,15 +138,6 @@ impl SqlBenchmarkRunner {
             if row_count.is_none() {
                 row_count = Some(result.row_count());
             }
-
-            if debug_counters_enabled() {
-                tracing::info!(
-                    query_idx,
-                    %format,
-                    counters = %snapshot_debug_counters(),
-                    "debug counters"
-                );
-            }
         }
 
         let row_count = row_count.expect("iterations must be > 0");
@@ -422,9 +363,6 @@ impl SqlBenchmarkRunner {
                         tracing::debug!(%format, query_idx, "Running query");
 
                         for _ in 0..iterations {
-                            if debug_counters_enabled() {
-                                reset_debug_counters();
-                            }
                             let start = Instant::now();
                             let (timing, result) = execute(query_idx, &ctx, query.as_str())
                                 .await
@@ -437,15 +375,6 @@ impl SqlBenchmarkRunner {
                             if row_count.is_none() {
                                 row_count = Some(result.row_count());
                             }
-
-                            if debug_counters_enabled() {
-                                tracing::info!(
-                                    query_idx,
-                                    %format,
-                                    counters = %snapshot_debug_counters(),
-                                    "debug counters"
-                                );
-                            }
                         }
 
                         let row_count = row_count.expect("iterations must be > 0");

vortex-bench/src/runner.rs

@@ -12,7 +12,6 @@ const IDEAL_SPLIT_SIZE: u64 = 100_000;
 
 pub mod api;
 pub mod arrow;
-pub mod debug;
 mod filter;
 pub mod row_mask;
 mod splits;

vortex-scan/src/debug.rs

@@ -6,8 +6,6 @@ use std::collections::BTreeMap;
 use std::collections::VecDeque;
 use std::ops::Range;
 use std::sync::Arc;
-use std::sync::atomic::AtomicU64;
-use std::sync::atomic::Ordering;
 use std::task::Context;
 use std::task::Poll;
 
@@ -44,22 +42,6 @@ use crate::tasks::split_exec;
 const ADAPTIVE_SELECTIVITY_SAMPLE_SPLITS: usize = 4;
 const HIGH_SURVIVOR_RATIO: f64 = 0.75;
 const MAX_PROJECTION_SUBSPLITS_PER_SPLIT: usize = 8;
-static LEGACY_STREAM_FALLBACK_COUNT: AtomicU64 = AtomicU64::new(0);
-
-#[derive(Debug, Clone, Copy, Default)]
-pub struct ScanDebugCounters {
-    pub legacy_stream_fallbacks: u64,
-}
-
-pub(crate) fn reset_scan_debug_counters() {
-    LEGACY_STREAM_FALLBACK_COUNT.store(0, Ordering::Relaxed);
-}
-
-pub(crate) fn scan_debug_counters() -> ScanDebugCounters {
-    ScanDebugCounters {
-        legacy_stream_fallbacks: LEGACY_STREAM_FALLBACK_COUNT.load(Ordering::Relaxed),
-    }
-}
 
 fn should_prefer_immediate_projection(
     observed_filter_splits: usize,
@@ -314,7 +296,6 @@ impl<A: 'static + Send> RepeatedScan<A> {
             split_ranges_exhausted,
             filter_ahead,
         ) {
-            LEGACY_STREAM_FALLBACK_COUNT.fetch_add(1, Ordering::Relaxed);
             return self.legacy_stream_from_ranges(
                 ctx,
                 prefetched_ranges,