add now executor builder

Author: Liang Zhao

proto/stream_plan.proto

@@ -413,6 +413,11 @@ message SortNode {
   uint32 sort_column_index = 2;
 }
 
+message NowNode {
+  // Persists emitted 'now'.
+  catalog.Table state_table = 1;
+}
+
 message StreamNode {
   oneof node_body {
     SourceNode source = 100;
@@ -442,6 +447,7 @@ message StreamNode {
     GroupTopNNode group_top_n = 124;
     SortNode sort = 125;
     WatermarkFilterNode watermark_filter = 126;
+    NowNode now = 127;
   }
   // The id for the operator. This is local per mview.
   // TODO: should better be a uint32.

src/stream/src/executor/mod.rs

@@ -106,6 +106,7 @@ pub use lookup::*;
 pub use lookup_union::LookupUnionExecutor;
 pub use merge::MergeExecutor;
 pub use mview::*;
+pub use now::NowExecutor;
 pub use project::ProjectExecutor;
 pub use project_set::*;
 pub use rearranged_chain::RearrangedChainExecutor;

src/stream/src/executor/now.rs

@@ -42,7 +42,6 @@ pub struct NowExecutor<S: StateStore> {
 }
 
 impl<S: StateStore> NowExecutor<S> {
-    #[allow(dead_code)]
     pub fn new(
         barrier_receiver: UnboundedReceiver<Barrier>,
         executor_id: u64,

src/stream/src/from_proto/mod.rs

@@ -30,6 +30,7 @@ mod lookup;
 mod lookup_union;
 mod merge;
 mod mview;
+mod now;
 mod project;
 mod project_set;
 mod sink;
@@ -62,6 +63,7 @@ use self::lookup::*;
 use self::lookup_union::*;
 use self::merge::*;
 use self::mview::*;
+use self::now::NowExecutorBuilder;
 use self::project::*;
 use self::project_set::*;
 use self::sink::*;
@@ -136,5 +138,6 @@ pub async fn create_executor(
         NodeBody::GroupTopN => GroupTopNExecutorBuilder,
         NodeBody::Sort => SortExecutorBuilder,
         NodeBody::WatermarkFilter => WatermarkFilterBuilder,
+        NodeBody::Now => NowExecutorBuilder,
     }
 }

src/stream/src/from_proto/now.rs

@@ -0,0 +1,54 @@
+// Copyright 2022 Singularity Data
+//
+// 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.
+
+use risingwave_common::try_match_expand;
+use risingwave_pb::stream_plan::stream_node::NodeBody;
+use risingwave_pb::stream_plan::StreamNode;
+use risingwave_storage::table::streaming_table::state_table::StateTable;
+use risingwave_storage::StateStore;
+use tokio::sync::mpsc::unbounded_channel;
+
+use super::ExecutorBuilder;
+use crate::error::StreamResult;
+use crate::executor::{BoxedExecutor, NowExecutor};
+use crate::task::{ExecutorParams, LocalStreamManagerCore};
+
+pub struct NowExecutorBuilder;
+
+#[async_trait::async_trait]
+impl ExecutorBuilder for NowExecutorBuilder {
+    async fn new_boxed_executor(
+        params: ExecutorParams,
+        node: &StreamNode,
+        store: impl StateStore,
+        stream: &mut LocalStreamManagerCore,
+    ) -> StreamResult<BoxedExecutor> {
+        let node = try_match_expand!(node.get_node_body().unwrap(), NodeBody::Now)?;
+
+        let (sender, barrier_receiver) = unbounded_channel();
+        stream
+            .context
+            .lock_barrier_manager()
+            .register_sender(params.actor_context.id, sender);
+
+        let state_table =
+            StateTable::from_table_catalog(node.get_state_table()?, store, None).await;
+
+        Ok(Box::new(NowExecutor::new(
+            barrier_receiver,
+            params.executor_id,
+            state_table,
+        )))
+    }
+}