chore: remove sql feature and sqlparser dependency from DataFusion

Cargo.toml

@@ -142,14 +142,27 @@ prost-types = "0.14"
 # The `axum` version must match the one used in `tonic` (replace `RELEASE` with the release we are using):
 #   https://github.com/hyperium/tonic/blob/vRELEASE/tonic/Cargo.toml
 axum = "0.8.8"
-datafusion = { version = "53.0.0", features = ["serde", "avro", "sql"] }
+datafusion = { version = "53.0.0", default-features = false, features = [
+    "nested_expressions",
+    "crypto_expressions",
+    "datetime_expressions",
+    "encoding_expressions",
+    "regex_expressions",
+    "string_expressions",
+    "unicode_expressions",
+    "compression",
+    "parquet",
+    "serde",
+    "recursive_protection",
+] }
 datafusion-common = { version = "53.0.0", features = ["object_store", "avro"] }
 datafusion-datasource = { version = "53.0.0" }
-datafusion-expr = { version = "53.0.0" }
+datafusion-datasource-avro = { version = "53.0.0" }
+datafusion-expr = { version = "53.0.0", default-features = false }
 datafusion-expr-common = { version = "53.0.0" }
 datafusion-proto = { version = "53.0.0" }
 datafusion-functions = { version = "53.0.0" }
-datafusion-functions-nested = { version = "53.0.0" }
+datafusion-functions-nested = { version = "53.0.0", default-features = false }
 datafusion-physical-expr = { version = "53.0.0" }
 datafusion-session = { version = "53.0.0" }
 datafusion-spark = { version = "53.0.0", features = ["core"] }

crates/sail-common-datafusion/src/error.rs

