Minimal build now that pyo3 has updated. Still need to resolve many deprecation warnings

Author: timsaucer

Cargo.lock

@@ -34,28 +34,27 @@ protoc = [ "datafusion-substrait/protoc" ]
 substrait = ["dep:datafusion-substrait"]
 
 [dependencies]
-tokio = { version = "1.47", features = ["macros", "rt", "rt-multi-thread", "sync"] }
-pyo3 = { version = "0.25", features = ["extension-module", "abi3", "abi3-py39"] }
-pyo3-async-runtimes = { version = "0.25", features = ["tokio-runtime"]}
-pyo3-log = "0.12.4"
-arrow = { version = "56", features = ["pyarrow"] }
-datafusion = { version = "50.3.0", features = ["avro", "unicode_expressions"] }
-datafusion-substrait = { version = "50.3.0", optional = true }
-datafusion-proto = { version = "50.3.0" }
-datafusion-ffi = { version = "50.3.0" }
-prost = "0.13.1" # keep in line with `datafusion-substrait`
-uuid = { version = "1.18", features = ["v4"] }
-mimalloc = { version = "0.1", optional = true, default-features = false, features = ["local_dynamic_tls"] }
+arrow = { version = "57", features = ["pyarrow"] }
 async-trait = "0.1.89"
+datafusion = { version = "51", features = ["avro", "unicode_expressions"] }
+datafusion-substrait = { version = "51", optional = true }
+datafusion-proto = { version = "51" }
+datafusion-ffi = { version = "51" }
 futures = "0.3"
-object_store = { version = "0.12.4", features = ["aws", "gcp", "azure", "http"] }
-url = "2"
 log = "0.4.27"
+mimalloc = { version = "0.1", optional = true, default-features = false, features = ["local_dynamic_tls"] }
+object_store = { version = "0.12.4", features = ["aws", "gcp", "azure", "http"] }
 parking_lot = "0.12"
+prost = "0.14" # keep in line with `datafusion-substrait`
+pyo3 = { version = "0.26", features = ["extension-module", "abi3", "abi3-py39"] }
+pyo3-async-runtimes = { version = "0.26", features = ["tokio-runtime"]}
+pyo3-log = "0.13"
+tokio = { version = "1.48", features = ["macros", "rt", "rt-multi-thread", "sync"] }
+uuid = { version = "1.18", features = ["v4"] }
+url = "2"
 
 [build-dependencies]
-prost-types = "0.13.1" # keep in line with `datafusion-substrait`
-pyo3-build-config = "0.25"
+pyo3-build-config = "0.26"
 
 [lib]
 name = "datafusion_python"
@@ -66,7 +65,7 @@ lto = true
 codegen-units = 1
 
 [patch.crates-io]
-datafusion = { git = "https://github.com/apache/datafusion", branch = "main" }
-datafusion-substrait = { git = "https://github.com/apache/datafusion", branch = "main" }
-datafusion-proto = { git = "https://github.com/apache/datafusion", branch = "main" }
-datafusion-ffi = { git = "https://github.com/apache/datafusion", branch = "main" }
+datafusion = { git = "https://github.com/apache/datafusion", branch = "branch-51" }
+datafusion-substrait = { git = "https://github.com/apache/datafusion", branch = "branch-51" }
+datafusion-proto = { git = "https://github.com/apache/datafusion", branch = "branch-51" }
+datafusion-ffi = { git = "https://github.com/apache/datafusion", branch = "branch-51" }

Cargo.toml

@@ -65,7 +65,7 @@ impl PyConfig {
     }
 
     /// Set a configuration option
