update

Author: luoluoyuyu

Cargo.lock

@@ -38,9 +38,6 @@ protocol = { path = "./protocol" }
 prost = "0.13"
 rdkafka = { version = "0.38", features = ["cmake-build", "ssl", "gssapi"] }
 crossbeam-channel = "0.5"
-pest = "2.7"
-pest_derive = "2.7"
-clap = { version = "4.5", features = ["derive"] }
 wasmtime = { version = "41.0.3", features = ["component-model", "async"] }
 base64 = "0.22"
 wasmtime-wasi = "41.0.3"
@@ -63,7 +60,6 @@ petgraph = "0.7"
 rand = { version = "0.8", features = ["small_rng"] }
 itertools = "0.14"
 strum = { version = "0.26", features = ["derive"] }
-datafusion-functions-aggregate = {git = 'https://github.com/ArroyoSystems/arrow-datafusion', branch = '48.0.1/arroyo'}
 
 typify = { git = 'https://github.com/ArroyoSystems/typify.git', branch = 'arroyo' }
 parquet = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '55.2.0/parquet'}

Cargo.toml

@@ -284,8 +284,9 @@ impl PlanVisitor for Executor {
             let fs_program: FsProgram = plan.program.clone().into();
             let job_manager: Arc<JobManager> = Arc::clone(&self.job_manager);
 
+            let job_id = plan.name.clone();
             let job_id = tokio::task::block_in_place(|| {
-                tokio::runtime::Handle::current().block_on(job_manager.submit_job(fs_program))
+                tokio::runtime::Handle::current().block_on(job_manager.submit_job(job_id, fs_program))
             })
             .map_err(|e| ExecuteError::Internal(format!("Failed to submit streaming job: {e}")))?;
 

src/coordinator/execution/executor.rs

@@ -155,7 +155,6 @@ impl LogicalPlanVisitor {
         Ok(StreamingTable {
             name: sink_table_name,
             comment: comment.clone(),
-            source_table: sink_definition,
             program: validated_program,
         })
     }

src/coordinator/plan/logical_plan_visitor.rs

@@ -12,14 +12,12 @@
 
 use super::{PlanNode, PlanVisitor, PlanVisitorContext, PlanVisitorResult};
 use crate::sql::logical_node::logical::LogicalProgram;
-use crate::sql::schema::source_table::SourceTable;
 
 /// Plan node representing a fully resolved streaming table (DDL).
 #[derive(Debug)]
 pub struct StreamingTable {
     pub name: String,
     pub comment: Option<String>,
-    pub source_table: SourceTable,
     pub program: LogicalProgram,
 }
 

src/coordinator/plan/streaming_table_plan.rs

@@ -15,6 +15,7 @@
 pub mod buffer_and_event;
 pub mod common;
 pub mod streaming;
+pub mod util;
 pub mod task;
 pub mod taskexecutor;
 pub mod wasm;

src/runtime/mod.rs

@@ -17,5 +17,5 @@ pub mod operator;
 pub mod source;
 
 pub use context::TaskContext;
-pub use operator::{ConstructedOperator, MessageOperator, Registry};
+pub use operator::{ConstructedOperator, MessageOperator};
 pub use source::{SourceEvent, SourceOffset, SourceOperator};

src/runtime/streaming/api/mod.rs

@@ -15,68 +15,9 @@ use crate::runtime::streaming::api::source::SourceOperator;
 use crate::runtime::streaming::protocol::stream_out::StreamOutput;
 use arrow_array::RecordBatch;
 use async_trait::async_trait;
-use datafusion::common::Result as DfResult;
-use datafusion::execution::context::SessionContext;
-use datafusion::execution::FunctionRegistry;
-use datafusion::logical_expr::{AggregateUDF, ScalarUDF, WindowUDF};
-use datafusion::logical_expr::planner::ExprPlanner;
-use std::collections::HashSet;
-use std::sync::Arc;
 use std::time::Duration;
 use crate::sql::common::{CheckpointBarrier, Watermark};
 
-// ---------------------------------------------------------------------------
-// Registry — 算子 / UDF 注册表（取代 tracing_subscriber::Registry）
-// ---------------------------------------------------------------------------
-
-/// 运行时函数与状态注册表。
-///
-/// 包装 DataFusion [`SessionContext`]，为物理计划反序列化提供 UDF / UDAF / UDWF 查询能力。
-/// `Arc<Registry>` 在工厂中创建后，由各构造器共享。
-pub struct Registry {
-    ctx: SessionContext,
-}
-
-impl Registry {
-    pub fn new() -> Self {
-        Self {
-            ctx: SessionContext::new(),
-        }
-    }
-
-    pub fn session_context(&self) -> &SessionContext {
-        &self.ctx
-    }
-}
-
-impl Default for Registry {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl FunctionRegistry for Registry {
-    fn udfs(&self) -> HashSet<String> {
-        self.ctx.udfs()
-    }
-
-    fn udf(&self, name: &str) -> DfResult<Arc<ScalarUDF>> {
-        self.ctx.udf(name)
-    }
-
-    fn udaf(&self, name: &str) -> DfResult<Arc<AggregateUDF>> {
-        self.ctx.udaf(name)
-    }
-
-    fn udwf(&self, name: &str) -> DfResult<Arc<WindowUDF>> {
-        self.ctx.udwf(name)
-    }
-
-    fn expr_planners(&self) -> Vec<Arc<dyn ExprPlanner>> {
-        self.ctx.expr_planners()
-    }
-}
-
 // ---------------------------------------------------------------------------
 // ConstructedOperator
 // ---------------------------------------------------------------------------

src/runtime/streaming/api/operator.rs

@@ -0,0 +1,58 @@
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Source / Sink 连接器协议：按 [`ConnectorOp::connector`] 分发到具体实现。
+
+use anyhow::{anyhow, Result};
+use prost::Message;
+use std::sync::Arc;
+
+use protocol::grpc::api::ConnectorOp;
+
+use crate::runtime::streaming::api::operator::ConstructedOperator;
+use crate::runtime::streaming::factory::global::Registry;
+use crate::runtime::streaming::factory::operator_constructor::OperatorConstructor;
+use crate::sql::common::constants::connector_type;
+
+use super::kafka::{KafkaSinkDispatcher, KafkaSourceDispatcher};
+
+pub struct ConnectorSourceDispatcher;
+
+impl OperatorConstructor for ConnectorSourceDispatcher {
+    fn with_config(&self, config: &[u8], registry: Arc<Registry>) -> Result<ConstructedOperator> {
+        let op = ConnectorOp::decode(config)
+            .map_err(|e| anyhow!("decode ConnectorOp (source): {e}"))?;
+
+        match op.connector.as_str() {
+            ct if ct == connector_type::KAFKA => KafkaSourceDispatcher.with_config(config, registry),
+            ct if ct == connector_type::REDIS => Err(anyhow!(
+                "ConnectorSource '{}' factory wiring not yet implemented",
+                op.connector
+            )),
+            other => Err(anyhow!("Unsupported source connector type: {}", other)),
+        }
+    }
+}
+
+pub struct ConnectorSinkDispatcher;
+
+impl OperatorConstructor for ConnectorSinkDispatcher {
+    fn with_config(&self, config: &[u8], registry: Arc<Registry>) -> Result<ConstructedOperator> {
+        let op = ConnectorOp::decode(config)
+            .map_err(|e| anyhow!("decode ConnectorOp (sink): {e}"))?;
+
+        match op.connector.as_str() {
+            ct if ct == connector_type::KAFKA => KafkaSinkDispatcher.with_config(config, registry),
+            other => Err(anyhow!("Unsupported sink connector type: {}", other)),
+        }
+    }
+}

src/runtime/streaming/factory/connector/dispatchers.rs

@@ -21,9 +21,10 @@ use std::sync::Arc;
 use protocol::grpc::api::ConnectorOp;
 use tracing::{info, warn};
 
-use super::OperatorConstructor;
-use crate::runtime::streaming::api::operator::{ConstructedOperator, Registry};
+use crate::runtime::streaming::api::operator::ConstructedOperator;
 use crate::runtime::streaming::api::source::SourceOffset;
+use crate::runtime::streaming::factory::global::Registry;
+use crate::runtime::streaming::factory::operator_constructor::OperatorConstructor;
 use crate::runtime::streaming::format::{
     BadDataPolicy, DataSerializer, DecimalEncoding as RtDecimalEncoding, Format as RuntimeFormat,
     JsonFormat as RuntimeJsonFormat, TimestampFormat as RtTimestampFormat,
@@ -332,10 +333,3 @@ impl OperatorConstructor for KafkaSinkDispatcher {
         Ok(ConstructedOperator::Operator(Box::new(sink_op)))
     }
 }
-
-/// 注册 `KafkaSource` / `KafkaSink` 构造器（由 [`super::OperatorFactory::register_builtins`] 调用）。
-pub fn register_kafka_plugins(factory: &mut super::OperatorFactory) {
-    factory.register("KafkaSource", Box::new(KafkaSourceDispatcher));
-    factory.register("KafkaSink", Box::new(KafkaSinkDispatcher));
-    info!("Registered Kafka connector plugins (KafkaSource, KafkaSink)");
-}