feat: upgrade arrow to 50 and datafusion to 35 (#1868)

Author: wjones127

python/Cargo.toml

@@ -11,10 +11,10 @@ name = "lance"
 crate-type = ["cdylib"]
 
 [dependencies]
-arrow = { version = "49.0.0", features = ["pyarrow"] }
-arrow-array = "49.0"
-arrow-data = "49.0"
-arrow-schema = "49.0"
+arrow = { version = "50.0.0", features = ["pyarrow"] }
+arrow-array = "50.0"
+arrow-data = "50.0"
+arrow-schema = "50.0"
 object_store = "0.9.0"
 async-trait = "0.1"
 chrono = "0.4.31"

python/src/arrow.rs

@@ -14,7 +14,7 @@
 
 use std::sync::Arc;
 
-use arrow::ffi_stream::{export_reader_into_raw, FFI_ArrowArrayStream};
+use arrow::ffi_stream::FFI_ArrowArrayStream;
 use arrow::pyarrow::*;
 use arrow_array::{RecordBatch, RecordBatchIterator, RecordBatchReader};
 use arrow_schema::{DataType, Field, Schema};
@@ -105,8 +105,7 @@ pub fn reader_to_pyarrow(
     py: Python,
     reader: Box<dyn RecordBatchReader + Send>,
 ) -> PyResult<PyObject> {
-    let mut stream = FFI_ArrowArrayStream::empty();
-    unsafe { export_reader_into_raw(reader, &mut stream) };
+    let mut stream = FFI_ArrowArrayStream::new(reader);
 
     let stream_ptr = (&mut stream) as *mut FFI_ArrowArrayStream;
     let module = py.import("pyarrow")?;

rust/Cargo.toml

@@ -53,17 +53,17 @@ lance-test-macros = { version = "=0.9.9", path = "./lance-test-macros" }
 lance-testing = { version = "=0.9.9", path = "./lance-testing" }
 approx = "0.5.1"
 # Note that this one does not include pyarrow
-arrow = { version = "49.0.0", optional = false, features = ["prettyprint"] }
-arrow-arith = "49.0"
-arrow-array = "49.0"
-arrow-buffer = "49.0"
-arrow-cast = "49.0"
-arrow-data = "49.0"
-arrow-ipc = { version = "49.0", features = ["zstd"] }
-arrow-ord = "49.0"
-arrow-row = "49.0"
-arrow-schema = "49.0"
-arrow-select = "49.0"
+arrow = { version = "50.0.0", optional = false, features = ["prettyprint"] }
+arrow-arith = "50.0"
+arrow-array = "50.0"
+arrow-buffer = "50.0"
+arrow-cast = "50.0"
+arrow-data = "50.0"
+arrow-ipc = { version = "50.0", features = ["zstd"] }
+arrow-ord = "50.0"
+arrow-row = "50.0"
+arrow-schema = "50.0"
+arrow-select = "50.0"
 async-recursion = "1.0"
 async-trait = "0.1"
 aws-config = "0.56"
@@ -77,13 +77,13 @@ bytes = "1.4"
 byteorder = "1.5"
 chrono = "0.4.23"
 criterion = { version = "0.5", features = ["async", "async_tokio"] }
-datafusion = { version = "34.0.0", default-features = false, features = [
+datafusion = { version = "35.0.0", default-features = false, features = [
     "regex_expressions",
 ] }
-datafusion-common = "34.0"
-datafusion-sql = "34.0"
-datafusion-expr = "34.0"
-datafusion-physical-expr = "34.0"
+datafusion-common = "35.0"
+datafusion-sql = "35.0"
+datafusion-expr = "35.0"
+datafusion-physical-expr = "35.0"
 either = "1.0"
 futures = "0.3"
 http = "0.2.9"

rust/lance-datafusion/src/expr.rs

@@ -17,7 +17,7 @@
 use std::sync::Arc;
 
 use arrow::compute::cast;
-use arrow_array::{cast::AsArray, FixedSizeListArray};
+use arrow_array::{cast::AsArray, ArrayRef};
 use arrow_schema::DataType;
 use datafusion_common::ScalarValue;
 
@@ -233,25 +233,51 @@ pub fn safe_coerce_scalar(value: &ScalarValue, ty: &DataType) -> Option<ScalarVa
             _ => None,
         },
         ScalarValue::Null => Some(value.clone()),
-        // In Arrow 50.0.0, we will have support to cast from list -> FSL, and we
-        // can remove the special case below.
-        ScalarValue::List(values) if matches!(ty, DataType::FixedSizeList(_, _)) => {
-            let values = values.as_list::<i32>().values();
-            if let DataType::FixedSizeList(field, size) = ty {
-                let new_values = cast(values, field.data_type()).ok()?;
-                Some(ScalarValue::FixedSizeList(Arc::new(
-                    FixedSizeListArray::new(field.clone(), *size, new_values, None),
-                )))
-            } else {
-                unreachable!()
+        ScalarValue::List(values) => {
+            let values = values.clone() as ArrayRef;
+            let new_values = cast(&values, ty).ok()?;
+            match ty {
+                DataType::List(_) => {
+                    Some(ScalarValue::List(Arc::new(new_values.as_list().clone())))
+                }
+                DataType::LargeList(_) => Some(ScalarValue::LargeList(Arc::new(
+                    new_values.as_list().clone(),
+                ))),
+                DataType::FixedSizeList(_, _) => Some(ScalarValue::FixedSizeList(Arc::new(
+                    new_values.as_fixed_size_list().clone(),
+                ))),
+                _ => None,
+            }
+        }
+        ScalarValue::LargeList(values) => {
+            let values = values.clone() as ArrayRef;
+            let new_values = cast(&values, ty).ok()?;
+            match ty {
+                DataType::List(_) => {
+                    Some(ScalarValue::List(Arc::new(new_values.as_list().clone())))
+                }
+                DataType::LargeList(_) => Some(ScalarValue::LargeList(Arc::new(
+                    new_values.as_list().clone(),
+                ))),
+                DataType::FixedSizeList(_, _) => Some(ScalarValue::FixedSizeList(Arc::new(
+                    new_values.as_fixed_size_list().clone(),
+                ))),
+                _ => None,
             }
         }
-        ScalarValue::List(values) | ScalarValue::FixedSizeList(values) => {
-            let new_values = cast(values, ty).ok()?;
+        ScalarValue::FixedSizeList(values) => {
+            let values = values.clone() as ArrayRef;
+            let new_values = cast(&values, ty).ok()?;
             match ty {
-                DataType::List(_) => Some(ScalarValue::List(new_values)),
-                DataType::LargeList(_) => Some(ScalarValue::LargeList(new_values)),
-                DataType::FixedSizeList(_, _) => Some(ScalarValue::FixedSizeList(new_values)),
+                DataType::List(_) => {
+                    Some(ScalarValue::List(Arc::new(new_values.as_list().clone())))
+                }
+                DataType::LargeList(_) => Some(ScalarValue::LargeList(Arc::new(
+                    new_values.as_list().clone(),
+                ))),
+                DataType::FixedSizeList(_, _) => Some(ScalarValue::FixedSizeList(Arc::new(
+                    new_values.as_fixed_size_list().clone(),
+                ))),
                 _ => None,
             }
         }

rust/lance-index/Cargo.toml

@@ -23,6 +23,7 @@ datafusion.workspace = true
 datafusion-common.workspace = true
 datafusion-expr.workspace = true
 datafusion-physical-expr.workspace = true
+datafusion-sql.workspace = true
 futures.workspace = true
 half.workspace = true
 lance-arrow.workspace = true

rust/lance/src/dataset/scanner.rs

@@ -560,7 +560,6 @@ impl Scanner {
             PhysicalGroupBy::new_single(Vec::new()),
             vec![count_expr],
             vec![None],
-            vec![None],
             plan,
             plan_schema,
         )?);

rust/lance/src/io/exec/planner.rs

@@ -561,7 +561,6 @@ impl Planner {
         Ok(datafusion::physical_expr::create_physical_expr(
             expr,
             df_schema.as_ref(),
-            &self.schema,
             &Default::default(),
         )?)
     }

rust/lance/src/io/exec/scan.rs

@@ -305,9 +305,15 @@ impl ExecutionPlan for LanceScanExec {
 
     fn with_new_children(
         self: Arc<Self>,
-        _children: Vec<Arc<dyn ExecutionPlan>>,
+        children: Vec<Arc<dyn ExecutionPlan>>,
     ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
-        todo!()
+        if children.is_empty() {
+            Ok(self)
+        } else {
+            Err(DataFusionError::Internal(
+                "LanceScanExec cannot be assigned children".to_string(),
+            ))
+        }
     }
 
     fn execute(