-    pub fn set(&self, key: &str, value: PyObject, py: Python) -> PyDataFusionResult<()> {
+    pub fn set(&self, key: &str, value: Bound<PyAny>, py: Python) -> PyDataFusionResult<()> {
         let scalar_value = py_obj_to_scalar_value(py, value)?;
         let mut options = self.config.write();
         options.set(key, scalar_value.to_string().as_str())?;

src/config.rs

@@ -309,11 +309,11 @@ impl PyDataFrame {
 
         let table_uuid = uuid::Uuid::new_v4().to_string();
 
-        // Convert record batches to PyObject list
+        // Convert record batches to PyAny list
         let py_batches = batches
             .iter()
             .map(|rb| rb.to_pyarrow(py))
-            .collect::<PyResult<Vec<PyObject>>>()?;
+            .collect::<PyResult<Vec<_>>>()?;
 
         let py_schema = self.schema().into_pyobject(py)?;
 
@@ -523,7 +523,7 @@ impl PyDataFrame {
     /// Executes the plan, returning a list of `RecordBatch`es.
     /// Unless some order is specified in the plan, there is no
     /// guarantee of the order of the result.
-    fn collect(&self, py: Python) -> PyResult<Vec<PyObject>> {
+    fn collect<'py>(&self, py: Python<'py>) -> PyResult<Vec<Bound<'py, PyAny>>> {
         let batches = wait_for_future(py, self.df.as_ref().clone().collect())?
             .map_err(PyDataFusionError::from)?;
         // cannot use PyResult<Vec<RecordBatch>> return type due to
@@ -539,7 +539,7 @@ impl PyDataFrame {
 
     /// Executes this DataFrame and collects all results into a vector of vector of RecordBatch
     /// maintaining the input partitioning.
-    fn collect_partitioned(&self, py: Python) -> PyResult<Vec<Vec<PyObject>>> {
+    fn collect_partitioned<'py>(&self, py: Python<'py>) -> PyResult<Vec<Vec<Bound<'py, PyAny>>>> {
         let batches = wait_for_future(py, self.df.as_ref().clone().collect_partitioned())?
             .map_err(PyDataFusionError::from)?;
 
@@ -907,15 +907,14 @@ impl PyDataFrame {
 
     /// Convert to Arrow Table
     /// Collect the batches and pass to Arrow Table
-    fn to_arrow_table(&self, py: Python<'_>) -> PyResult<PyObject> {
+    fn to_arrow_table<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
         let batches = self.collect(py)?.into_pyobject(py)?;
         let schema = self.schema().into_pyobject(py)?;
 
         // Instantiate pyarrow Table object and use its from_batches method
         let table_class = py.import("pyarrow")?.getattr("Table")?;
         let args = PyTuple::new(py, &[batches, schema])?;
-        let table: PyObject = table_class.call_method1("from_batches", args)?.into();
-        Ok(table)
+        table_class.call_method1("from_batches", args)
     }
 
     #[pyo3(signature = (requested_schema=None))]
@@ -976,42 +975,38 @@ impl PyDataFrame {
 
     /// Convert to pandas dataframe with pyarrow
     /// Collect the batches, pass to Arrow Table & then convert to Pandas DataFrame
-    fn to_pandas(&self, py: Python<'_>) -> PyResult<PyObject> {
+    fn to_pandas<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
         let table = self.to_arrow_table(py)?;
 
         // See also: https://arrow.apache.org/docs/python/generated/pyarrow.Table.html#pyarrow.Table.to_pandas
-        let result = table.call_method0(py, "to_pandas")?;
-        Ok(result)
+        table.call_method0("to_pandas")
     }
 
     /// Convert to Python list using pyarrow
     /// Each list item represents one row encoded as dictionary
-    fn to_pylist(&self, py: Python<'_>) -> PyResult<PyObject> {
+    fn to_pylist<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
         let table = self.to_arrow_table(py)?;
 
         // See also: https://arrow.apache.org/docs/python/generated/pyarrow.Table.html#pyarrow.Table.to_pylist
-        let result = table.call_method0(py, "to_pylist")?;
-        Ok(result)
+        table.call_method0("to_pylist")
     }
 
     /// Convert to Python dictionary using pyarrow
     /// Each dictionary key is a column and the dictionary value represents the column values
-    fn to_pydict(&self, py: Python) -> PyResult<PyObject> {
+    fn to_pydict<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
         let table = self.to_arrow_table(py)?;
 
         // See also: https://arrow.apache.org/docs/python/generated/pyarrow.Table.html#pyarrow.Table.to_pydict
-        let result = table.call_method0(py, "to_pydict")?;
-        Ok(result)
+        table.call_method0("to_pydict")
     }
 
     /// Convert to polars dataframe with pyarrow
     /// Collect the batches, pass to Arrow Table & then convert to polars DataFrame
-    fn to_polars(&self, py: Python<'_>) -> PyResult<PyObject> {
+    fn to_polars<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
         let table = self.to_arrow_table(py)?;
         let dataframe = py.import("polars")?.getattr("DataFrame")?;
         let args = PyTuple::new(py, &[table])?;
-        let result: PyObject = dataframe.call1(args)?.into();
-        Ok(result)
+        dataframe.call1(args)
     }
 
     // Executes this DataFrame to get the total number of rows.
@@ -1023,7 +1018,7 @@ impl PyDataFrame {
     #[pyo3(signature = (value, columns=None))]
     fn fill_null(
         &self,
-        value: PyObject,
+        value: Bound<PyAny>,
         columns: Option<Vec<PyBackedStr>>,
         py: Python,
     ) -> PyDataFusionResult<Self> {

src/dataframe.rs

@@ -381,7 +381,7 @@ impl PyExpr {
     }
 
     /// Extracts the Expr value into a PyObject that can be shared with Python
-    pub fn python_value(&self, py: Python) -> PyResult<PyObject> {
+    pub fn python_value<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
         match &self.expr {
             Expr::Literal(scalar_value, _) => scalar_to_pyarrow(scalar_value, py),
             _ => Err(py_type_err(format!(

src/expr.rs

@@ -106,7 +106,7 @@ impl TryFrom<&Expr> for PyArrowFilterExpression {
             let op_module = Python::import(py, "operator")?;
             let pc_expr: PyDataFusionResult<Bound<'_, PyAny>> = match expr {
                 Expr::Column(Column { name, .. }) => Ok(pc.getattr("field")?.call1((name,))?),
-                Expr::Literal(scalar, _) => Ok(scalar_to_pyarrow(scalar, py)?.into_bound(py)),
+                Expr::Literal(scalar, _) => Ok(scalar_to_pyarrow(scalar, py)?),
                 Expr::BinaryExpr(BinaryExpr { left, op, right }) => {
                     let operator = operator_to_py(op, &op_module)?;
                     let left = PyArrowFilterExpression::try_from(left.as_ref())?.0;

src/pyarrow_filter_expression.rs

@@ -51,11 +51,11 @@ impl<'source> FromPyObject<'source> for PyScalarValue {
     }
 }
 
-pub fn scalar_to_pyarrow(scalar: &ScalarValue, py: Python) -> PyResult<PyObject> {
+pub fn scalar_to_pyarrow<'py>(scalar: &ScalarValue, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
     let array = scalar.to_array().map_err(PyDataFusionError::from)?;
     // convert to pyarrow array using C data interface
     let pyarray = array.to_data().to_pyarrow(py)?;
-    let pyscalar = pyarray.call_method1(py, "__getitem__", (0,))?;
+    let pyscalar = pyarray.call_method1("__getitem__", (0,))?;
 
     Ok(pyscalar)
 }

src/pyarrow_util.rs

@@ -35,7 +35,7 @@ pub struct PyRecordBatch {
 
 #[pymethods]
 impl PyRecordBatch {
-    fn to_pyarrow(&self, py: Python) -> PyResult<PyObject> {
+    fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
         self.batch.to_pyarrow(py)
     }
 }

src/record_batch.rs

@@ -86,7 +86,7 @@ impl PyTable {
 
     /// Get a reference to the schema for this table
     #[getter]
-    fn schema(&self, py: Python) -> PyResult<PyObject> {
+    fn schema<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
         self.table.schema().to_pyarrow(py)
     }
 

src/table.rs

@@ -90,7 +90,7 @@ impl Accumulator for RustAccumulator {
     fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
         Python::with_gil(|py| {
             // // 1. cast states to Pyarrow arrays
-            let py_states: Result<Vec<PyObject>> = states
+            let py_states: Result<Vec<Bound<PyAny>>> = states
                 .iter()
                 .map(|state| {
                     state