Merge remote-tracking branch 'origin/develop' into ji/better-execute

Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>

# Conflicts:
#	vortex-array/src/arrays/filter/vtable.rs
#	vortex-array/src/arrays/masked/vtable/mod.rs
#	vortex-array/src/arrays/slice/vtable.rs

Author: joseph-isaacs

Cargo.toml

@@ -349,6 +349,7 @@ needless_range_loop = "allow"
 or_fun_call = "deny"
 panic = "deny"
 # panic_in_result_fn = "deny" -- we cannot disable this for tests to use assertions
+clone_on_ref_ptr = "deny"
 redundant_clone = "deny"
 same_name_method = "deny"
 tests_outside_test_module = "deny"

benchmarks/compress-bench/src/parquet.rs

@@ -53,7 +53,7 @@ impl Compressor for ParquetCompressor {
         // Read the input parquet file
         let file = File::open(parquet_path)?;
         let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
-        let schema = builder.schema().clone();
+        let schema = Arc::clone(builder.schema());
         let reader = builder.build()?;
         let batches: Vec<RecordBatch> = reader.collect::<Result<Vec<_>, _>>()?;
 
@@ -69,7 +69,7 @@ impl Compressor for ParquetCompressor {
         // First compress to get the bytes we'll decompress
         let file = File::open(parquet_path)?;
         let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
-        let schema = builder.schema().clone();
+        let schema = Arc::clone(builder.schema());
         let reader = builder.build()?;
         let batches: Vec<RecordBatch> = reader.collect::<Result<Vec<_>, _>>()?;
 

benchmarks/datafusion-bench/src/lib.rs

@@ -81,8 +81,10 @@ pub fn make_object_store(
                     .with_bucket_name(bucket_name)
                     .build()?,
             );
-            session
-                .register_object_store(&Url::parse(&format!("s3://{bucket_name}/"))?, s3.clone());
+            session.register_object_store(
+                &Url::parse(&format!("s3://{bucket_name}/"))?,
+                Arc::<object_store::aws::AmazonS3>::clone(&s3),
+            );
             Ok(s3)
         }
         "gs" => {
@@ -92,13 +94,16 @@ pub fn make_object_store(
                     .with_bucket_name(bucket_name)
                     .build()?,
             );
-            session
-                .register_object_store(&Url::parse(&format!("gs://{bucket_name}/"))?, gcs.clone());
+            session.register_object_store(
+                &Url::parse(&format!("gs://{bucket_name}/"))?,
+                Arc::<object_store::gcp::GoogleCloudStorage>::clone(&gcs),
+            );
             Ok(gcs)
         }
         _ => {
             let fs = Arc::new(LocalFileSystem::default());
-            session.register_object_store(&Url::parse("file:/")?, fs.clone());
+            session
+                .register_object_store(&Url::parse("file:/")?, Arc::<LocalFileSystem>::clone(&fs));
             Ok(fs)
         }
     }

benchmarks/datafusion-bench/src/main.rs

@@ -203,7 +203,7 @@ async fn main() -> anyhow::Result<()> {
                                 .iter()
                                 .any(|(idx, f, _)| *idx == query_idx && *f == *format)
                             {
-                                plans_mut.push((query_idx, *format, plan.clone()));
+                                plans_mut.push((query_idx, *format, Arc::clone(&plan)));
                             }
                         }
 
