Array VTables (#7215)

See #7181

This is a large break that changes how Vortex arrays are held in memory.
Previously they were modeled as Arc<dyn Array> where all array impls
used a vtable macro to unsafely transmute into a dyn Array.

This was cute, but meant that we were performing a lot of unnecessary
clones and Arc allocations for things that should never have required
them. We were also limited by how expressive we could make APIs since
Arc is considered a foreign type.

This PR makes ArrayRef a struct, makes typed arrays a struct of
`Array<V: VTable>`, and adds an `ArrayView<'a, V>`. This means all
arrays are Arc allocated up-front in any of these forms, and we pass the
Arc allocation around everywhere allowing us to cheaply clone in either
the type-erased or typed structs.

The main user-facing change is that array constructors should now live
on the vtable struct, e.g. `Primitive::new(...)` instead of
`PrimitiveArray::new(...)`

---------

Signed-off-by: Nicholas Gates <nick@nickgates.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: gatesn

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