update

Author: luoluoyuyu

src/runtime.rs

@@ -0,0 +1,27 @@
+// 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.
+
+// In-tree runtime: streaming engine, util helpers, and WASM task runtime.
+// Paths are relative to `src/` (this file lives at `src/runtime.rs`).
+
+pub use function_stream_runtime_common::{common, memory};
+
+#[path = "streaming_runtime/src/streaming/mod.rs"]
+pub mod streaming;
+
+#[path = "streaming_runtime/src/util/mod.rs"]
+pub mod util;
+
+#[path = "wasm_runtime/src/wasm/mod.rs"]
+pub mod wasm;
+
+pub use wasm::{input, output, processor};

src/runtime_common/src/streaming_protocol.rs

@@ -0,0 +1,105 @@
+// 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.
+
+//! Streaming control-plane types shared by the SQL planner and the execution runtime.
+
+use bincode::{Decode, Encode};
+use serde::{Deserialize, Serialize};
+use std::time::SystemTime;
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Encode, Decode, Serialize, Deserialize)]
+pub enum Watermark {
+    EventTime(SystemTime),
+    Idle,
+}
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Encode, Decode, Serialize, Deserialize)]
+pub struct CheckpointBarrier {
+    pub epoch: u64,
+    pub min_epoch: u64,
+    pub timestamp: SystemTime,
+    pub then_stop: bool,
+}
+
+pub fn merge_watermarks(per_input: &[Option<Watermark>]) -> Option<Watermark> {
+    if per_input.iter().any(|w| w.is_none()) {
+        return None;
+    }
+
+    let mut min_event: Option<SystemTime> = None;
+    let mut all_idle = true;
+
+    for w in per_input.iter().flatten() {
+        match w {
+            Watermark::Idle => {}
+            Watermark::EventTime(t) => {
+                all_idle = false;
+                min_event = Some(match min_event {
+                    None => *t,
+                    Some(m) => m.min(*t),
+                });
+            }
+        }
+    }
+
+    if all_idle {
+        Some(Watermark::Idle)
+    } else {
+        Some(Watermark::EventTime(min_event.expect(
+            "non-idle alignment must have at least one EventTime",
+        )))
+    }
+}
+
+pub fn watermark_strictly_advances(new: Watermark, previous: Option<Watermark>) -> bool {
+    match previous {
+        None => true,
+        Some(prev) => match (new, prev) {
+            (Watermark::EventTime(tn), Watermark::EventTime(tp)) => tn > tp,
+            (Watermark::Idle, Watermark::Idle) => false,
+            (Watermark::Idle, Watermark::EventTime(_)) => true,
+            (Watermark::EventTime(_), Watermark::Idle) => true,
+        },
+    }
+}
+
+#[cfg(test)]
+mod watermark_tests {
+    use super::*;
+    use std::time::Duration;
+
+    #[test]
+    fn merge_waits_for_all_channels() {
+        let wms = vec![Some(Watermark::EventTime(SystemTime::UNIX_EPOCH)), None];
+        assert!(merge_watermarks(&wms).is_none());
+    }
+
+    #[test]
+    fn merge_min_event_time_ignores_idle() {
+        let t1 = SystemTime::UNIX_EPOCH + Duration::from_secs(10);
+        let t2 = SystemTime::UNIX_EPOCH + Duration::from_secs(5);
+        let wms = vec![Some(Watermark::EventTime(t1)), Some(Watermark::Idle)];
+        assert_eq!(merge_watermarks(&wms), Some(Watermark::EventTime(t1)));
+
+        let wms = vec![
+            Some(Watermark::EventTime(t1)),
+            Some(Watermark::EventTime(t2)),
+        ];
+        assert_eq!(merge_watermarks(&wms), Some(Watermark::EventTime(t2)));
+    }
+
+    #[test]
+    fn merge_all_idle() {
+        let wms = vec![Some(Watermark::Idle), Some(Watermark::Idle)];
+        assert_eq!(merge_watermarks(&wms), Some(Watermark::Idle));
+    }
+}

src/storage.rs

