Merge branch 'develop' into ngates/arrow-executor

Author: gatesn

.github/workflows/ci.yml

@@ -662,13 +662,10 @@ jobs:
       - uses: ./.github/actions/setup-rust
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
-          toolchain: nightly-2025-06-26
           targets: "wasm32-wasip1"
-          components: "rust-src"
       - name: Setup Wasmer
         uses: wasmerio/[email protected]
-        # there is a compiler bug in nightly (but not in nightly-2025-06-26)
-      - run: cargo +nightly-2025-06-26 -Zbuild-std=panic_abort,std build --target wasm32-wasip1
+      - run: cargo build --target wasm32-wasip1
         working-directory: ./wasm-test
       - run: wasmer run ./target/wasm32-wasip1/debug/wasm-test.wasm
         working-directory: ./wasm-test

bench-vortex/src/bin/compress.rs

@@ -109,7 +109,7 @@ fn compress(
         .map(|f| Target::new(Engine::default(), *f))
         .collect_vec();
 
-    let structlistofints = vec![
+    let structlistofints = [
         StructListOfInts::new(100, 1000, 1),
         StructListOfInts::new(1000, 1000, 1),
         StructListOfInts::new(10000, 1000, 1),

bench-vortex/src/compress/bench.rs

@@ -24,6 +24,7 @@ use {
     crate::bench_run::run_with_setup,
     crate::utils::parquet::convert_utf8view_batch,
     crate::utils::parquet::convert_utf8view_schema,
+    anyhow::Context,
     arrow_array::RecordBatch,
     parking_lot::Mutex,
     std::fs,
@@ -131,9 +132,7 @@ pub fn benchmark_vortex_decompress(
     bench_name: &str,
 ) -> Result<(Duration, CompressionTimingMeasurement)> {
     let mut buf = Vec::new();
-    runtime
-        .block_on(vortex_compress_write(uncompressed, &mut buf))
-        .expect("Failed to compress with vortex for decompression test");
+    runtime.block_on(vortex_compress_write(uncompressed, &mut buf))?;
     let buffer = Bytes::from(buf);
 
     // Run the benchmark and measure time.
@@ -255,7 +254,7 @@ pub fn benchmark_lance_compress(
         .collect::<Result<Vec<_>, _>>()?;
     let converted_schema = convert_utf8view_schema(&schema);
 
-    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
+    let temp_dir = tempfile::tempdir().context("Failed to create temp dir")?;
     let iteration_paths: Arc<Mutex<Vec<PathBuf>>> = Arc::new(Mutex::new(Vec::new()));
     let iteration_counter = AtomicU64::new(0);
 
@@ -289,7 +288,7 @@ pub fn benchmark_lance_compress(
     // Calculate size from the last iteration.
     let paths = iteration_paths.lock();
     let lance_compressed_size_val = if let Some(last_path) = paths.last() {
-        calculate_lance_size(last_path).expect("Failed to calculate Lance size")
+        calculate_lance_size(last_path).context("Failed to calculate Lance size")?
     } else {
         0
     };
@@ -320,7 +319,7 @@ pub fn benchmark_lance_decompress(
     // NOTE: Lance requires filesystem access unlike Parquet/Vortex which use in-memory buffers.
     let chunked = uncompressed.as_::<ChunkedVTable>().clone();
     let (batches, schema) = chunked_to_vec_record_batch(chunked);
-    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
+    let temp_dir = tempfile::tempdir().context("Failed to create temp dir")?;
 
     // Write the Lance dataset once for all iterations.
     let dataset_path = runtime.block_on(async {

bench-vortex/src/engines/df/mod.rs

@@ -100,8 +100,7 @@ pub fn make_object_store(
             let s3 = Arc::new(
                 AmazonS3Builder::from_env()
                     .with_bucket_name(bucket_name)
-                    .build()
-                    .unwrap(),
+                    .build()?,
             );
             df.register_object_store(&Url::parse(&format!("s3://{bucket_name}/"))?, s3.clone());
             Ok(s3)
@@ -111,8 +110,7 @@ pub fn make_object_store(
             let gcs = Arc::new(
                 GoogleCloudStorageBuilder::from_env()
                     .with_bucket_name(bucket_name)
-                    .build()
-                    .unwrap(),
+                    .build()?,
             );
             df.register_object_store(&Url::parse(&format!("gs://{bucket_name}/"))?, gcs.clone());
             Ok(gcs)

encodings/fastlanes/benches/bitpacking_take.rs

@@ -237,7 +237,7 @@ fn patched_take_10k_adversarial(bencher: Bencher) {
         (0..(NUM_EXCEPTIONS + 1024) / 1024)
             .cycle()
             .map(|chunk_idx| BIG_BASE2 - 1024 + chunk_idx * 1024)
-            .flat_map(|base_idx| (base_idx..(base_idx + per_chunk_count)))
+            .flat_map(|base_idx| base_idx..(base_idx + per_chunk_count))
             .take(10000),
     );
 

encodings/fastlanes/src/delta/array/mod.rs

@@ -122,7 +122,7 @@ impl DeltaArray {
 
         let lanes = lane_count(ptype);
 
-        if (deltas.len() % 1024 == 0) != (bases.len() % lanes == 0) {
+        if deltas.len().is_multiple_of(1024) != bases.len().is_multiple_of(lanes) {
             vortex_bail!(
                 "deltas length ({}) is a multiple of 1024 iff bases length ({}) is a multiple of LANES ({})",
                 deltas.len(),

encodings/runend/src/array.rs

@@ -45,6 +45,7 @@ use vortex_scalar::PValue;
 use crate::compress::runend_decode_bools;
 use crate::compress::runend_decode_primitive;
 use crate::compress::runend_encode;
+use crate::rules::RULES;
 
 vtable!(RunEnd);
 
@@ -132,6 +133,14 @@ impl VTable for RunEndVTable {
 
         Ok(())
     }
+
+    fn reduce_parent(
+        array: &Self::Array,
+        parent: &ArrayRef,
+        child_idx: usize,
+    ) -> VortexResult<Option<ArrayRef>> {
+        RULES.evaluate(array, parent, child_idx)
+    }
 }
 
 #[derive(Clone, Debug)]

encodings/runend/src/lib.rs

@@ -24,17 +24,14 @@ pub mod _benchmarking {
 use vortex_array::ArrayBufferVisitor;
 use vortex_array::ArrayChildVisitor;
 use vortex_array::Canonical;
-use vortex_array::session::ArraySession;
 use vortex_array::session::ArraySessionExt;
 use vortex_array::vtable::ArrayVTableExt;
 use vortex_array::vtable::EncodeVTable;
 use vortex_array::vtable::VisitorVTable;
 use vortex_error::VortexResult;
-use vortex_session::SessionExt;
 use vortex_session::VortexSession;
 
 use crate::compress::runend_encode;
-use crate::rules::RunEndScalarFnRule;
 
 impl EncodeVTable<RunEndVTable> for RunEndVTable {
     fn encode(
@@ -69,10 +66,6 @@ impl VisitorVTable<RunEndVTable> for RunEndVTable {
 /// Initialize run-end encoding in the given session.
 pub fn initialize(session: &mut VortexSession) {
     session.arrays().register(RunEndVTable.as_vtable());
-    session
-        .get_mut::<ArraySession>()
-        .optimizer_mut()
-        .register_parent_rule(RunEndScalarFnRule);
 }
 
 #[cfg(test)]

encodings/runend/src/rules.rs

@@ -8,23 +8,24 @@ use vortex_array::arrays::ConstantArray;
 use vortex_array::arrays::ConstantVTable;
 use vortex_array::arrays::ScalarFnArray;
 use vortex_array::optimizer::rules::ArrayParentReduceRule;
-use vortex_array::optimizer::rules::Exact;
+use vortex_array::optimizer::rules::ParentRuleSet;
 use vortex_dtype::DType;
 use vortex_error::VortexResult;
 
 use crate::RunEndArray;
 use crate::RunEndVTable;
 
+pub(super) const RULES: ParentRuleSet<RunEndVTable> =
+    ParentRuleSet::new(&[ParentRuleSet::lift(&RunEndScalarFnRule)]);
+
 /// A rule to push down scalar functions through run-end encoding into the values array.
 ///
 /// This only works if all other children of the scalar function array are constants.
 #[derive(Debug)]
 pub(crate) struct RunEndScalarFnRule;
 
-impl ArrayParentReduceRule<Exact<RunEndVTable>, AnyScalarFn> for RunEndScalarFnRule {
-    fn child(&self) -> Exact<RunEndVTable> {
-        Exact::from(&RunEndVTable)
-    }
+impl ArrayParentReduceRule<RunEndVTable> for RunEndScalarFnRule {
+    type Parent = AnyScalarFn;
 
     fn parent(&self) -> AnyScalarFn {
         AnyScalarFn

vortex-array/benches/chunk_array_builder.rs

@@ -116,7 +116,7 @@ fn make_opt_bool_chunks(len: usize, chunk_count: usize) -> ArrayRef {
     let mut rng = StdRng::seed_from_u64(0);
 
     const SPAN_LEN: usize = 10;
-    assert!(len % SPAN_LEN == 0);
+    assert!(len.is_multiple_of(SPAN_LEN));
 
     (0..chunk_count)
         .map(|_| {