chore: update datafusion to 42.0 and arrow to 53.2

Cargo.toml

@@ -61,17 +61,17 @@ lance-test-macros = { version = "=0.19.2", path = "./rust/lance-test-macros" }
 lance-testing = { version = "=0.19.2", path = "./rust/lance-testing" }
 approx = "0.5.1"
 # Note that this one does not include pyarrow
-arrow = { version = "52.2", optional = false, features = ["prettyprint"] }
-arrow-arith = "52.2"
-arrow-array = "52.2"
-arrow-buffer = "52.2"
-arrow-cast = "52.2"
-arrow-data = "52.2"
-arrow-ipc = { version = "52.2", features = ["zstd"] }
-arrow-ord = "52.2"
-arrow-row = "52.2"
-arrow-schema = "52.2"
-arrow-select = "52.2"
+arrow = { version = "53.2", optional = false, features = ["prettyprint"] }
+arrow-arith = "53.2"
+arrow-array = "53.2"
+arrow-buffer = "53.2"
+arrow-cast = "53.2"
+arrow-data = "53.2"
+arrow-ipc = { version = "53.2", features = ["zstd"] }
+arrow-ord = "53.2"
+arrow-row = "53.2"
+arrow-schema = "53.2"
+arrow-select = "53.2"
 async-recursion = "1.0"
 async-trait = "0.1"
 aws-config = "1.2.0"
@@ -95,18 +95,18 @@ criterion = { version = "0.5", features = [
     "html_reports",
 ] }
 crossbeam-queue = "0.3"
