refactor: remove unecessary uses of datafusion subcrates

Signed-off-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: alamb

Cargo.toml

@@ -45,15 +45,8 @@ parquet = { version = "=55.0.0" }
 
 # datafusion
 datafusion = "47.0.0"
-datafusion-expr = "47.0.0"
-datafusion-common = "47.0.0"
 datafusion-ffi = "47.0.0"
-datafusion-functions = "47.0.0"
-datafusion-functions-aggregate = "47.0.0"
-datafusion-physical-expr = "47.0.0"
-datafusion-physical-plan = "47.0.0"
 datafusion-proto = "47.0.0"
-datafusion-sql = "47.0.0"
 
 # serde
 serde = { version = "1.0.194", features = ["derive"] }

crates/benchmarks/Cargo.toml

@@ -34,11 +34,6 @@ serde_json = { workspace = true }
 
 # datafusion
 datafusion = { workspace = true }
-datafusion-expr = { workspace = true }
-datafusion-common = { workspace = true }
-datafusion-proto = { workspace = true }
-datafusion-sql = { workspace = true }
-datafusion-physical-expr = { workspace = true }
 
 [dependencies.deltalake-core]
 path = "../core"

crates/benchmarks/src/bin/merge.rs

@@ -7,10 +7,10 @@ use arrow::datatypes::Schema as ArrowSchema;
 use arrow_array::{RecordBatch, StringArray, UInt32Array};
 use chrono::Duration;
 use clap::{command, Args, Parser, Subcommand};
+use datafusion::common::DataFusionError;
 use datafusion::functions::expr_fn::random;
+use datafusion::logical_expr::{cast, col, lit};
 use datafusion::{datasource::MemTable, prelude::DataFrame};