@@ -128,7 +128,6 @@ impl CommonErrorCause {
                 DataFusionError::AvroError(e) => Self::FormatAvro(e.to_string()),
                 DataFusionError::ObjectStore(e) => Self::Io(e.to_string()),
                 DataFusionError::IoError(e) => Self::Io(e.to_string()),
-                DataFusionError::SQL(e, _) => Self::Unknown(e.to_string()),
                 DataFusionError::NotImplemented(x) => Self::NotImplemented(x.clone()),
                 DataFusionError::Internal(x) => Self::Internal(x.clone()),
                 DataFusionError::Plan(x) => Self::Plan(x.clone()),

crates/sail-data-source/Cargo.toml

@@ -17,6 +17,7 @@ futures = { workspace = true }
 datafusion = { workspace = true }
 datafusion-common = { workspace = true }
 datafusion-datasource = { workspace = true }
+datafusion-datasource-avro = { workspace = true }
 datafusion-session = { workspace = true }
 object_store = { workspace = true }
 chumsky = { workspace = true }

crates/sail-data-source/src/formats/avro/mod.rs

@@ -2,9 +2,9 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use datafusion::catalog::Session;
-use datafusion::datasource::file_format::avro::AvroFormat;
 use datafusion_common::parsers::CompressionTypeVariant;
 use datafusion_datasource::file_format::FileFormat;
+use datafusion_datasource_avro::AvroFormat;
 
 use crate::formats::listing::{DefaultSchemaInfer, ListingFormat, ListingTableFormat, SchemaInfer};
 

crates/sail-execution/Cargo.toml

@@ -30,6 +30,7 @@ fastrace = { workspace = true }
 tower = { workspace = true }
 prost = { workspace = true }
 datafusion = { workspace = true }
+datafusion-datasource-avro = { workspace = true }
 datafusion-proto = { workspace = true }
 datafusion-spark = { workspace = true }
 arrow-flight = { workspace = true }

crates/sail-execution/src/codec.rs

@@ -12,8 +12,7 @@ use datafusion::common::{
 use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
 use datafusion::datasource::memory::MemorySourceConfig;
 use datafusion::datasource::physical_plan::{
-    ArrowSource, AvroSource, FileScanConfig, FileScanConfigBuilder, FileSink, FileSinkConfig,
-    JsonSource,
+    ArrowSource, FileScanConfig, FileScanConfigBuilder, FileSink, FileSinkConfig, JsonSource,
 };
 use datafusion::datasource::sink::DataSinkExec;
 use datafusion::datasource::source::{DataSource, DataSourceExec};
@@ -34,6 +33,7 @@ use datafusion::physical_plan::recursive_query::RecursiveQueryExec;
 use datafusion::physical_plan::sorts::partial_sort::PartialSortExec;
 use datafusion::physical_plan::work_table::WorkTableExec;
 use datafusion::physical_plan::{ExecutionPlan, PlanProperties};
+use datafusion_datasource_avro::source::AvroSource;
 use datafusion_proto::generated::datafusion_common as gen_datafusion_common;
 use datafusion_proto::physical_plan::from_proto::{
     parse_physical_expr, parse_physical_sort_exprs, parse_protobuf_file_scan_config,

crates/sail-plan/src/resolver/expression/wildcard.rs

@@ -1,7 +1,6 @@
 use std::collections::VecDeque;
 
 use arrow::datatypes::DataType;
-use datafusion::sql::unparser::expr_to_sql;
 use datafusion_common::{DFSchemaRef, TableReference};
 use datafusion_expr::expr::ScalarFunction;
 use datafusion_expr::{col, expr, lit, ScalarUDF};
@@ -148,18 +147,24 @@ impl PlanResolver<'_> {
         schema: &DFSchemaRef,
         state: &mut PlanResolverState,
     ) -> PlanResult<expr::WildcardOptions> {
-        use datafusion::sql::sqlparser::ast;
+        fn make_ident(value: impl Into<String>) -> expr::Ident {
+            expr::Ident {
+                value: value.into(),
+                quote_style: None,
+                span: String::new(),
+            }
+        }
 
         let ilike = wildcard_options
             .ilike_pattern
-            .map(|x| ast::IlikeSelectItem { pattern: x });
+            .map(|x| expr::IlikeSelectItem { pattern: x });
         let exclude = wildcard_options
             .exclude_columns
             .map(|x| {
                 let exclude = if x.len() > 1 {
-                    ast::ExcludeSelectItem::Multiple(x.into_iter().map(ast::Ident::new).collect())
+                    expr::ExcludeSelectItem::Multiple(x.into_iter().map(make_ident).collect())
                 } else if let Some(x) = x.into_iter().next() {
-                    ast::ExcludeSelectItem::Single(ast::Ident::new(x))
+                    expr::ExcludeSelectItem::Single(make_ident(x))
                 } else {
                     return Err(PlanError::invalid(
                         "exclude columns must have at least one column",
@@ -176,14 +181,14 @@ impl PlanResolver<'_> {
                     let first_element = deque.pop_front().ok_or_else(|| {
                         PlanError::invalid("except columns must have at least one column")
                     })?;
-                    let additional_elements = deque.into_iter().map(ast::Ident::new).collect();
-                    ast::ExceptSelectItem {
-                        first_element: ast::Ident::new(first_element),
+                    let additional_elements = deque.into_iter().map(make_ident).collect();
+                    expr::ExceptSelectItem {
+                        first_element: make_ident(first_element),
                         additional_elements,
                     }
                 } else if let Some(x) = x.into_iter().next() {
-                    ast::ExceptSelectItem {
-                        first_element: ast::Ident::new(x),
+                    expr::ExceptSelectItem {
+                        first_element: make_ident(x),
                         additional_elements: vec![],
                     }
                 } else {
@@ -202,9 +207,9 @@ impl PlanResolver<'_> {
                     let expression = self
                         .resolve_expression(*elem.expression, schema, state)
                         .await?;
-                    let item = ast::ReplaceSelectElement {
-                        expr: expr_to_sql(&expression)?,
-                        column_name: ast::Ident::new(elem.column_name),
+                    let item = expr::ReplaceSelectElement {
+                        expr: expression.to_string(),
+                        column_name: make_ident(elem.column_name),
                         as_keyword: elem.as_keyword,
                     };
                     items.push(item);
@@ -220,26 +225,30 @@ impl PlanResolver<'_> {
         let rename = wildcard_options
             .rename_columns
             .map(|x| {
-                let exclude = if x.len() > 1 {
-                    ast::RenameSelectItem::Multiple(
+                let rename = if x.len() > 1 {
+                    expr::RenameSelectItem::Multiple(
                         x.into_iter()
-                            .map(|x| ast::IdentWithAlias {
-                                ident: ast::Ident::new(x.identifier),
-                                alias: ast::Ident::new(x.alias),
+                            .map(|x| {
+                                format!(
+                                    "{} AS {}",
+                                    String::from(x.identifier),
+                                    String::from(x.alias)
+                                )
                             })
                             .collect(),
                     )
                 } else if let Some(x) = x.into_iter().next() {
-                    ast::RenameSelectItem::Single(ast::IdentWithAlias {
-                        ident: ast::Ident::new(x.identifier),
-                        alias: ast::Ident::new(x.alias),
-                    })
+                    expr::RenameSelectItem::Single(format!(
+                        "{} AS {}",
+                        String::from(x.identifier),
+                        String::from(x.alias)
+                    ))
                 } else {
                     return Err(PlanError::invalid(
                         "exclude columns must have at least one column",
                     ));
                 };
-                Ok(exclude)
+                Ok(rename)
             })
             .transpose()?;
         Ok(expr::WildcardOptions {