refactor(planner): move DDL compiler into streaming_planner

Relocate ast_utils and ddl_compiler from the coordinator plan module
into streaming_planner (common and schema/introspection). Add
try_compile_connector_create_table so coordinator logical plan
visitors compile connector CREATE TABLE statements via the planner
crate without duplicating AST routing logic.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: luoluoyuyu

src/coordinator/src/plan/logical_plan_visitor.rs

@@ -11,7 +11,6 @@
 // limitations under the License.
 
 use datafusion::sql::sqlparser::ast::{ObjectType, Statement as DFStatement};
-use sqlparser::ast::Statement;
 use tracing::debug;
 
 use crate::coordinator::analyze::analysis::Analysis;
@@ -28,9 +27,8 @@ use crate::coordinator::statement::{
     StatementVisitorResult, StopFunction, StreamingTableStatement,
 };
 use crate::sql::analysis::StreamSchemaProvider;
+use crate::sql::schema::try_compile_connector_create_table;
 
-use super::ast_utils::AstUtils;
-use super::ddl_compiler::DdlCompiler;
 use super::streaming_compiler::StreamingCompiler;
 
 #[derive(Clone)]
@@ -135,15 +133,12 @@ impl StatementVisitor for LogicalPlanVisitor {
         stmt: &CreateTable,
         _ctx: &StatementVisitorContext,
     ) -> StatementVisitorResult {
-        if let Statement::CreateTable(ast_node) = &stmt.statement
-            && ast_node.query.is_none()
-            && AstUtils::contains_connector_property(&ast_node.with_options)
+        if let Some(result) =
+            try_compile_connector_create_table(&self.schema_provider, &stmt.statement)
         {
-            let declared_role = AstUtils::peek_table_role(&ast_node.with_options);
-            let compiler = DdlCompiler::new(&self.schema_provider);
-            return match compiler.compile(ast_node, declared_role.as_deref()) {
-                Ok(external_table) => StatementVisitorResult::Plan(Box::new(
-                    CreateTablePlan::external_table(external_table, ast_node.if_not_exists),
+            return match result {
+                Ok((external_table, if_not_exists)) => StatementVisitorResult::Plan(Box::new(
+                    CreateTablePlan::external_table(external_table, if_not_exists),
                 )),
                 Err(err) => StatementVisitorResult::Plan(Box::new(CompileErrorPlan::new(format!(
                     "Ingest table resolution failed - {err:#}"

src/coordinator/src/plan/mod.rs

@@ -10,12 +10,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod ast_utils;
 mod compile_error_plan;
 mod create_function_plan;
 mod create_python_function_plan;
 mod create_table_plan;
-mod ddl_compiler;
 mod drop_function_plan;
 mod drop_streaming_table_plan;
 mod drop_table_plan;

src/coordinator/src/plan/ast_utils.rs …treaming_planner/src/common/ast_utils.rssrc/coordinator/src/plan/ast_utils.rs renamed to src/streaming_planner/src/common/ast_utils.rs src/coordinator/src/plan/ast_utils.rs renamed to src/streaming_planner/src/common/ast_utils.rs

@@ -20,7 +20,7 @@
 use datafusion::common::{Result, plan_err};
 use datafusion::sql::sqlparser::ast::{Expr as SqlExpr, SqlOption, TableConstraint};
 
-use crate::sql::common::with_option_keys as opt;
+use crate::common::with_option_keys as opt;
 
 /// Namespace for AST extraction helpers.
 pub struct AstUtils;

src/streaming_planner/src/common/mod.rs

@@ -16,6 +16,7 @@
 //! analogous to `arroyo-types` + `arroyo-rpc` in Arroyo.
 
 pub mod arrow_ext;
+pub mod ast_utils;
 pub mod connector_options;
 pub mod constants;
 pub mod control;
@@ -38,6 +39,7 @@ pub use function_stream_runtime_common::streaming_protocol::{CheckpointBarrier,
 pub use time_utils::{from_nanos, to_micros, to_millis, to_nanos};
 
 // ── Re-exports from new modules ──
+pub use ast_utils::AstUtils;
 pub use connector_options::ConnectorOptions;
 pub use formats::{BadData, Format, Framing, JsonCompression, JsonFormat};
 pub use fs_schema::{FsSchema, FsSchemaRef};

src/coordinator/src/plan/ddl_compiler.rs …src/schema/introspection/ddl_compiler.rssrc/coordinator/src/plan/ddl_compiler.rs renamed to src/streaming_planner/src/schema/introspection/ddl_compiler.rs src/coordinator/src/plan/ddl_compiler.rs renamed to src/streaming_planner/src/schema/introspection/ddl_compiler.rs

@@ -22,21 +22,21 @@ use datafusion::sql::TableReference;
 use datafusion::sql::planner::{PlannerContext, SqlToRel};
 use datafusion::sql::sqlparser::ast;
 use datafusion::sql::sqlparser::ast::CreateTable as SqlCreateTable;
+use datafusion::sql::sqlparser::ast::Statement as DfStatement;
 use datafusion_expr::ExprSchemable;
 use tracing::warn;
 
-use super::ast_utils::AstUtils;
-use crate::coordinator::tool::ConnectorOptions;
-use crate::sql::analysis::StreamSchemaProvider;
-use crate::sql::common::constants::{connection_table_role, connector_type, sql_field};
-use crate::sql::common::with_option_keys as opt;
-use crate::sql::common::{BadData, Format, Framing, JsonCompression, JsonFormat};
-use crate::sql::connector::registry::REGISTRY;
-use crate::sql::schema::ColumnDescriptor;
-use crate::sql::schema::catalog::{ExternalTable, LookupTable, SourceTable};
-use crate::sql::schema::data_encoding_format::DataEncodingFormat;
-use crate::sql::schema::table_role::{apply_adapter_specific_rules, validate_adapter_availability};
-use crate::sql::schema::temporal_pipeline_config::TemporalPipelineConfig;
+use crate::common::ast_utils::AstUtils;
+use crate::common::constants::{connection_table_role, connector_type, sql_field};
+use crate::common::with_option_keys as opt;
+use crate::common::{BadData, ConnectorOptions, Format, Framing, JsonCompression, JsonFormat};
+use crate::connector::registry::REGISTRY;
+use crate::schema::ColumnDescriptor;
+use crate::schema::catalog::{ExternalTable, LookupTable, SourceTable};
+use crate::schema::data_encoding_format::DataEncodingFormat;
+use crate::schema::schema_provider::StreamSchemaProvider;
+use crate::schema::table_role::{apply_adapter_specific_rules, validate_adapter_availability};
+use crate::schema::temporal_pipeline_config::TemporalPipelineConfig;
 
 pub struct DdlCompiler<'a> {
     schema_provider: &'a StreamSchemaProvider,
@@ -293,6 +293,30 @@ impl<'a> DdlCompiler<'a> {
     }
 }
 
+/// If `stmt` is `CREATE TABLE ... WITH (connector = ...)`, compile it to [`ExternalTable`].
+pub fn try_compile_connector_create_table(
+    schema_provider: &StreamSchemaProvider,
+    stmt: &DfStatement,
+) -> Option<Result<(ExternalTable, bool)>> {
+    let DfStatement::CreateTable(ast_node) = stmt else {
+        return None;
+    };
+    if ast_node.query.is_some() {
+        return None;
+    }
+    if !AstUtils::contains_connector_property(&ast_node.with_options) {
+        return None;
+    }
+    let declared_role = AstUtils::peek_table_role(&ast_node.with_options);
+    let if_not_exists = ast_node.if_not_exists;
+    let compiler = DdlCompiler::new(schema_provider);
+    Some(
+        compiler
+            .compile(ast_node, declared_role.as_deref())
+            .map(|table| (table, if_not_exists)),
+    )
+}
+
 fn resolve_source_watermark(
     table_identifier: &str,
     columns: &mut Vec<ColumnDescriptor>,

src/streaming_planner/src/schema/introspection/mod.rs

@@ -10,10 +10,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod ddl_compiler;
 pub mod ddl_formatter;
 pub mod show_formatter;
 pub mod stream_formatter;
 
+pub use ddl_compiler::{DdlCompiler, try_compile_connector_create_table};
 #[allow(unused_imports)]
 pub use ddl_formatter::{DdlBuilder, format_data_type, schema_columns_one_line};
 pub use show_formatter::{catalog_table_row_detail, show_create_catalog_table};

src/streaming_planner/src/schema/mod.rs

@@ -24,7 +24,8 @@ pub mod utils;
 pub use catalog::{ExternalTable, LookupTable, SinkTable, SourceTable};
 pub use column_descriptor::ColumnDescriptor;
 pub use introspection::{
-    catalog_table_row_detail, schema_columns_one_line, show_create_catalog_table,
+    catalog_table_row_detail, schema_columns_one_line, show_create_catalog_table, DdlCompiler,
+    try_compile_connector_create_table,
 };
 pub use schema_provider::{ObjectName, StreamPlanningContext, StreamSchemaProvider, StreamTable};
 pub use table::CatalogEntity;