chore: upgrade to datafusion 40 (#2661)

Cargo.toml

@@ -45,14 +45,14 @@ object_store = { version = "0.10.1" }
 parquet = { version = "52" }
 
 # datafusion
-datafusion = { version = "39" }
-datafusion-expr = { version = "39" }
-datafusion-common = { version = "39" }
-datafusion-proto = { version = "39" }
-datafusion-sql = { version = "39" }
-datafusion-physical-expr = { version = "39" }
-datafusion-functions = { version = "39" }
-datafusion-functions-array = { version = "39" }
+datafusion = { version = "40" }
+datafusion-expr = { version = "40" }
+datafusion-common = { version = "40" }
+datafusion-proto = { version = "40" }
+datafusion-sql = { version = "40" }
+datafusion-physical-expr = { version = "40" }
+datafusion-functions = { version = "40" }
+datafusion-functions-array = { version = "40" }
 
 # serde
 serde = { version = "1.0.194", features = ["derive"] }

crates/core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltalake-core"
-version = "0.18.1"
+version = "0.18.2"
 authors.workspace = true
 keywords.workspace = true
 readme.workspace = true

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

@@ -26,6 +26,10 @@ impl DisplayAs for DeltaCdfScan {
 }
 
 impl ExecutionPlan for DeltaCdfScan {
+ fn name(&self) -> &str {
+ Self::static_name()
+ }
+
  fn as_any(&self) -> &dyn Any {
  self
  }

crates/core/src/delta_datafusion/expr.rs

@@ -29,6 +29,7 @@ use std::{
 use arrow_schema::DataType;
 use chrono::{DateTime, NaiveDate};
 use datafusion::execution::context::SessionState;
+use datafusion::execution::FunctionRegistry;
 use datafusion_common::Result as DFResult;
 use datafusion_common::{config::ConfigOptions, DFSchema, Result, ScalarValue, TableReference};
 use datafusion_expr::{
@@ -107,9 +108,15 @@ pub(crate) fn parse_predicate_expression(
  })?;
 
  let context_provider = DeltaContextProvider { state: df_state };
- let sql_to_rel =
+ let mut sql_to_rel =
  SqlToRel::new_with_options(&context_provider, DeltaParserOptions::default().into());
 
+ // NOTE: This can be probably removed with Datafusion 41 once
+ // <https://github.com/apache/datafusion/pull/11485> is released
+ for planner in context_provider.state.expr_planners() {
+ sql_to_rel = sql_to_rel.with_user_defined_planner(planner.clone());
+ }
+
  Ok(sql_to_rel.sql_to_expr(sql, schema, &mut Default::default())?)
 }
 

crates/core/src/delta_datafusion/find_files/mod.rs

@@ -141,7 +141,7 @@ async fn scan_table_by_files(
 
  // Identify which columns we need to project
  let mut used_columns = expression
- .to_columns()?
+ .column_refs()
  .into_iter()
  .map(|column| logical_schema.index_of(&column.name))
  .collect::<std::result::Result<Vec<usize>, ArrowError>>()?;

crates/core/src/delta_datafusion/find_files/physical.rs

@@ -85,6 +85,10 @@ impl DisplayAs for FindFilesExec {
 }
 
 impl ExecutionPlan for FindFilesExec {
+ fn name(&self) -> &str {
+ Self::static_name()
+ }
+
  fn as_any(&self) -> &dyn Any {
  self
  }

crates/core/src/delta_datafusion/mod.rs

@@ -61,7 +61,7 @@ use datafusion_common::scalar::ScalarValue;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion, TreeNodeVisitor};
 use datafusion_common::{
  config::ConfigOptions, Column, DFSchema, DataFusionError, Result as DataFusionResult,
- ToDFSchema,
+ TableReference, ToDFSchema,
 };
 use datafusion_expr::logical_plan::CreateExternalTable;
 use datafusion_expr::utils::conjunction;
@@ -845,6 +845,10 @@ impl DisplayAs for DeltaScan {
 }
 
 impl ExecutionPlan for DeltaScan {
+ fn name(&self) -> &str {
+ Self::static_name()
+ }
+
  fn as_any(&self) -> &dyn Any {
  self
  }
@@ -1316,6 +1320,7 @@ impl LogicalExtensionCodec for DeltaLogicalCodec {
  fn try_decode_table_provider(
  &self,
  buf: &[u8],
+ _table_ref: &TableReference,
  _schema: SchemaRef,
  _ctx: &SessionContext,
  ) -> Result<Arc<dyn TableProvider>, DataFusionError> {
@@ -1326,6 +1331,7 @@ impl LogicalExtensionCodec for DeltaLogicalCodec {
 
  fn try_encode_table_provider(
  &self,
+ _table_ref: &TableReference,
  node: Arc<dyn TableProvider>,
  buf: &mut Vec<u8>,
  ) -> Result<(), DataFusionError> {
@@ -1506,7 +1512,7 @@ pub(crate) async fn find_files_scan<'a>(
 
  // Identify which columns we need to project
  let mut used_columns = expression
- .to_columns()?
+ .column_refs()
  .into_iter()
  .map(|column| logical_schema.index_of(&column.name))
  .collect::<Result<Vec<usize>, ArrowError>>()?;
@@ -1756,9 +1762,8 @@ mod tests {
  use datafusion::assert_batches_sorted_eq;
  use datafusion::datasource::physical_plan::ParquetExec;
  use datafusion::physical_plan::empty::EmptyExec;
- use datafusion::physical_plan::{visit_execution_plan, ExecutionPlanVisitor};
+ use datafusion::physical_plan::{visit_execution_plan, ExecutionPlanVisitor, PhysicalExpr};
  use datafusion_expr::lit;
- use datafusion_physical_expr::PhysicalExpr;
  use datafusion_proto::physical_plan::AsExecutionPlan;
  use datafusion_proto::protobuf;
  use object_store::path::Path;

crates/core/src/delta_datafusion/physical.rs

@@ -74,6 +74,10 @@ impl DisplayAs for MetricObserverExec {
 }
 
 impl ExecutionPlan for MetricObserverExec {
+ fn name(&self) -> &str {
+ Self::static_name()
+ }
+
  fn as_any(&self) -> &dyn std::any::Any {
  self
  }

crates/core/src/operations/cdc.rs

@@ -196,6 +196,10 @@ impl DisplayAs for CDCObserver {
 }
 
 impl ExecutionPlan for CDCObserver {
+ fn name(&self) -> &str {
+ Self::static_name()
+ }
+
  fn as_any(&self) -> &dyn std::any::Any {
  self
  }

crates/core/src/operations/merge/barrier.rs

@@ -67,6 +67,10 @@ impl MergeBarrierExec {
 }
 
 impl ExecutionPlan for MergeBarrierExec {
+ fn name(&self) -> &str {
+ Self::static_name()
+ }
+
  fn as_any(&self) -> &dyn std::any::Any {
  self
  }

crates/deltalake/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltalake"
-version = "0.18.1"
+version = "0.18.2"
 authors.workspace = true
 keywords.workspace = true
 readme.workspace = true

crates/sql/src/planner.rs

@@ -42,6 +42,7 @@ impl<'a, S: ContextProvider> DeltaSqlToRel<'a, S> {
  ParserOptions {
  parse_float_as_decimal: self.options.parse_float_as_decimal,
  enable_ident_normalization: self.options.enable_ident_normalization,
+ support_varchar_with_length: false,
  },
  );
  planner.statement_to_plan(s)