-use datafusion_common::DataFusionError;
-use datafusion_expr::{cast, col, lit};
 use deltalake_core::protocol::SaveMode;
 use deltalake_core::{
     arrow::{
@@ -223,7 +223,7 @@ async fn benchmark_merge_tpcds(
 
     let row_sample = ctx.table("t1").await?.join(
         ctx.table("file_sample").await?,
-        datafusion_common::JoinType::Inner,
+        datafusion::common::JoinType::Inner,
         &["file_path"],
         &["file"],
         None,
@@ -619,7 +619,7 @@ async fn main() {
             before_stats
                 .join(
                     after_stats,
-                    datafusion_common::JoinType::Inner,
+                    datafusion::common::JoinType::Inner,
                     &["before_name"],
                     &["after_name"],
                     None,

crates/catalog-unity/Cargo.toml

@@ -30,7 +30,6 @@ reqwest-middleware = { version = "0.4.0", features = ["json"] }
 rand = "0.8"
 dashmap = "6"
 datafusion = { workspace = true, optional = true }
-datafusion-common = { workspace = true, optional = true }
 moka = { version = "0.12", optional = true, features = ["future"] }
 
 [dev-dependencies]
@@ -45,7 +44,7 @@ aws = ["deltalake-aws"]
 azure = ["deltalake-azure"]
 gcp = ["deltalake-gcp"]
 r2 = ["deltalake-aws"]
-datafusion = ["dep:datafusion", "datafusion-common", "deltalake-core/datafusion", "moka"]
+datafusion = ["dep:datafusion", "deltalake-core/datafusion", "moka"]
 
 [[example]]
 name = "uc_example"

crates/catalog-unity/src/datafusion.rs

@@ -4,8 +4,8 @@ use chrono::prelude::*;
 use dashmap::DashMap;
 use datafusion::catalog::SchemaProvider;
 use datafusion::catalog::{CatalogProvider, CatalogProviderList};
+use datafusion::common::DataFusionError;
 use datafusion::datasource::TableProvider;
-use datafusion_common::DataFusionError;
 use futures::FutureExt;
 use moka::future::Cache;
 use moka::Expiry;
@@ -204,7 +204,10 @@ impl SchemaProvider for UnitySchemaProvider {
         self.table_names.clone()
     }
 
-    async fn table(&self, name: &str) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>> {
+    async fn table(
+        &self,
+        name: &str,
+    ) -> datafusion::common::Result<Option<Arc<dyn TableProvider>>> {
         let maybe_table = self
             .client
             .get_table(&self.catalog_name, &self.schema_name, name)

crates/catalog-unity/src/lib.rs

@@ -109,7 +109,7 @@ pub enum UnityCatalogError {
 
     #[cfg(feature = "datafusion")]
     #[error("Datafusion error: {0}")]
-    DatafusionError(#[from] datafusion_common::DataFusionError),
+    DatafusionError(#[from] ::datafusion::common::DataFusionError),
 
     /// Cannot initialize DynamoDbConfiguration due to some sort of threading issue
     #[error("Unable to initialize Unity Catalog, potentially a threading issue")]

crates/core/Cargo.toml

@@ -37,14 +37,7 @@ pin-project-lite = "^0.2.7"
 
 # datafusion
 datafusion = { workspace = true, optional = true }
-datafusion-expr = { workspace = true, optional = true }
-datafusion-common = { workspace = true, optional = true }
 datafusion-proto = { workspace = true, optional = true }
-datafusion-sql = { workspace = true, optional = true }
-datafusion-physical-expr = { workspace = true, optional = true }
-datafusion-physical-plan = { workspace = true, optional = true }
-datafusion-functions = { workspace = true, optional = true }
-datafusion-functions-aggregate = { workspace = true, optional = true }
 
 # serde
 serde = { workspace = true, features = ["derive"] }
@@ -115,14 +108,7 @@ tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 default = ["rustls"]
 datafusion = [
     "dep:datafusion",
-    "datafusion-expr",
-    "datafusion-common",
     "datafusion-proto",
-    "datafusion-physical-expr",
-    "datafusion-physical-plan",
-    "datafusion-sql",
-    "datafusion-functions",
-    "datafusion-functions-aggregate",
 ]
 datafusion-ext = ["datafusion"]
 json = ["parquet/json"]

crates/core/src/data_catalog/storage/mod.rs

@@ -7,8 +7,8 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use dashmap::DashMap;
 use datafusion::catalog::SchemaProvider;
+use datafusion::common::DataFusionError;
 use datafusion::datasource::TableProvider;
-use datafusion_common::DataFusionError;
 use futures::TryStreamExt;
 use object_store::ObjectStore;
 
@@ -59,7 +59,7 @@ impl ListingSchemaProvider {
     }
 
     /// Reload table information from ObjectStore
-    pub async fn refresh(&self) -> datafusion_common::Result<()> {
+    pub async fn refresh(&self) -> datafusion::common::Result<()> {
         let entries: Vec<_> = self.store.list(None).try_collect().await?;
         let mut tables = HashSet::new();
         for file in entries.iter() {
@@ -110,7 +110,10 @@ impl SchemaProvider for ListingSchemaProvider {
         self.tables.iter().map(|t| t.key().clone()).collect()
     }
 
-    async fn table(&self, name: &str) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>> {
+    async fn table(
+        &self,
+        name: &str,
+    ) -> datafusion::common::Result<Option<Arc<dyn TableProvider>>> {
         let Some(location) = self.tables.get(name).map(|t| t.clone()) else {
             return Ok(None);
         };
@@ -123,7 +126,7 @@ impl SchemaProvider for ListingSchemaProvider {
         &self,
         _name: String,
         _table: Arc<dyn TableProvider>,
-    ) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>> {
+    ) -> datafusion::common::Result<Option<Arc<dyn TableProvider>>> {
         Err(DataFusionError::Execution(
             "schema provider does not support registering tables".to_owned(),
         ))
@@ -132,7 +135,7 @@ impl SchemaProvider for ListingSchemaProvider {
     fn deregister_table(
         &self,
         _name: &str,
-    ) -> datafusion_common::Result<Option<Arc<dyn TableProvider>>> {
+    ) -> datafusion::common::Result<Option<Arc<dyn TableProvider>>> {
         Err(DataFusionError::Execution(
             "schema provider does not support deregistering tables".to_owned(),
         ))

crates/core/src/delta_datafusion/cdf/scan.rs

@@ -5,15 +5,15 @@ use arrow_schema::{Schema, SchemaRef};
 use async_trait::async_trait;
 use datafusion::catalog::Session;
 use datafusion::catalog::TableProvider;
+use datafusion::common::{exec_datafusion_err, Column, DFSchema, Result as DataFusionResult};
 use datafusion::execution::SessionState;
-use datafusion_common::{exec_datafusion_err, Column, DFSchema, Result as DataFusionResult};
-use datafusion_expr::utils::conjunction;
-use datafusion_expr::{Expr, TableProviderFilterPushDown, TableType};
-use datafusion_physical_expr::PhysicalExpr;
-use datafusion_physical_plan::filter::FilterExec;
-use datafusion_physical_plan::limit::GlobalLimitExec;
-use datafusion_physical_plan::projection::ProjectionExec;
-use datafusion_physical_plan::ExecutionPlan;
+use datafusion::logical_expr::utils::conjunction;
+use datafusion::logical_expr::{Expr, TableProviderFilterPushDown, TableType};
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::filter::FilterExec;
+use datafusion::physical_plan::limit::GlobalLimitExec;
+use datafusion::physical_plan::projection::ProjectionExec;
+use datafusion::physical_plan::ExecutionPlan;
 
 use crate::DeltaTableError;
 use crate::{

crates/core/src/delta_datafusion/cdf/scan_utils.rs

@@ -3,8 +3,8 @@ use std::sync::Arc;
 
 use arrow_schema::{DataType, Field, Schema, SchemaRef};
 use chrono::TimeZone;
+use datafusion::common::ScalarValue;
 use datafusion::datasource::listing::PartitionedFile;
-use datafusion_common::ScalarValue;
 use object_store::path::Path;
 use object_store::ObjectMeta;
 use serde_json::Value;