chore: upgrade datafusion to 44 (#103)

Author: xhwhis

Cargo.toml

@@ -20,9 +20,9 @@ arrow-flight = { version = "53.3", features = ["flight-sql-experimental"] }
 arrow-json = "53.3"
 async-stream = "0.3.5"
 async-trait = "0.1.83"
-datafusion = "43.0.0"
+datafusion = "44.0.0"
 datafusion-federation = { path = "./datafusion-federation", version = "0.3.3" }
-datafusion-substrait = "43.0.0"
+datafusion-substrait = "44.0.0"
 futures = "0.3.31"
 tokio = { version = "1.41", features = ["full"] }
 tonic = { version = "0.12", features = [

datafusion-federation/src/schema_cast/mod.rs

@@ -28,11 +28,13 @@ pub struct SchemaCastScanExec {
 impl SchemaCastScanExec {
     pub fn new(input: Arc<dyn ExecutionPlan>, schema: SchemaRef) -> Self {
         let eq_properties = input.equivalence_properties().clone();
-        let execution_mode = input.execution_mode();
+        let emission_type = input.pipeline_behavior();
+        let boundedness = input.boundedness();
         let properties = PlanProperties::new(
             eq_properties,
             input.output_partitioning().clone(),
-            execution_mode,
+            emission_type,
+            boundedness,
         );
         Self {
             input,

datafusion-federation/src/sql/mod.rs

@@ -20,7 +20,8 @@ use datafusion::{
     optimizer::{optimizer::Optimizer, OptimizerConfig, OptimizerRule},
     physical_expr::EquivalenceProperties,
     physical_plan::{
-        DisplayAs, DisplayFormatType, ExecutionMode, ExecutionPlan, Partitioning, PlanProperties,
+        execution_plan::{Boundedness, EmissionType},
+        DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties,
         SendableRecordBatchStream,
     },
     sql::{
@@ -579,15 +580,18 @@ fn rewrite_table_scans_in_expr(
                         })
                     })
                     .transpose()?,
-                ..options
+                ..*options
             };
             if let Some(rewrite) = qualifier.as_ref().and_then(|q| known_rewrites.get(q)) {
                 Ok(Expr::Wildcard {
                     qualifier: Some(rewrite.clone()),
-                    options,
+                    options: Box::new(options),
                 })
             } else {
-                Ok(Expr::Wildcard { qualifier, options })
+                Ok(Expr::Wildcard {
+                    qualifier,
+                    options: Box::new(options),
+                })
             }
         }
         Expr::GroupingSet(gs) => match gs {
@@ -676,7 +680,8 @@ impl VirtualExecutionPlan {
         let props = PlanProperties::new(
             EquivalenceProperties::new(Arc::new(schema)),
             Partitioning::UnknownPartitioning(1),
-            ExecutionMode::Bounded,
+            EmissionType::Incremental,
+            Boundedness::Bounded,
         );
         Self {
             plan,
@@ -952,7 +957,7 @@ mod tests {
             // different tables in single aggregation expression
             (
                 "SELECT COUNT(CASE WHEN appt.a > 0 THEN appt.a ELSE dft.a END) FROM app_table as appt, foo.df_table as dft",
-                "SELECT count(CASE WHEN (appt.a > 0) THEN appt.a ELSE dft.a END) FROM remote_table AS appt JOIN remote_table AS dft"
+                "SELECT count(CASE WHEN (appt.a > 0) THEN appt.a ELSE dft.a END) FROM remote_table AS appt CROSS JOIN remote_table AS dft"
             ),
         ];
 

datafusion-flight-sql-server/src/service.rs

@@ -1002,7 +1002,7 @@ async fn parse_substrait_bytes(
         .await
         .map_err(df_error_to_status)?;
 
-    from_substrait_plan(&ctx.inner, &substrait_plan)
+    from_substrait_plan(&ctx.inner.state(), &substrait_plan)
         .await
         .map_err(df_error_to_status)
 }