@@ -252,7 +252,7 @@ async fn register_benchmark_tables<B: Benchmark + ?Sized>(
                 let pattern = benchmark.pattern(table.name, format);
                 let table_url = ListingTableUrl::try_new(benchmark_base.clone(), pattern)?;
 
-                let mut listing_options = ListingOptions::new(file_format.clone())
+                let mut listing_options = ListingOptions::new(Arc::clone(&file_format))
                     .with_session_config_options(session.state().config());
                 if benchmark.dataset_name() == "polarsignals" && format == Format::Parquet {
                     // Work around a DataFusion bug (fixed in 53.0.0) where the
@@ -304,8 +304,10 @@ async fn register_v2_tables<B: Benchmark + ?Sized>(
             .runtime_env()
             .object_store(table_url.object_store())?;
 
-        let fs: vortex::io::filesystem::FileSystemRef =
-            Arc::new(ObjectStoreFileSystem::new(store.clone(), SESSION.handle()));
+        let fs: vortex::io::filesystem::FileSystemRef = Arc::new(ObjectStoreFileSystem::new(
+            Arc::clone(&store),
+            SESSION.handle(),
+        ));
         let base_prefix = benchmark_base.path().trim_start_matches('/').to_string();
         let fs = fs.with_prefix(base_prefix);
 
@@ -416,7 +418,7 @@ pub async fn execute_query(
         .create_physical_plan()
         .with_labelset(get_labelset_from_global())
         .await?;
-    let result = collect(plan.clone(), task_ctx)
+    let result = collect(Arc::clone(&plan), task_ctx)
         .with_labelset(get_labelset_from_global())
         .await?;
 

benchmarks/lance-bench/src/compress.rs

@@ -4,6 +4,7 @@
 use std::fs;
 use std::fs::File;
 use std::path::Path;
+use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
 
@@ -92,7 +93,7 @@ impl Compressor for LanceCompressor {
         // Read the input parquet file
         let file = File::open(parquet_path)?;
         let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
-        let schema = builder.schema().clone();
+        let schema = Arc::clone(builder.schema());
         let reader = builder.build()?;
         let batches: Vec<RecordBatch> = reader.collect::<Result<Vec<_>, _>>()?;
 
@@ -131,7 +132,7 @@ impl Compressor for LanceCompressor {
         // First compress to get the Lance dataset
         let file = File::open(parquet_path)?;
         let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
-        let schema = builder.schema().clone();
+        let schema = Arc::clone(builder.schema());
         let reader = builder.build()?;
         let batches: Vec<RecordBatch> = reader.collect::<Result<Vec<_>, _>>()?;
 

benchmarks/lance-bench/src/convert.rs

@@ -84,7 +84,7 @@ impl Iterator for ParquetFilesIterator {
 
 impl RecordBatchReader for ParquetFilesIterator {
     fn schema(&self) -> SchemaRef {
-        self.schema.clone()
+        Arc::clone(&self.schema)
     }
 }
 
@@ -161,7 +161,7 @@ pub async fn convert_parquet_to_lance<'p>(
             // Get schema from the first Parquet file
             let first_file = File::open(&parquet_files[0])?;
             let first_builder = ParquetRecordBatchReaderBuilder::try_new(first_file)?;
-            let schema = first_builder.schema().clone();
+            let schema = Arc::clone(first_builder.schema());
 
             // Create a streaming iterator that reads from all Parquet files
             let batch_iter = ParquetFilesIterator::new(parquet_files, schema)?;
@@ -237,7 +237,7 @@ pub fn convert_utf8view_batch(batch: RecordBatch) -> anyhow::Result<RecordBatch>
             // Cast Utf8View to Utf8.
             cast(column, &DataType::Utf8)?
         } else {
-            column.clone()
+            Arc::clone(column)
         };
         new_columns.push(new_column);
     }
@@ -277,6 +277,6 @@ impl Iterator for ConvertingParquetFilesIterator {
 
 impl RecordBatchReader for ConvertingParquetFilesIterator {
     fn schema(&self) -> SchemaRef {
-        self.converted_schema.clone()
+        Arc::clone(&self.converted_schema)
     }
 }

encodings/alp/src/alp/array.rs

@@ -172,7 +172,6 @@ impl VTable for ALP {
         let array = require_child!(array, array.encoded(), ENCODED_SLOT => Primitive);
         require_patches!(
             array,
-            array.patches(),
             PATCH_INDICES_SLOT,
             PATCH_VALUES_SLOT,
             PATCH_CHUNK_OFFSETS_SLOT

encodings/alp/src/alp_rd/array.rs

@@ -228,7 +228,6 @@ impl VTable for ALPRD {
         let array = require_child!(array, array.right_parts(), 1 => Primitive);
         require_patches!(
             array,
-            array.left_parts_patches(),
             LP_PATCH_INDICES_SLOT,
             LP_PATCH_VALUES_SLOT,
             LP_PATCH_CHUNK_OFFSETS_SLOT

encodings/fastlanes/src/bitpacking/compute/filter.rs

@@ -82,7 +82,7 @@ impl FilterKernel for BitPacked {
 
         let patches = array
             .patches()
-            .map(|patches| patches.filter(&Mask::Values(values.clone()), ctx))
+            .map(|patches| patches.filter(&Mask::Values(Arc::clone(values)), ctx))
             .transpose()?
             .flatten();
 
@@ -112,7 +112,9 @@ fn filter_primitive_without_patches<U: UnsignedPType + BitPacking>(
     selection: &Arc<MaskValues>,
 ) -> VortexResult<(Buffer<U>, Validity)> {
     let values = filter_with_indices(array.data(), selection.indices());
-    let validity = array.validity()?.filter(&Mask::Values(selection.clone()))?;
+    let validity = array
+        .validity()?
+        .filter(&Mask::Values(Arc::clone(selection)))?;
 
     Ok((values.freeze(), validity))
 }

encodings/fastlanes/src/bitpacking/vtable/mod.rs

@@ -5,7 +5,6 @@ use std::hash::Hash;
 use std::hash::Hasher;
 
 use prost::Message;
-use vortex_array::AnyCanonical;
 use vortex_array::Array;
 use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
@@ -283,17 +282,11 @@ impl VTable for BitPacked {
     fn execute(array: Array<Self>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         require_patches!(
             array,
-            array.patches(),
             PATCH_INDICES_SLOT,
             PATCH_VALUES_SLOT,
             PATCH_CHUNK_OFFSETS_SLOT
         );
-        let validity = array.validity()?;
-        require_validity!(
-            array,
-            &validity,
-            VALIDITY_SLOT => AnyCanonical
-        );
+        require_validity!(array, VALIDITY_SLOT);
 
         Ok(ExecutionResult::done(
             unpack_array(array.as_view(), ctx)?.into_array(),