update

Author: luoluoyuyu

src/runtime/streaming/api/context.rs

@@ -52,15 +52,12 @@ impl TaskContext {
     }
 
     // ========================================================================
-    // 水位线与时间流管理 API
     // ========================================================================
 
-    /// 供业务算子调用：获取当前任务的安全水位线
     pub fn last_present_watermark(&self) -> Option<std::time::SystemTime> {
         self.current_watermark
     }
 
-    /// 供底座框架 (SubtaskRunner) 调用：推进本地时间，保证单调递增
     pub fn advance_watermark(&mut self, watermark: std::time::SystemTime) {
         if let Some(current) = self.current_watermark {
             if watermark > current {
@@ -72,10 +69,8 @@ impl TaskContext {
     }
 
     // ========================================================================
-    // 可观测性 API (Observability)
     // ========================================================================
 
-    /// 格式化当前 Task 的唯一标识，用于分布式追踪和日志打印
     pub fn task_identity(&self) -> String {
         format!(
             "Job[{}], Vertex[{}], Subtask[{}/{}]",
@@ -84,10 +79,8 @@ impl TaskContext {
     }
 
     // ========================================================================
-    // 背压网络发送 API
     // ========================================================================
 
-    /// 受内存池管控的数据发送：申请精准字节的内存船票后广播到所有下游
     pub async fn collect(&self, batch: RecordBatch) -> anyhow::Result<()> {
         if self.outboxes.is_empty() {
             return Ok(());
@@ -103,7 +96,6 @@ impl TaskContext {
         Ok(())
     }
 
-    /// 按 Key 哈希路由到单分区（用于 Shuffle / KeyBy）
     pub async fn collect_keyed(
         &self,
         key_hash: u64,
@@ -122,7 +114,6 @@ impl TaskContext {
         Ok(())
     }
 
-    /// 广播控制信号（如 Watermark, Barrier：不申请内存船票，保证在拥堵时畅通无阻）
     pub async fn broadcast(&self, event: StreamEvent) -> anyhow::Result<()> {
         let tracked_event = TrackedEvent::control(event);
         for outbox in &self.outboxes {

src/runtime/streaming/api/mod.rs

@@ -10,7 +10,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! 接口层：算子与源实现需遵循的 trait 与运行时上下文。
 
 pub mod context;
 pub mod operator;

src/runtime/streaming/api/operator.rs

@@ -22,13 +22,11 @@ use crate::sql::common::{CheckpointBarrier, Watermark};
 // ConstructedOperator
 // ---------------------------------------------------------------------------
 
-/// 工厂反射产出的具体算子实例
 pub enum ConstructedOperator {
     Source(Box<dyn SourceOperator>),
     Operator(Box<dyn MessageOperator>),
 }
 
-/// 多上游、被动驱动的消息算子。
 #[async_trait]
 pub trait MessageOperator: Send + 'static {
     fn name(&self) -> &str;
@@ -37,7 +35,6 @@ pub trait MessageOperator: Send + 'static {
         Ok(())
     }
 
-    /// `input_idx`：多输入拓扑下第几条边（与 `SubtaskRunner` 的 inbox 下标一致；单输入恒为 0）。
     async fn process_data(
         &mut self,
         input_idx: usize,
@@ -57,7 +54,6 @@ pub trait MessageOperator: Send + 'static {
         ctx: &mut TaskContext,
     ) -> anyhow::Result<()>;
 
-    /// 全局 checkpoint 确认后由 `SubtaskRunner` 在 [`ControlCommand::Commit`] 上调用（如 Kafka EOS 二阶段提交）。
     async fn commit_checkpoint(
         &mut self,
         _epoch: u32,
@@ -66,12 +62,10 @@ pub trait MessageOperator: Send + 'static {
         Ok(())
     }
 
-    /// 周期性时钟（如 Idle 检测）；`None` 表示不注册 tick。
     fn tick_interval(&self) -> Option<Duration> {
         None
     }
 
-    /// 与 [`Self::tick_interval`] 配套，由 `SubtaskRunner` 按固定间隔调用。
     async fn process_tick(
         &mut self,
         _tick_index: u64,

src/runtime/streaming/api/source.rs

@@ -10,14 +10,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! 源算子：由 [`crate::runtime::streaming::execution::SourceRunner`] 驱动 `fetch_next`，不得在内部死循环阻塞控制面。
 
 use crate::runtime::streaming::api::context::TaskContext;
 use arrow_array::RecordBatch;
 use async_trait::async_trait;
 use crate::sql::common::{CheckpointBarrier, Watermark};
 
-/// Kafka 等外部源在 **无已存位点** 时的起始消费策略（与 `arroyo-connectors` 语义对齐）。
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 pub enum SourceOffset {
     Earliest,
@@ -30,7 +28,6 @@ pub enum SourceOffset {
 pub enum SourceEvent {
     Data(RecordBatch),
     Watermark(Watermark),
-    /// 无数据可读：必须由 Runner 调度退避，禁止在 `fetch_next` 内长时间阻塞。
     Idle,
     EndOfStream,
 }
@@ -43,10 +40,8 @@ pub trait SourceOperator: Send + 'static {
         Ok(())
     }
 
-    /// 核心拉取：无数据时必须返回 [`SourceEvent::Idle`]，严禁内部阻塞控制面。
     async fn fetch_next(&mut self, ctx: &mut TaskContext) -> anyhow::Result<SourceEvent>;
 
-    /// 独立于 `fetch_next` 的水位线脉搏（例如解决 Idle 时仍要推进水印）。
     fn poll_watermark(&mut self) -> Option<Watermark> {
         None
     }

src/runtime/streaming/error.rs

@@ -13,42 +13,33 @@
 use std::fmt::Display;
 use thiserror::Error;
 
-/// 流水线 / 子任务运行期间的错误定义。
 #[derive(Debug, Error)]
 pub enum RunError {
-    /// 算子内部业务逻辑抛出的错误
     #[error("Operator execution failed: {0:#}")]
     Operator(#[from] anyhow::Error),
 
-    /// 向下游 Task 发送数据/信号时通道阻塞或断开
     #[error("Downstream send failed: {0}")]
     DownstreamSend(String),
 
-    /// 引擎内部状态机错误或拓扑规划错误（如：DAG 为空、在链条中间发生 Shuffle）
     #[error("Internal engine error: {0}")]
     Internal(String),
 
-    /// Checkpoint 状态持久化或恢复时发生的错误
     #[error("State backend error: {0}")]
     State(String),
 
-    /// 底层网络或文件 I/O 错误
     #[error("I/O error: {0}")]
     Io(#[from] std::io::Error),
 }
 
 impl RunError {
-    /// 快捷构造器：引擎内部错误（常用于防御性编程和边界校验）
     pub fn internal<T: Display>(msg: T) -> Self {
         Self::Internal(msg.to_string())
     }
 
-    /// 快捷构造器：下游发送异常
     pub fn downstream<T: Display>(msg: T) -> Self {
         Self::DownstreamSend(msg.to_string())
     }
 
-    /// 快捷构造器：状态后端异常
     pub fn state<T: Display>(msg: T) -> Self {
         Self::State(msg.to_string())
     }

src/runtime/streaming/execution/mod.rs

@@ -10,7 +10,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! 执行层：Tokio Actor 运行容器。
 
 pub mod runner;
 pub mod source;

src/runtime/streaming/execution/runner.rs

@@ -32,7 +32,6 @@ use crate::runtime::streaming::execution::tracker::{
 use crate::sql::common::{CheckpointBarrier, Watermark};
 
 // ==========================================
-// 第一部分：逻辑处理层 - 算子融合链 (Logical Driver)
 // ==========================================
 
 #[async_trait]
@@ -62,7 +61,6 @@ impl ChainedDriver {
         Self { operator, next }
     }
 
-    /// 从后往前组装算子，构建责任链
     pub fn build_chain(mut operators: Vec<Box<dyn MessageOperator>>) -> Option<Box<dyn OperatorDrive>> {
         if operators.is_empty() {
             return None;
@@ -227,7 +225,6 @@ impl OperatorDrive for ChainedDriver {
 }
 
 // ==========================================
-// 第二部分：物理执行层 - 流水线 (Physical Driver)
 // ==========================================
 
 pub struct Pipeline {
@@ -238,7 +235,6 @@ pub struct Pipeline {
 
     wm_tracker: WatermarkTracker,
     barrier_aligner: BarrierAligner,
-    /// Barrier 未对齐时从轮询池移除的输入流（背压）
     paused_streams: Vec<Option<BoxedEventStream>>,
 }
 
@@ -376,5 +372,4 @@ impl Pipeline {
     }
 }
 
-/// 与执行引擎语义对齐的别名
 pub type SubtaskRunner = Pipeline;

src/runtime/streaming/execution/source.rs

@@ -10,7 +10,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! 源任务物理驱动：控制面优先、`fetch_next` 非阻塞契约、可选融合算子链下推。
 
 use crate::runtime::streaming::api::context::TaskContext;
 use crate::runtime::streaming::api::source::{SourceEvent, SourceOperator};
@@ -30,7 +29,6 @@ pub const WATERMARK_EMIT_INTERVAL: Duration = Duration::from_millis(200);
 
 pub struct SourceRunner {
     operator: Box<dyn SourceOperator>,
-    /// 有链时数据与信号经链尾再 `collect` / `broadcast`；无链则直接走 `TaskContext`。
     chain_head: Option<Box<dyn OperatorDrive>>,
     ctx: TaskContext,
     control_rx: Receiver<ControlCommand>,

src/runtime/streaming/execution/tracker/barrier_aligner.rs

@@ -10,17 +10,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Chandy–Lamport 风格屏障对齐（零内存缓冲：未对齐时从轮询池移除输入流，依赖底层背压）。
 
 use std::collections::HashSet;
 
 use crate::sql::common::CheckpointBarrier;
 
 #[derive(Debug)]
 pub enum AlignmentStatus {
-    /// 未对齐：外层应将当前通道从 `StreamMap` 挂起（Pause）。
     Pending,
-    /// 已对齐：外层触发快照并唤醒所有挂起通道（Resume）。
     Complete,
 }
 

src/runtime/streaming/execution/tracker/mod.rs

@@ -10,7 +10,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! 协调层：屏障对齐与多路水位线追踪。
 
 pub mod barrier_aligner;
 pub mod watermark_tracker;