Adapt planner to DataFusion 51 APIs

Author: wi11dey

Cargo.lock

@@ -60,10 +60,10 @@ delta_kernel = { version = "0.19.0" }
 cornucopia = { version = "0.9.0" }
 cornucopia_async = {version = "0.6.0"}
 deadpool-postgres = "0.14"
-prost = { version = "0.13", features = ["no-recursion-limit"] }
-prost-reflect = "0.14.0"
-prost-build = { version = "0.13" }
-prost-types = "0.13"
+prost = { version = "0.14", features = ["no-recursion-limit"] }
+prost-reflect = "0.16.0"
+prost-build = { version = "0.14" }
+prost-types = "0.14"
 aws-config = "=1.8.0"
 reqwest = "0.12"
 rusqlite = "0.31"

Cargo.toml

@@ -13,7 +13,6 @@ use datafusion::common::{
     DFSchema, DFSchemaRef, DataFusionError, Result, ScalarValue, Spans, TableReference, plan_err,
 };
 use datafusion::execution::context::SessionState;
-use datafusion::execution::runtime_env::RuntimeEnvBuilder;
 use datafusion::functions::datetime::date_bin;
 use datafusion::logical_expr::{Expr, Extension, LogicalPlan, UserDefinedLogicalNode};
 use datafusion::physical_expr::PhysicalExpr;
@@ -161,11 +160,8 @@ impl<'a> Planner<'a> {
             .ok_or_else(|| DataFusionError::Plan("missing input".to_string()))?;
 
         // need to convert to ExecutionPlan to get the partial schema.
-        let partial_aggregation_exec_plan = partial_aggregation_plan.try_into_physical_plan(
-            self.schema_provider,
-            &RuntimeEnvBuilder::new().build().unwrap(),
-            &codec,
-        )?;
+        let partial_aggregation_exec_plan =
+            partial_aggregation_plan.try_into_physical_plan(self.schema_provider, &codec)?;
 
         let partial_schema = partial_aggregation_exec_plan.schema();
         let final_input_table_provider =
@@ -244,7 +240,7 @@ impl ExtensionPlanner for ArroyoExtensionPlanner {
         physical_inputs: &[Arc<dyn ExecutionPlan>],
         _session_state: &SessionState,
     ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
-        let schema = node.schema().as_ref().into();
+        let schema = node.schema().as_arrow().clone();
         if let Ok::<&dyn ArroyoExtension, _>(arroyo_extension) = node.try_into()
             && arroyo_extension.transparent()
         {