@@ -0,0 +1,61 @@
+// 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.
+
+// Persistent / catalog-related modules compiled into the root crate.
+// Paths are relative to `src/` (this file lives at `src/storage.rs`).
+
+use std::sync::Arc;
+
+use anyhow::Context;
+
+#[path = "wasm_runtime/src/state_backend/mod.rs"]
+pub mod state_backend;
+
+#[path = "catalog_storage/src/stream_catalog/mod.rs"]
+pub mod stream_catalog;
+
+#[path = "catalog_storage/src/task/mod.rs"]
+pub mod task;
+
+/// Install the process-global [`stream_catalog::CatalogManager`] from configuration.
+/// In-memory when `config.stream_catalog.persist` is `false`, otherwise a durable
+/// [`stream_catalog::RocksDbMetaStore`] (default path: `{data_dir}/catalog.db`).
+pub fn initialize_stream_catalog(config: &crate::config::GlobalConfig) -> anyhow::Result<()> {
+    use stream_catalog::{CatalogManager, InMemoryMetaStore, MetaStore, RocksDbMetaStore};
+
+    let store: Arc<dyn MetaStore> = if !config.stream_catalog.persist {
+        Arc::new(InMemoryMetaStore::new())
+    } else {
+        let path = config
+            .stream_catalog
+            .db_path
+            .as_ref()
+            .map(|p| crate::config::resolve_path(p))
+            .unwrap_or_else(|| crate::config::get_data_dir().join("catalog.db"));
+
+        std::fs::create_dir_all(&path).with_context(|| {
+            format!(
+                "Failed to create stream catalog RocksDB directory {}",
+                path.display()
+            )
+        })?;
+
+        Arc::new(RocksDbMetaStore::open(&path).with_context(|| {
+            format!(
+                "Failed to open stream catalog RocksDB at {}",
+                path.display()
+            )
+        })?)
+    };
+
+    CatalogManager::init_global(store).context("Stream catalog (CatalogManager) global init failed")
+}

src/streaming_planner/Cargo.toml

@@ -0,0 +1,49 @@
+[package]
+name = "function-stream-streaming-planner"
+version = "0.6.0"
+edition = "2024"
+
+[lib]
+name = "function_stream_streaming_planner"
+path = "src/lib.rs"
+
+[dependencies]
+protocol = { path = "../../protocol" }
+prost = "0.13"
+function-stream-config = { path = "../config" }
+function-stream-runtime-common = { path = "../runtime_common" }
+tokio = { version = "1.0", features = ["macros", "rt-multi-thread", "sync", "time", "net"] }
+tokio-stream = "0.1.18"
+anyhow = "1.0"
+xxhash-rust = { version = "0.8", features = ["xxh3"] }
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+thiserror = "2"
+tracing = "0.1"
+async-trait = "0.1"
+futures = "0.3"
+itertools = "0.14"
+petgraph = "0.7"
+unicase = "2.7"
+rand = { version = "0.8", features = ["small_rng"] }
+bincode = { version = "2", features = ["serde"] }
+chrono = "0.4"
+bytes = "1"
+ahash = "0.8"
+strum = { version = "0.26", features = ["derive"] }
+serde_json_path = "0.7"
+
+arrow = { version = "55", default-features = false }
+arrow-array = "55"
+arrow-schema = { version = "55", features = ["serde"] }
+arrow-json = { version = "55.2.0" }
+apache-avro = "0.21"
+
+datafusion = { git = "https://github.com/FunctionStream/datafusion", branch = "48.0.1/fs" }
+datafusion-common = { git = "https://github.com/FunctionStream/datafusion", branch = "48.0.1/fs" }
+datafusion-execution = { git = "https://github.com/FunctionStream/datafusion", branch = "48.0.1/fs" }
+datafusion-expr = { git = "https://github.com/FunctionStream/datafusion", branch = "48.0.1/fs" }
+datafusion-physical-expr = { git = "https://github.com/FunctionStream/datafusion", branch = "48.0.1/fs" }
+datafusion-proto = { git = "https://github.com/FunctionStream/datafusion", branch = "48.0.1/fs" }
+
+sqlparser = { git = "https://github.com/FunctionStream/sqlparser-rs", branch = "0.58.0/fs" }