chore: update lance dependency to v9.0.0-rc.1

rust/lancedb/src/dataloader/permutation/shuffle.rs

@@ -8,13 +8,13 @@ use arrow_array::{RecordBatch, UInt64Array};
 use futures::{StreamExt, TryStreamExt};
 use lance::io::ObjectStore;
 use lance_core::{cache::LanceCache, utils::futures::FinallyStreamExt};
-use lance_encoding::decoder::DecoderPlugins;
+use lance_encoding::decoder::{DecoderPlugins, FilterExpression};
 use lance_file::{
     reader::{FileReader, FileReaderOptions},
     writer::{FileWriter, FileWriterOptions},
 };
-use lance_index::scalar::IndexReader;
 use lance_io::{
+    ReadBatchParams,
     scheduler::{ScanScheduler, SchedulerConfig},
     utils::CachedFileSize,
 };
@@ -239,7 +239,17 @@ impl Shuffler {
                     )
                     .await?;
                     // Need to read the entire file in a single batch for in-memory shuffling
-                    let batch = reader.read_record_batch(0, reader.num_rows()).await?;
+                    let batches = reader
+                        .read_stream(
+                            ReadBatchParams::RangeFull,
+                            reader.num_rows() as u32,
+                            1,
+                            FilterExpression::no_filter(),
+                        )
+                        .await?
+                        .try_collect::<Vec<_>>()
+                        .await?;
+                    let batch = concat_batches(&batches[0].schema(), &batches)?;
                     let mut rng = rng.lock().unwrap_or_else(|e| e.into_inner());
                     Self::shuffle_batch(&batch, &mut rng, clump_size)
                 }

rust/lancedb/src/dataloader/permutation/split.rs

@@ -8,7 +8,7 @@ use std::sync::{
 
 use arrow_array::{Array, BooleanArray, RecordBatch, UInt64Array};
 use arrow_schema::{DataType, Field, Schema};
-use datafusion_common::hash_utils::create_hashes;
+use datafusion_common::hash_utils::{RandomState, create_hashes};
 use futures::{StreamExt, TryStreamExt};
 use lance_arrow::SchemaExt;
 
@@ -234,7 +234,7 @@ impl Splitter {
             .cloned()
             .collect::<Vec<_>>();
         let mut hashes = vec![0; batch.num_rows()];
-        let random_state = ahash::RandomState::with_seeds(0, 0, 0, 0);
+        let random_state = RandomState::with_seed(0);
         create_hashes(&arrays, &random_state, &mut hashes).unwrap();
         // As an example, let's assume the weights are 1, 2.  Our total weight is 3.
         //

rust/lancedb/src/remote/table.rs

@@ -1250,8 +1250,7 @@ impl<S: HttpSend + 'static> RemoteTable<S> {
 
             match result {
                 Ok(_) => {
-                    let add_result = insert
-                        .as_any()
+                    let add_result = (insert.as_ref() as &dyn std::any::Any)
                         .downcast_ref::<RemoteInsertExec<S>>()
                         .and_then(|i| i.add_result())
                         .unwrap_or(AddResult { version: 0 });

rust/lancedb/src/remote/table/insert.rs

@@ -3,7 +3,6 @@
 
 //! DataFusion ExecutionPlan for inserting data into remote LanceDB tables.
 
-use std::any::Any;
 use std::sync::{Arc, Mutex};
 
 use arrow_array::{ArrayRef, RecordBatch, UInt64Array};
@@ -237,10 +236,6 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteInsertExec<S> {
         Self::static_name()
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }

rust/lancedb/src/table/datafusion.rs

@@ -89,10 +89,6 @@ impl ExecutionPlan for MetadataEraserExec {
         "MetadataEraserExec"
     }
 
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
     fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }
@@ -138,7 +134,7 @@ impl ExecutionPlan for MetadataEraserExec {
         )
     }
 
-    fn partition_statistics(&self, partition: Option<usize>) -> DataFusionResult<Statistics> {
+    fn partition_statistics(&self, partition: Option<usize>) -> DataFusionResult<Arc<Statistics>> {
         self.input.partition_statistics(partition)
     }
 
@@ -188,10 +184,6 @@ impl BaseTableAdapter {
 
 #[async_trait]
 impl TableProvider for BaseTableAdapter {
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
     fn schema(&self) -> Arc<ArrowSchema> {
         self.schema.clone()
     }

rust/lancedb/src/table/datafusion/insert.rs

@@ -3,7 +3,6 @@
 
 //! DataFusion ExecutionPlan for inserting data into LanceDB tables.
 
-use std::any::Any;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, LazyLock, Mutex};
 
@@ -146,10 +145,6 @@ impl ExecutionPlan for InsertExec {
         Self::static_name()
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }

rust/lancedb/src/table/datafusion/reject_nan.rs

@@ -3,7 +3,6 @@
 
 //! A DataFusion projection that rejects vectors containing NaN values.
 
-use std::any::Any;
 use std::sync::{Arc, LazyLock};
 
 use arrow_array::{Array, FixedSizeListArray};
@@ -86,10 +85,6 @@ impl RejectNanUdf {
 }
 
 impl ScalarUDFImpl for RejectNanUdf {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "reject_nan"
     }

rust/lancedb/src/table/datafusion/scannable_exec.rs

@@ -66,10 +66,6 @@ impl ExecutionPlan for ScannableExec {
         "ScannableExec"
     }
 
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
     fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }
@@ -121,14 +117,14 @@ impl ExecutionPlan for ScannableExec {
         )))
     }
 
-    fn partition_statistics(&self, _partition: Option<usize>) -> DFResult<Statistics> {
-        Ok(Statistics {
+    fn partition_statistics(&self, _partition: Option<usize>) -> DFResult<Arc<Statistics>> {
+        Ok(Arc::new(Statistics {
             num_rows: self
                 .num_rows
                 .map(Precision::Exact)
                 .unwrap_or(Precision::Absent),
             total_byte_size: Precision::Absent,
             column_statistics: vec![],
-        })
+        }))
     }
 }

rust/lancedb/src/table/datafusion/udtf/fts.rs

@@ -137,8 +137,7 @@ mod tests {
             };
 
             // Downcast to BaseTableAdapter and apply FTS query
-            let base_adapter = table_provider
-                .as_any()
+            let base_adapter = (table_provider.as_ref() as &dyn std::any::Any)
                 .downcast_ref::<BaseTableAdapter>()
                 .ok_or_else(|| {
                     DataFusionError::Internal(

rust/lancedb/src/table/query.rs

@@ -982,10 +982,10 @@ mod tests {
         use crate::table::query::create_plan;
 
         fn find_ann_approx_mode(plan: &dyn ExecutionPlan) -> Option<ApproxMode> {
-            if let Some(ann) = plan.as_any().downcast_ref::<ANNIvfSubIndexExec>() {
+            if let Some(ann) = (plan as &dyn std::any::Any).downcast_ref::<ANNIvfSubIndexExec>() {
                 return Some(ann.query().approx_mode);
             }
-            if let Some(ann) = plan.as_any().downcast_ref::<ANNIvfPartitionExec>() {
+            if let Some(ann) = (plan as &dyn std::any::Any).downcast_ref::<ANNIvfPartitionExec>() {
                 return Some(ann.query.approx_mode);
             }
             plan.children()