TMP: Updates for newer DataFusion

Author: davisp

src/common/data_type.rs

@@ -347,6 +347,7 @@ impl DataTypeMap {
             ScalarValue::Map(_) => Err(PyNotImplementedError::new_err(
                 "ScalarValue::Map".to_string(),
             )),
+            ScalarValue::RunEndEncoded(_, _, _) => todo!(),
         }
     }
 }

src/context.rs

@@ -44,7 +44,7 @@ use datafusion::execution::options::ReadOptions;
 use datafusion::execution::runtime_env::RuntimeEnvBuilder;
 use datafusion::execution::session_state::SessionStateBuilder;
 use datafusion::prelude::{
-    AvroReadOptions, CsvReadOptions, DataFrame, NdJsonReadOptions, ParquetReadOptions,
+    AvroReadOptions, CsvReadOptions, DataFrame, JsonReadOptions, ParquetReadOptions,
 };
 use datafusion_ffi::catalog_provider::FFI_CatalogProvider;
 use datafusion_ffi::catalog_provider_list::FFI_CatalogProviderList;
@@ -815,7 +815,7 @@ impl PySessionContext {
             .to_str()
             .ok_or_else(|| PyValueError::new_err("Unable to convert path to a string"))?;
 
-        let mut options = NdJsonReadOptions::default()
+        let mut options = JsonReadOptions::default()
             .file_compression_type(parse_file_compression_type(file_compression_type)?)
             .table_partition_cols(
                 table_partition_cols
@@ -976,7 +976,7 @@ impl PySessionContext {
         let path = path
             .to_str()
             .ok_or_else(|| PyValueError::new_err("Unable to convert path to a string"))?;
-        let mut options = NdJsonReadOptions::default()
+        let mut options = JsonReadOptions::default()
             .table_partition_cols(
                 table_partition_cols
                     .into_iter()

src/dataset_exec.rs

@@ -31,7 +31,7 @@ use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
 use datafusion::physical_plan::{
     DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, Partitioning,
-    SendableRecordBatchStream, Statistics,
+    SendableRecordBatchStream,
 };
 use futures::{TryStreamExt, stream};
 /// Implements a Datafusion physical ExecutionPlan that delegates to a PyArrow Dataset
@@ -70,7 +70,6 @@ pub(crate) struct DatasetExec {
     fragments: Py<PyList>,
     columns: Option<Vec<String>>,
     filter_expr: Option<Py<PyAny>>,
-    projected_statistics: Statistics,
     plan_properties: datafusion::physical_plan::PlanProperties,
 }
 
@@ -111,7 +110,7 @@ impl DatasetExec {
 
         let scanner = dataset.call_method("scanner", (), Some(&kwargs))?;
 
-        let schema = Arc::new(
+        let schema: SchemaRef = Arc::new(
             scanner
                 .getattr("projected_schema")?
                 .extract::<PyArrowType<_>>()?
@@ -130,7 +129,6 @@ impl DatasetExec {
         let fragments_iter = pylist.call1((fragments_iterator,))?;
         let fragments = fragments_iter.downcast::<PyList>().map_err(PyErr::from)?;
 
-        let projected_statistics = Statistics::new_unknown(&schema);
         let plan_properties = datafusion::physical_plan::PlanProperties::new(
             EquivalenceProperties::new(schema.clone()),
             Partitioning::UnknownPartitioning(fragments.len()),
@@ -144,7 +142,6 @@ impl DatasetExec {
             fragments: fragments.clone().unbind(),
             columns,
             filter_expr,
-            projected_statistics,
             plan_properties,
         })
     }
@@ -235,10 +232,6 @@ impl ExecutionPlan for DatasetExec {
         })
     }
 
-    fn statistics(&self) -> DFResult<Statistics> {
-        Ok(self.projected_statistics.clone())
-    }
-
     fn properties(&self) -> &datafusion::physical_plan::PlanProperties {
         &self.plan_properties
     }

src/expr.rs

@@ -210,6 +210,7 @@ impl PyExpr {
             Expr::Unnest(value) => {
                 Ok(unnest_expr::PyUnnestExpr::from(value.clone()).into_bound_py_any(py)?)
             }
+            Expr::SetComparison(_) => todo!(),
         })
     }
 
@@ -382,6 +383,7 @@ impl PyExpr {
             Expr::Wildcard { .. } => {
                 return Err(py_unsupported_variant_err("Expr::Wildcard is unsupported"));
             }
+            Expr::SetComparison(_) => todo!(),
         })
     }
 
@@ -518,6 +520,7 @@ impl PyExpr {
             Expr::Wildcard { .. } => {
                 Err(py_unsupported_variant_err("Expr::Wildcard is unsupported"))
             }
+            Expr::SetComparison(_) => todo!(),
         }
     }
 

src/expr/dml.rs

@@ -110,6 +110,7 @@ impl From<WriteOp> for PyWriteOp {
             WriteOp::Update => PyWriteOp::Update,
             WriteOp::Delete => PyWriteOp::Delete,
             WriteOp::Ctas => PyWriteOp::Ctas,
+            WriteOp::Truncate => todo!(),
         }
     }
 }