-datafusion = { version = "41.0", default-features = false, features = [
+datafusion = { version = "42.0", default-features = false, features = [
     "nested_expressions",
     "regex_expressions",
     "unicode_expressions",
 ] }
-datafusion-common = "41.0"
-datafusion-functions = { version = "41.0", features = ["regex_expressions"] }
-datafusion-sql = "41.0"
-datafusion-expr = "41.0"
-datafusion-execution = "41.0"
-datafusion-optimizer = "41.0"
-datafusion-physical-expr = { version = "41.0", features = [
+datafusion-common = "42.0"
+datafusion-functions = { version = "42.0", features = ["regex_expressions"] }
+datafusion-sql = "42.0"
+datafusion-expr = "42.0"
+datafusion-execution = "42.0"
+datafusion-optimizer = "42.0"
+datafusion-physical-expr = { version = "42.0", features = [
     "regex_expressions",
 ] }
 deepsize = "0.2.0"
@@ -129,9 +129,9 @@ pin-project = "1.0"
 path_abs = "0.5"
 pprof = { version = "0.13", features = ["flamegraph", "criterion"] }
 proptest = "1.3.1"
-prost = "0.12.2"
-prost-build = "0.12.2"
-prost-types = "0.12.2"
+prost = "0.13.2"
+prost-build = "0.13.2"
+prost-types = "0.13.2"
 rand = { version = "0.8.3", features = ["small_rng"] }
 rangemap = { version = "1.0" }
 rayon = "1.10"

rust/lance-core/src/error.rs

@@ -224,6 +224,16 @@ impl From<prost::EncodeError> for Error {
     }
 }
 
+impl From<prost::UnknownEnumValue> for Error {
+    #[track_caller]
+    fn from(e: prost::UnknownEnumValue) -> Self {
+        Self::IO {
+            source: box_error(e),
+            location: std::panic::Location::caller().to_snafu_location(),
+        }
+    }
+}
+
 impl From<tokio::task::JoinError> for Error {
     #[track_caller]
     fn from(e: tokio::task::JoinError) -> Self {

rust/lance-datafusion/Cargo.toml

@@ -21,7 +21,7 @@ datafusion.workspace = true
 datafusion-common.workspace = true
 datafusion-functions.workspace = true
 datafusion-physical-expr.workspace = true
-datafusion-substrait = { version = "41.0", optional = true }
+datafusion-substrait = { version = "42.0", optional = true }
 futures.workspace = true
 lance-arrow.workspace = true
 lance-core = { workspace = true, features = ["datafusion"] }
@@ -32,7 +32,8 @@ snafu.workspace = true
 tokio.workspace = true
 
 [dev-dependencies]
-substrait-expr = { version = "0.2.1" }
+# TODO: This is too old
+#substrait-expr = { version = "0.2.1" }
 lance-datagen.workspace = true
 
 [features]

rust/lance-datafusion/src/substrait.rs

@@ -382,6 +382,7 @@ pub async fn parse_substrait(expr: &[u8], input_schema: Arc<Schema>) -> Result<E
     Ok(expr.data)
 }
 
+/*
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
@@ -440,3 +441,4 @@ mod tests {
         assert_eq!(df_expr, expected);
     }
 }
+*/

rust/lance-io/src/encodings/binary.rs

@@ -224,7 +224,7 @@ impl<'a, T: ByteArrayType> BinaryDecoder<'a, T> {
                 .null_bit_buffer(null_buf);
         }
 
-        let buf = bytes.into();
+        let buf = Buffer::from_vec(bytes.to_vec());
         let array_data = data_builder
             .add_buffer(offset_data.buffers()[0].clone())
             .add_buffer(buf)

rust/lance-io/src/encodings/plain.rs

@@ -205,7 +205,7 @@ pub fn bytes_to_array(
 
         // alignment or size isn't right -- just make a copy
         if (bytes.len() < min_buffer_size) || (bytes.as_ptr().align_offset(*alignment) != 0) {
-            bytes.into()
+            Buffer::from_vec(bytes.to_vec())
         } else {
             // SAFETY: the alignment is correct we can make this conversion
             unsafe {
@@ -218,7 +218,7 @@ pub fn bytes_to_array(
         }
     } else {
         // cases we don't handle, just copy
-        bytes.into()
+        Buffer::from_vec(bytes.to_vec())
     };
 
     let array_data = ArrayDataBuilder::new(data_type.clone())

rust/lance/Cargo.toml

@@ -59,6 +59,7 @@ arrow.workspace = true
 datafusion.workspace = true
 datafusion-functions.workspace = true
 datafusion-physical-expr.workspace = true
+datafusion-expr.workspace = true
 lapack = { version = "0.19.0", optional = true }
 snafu = { workspace = true }
 log = { workspace = true }

rust/lance/src/datafusion/logical_plan.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The Lance Authors
 
-use std::{any::Any, sync::Arc};
+use std::{any::Any, borrow::Cow, sync::Arc};
 
 use arrow_schema::Schema as ArrowSchema;
 use async_trait::async_trait;
@@ -34,7 +34,7 @@ impl TableProvider for Dataset {
         None
     }
 
-    fn get_logical_plan(&self) -> Option<&LogicalPlan> {
+    fn get_logical_plan(&self) -> Option<Cow<LogicalPlan>> {
         None
     }
 

rust/lance/src/dataset/scanner.rs

@@ -26,11 +26,11 @@ use datafusion::physical_plan::{
     filter::FilterExec,
     limit::GlobalLimitExec,
     repartition::RepartitionExec,
-    udaf::create_aggregate_expr,
     union::UnionExec,
     ExecutionPlan, SendableRecordBatchStream,
 };
 use datafusion::scalar::ScalarValue;
+use datafusion_physical_expr::aggregate::AggregateExprBuilder;
 use datafusion_physical_expr::{Partitioning, PhysicalExpr};
 use futures::stream::{Stream, StreamExt};
 use futures::TryStreamExt;
@@ -957,17 +957,18 @@ impl Scanner {
         let plan = self.create_plan().await?;
         // Datafusion interprets COUNT(*) as COUNT(1)
         let one = Arc::new(Literal::new(ScalarValue::UInt8(Some(1))));
-        let count_expr = create_aggregate_expr(
-            &count_udaf(),
-            &[one],
-            &[lit(1)],
-            &[],
-            &[],
-            &plan.schema(),
-            None,
-            false,
-            false,
-        )?;
+
+        let input_phy_exprs: &[Arc<dyn PhysicalExpr>] = &[one];
+        let schema = plan.schema();
+
+        let mut builder = AggregateExprBuilder::new(count_udaf(), input_phy_exprs.to_vec());
+        //builder = builder.logical_exprs(input_exprs.to_vec());
+        builder = builder.schema(schema);
+        // TODO: This alias seem to be required?
+        builder = builder.alias("count".to_string());
+
+        let count_expr = builder.build()?;
+
         let plan_schema = plan.schema();
         let count_plan = Arc::new(AggregateExec::try_new(
             AggregateMode::Single,