Upgrade Arroyo SQL parser to 0.59

Author: wi11dey

Cargo.lock

@@ -71,7 +71,7 @@ reqwest = "0.12"
 rusqlite = "0.31"
 apache-avro = "0.18.0"
 k8s-openapi = "0.24.0"
-sqlparser = "0.55.0"
+sqlparser = "0.59.0"
 petgraph = "0.8"
 rand = "0.9"
 prometheus = "0.14.0"
@@ -119,7 +119,7 @@ split-debuginfo = "unpacked"
 [patch.crates-io]
 typify = { git = 'https://github.com/ArroyoSystems/typify.git', branch = 'arroyo' }
 
-sqlparser = { git = "https://github.com/ArroyoSystems/sqlparser-rs", branch = "0.55.0/arroyo" }
+sqlparser = { git = "https://github.com/wi11dey/sqlparser-rs", branch = "0.59.0/arroyo" }
 
 cornucopia_async = { git = "https://github.com/ArroyoSystems/cornucopia", branch = "sqlite" }
 cornucopia = { git = "https://github.com/ArroyoSystems/cornucopia", branch = "sqlite" }

Cargo.toml

@@ -160,8 +160,9 @@ impl<'a> Planner<'a> {
             .ok_or_else(|| DataFusionError::Plan("missing input".to_string()))?;
 
         // need to convert to ExecutionPlan to get the partial schema.
+        let task_ctx = self.session_state.task_ctx();
         let partial_aggregation_exec_plan =
-            partial_aggregation_plan.try_into_physical_plan(self.schema_provider, &codec)?;
+            partial_aggregation_plan.try_into_physical_plan(task_ctx.as_ref(), &codec)?;
 
         let partial_schema = partial_aggregation_exec_plan.schema();
         let final_input_table_provider =

crates/arroyo-planner/src/builder.rs

@@ -177,9 +177,11 @@ impl ArroyoExtension for KeyCalculationExtension {
         let arrow_schema = self.input.schema().as_ref();
 
         match &self.keys {
-            KeysOrExprs::Keys(keys) => {
-                ArroyoSchema::from_schema_keys(Arc::new(arrow_schema.into()), keys.clone()).unwrap()
-            }
+            KeysOrExprs::Keys(keys) => ArroyoSchema::from_schema_keys(
+                Arc::new(arrow_schema.as_arrow().clone()),
+                keys.clone(),
+            )
+            .unwrap(),
             KeysOrExprs::Exprs(exprs) => {
                 let mut fields = vec![];
 

crates/arroyo-planner/src/extension/key_calculation.rs

@@ -48,7 +48,7 @@ use datafusion::optimizer::replace_distinct_aggregate::ReplaceDistinctWithAggreg
 use datafusion::optimizer::scalar_subquery_to_join::ScalarSubqueryToJoin;
 use datafusion::optimizer::simplify_expressions::SimplifyExpressions;
 use datafusion::sql::sqlparser;
-use datafusion::sql::sqlparser::ast::{CreateTable, Query};
+use datafusion::sql::sqlparser::ast::{CreateTable, CreateTableOptions, Query};
 use datafusion::{
     optimizer::{OptimizerContext, optimizer::Optimizer},
     sql::{
@@ -773,14 +773,19 @@ impl Table {
         if let Statement::CreateTable(CreateTable {
             name,
             columns,
-            with_options,
+            table_options,
             query: None,
             temporary,
             constraints,
             arroyo_partitions,
             ..
         }) = statement
         {
+            let with_options = match table_options {
+                CreateTableOptions::None => &[],
+                CreateTableOptions::With(options) => options.as_slice(),
+                options => return plan_err!("Unsupported create table options: {options}"),
+            };
             let name: String = name.to_string();
             let mut connector_opts = ConnectorOptions::new(with_options, arroyo_partitions)?;