Enable clippy::clone_on_ref_ptr lint and fix all violations

Adds `clone_on_ref_ptr = "deny"` to workspace clippy lints, enforcing
explicit `Arc::clone(&x)` instead of `x.clone()` on ref-counted pointers
to make refcount bumps visually distinct from deep clones.

Signed-off-by: Joseph Isaacs <joseph@spiraldb.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

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/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/sparse/src/canonical.rs

@@ -113,7 +113,7 @@ pub(super) fn execute_sparse(
             execute_varbin(array, dtype.clone(), fill_value, ctx)?
         }
         DType::List(values_dtype, nullability) => {
-            execute_sparse_lists(array, values_dtype.clone(), *nullability, ctx)?
+            execute_sparse_lists(array, Arc::clone(values_dtype), *nullability, ctx)?
         }
         DType::FixedSizeList(.., nullability) => {
             execute_sparse_fixed_size_list(array, *nullability, ctx)?

fuzz/src/array/fill_null.rs

@@ -1,6 +1,8 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::sync::Arc;
+
 use vortex_array::ArrayRef;
 use vortex_array::Canonical;
 use vortex_array::IntoArray;
@@ -221,7 +223,7 @@ fn fill_varbinview_array(
                     if result_nullability == Nullability::Nullable {
                         VarBinViewArray::new_handle(
                             result.to_varbinview().views_handle().clone(),
-                            result.to_varbinview().data_buffers().clone(),
+                            Arc::clone(result.to_varbinview().data_buffers()),
                             result.dtype().as_nullable(),
                             result_nullability.into(),
                         )
@@ -255,7 +257,7 @@ fn fill_varbinview_array(
                     if result_nullability == Nullability::Nullable {
                         VarBinViewArray::new_handle(
                             result.to_varbinview().views_handle().clone(),
-                            result.to_varbinview().data_buffers().clone(),
+                            Arc::clone(result.to_varbinview().data_buffers()),
                             result.dtype().as_nullable(),
                             result_nullability.into(),
                         )

fuzz/src/array/mask.rs

@@ -1,6 +1,8 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::sync::Arc;
+
 use vortex_array::ArrayRef;
 use vortex_array::Canonical;
 use vortex_array::IntoArray;
@@ -86,7 +88,7 @@ pub fn mask_canonical_array(canonical: Canonical, mask: &Mask) -> VortexResult<A
             let new_validity = mask_validity(&array.validity()?, mask);
             VarBinViewArray::new_handle(
                 array.views_handle().clone(),
-                array.data_buffers().clone(),
+                Arc::clone(array.data_buffers()),
                 array.dtype().with_nullability(new_validity.nullability()),
                 new_validity,
             )