feat: add upstream variant and method params

- `WindowFunction` and `AggregateFunction` have `null_treatment` options.
- `ScalarValue` and `DataType` have new variants
- `SchemaProvider::table` now returns a `Result`

Author: Michael-J-Ward

src/catalog.rs

@@ -97,7 +97,7 @@ impl PyDatabase {
     }
 
     fn table(&self, name: &str, py: Python) -> PyResult<PyTable> {
-        if let Some(table) = wait_for_future(py, self.database.table(name)) {
+        if let Some(table) = wait_for_future(py, self.database.table(name))? {
             Ok(PyTable::new(table))
         } else {
             Err(DataFusionError::Common(format!("Table not found: {name}")).into())

src/common/data_type.rs

@@ -226,6 +226,19 @@ impl DataTypeMap {
             DataType::RunEndEncoded(_, _) => Err(py_datafusion_err(
                 DataFusionError::NotImplemented(format!("{:?}", arrow_type)),
             )),
+            DataType::BinaryView => Err(py_datafusion_err(DataFusionError::NotImplemented(
+                format!("{:?}", arrow_type),
+            ))),
+            DataType::Utf8View => Err(py_datafusion_err(DataFusionError::NotImplemented(format!(
+                "{:?}",
+                arrow_type
+            )))),
+            DataType::ListView(_) => Err(py_datafusion_err(DataFusionError::NotImplemented(
+                format!("{:?}", arrow_type),
+            ))),
+            DataType::LargeListView(_) => Err(py_datafusion_err(DataFusionError::NotImplemented(
+                format!("{:?}", arrow_type),
+            ))),
         }
     }
 
@@ -309,6 +322,9 @@ impl DataTypeMap {
             ScalarValue::DurationMillisecond(_) => Ok(DataType::Duration(TimeUnit::Millisecond)),
             ScalarValue::DurationMicrosecond(_) => Ok(DataType::Duration(TimeUnit::Microsecond)),
             ScalarValue::DurationNanosecond(_) => Ok(DataType::Duration(TimeUnit::Nanosecond)),
+            ScalarValue::Union(_, _, _) => Err(py_datafusion_err(DataFusionError::NotImplemented(
+                "ScalarValue::LargeList".to_string(),
+            ))),
         }
     }
 }
@@ -598,6 +614,10 @@ impl DataTypeMap {
             DataType::Decimal256(_, _) => "Decimal256",
             DataType::Map(_, _) => "Map",
             DataType::RunEndEncoded(_, _) => "RunEndEncoded",
+            DataType::BinaryView => "BinaryView",
+            DataType::Utf8View => "Utf8View",
+            DataType::ListView(_) => "ListView",
+            DataType::LargeListView(_) => "LargeListView",
         })
     }
 }

src/expr.rs

@@ -382,6 +382,11 @@ impl PyExpr {
                         "ScalarValue::LargeList".to_string(),
                     ),
                 )),
+                ScalarValue::Union(_, _, _) => Err(py_datafusion_err(
+                    datafusion_common::DataFusionError::NotImplemented(
+                        "ScalarValue::Union".to_string(),
+                    ),
+                )),
             },
             _ => Err(py_type_err(format!(
                 "Non Expr::Literal encountered in types: {:?}",

src/functions.rs

@@ -162,6 +162,7 @@ fn count_star() -> PyResult<PyExpr> {
             distinct: false,
             filter: None,
             order_by: None,
+            null_treatment: None,
         }),
     })
 }
@@ -214,6 +215,7 @@ fn window(
                 .map(|x| x.expr)
                 .collect::<Vec<_>>(),
             window_frame,
+            null_treatment: None,
         }),
     })
 }
@@ -256,6 +258,7 @@ macro_rules! aggregate_function {
                 distinct,
                 filter: None,
                 order_by: None,
+                null_treatment: None,
             });
             expr.into()
         }

src/utils.rs

@@ -37,9 +37,9 @@ pub(crate) fn get_tokio_runtime(py: Python) -> PyRef<TokioRuntime> {
 }
 
 /// Utility to collect rust futures with GIL released
-pub fn wait_for_future<F: Future>(py: Python, f: F) -> F::Output
+pub fn wait_for_future<F>(py: Python, f: F) -> F::Output
 where
-    F: Send,
+    F: Future + Send,
     F::Output: Send,
 {
     let runtime: &Runtime = &get_tokio_runtime(py).0;