chore: Use datafusion re-exported dependencies (#856)

* Use datafusion re-exported crates

* update dependency export syntax

Author: emgeee

.gitignore

@@ -25,6 +25,7 @@ dist
 #   intended to run in multiple environments; otherwise, check them in:
 .python-version
 venv
+.venv
 
 apache-rat-*.jar
 *rat.txt

Cargo.lock

@@ -39,11 +39,6 @@ rand = "0.8"
 pyo3 = { version = "0.21", features = ["extension-module", "abi3", "abi3-py38"] }
 arrow = { version = "52", feature = ["pyarrow"] }
 datafusion = { version = "41.0.0", features = ["pyarrow", "avro", "unicode_expressions"] }
-datafusion-common = { version = "41.0.0", features = ["pyarrow"] }
-datafusion-expr = { version = "41.0.0" }
-datafusion-functions-nested = { version = "41.0.0" }
-datafusion-optimizer = { version = "41.0.0" }
-datafusion-sql = { version = "41.0.0" }
 datafusion-substrait = { version = "41.0.0", optional = true }
 prost = "0.12" # keep in line with `datafusion-substrait`
 prost-types = "0.12" # keep in line with `datafusion-substrait`
@@ -67,4 +62,4 @@ crate-type = ["cdylib", "rlib"]
 [profile.release]
 lto = true
 codegen-units = 1
-    
+    

Cargo.toml

@@ -17,8 +17,8 @@
 
 use datafusion::arrow::array::Array;
 use datafusion::arrow::datatypes::{DataType, IntervalUnit, TimeUnit};
-use datafusion_common::{DataFusionError, ScalarValue};
-use datafusion_expr::sqlparser::ast::NullTreatment as DFNullTreatment;
+use datafusion::common::{DataFusionError, ScalarValue};
+use datafusion::logical_expr::sqlparser::ast::NullTreatment as DFNullTreatment;
 use pyo3::{exceptions::PyValueError, prelude::*};
 
 use crate::errors::py_datafusion_err;

src/common/data_type.rs

@@ -17,7 +17,7 @@
 
 use std::sync::Arc;
 
-use datafusion_common::DFSchema;
+use datafusion::common::DFSchema;
 use pyo3::prelude::*;
 
 #[derive(Debug, Clone)]

src/common/df_schema.rs

@@ -18,10 +18,10 @@
 use std::any::Any;
 
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion_expr::{Expr, TableProviderFilterPushDown, TableSource};
+use datafusion::logical_expr::{Expr, TableProviderFilterPushDown, TableSource};
 use pyo3::prelude::*;
 
-use datafusion_expr::utils::split_conjunction;
+use datafusion::logical_expr::utils::split_conjunction;
 
 use super::{data_type::DataTypeMap, function::SqlFunction};
 
@@ -166,7 +166,7 @@ impl TableSource for SqlTableSource {
     fn supports_filter_pushdown(
         &self,
         filter: &Expr,
-    ) -> datafusion_common::Result<TableProviderFilterPushDown> {
+    ) -> datafusion::common::Result<TableProviderFilterPushDown> {
         let filters = split_conjunction(filter);
         if filters.iter().all(|f| is_supported_push_down_expr(f)) {
             // Push down filters to the tablescan operation if all are supported
@@ -180,22 +180,22 @@ impl TableSource for SqlTableSource {
         }
     }
 
-    fn table_type(&self) -> datafusion_expr::TableType {
-        datafusion_expr::TableType::Base
+    fn table_type(&self) -> datafusion::logical_expr::TableType {
+        datafusion::logical_expr::TableType::Base
     }
 
     #[allow(deprecated)]
     fn supports_filters_pushdown(
         &self,
         filters: &[&Expr],
-    ) -> datafusion_common::Result<Vec<TableProviderFilterPushDown>> {
+    ) -> datafusion::common::Result<Vec<TableProviderFilterPushDown>> {
         filters
             .iter()
             .map(|f| self.supports_filter_pushdown(f))
             .collect()
     }
 
-    fn get_logical_plan(&self) -> Option<&datafusion_expr::LogicalPlan> {
+    fn get_logical_plan(&self) -> Option<&datafusion::logical_expr::LogicalPlan> {
         None
     }
 }

src/common/schema.rs

@@ -18,8 +18,8 @@
 use pyo3::prelude::*;
 use pyo3::types::*;
 
+use datafusion::common::ScalarValue;
 use datafusion::config::ConfigOptions;
-use datafusion_common::ScalarValue;
 
 #[pyclass(name = "Config", module = "datafusion", subclass)]
 #[derive(Clone)]

src/config.rs

@@ -46,6 +46,7 @@ use crate::utils::{get_tokio_runtime, wait_for_future};
 use datafusion::arrow::datatypes::{DataType, Schema, SchemaRef};
 use datafusion::arrow::pyarrow::PyArrowType;
 use datafusion::arrow::record_batch::RecordBatch;
+use datafusion::common::ScalarValue;
 use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
 use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::datasource::listing::{
@@ -61,7 +62,6 @@ use datafusion::physical_plan::SendableRecordBatchStream;
 use datafusion::prelude::{
     AvroReadOptions, CsvReadOptions, DataFrame, NdJsonReadOptions, ParquetReadOptions,
 };
-use datafusion_common::ScalarValue;
 use pyo3::types::{PyDict, PyList, PyTuple};
 use tokio::task::JoinHandle;
 
@@ -962,15 +962,15 @@ impl PySessionContext {
         // create a Tokio runtime to run the async code
         let rt = &get_tokio_runtime(py).0;
         let plan = plan.plan.clone();
-        let fut: JoinHandle<datafusion_common::Result<SendableRecordBatchStream>> =
+        let fut: JoinHandle<datafusion::common::Result<SendableRecordBatchStream>> =
             rt.spawn(async move { plan.execute(part, Arc::new(ctx)) });
         let stream = wait_for_future(py, fut).map_err(py_datafusion_err)?;
         Ok(PyRecordBatchStream::new(stream?))
     }
 }
 
 impl PySessionContext {
-    async fn _table(&self, name: &str) -> datafusion_common::Result<DataFrame> {
+    async fn _table(&self, name: &str) -> datafusion::common::Result<DataFrame> {
         self.ctx.table(name).await
     }
 }

src/context.rs

@@ -27,12 +27,12 @@ use arrow::util::display::{ArrayFormatter, FormatOptions};
 use datafusion::arrow::datatypes::Schema;
 use datafusion::arrow::pyarrow::{PyArrowType, ToPyArrow};
 use datafusion::arrow::util::pretty;
+use datafusion::common::UnnestOptions;
 use datafusion::config::{CsvOptions, TableParquetOptions};
 use datafusion::dataframe::{DataFrame, DataFrameWriteOptions};
 use datafusion::execution::SendableRecordBatchStream;
 use datafusion::parquet::basic::{BrotliLevel, Compression, GzipLevel, ZstdLevel};
 use datafusion::prelude::*;
-use datafusion_common::UnnestOptions;
 use pyo3::exceptions::{PyTypeError, PyValueError};
 use pyo3::prelude::*;
 use pyo3::pybacked::PyBackedStr;
@@ -541,7 +541,7 @@ impl PyDataFrame {
         // create a Tokio runtime to run the async code
         let rt = &get_tokio_runtime(py).0;
         let df = self.df.as_ref().clone();
-        let fut: JoinHandle<datafusion_common::Result<SendableRecordBatchStream>> =
+        let fut: JoinHandle<datafusion::common::Result<SendableRecordBatchStream>> =
             rt.spawn(async move { df.execute_stream().await });
         let stream = wait_for_future(py, fut).map_err(py_datafusion_err)?;
         Ok(PyRecordBatchStream::new(stream?))
@@ -551,7 +551,7 @@ impl PyDataFrame {
         // create a Tokio runtime to run the async code
         let rt = &get_tokio_runtime(py).0;
         let df = self.df.as_ref().clone();
-        let fut: JoinHandle<datafusion_common::Result<Vec<SendableRecordBatchStream>>> =
+        let fut: JoinHandle<datafusion::common::Result<Vec<SendableRecordBatchStream>>> =
             rt.spawn(async move { df.execute_stream_partitioned().await });
         let stream = wait_for_future(py, fut).map_err(py_datafusion_err)?;
 

src/dataframe.rs

@@ -31,9 +31,9 @@ use datafusion::arrow::datatypes::SchemaRef;
 use datafusion::arrow::pyarrow::PyArrowType;
 use datafusion::datasource::{TableProvider, TableType};
 use datafusion::error::{DataFusionError, Result as DFResult};
+use datafusion::logical_expr::Expr;
 use datafusion::logical_expr::TableProviderFilterPushDown;
 use datafusion::physical_plan::ExecutionPlan;
-use datafusion_expr::Expr;
 
 use crate::dataset_exec::DatasetExec;
 use crate::pyarrow_filter_expression::PyArrowFilterExpression;