feat(streaming): persist Join operator state with LSM-Tree and 3-phase watermark harvesting

- Extend TaskContext with state_dir, memory_controller, io_manager,
  and safe_epoch to bridge operators with the state engine
- Refactor JoinWithExpirationOperator: replace in-memory VecDeque with
  PersistentStateBuffer backed by OperatorStateStore, using composite
  keys [Side(1B) + Timestamp(8B BE)] and BTreeSet timeline index
- Refactor InstantJoinOperator: replace in-memory BTreeMap<SystemTime,
  JoinInstance> with LSM-Tree persistence, split process_watermark into
  3-phase pipeline (harvest -> compute -> cleanup) to eliminate
  interleaved mutable/immutable borrow conflicts
- Both operators now support on_start recovery via restore_metadata and
  snapshot_state via snapshot_epoch for exactly-once semantics

Made-with: Cursor

Author: luoluoyuyu

src/coordinator/execution/executor.rs

@@ -337,15 +337,16 @@ impl PlanVisitor for Executor {
                     custom_interval.unwrap_or(0),
                 )
                 .map_err(|e| {
-                    ExecuteError::Internal(format!(
-                        "Streaming job persistence failed: {e}",
-                    ))
+                    ExecuteError::Internal(format!("Streaming job persistence failed: {e}",))
                 })?;
 
             let job_id = tokio::task::block_in_place(|| {
-                tokio::runtime::Handle::current().block_on(
-                    job_manager.submit_job(job_id, fs_program, custom_interval, None),
-                )
+                tokio::runtime::Handle::current().block_on(job_manager.submit_job(
+                    job_id,
+                    fs_program,
+                    custom_interval,
+                    None,
+                ))
             })
             .map_err(|e| ExecuteError::Internal(format!("Failed to submit streaming job: {e}")))?;
 

src/runtime/streaming/api/context.rs

@@ -10,6 +10,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime};
 
@@ -19,6 +20,7 @@ use arrow_array::RecordBatch;
 use crate::runtime::streaming::memory::MemoryPool;
 use crate::runtime::streaming::network::endpoint::PhysicalSender;
 use crate::runtime::streaming::protocol::event::{StreamEvent, TrackedEvent};
+use crate::runtime::streaming::state::{IoManager, MemoryController};
 
 #[derive(Debug, Clone)]
 pub struct TaskContextConfig {
@@ -61,6 +63,18 @@ pub struct TaskContext {
 
     /// Subtask-level tunables.
     config: TaskContextConfig,
+
+    /// Root directory for operator state persistence (LSM-Tree data/tombstone files).
+    pub state_dir: PathBuf,
+
+    /// Shared memory controller for state engine back-pressure.
+    pub memory_controller: Arc<MemoryController>,
+
+    /// I/O thread pool handle for background spill/compaction.
+    pub io_manager: IoManager,
+
+    /// Last globally-committed safe epoch for crash recovery.
+    safe_epoch: u64,
 }
 
 impl TaskContext {
@@ -71,6 +85,10 @@ impl TaskContext {
         parallelism: u32,
         downstream_senders: Vec<PhysicalSender>,
         memory_pool: Arc<MemoryPool>,
+        memory_controller: Arc<MemoryController>,
+        io_manager: IoManager,
+        state_dir: PathBuf,
+        safe_epoch: u64,
     ) -> Self {
         let task_name = format!(
             "Task-[{}]-Pipe[{}]-Sub[{}/{}]",
@@ -87,9 +105,18 @@ impl TaskContext {
             memory_pool,
             current_watermark: None,
             config: TaskContextConfig::default(),
+            state_dir,
+            memory_controller,
+            io_manager,
+            safe_epoch,
         }
     }
 
+    #[inline]
+    pub fn latest_safe_epoch(&self) -> u64 {
+        self.safe_epoch
+    }
+
     #[inline]
     pub fn config(&self) -> &TaskContextConfig {
         &self.config

src/runtime/streaming/job/job_manager.rs

@@ -19,7 +19,7 @@ use anyhow::{Context, Result, anyhow, bail, ensure};
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle as TokioJoinHandle;
 use tokio_stream::wrappers::ReceiverStream;
-use tracing::{error, info, warn, debug};
+use tracing::{debug, error, info, warn};
 
 use protocol::function_stream_graph::{ChainedOperator, FsProgram};
 
@@ -34,7 +34,7 @@ use crate::runtime::streaming::job::models::{
 };
 use crate::runtime::streaming::memory::MemoryPool;
 use crate::runtime::streaming::network::endpoint::{BoxedEventStream, PhysicalSender};
-use crate::runtime::streaming::protocol::control::{ControlCommand, StopMode, JobMasterEvent};
+use crate::runtime::streaming::protocol::control::{ControlCommand, JobMasterEvent, StopMode};
 use crate::runtime::streaming::protocol::event::CheckpointBarrier;
 use crate::runtime::streaming::state::{IoManager, IoPool, MemoryController, NoopMetricsCollector};
 use crate::storage::stream_catalog::CatalogManager;
@@ -132,7 +132,8 @@ impl JobManager {
             state_config.max_background_spills,
             state_config.max_background_compactions,
             metrics,
-        ).context("Failed to initialize state engine I/O pool")?;
+        )
+        .context("Failed to initialize state engine I/O pool")?;
 
         Ok(Self {
             active_jobs: Arc::new(RwLock::new(HashMap::new())),
@@ -153,7 +154,12 @@ impl JobManager {
         state_config: StateConfig,
     ) -> Result<()> {
         GLOBAL_JOB_MANAGER
-            .set(Arc::new(Self::new(factory, memory_bytes, state_base_dir, state_config)?))
+            .set(Arc::new(Self::new(
+                factory,
+                memory_bytes,
+                state_base_dir,
+                state_config,
+            )?))
             .map_err(|_| anyhow!("JobManager singleton already initialized"))
     }
 
@@ -217,8 +223,8 @@ impl JobManager {
             pipelines.insert(pipeline_id, pipeline);
         }
 
-        let interval_ms = custom_checkpoint_interval_ms
-            .unwrap_or(self.state_config.checkpoint_interval_ms);
+        let interval_ms =
+            custom_checkpoint_interval_ms.unwrap_or(self.state_config.checkpoint_interval_ms);
 
         self.spawn_checkpoint_coordinator(
             job_id.clone(),
@@ -425,9 +431,9 @@ impl JobManager {
         pipeline_id: u32,
         operators: &[ChainedOperator],
         edge_manager: &mut EdgeManager,
-        _job_state_dir: &Path,
+        job_state_dir: &Path,
         _job_master_tx: mpsc::Sender<JobMasterEvent>,
-        _recovery_epoch: u64,
+        recovery_epoch: u64,
     ) -> Result<(PhysicalPipeline, bool)> {
         let (raw_inboxes, raw_outboxes) =
             edge_manager.take_endpoints(pipeline_id).with_context(|| {
@@ -479,6 +485,10 @@ impl JobManager {
             parallelism,
             physical_outboxes,
             Arc::clone(&self.memory_pool),
+            Arc::clone(&self.memory_controller),
+            self.io_manager_client.clone(),
+            job_state_dir.to_path_buf(),
+            recovery_epoch,
         );
 
         let runner = if let Some(source) = chain.source {