Upgrade to DataFusion 33.0.0-rc1

Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "datafusion-python"
-version = "32.0.0"
+version = "33.0.0"
 homepage = "https://github.com/apache/arrow-datafusion-python"
 repository = "https://github.com/apache/arrow-datafusion-python"
 authors = ["Apache Arrow <[email protected]>"]
@@ -36,15 +36,15 @@ substrait = ["dep:datafusion-substrait"]
 [dependencies]
 tokio = { version = "1.24", features = ["macros", "rt", "rt-multi-thread", "sync"] }
 rand = "0.8"
-pyo3 = { version = "0.19", features = ["extension-module", "abi3", "abi3-py38"] }
-datafusion = { version = "32.0.0", features = ["pyarrow", "avro"] }
-datafusion-common = { version = "32.0.0", features = ["pyarrow"] }
-datafusion-expr = { version = "32.0.0" }
-datafusion-optimizer = { version = "32.0.0" }
-datafusion-sql = { version = "32.0.0" }
-datafusion-substrait = { version = "32.0.0", optional = true }
-prost = "0.11"
-prost-types = "0.11"
+pyo3 = { version = "0.20", features = ["extension-module", "abi3", "abi3-py38"] }
+datafusion = { git = "https://github.com/apache/arrow-datafusion", rev = "33.0.0-rc1", features = ["pyarrow", "avro"] }
+datafusion-common = { git = "https://github.com/apache/arrow-datafusion", rev = "33.0.0-rc1", features = ["pyarrow"] }
+datafusion-expr = { git = "https://github.com/apache/arrow-datafusion", rev = "33.0.0-rc1" }
+datafusion-optimizer = { git = "https://github.com/apache/arrow-datafusion", rev = "33.0.0-rc1" }
+datafusion-sql = { git = "https://github.com/apache/arrow-datafusion", rev = "33.0.0-rc1" }
+datafusion-substrait = { git = "https://github.com/apache/arrow-datafusion", rev = "33.0.0-rc1", optional = true }
+prost = "0.12"
+prost-types = "0.12"
 uuid = { version = "1.3", features = ["v4"] }
 mimalloc = { version = "0.1", optional = true, default-features = false }
 async-trait = "0.1"

src/common/data_type.rs

@@ -288,7 +288,7 @@ impl DataTypeMap {
             ScalarValue::IntervalMonthDayNano(..) => {
                 Ok(DataType::Interval(IntervalUnit::MonthDayNano))
             }
-            ScalarValue::List(_val, field_ref) => Ok(DataType::List(field_ref.to_owned())),
+            ScalarValue::List(arr) => Ok(arr.data_type().to_owned()),
             ScalarValue::Struct(_, fields) => Ok(DataType::Struct(fields.to_owned())),
             ScalarValue::FixedSizeBinary(size, _) => Ok(DataType::FixedSizeBinary(*size)),
             ScalarValue::Fixedsizelist(_, field_ref, size) => {

src/dataset_exec.rs

@@ -133,13 +133,14 @@ impl DatasetExec {
             .downcast()
             .map_err(PyErr::from)?;
 
+        let projected_statistics = Statistics::new_unknown(&schema);
         Ok(DatasetExec {
             dataset: dataset.into(),
             schema,
             fragments: fragments.into(),
             columns,
             filter_expr,
-            projected_statistics: Default::default(),
+            projected_statistics,
         })
     }
 }
@@ -236,8 +237,8 @@ impl ExecutionPlan for DatasetExec {
         })
     }
 
-    fn statistics(&self) -> Statistics {
-        self.projected_statistics.clone()
+    fn statistics(&self) -> DFResult<Statistics> {
+        Ok(self.projected_statistics.clone())
     }
 }
 

src/expr.rs

@@ -332,7 +332,7 @@ impl PyExpr {
                 ScalarValue::Binary(v) => v.clone().into_py(py),
                 ScalarValue::FixedSizeBinary(_, _) => todo!(),
                 ScalarValue::LargeBinary(v) => v.clone().into_py(py),
-                ScalarValue::List(_, _) => todo!(),
+                ScalarValue::List(_) => todo!(),
                 ScalarValue::Date32(v) => v.into_py(py),
                 ScalarValue::Date64(v) => v.into_py(py),
                 ScalarValue::Time32Second(v) => v.into_py(py),

src/expr/window.rs

@@ -166,9 +166,10 @@ fn not_window_function_err(expr: Expr) -> PyErr {
 
 #[pymethods]
 impl PyWindowFrame {
-    #[new(unit, start_bound, end_bound)]
-    pub fn new(units: &str, start_bound: Option<u64>, end_bound: Option<u64>) -> PyResult<Self> {
-        let units = units.to_ascii_lowercase();
+    #[new]
+    #[pyo3(signature=(unit, start_bound, end_bound))]
+    pub fn new(unit: &str, start_bound: Option<u64>, end_bound: Option<u64>) -> PyResult<Self> {
+        let units = unit.to_ascii_lowercase();
         let units = match units.as_str() {
             "rows" => WindowFrameUnits::Rows,
             "range" => WindowFrameUnits::Range,

src/substrait.rs

@@ -134,7 +134,7 @@ impl PySubstraitConsumer {
         plan: PyPlan,
         py: Python,
     ) -> PyResult<PyLogicalPlan> {
-        let result = consumer::from_substrait_plan(&mut ctx.ctx, &plan.plan);
+        let result = consumer::from_substrait_plan(&ctx.ctx, &plan.plan);
         let logical_plan = wait_for_future(py, result).map_err(DataFusionError::from)?;
         Ok(PyLogicalPlan::new(logical_plan))
     }

src/udaf.rs

@@ -143,7 +143,8 @@ pub struct PyAggregateUDF {
 
 #[pymethods]
 impl PyAggregateUDF {
-    #[new(name, accumulator, input_type, return_type, state_type, volatility)]
+    #[new]
+    #[pyo3(signature=(name, accumulator, input_type, return_type, state_type, volatility))]
     fn new(
         name: &str,
         accumulator: PyObject,

src/udf.rs

@@ -68,7 +68,8 @@ pub struct PyScalarUDF {
 
 #[pymethods]
 impl PyScalarUDF {
-    #[new(name, func, input_types, return_type, volatility)]
+    #[new]
+    #[pyo3(signature=(name, func, input_types, return_type, volatility))]
     fn new(
         name: &str,
         func: PyObject,