feat(vortex-geo): geometry Bbox zone-map statistic + distance-filter pruning

benchmarks/duckdb-bench/src/lib.rs

@@ -78,6 +78,11 @@ impl DuckClient {
         for stmt in &statements {
             self.connection().query(stmt)?;
         }
+        // After `LOAD spatial`, shadow `ST_DWithin` so radius filters push. No-op without it.
+        self.db
+            .as_ref()
+            .vortex_expect("DuckClient database accessed after close")
+            .register_st_dwithin_override()?;
         self.init_sql = statements;
         Ok(())
     }
@@ -127,6 +132,11 @@ impl DuckClient {
                 .vortex_expect("connection just opened")
                 .query(stmt)?;
         }
+        // Re-shadow `ST_DWithin` against the fresh instance.
+        self.db
+            .as_ref()
+            .vortex_expect("database just opened")
+            .register_st_dwithin_override()?;
 
         Ok(())
     }

vortex-array/src/aggregate_fn/plugin.rs

@@ -10,6 +10,7 @@ use crate::aggregate_fn::AggregateFn;
 use crate::aggregate_fn::AggregateFnId;
 use crate::aggregate_fn::AggregateFnRef;
 use crate::aggregate_fn::AggregateFnVTable;
+use crate::dtype::DType;
 
 /// Reference-counted pointer to an aggregate function plugin.
 pub type AggregateFnPluginRef = Arc<dyn AggregateFnPlugin>;
@@ -28,6 +29,12 @@ pub trait AggregateFnPlugin: 'static + Send + Sync {
     /// Deserialize an aggregate function from serialized metadata.
     fn deserialize(&self, metadata: &[u8], session: &VortexSession)
     -> VortexResult<AggregateFnRef>;
+
+    /// The default per-chunk zone statistic to store for a column of `input_dtype`, or `None` if
+    /// this aggregate isn't one.
+    fn zone_stat_default(&self, _input_dtype: &DType) -> Option<AggregateFnRef> {
+        None
+    }
 }
 
 impl std::fmt::Debug for dyn AggregateFnPlugin {
@@ -51,4 +58,8 @@ impl<V: AggregateFnVTable> AggregateFnPlugin for V {
         let options = AggregateFnVTable::deserialize(self, metadata, session)?;
         Ok(AggregateFn::new(self.clone(), options).erased())
     }
+
+    fn zone_stat_default(&self, input_dtype: &DType) -> Option<AggregateFnRef> {
+        AggregateFnVTable::zone_stat_default(self, input_dtype)
+    }
 }

vortex-array/src/aggregate_fn/session.rs

@@ -10,6 +10,7 @@ use vortex_session::SessionVar;
 
 use crate::aggregate_fn::AggregateFnId;
 use crate::aggregate_fn::AggregateFnPluginRef;
+use crate::aggregate_fn::AggregateFnRef;
 use crate::aggregate_fn::AggregateFnVTable;
 use crate::aggregate_fn::fns::all_nan::AllNan;
 use crate::aggregate_fn::fns::all_non_distinct::AllNonDistinct;
@@ -44,6 +45,7 @@ use crate::arrays::chunked::compute::aggregate::ChunkedArrayAggregate;
 use crate::arrays::dict::compute::is_constant::DictIsConstantKernel;
 use crate::arrays::dict::compute::is_sorted::DictIsSortedKernel;
 use crate::arrays::dict::compute::min_max::DictMinMaxKernel;
+use crate::dtype::DType;
 
 /// Session state for aggregate functions and encoding-specific aggregate kernels.
 ///
@@ -134,6 +136,19 @@ impl AggregateFnSession {
         self.registry.insert(id, pluginref);
     }
 
+    /// The default per-chunk zone statistics for a column of `input_dtype`, collected from every
+    /// registered aggregate's `zone_stat_default`.
+    pub fn zone_stat_defaults(&self, input_dtype: &DType) -> Vec<AggregateFnRef> {
+        self.registry.read(|registry| {
+            let mut fns: Vec<AggregateFnRef> = registry
+                .values()
+                .filter_map(|plugin| plugin.zone_stat_default(input_dtype))
+                .collect();
+            fns.sort_by_key(|f| f.id());
+            fns
+        })
+    }
+
     /// Returns the aggregate kernel registered for `array_id` and `agg_fn_id`, if any.
     ///
     /// Lookup first checks for a kernel registered for the exact aggregate function, then falls

vortex-array/src/aggregate_fn/vtable.rs

@@ -93,6 +93,12 @@ pub trait AggregateFnVTable: 'static + Sized + Clone + Send + Sync {
     /// Returns `None` if the aggregate function cannot be applied to the input dtype.
     fn return_dtype(&self, options: &Self::Options, input_dtype: &DType) -> Option<DType>;
 
+    /// If this aggregate should be computed as a default zone statistic for `input_dtype`, return
+    /// the bound aggregate to store. Default: not a zone-map default.
+    fn zone_stat_default(&self, _input_dtype: &DType) -> Option<AggregateFnRef> {
+        None
+    }
+
     /// DType of the intermediate partial accumulator state.
     ///
     /// Use a struct dtype when multiple fields are needed

vortex-bench/Cargo.toml

@@ -34,6 +34,8 @@ async-trait = { workspace = true }
 bzip2 = { workspace = true }
 clap = { workspace = true, features = ["derive"] }
 futures = { workspace = true }
+geo = { workspace = true }
+geo-traits = { workspace = true }
 geoarrow = { workspace = true }
 geoarrow-cast = { workspace = true }
 get_dir = { workspace = true }

vortex-bench/src/spatialbench/benchmark.rs

@@ -95,6 +95,16 @@ impl Benchmark for SpatialBenchBenchmark {
                 crate::conversions::add_geoparquet_metadata(&zone_file, &geo).await?;
             }
         }
+
+        // Cluster the source parquet along a spatial curve so all lanes read the same layout and
+        // the geometry zone-map prune can skip chunks.
+        let derived_dirs = [
+            base_data_dir.join(Format::OnDiskVortex.name()),
+            base_data_dir.join(Format::VortexCompact.name()),
+            base_data_dir.join(NATIVE_DIR),
+        ];
+        datagen::spatially_sort_tables(&base_data_dir.join(Format::Parquet.name()), &derived_dirs)
+            .await?;
         Ok(())
     }
 

vortex-bench/src/spatialbench/datagen/mod.rs

@@ -6,9 +6,11 @@
 //! source of truth for the base tables both stages share.
 
 pub mod native;
+pub mod spatial_sort;
 pub mod table;
 pub mod wkb;
 
 pub use native::write_native_vortex;
+pub use spatial_sort::spatially_sort_tables;
 pub use table::Table;
 pub use wkb::generate_tables;