merge

Signed-off-by: Nicholas Gates <nick@nickgates.com>

Author: gatesn

.github/workflows/ci.yml

@@ -395,7 +395,7 @@ jobs:
     name: "Rust tests (${{ matrix.sanitizer }})"
     runs-on: >-
       ${{ github.repository == 'vortex-data/vortex'
-          && format('runs-on={0}/runner=amd64-medium/image=ubuntu24-full-x64-pre-v2/tag=rust-test-sanitizer', github.run_id)
+          && format('runs-on={0}/pool=amd64-medium-pre-v2/tag=rust-test-sanitizer', github.run_id)
           || 'ubuntu-latest' }}
     timeout-minutes: 40
     env:
@@ -699,7 +699,7 @@ jobs:
     name: "Java"
     runs-on: >-
       ${{ github.repository == 'vortex-data/vortex'
-          && format('runs-on={0}/runner=amd64-medium/image=ubuntu24-full-x64-pre-v2/tag=java', github.run_id)
+          && format('runs-on={0}/pool=amd64-medium-pre-v2/tag=java', github.run_id)
           || 'ubuntu-latest' }}
     timeout-minutes: 40
     steps:

Cargo.lock

@@ -23,7 +23,7 @@ pub fn chunked_to_vec_record_batch(
             // TODO(connor)[ListView]: The rust Parquet implementation does not support writing
             // `ListView` to Parquet files yet.
             let converted_array = recursive_list_from_list_view(array.clone())?;
-            Ok(RecordBatch::try_from(converted_array.as_ref())?)
+            Ok(RecordBatch::try_from(&converted_array)?)
         })
         .collect::<anyhow::Result<Vec<_>>>()?;
 

benchmarks/compress-bench/src/lib.rs

@@ -12,6 +12,7 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use futures::StreamExt;
 use futures::pin_mut;
+use vortex::array::IntoArray;
 use vortex::file::OpenOptionsSessionExt;
 use vortex::file::WriteOptionsSessionExt;
 use vortex_bench::Format;
@@ -37,7 +38,7 @@ impl Compressor for VortexCompressor {
         let mut cursor = Cursor::new(&mut buf);
         SESSION
             .write_options()
-            .write(&mut cursor, uncompressed.to_array_stream())
+            .write(&mut cursor, uncompressed.into_array().to_array_stream())
             .await?;
         let elapsed = start.elapsed();
 
@@ -51,7 +52,7 @@ impl Compressor for VortexCompressor {
         let mut cursor = Cursor::new(&mut buf);
         SESSION
             .write_options()
-            .write(&mut cursor, uncompressed.to_array_stream())
+            .write(&mut cursor, uncompressed.into_array().to_array_stream())
             .await?;
 
         // Now decompress

benchmarks/compress-bench/src/vortex.rs

@@ -78,9 +78,6 @@ Struct Fields
 .. c:autotype:: vx_struct_fields
     :file: vortex.h
 
-.. c:autofunction:: vx_struct_fields_clone
-    :file: vortex.h
-
 .. c:autofunction:: vx_struct_fields_free
     :file: vortex.h
 

docs/api/c/dtypes.rst

@@ -103,20 +103,47 @@ fn decompress_alp<T: ALPFloat + NativePType>(bencher: Bencher, args: (usize, f64
         .bench_values(|(v, mut ctx)| decompress_into_array(v, &mut ctx));
 }
 
-#[divan::bench(types = [f32, f64], args = [10_000, 100_000])]
-fn compress_rd<T: ALPRDFloat>(bencher: Bencher, n: usize) {
-    let primitive = PrimitiveArray::new(buffer![T::from(1.23).unwrap(); n], Validity::NonNullable);
-    let encoder = RDEncoder::new(&[T::from(1.23).unwrap()]);
+const RD_BENCH_ARGS: &[(usize, f64)] = &[
+    // length, fraction_patch
+    (10_000, 0.0),
+    (10_000, 0.01),
+    (10_000, 0.1),
+    (100_000, 0.0),
+    (100_000, 0.01),
+    (100_000, 0.1),
+];
+
+fn make_rd_array<T: ALPRDFloat + NativePType>(n: usize, fraction_patch: f64) -> PrimitiveArray {
+    let base_val = T::from(1.23).unwrap();
+    let mut rng = StdRng::seed_from_u64(42);
+    let mut values = buffer![base_val; n].into_mut();
+    if fraction_patch > 0.0 {
+        let outlier = T::from(1000.0).unwrap();
+        for index in 0..values.len() {
+            if rng.random_bool(fraction_patch) {
+                values[index] = outlier;
+            }
+        }
+    }
+    PrimitiveArray::new(values.freeze(), Validity::NonNullable)
+}
+
+#[divan::bench(types = [f32, f64], args = RD_BENCH_ARGS)]
+fn compress_rd<T: ALPRDFloat + NativePType>(bencher: Bencher, args: (usize, f64)) {
+    let (n, fraction_patch) = args;
+    let primitive = make_rd_array::<T>(n, fraction_patch);
+    let encoder = RDEncoder::new(primitive.as_slice::<T>());
 
     bencher
         .with_inputs(|| (&primitive, &encoder))
         .bench_refs(|(primitive, encoder)| encoder.encode(primitive))
 }
 
-#[divan::bench(types = [f32, f64], args = [10_000, 100_000])]
-fn decompress_rd<T: ALPRDFloat>(bencher: Bencher, n: usize) {
-    let primitive = PrimitiveArray::new(buffer![T::from(1.23).unwrap(); n], Validity::NonNullable);
-    let encoder = RDEncoder::new(&[T::from(1.23).unwrap()]);
+#[divan::bench(types = [f32, f64], args = RD_BENCH_ARGS)]
+fn decompress_rd<T: ALPRDFloat + NativePType>(bencher: Bencher, args: (usize, f64)) {
+    let (n, fraction_patch) = args;
+    let primitive = make_rd_array::<T>(n, fraction_patch);
+    let encoder = RDEncoder::new(primitive.as_slice::<T>());
     let encoded = encoder.encode(&primitive);
 
     bencher