risingwave_stream/from_proto/
now.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use anyhow::Context;
16use risingwave_common::types::{DataType, Datum};
17use risingwave_common::util::value_encoding::DatumFromProtoExt;
18use risingwave_pb::stream_plan::now_node::PbMode as PbNowMode;
19use risingwave_pb::stream_plan::{NowNode, PbNowModeGenerateSeries};
20use risingwave_storage::StateStore;
21
22use super::ExecutorBuilder;
23use crate::common::table::state_table::StateTableBuilder;
24use crate::error::StreamResult;
25use crate::executor::{Executor, NowExecutor, NowMode};
26use crate::task::ExecutorParams;
27
28pub struct NowExecutorBuilder;
29
30impl_stream_node_body!(Now(NowNode) => NowExecutorBuilder);
31
32impl ExecutorBuilder for NowExecutorBuilder {
33    type Node = NowNode;
34
35    async fn new_boxed_executor(
36        params: ExecutorParams,
37        node: &NowNode,
38        store: impl StateStore,
39    ) -> StreamResult<Executor> {
40        let barrier_receiver = params
41            .local_barrier_manager
42            .subscribe_barrier(params.actor_context.id);
43
44        let mode = if let Ok(pb_mode) = node.get_mode() {
45            match pb_mode {
46                PbNowMode::UpdateCurrent(_) => NowMode::UpdateCurrent,
47                PbNowMode::GenerateSeries(PbNowModeGenerateSeries {
48                    start_timestamp,
49                    interval,
50                }) => {
51                    let start_timestamp = Datum::from_protobuf(
52                        start_timestamp.as_ref().unwrap(),
53                        &DataType::Timestamptz,
54                    )
55                    .context("`start_timestamp` field is not decodable")?
56                    .context("`start_timestamp` field should not be NULL")?
57                    .into_timestamptz();
58                    let interval =
59                        Datum::from_protobuf(interval.as_ref().unwrap(), &DataType::Interval)
60                            .context("`interval` field is not decodable")?
61                            .context("`interval` field should not be NULL")?
62                            .into_interval();
63                    NowMode::GenerateSeries {
64                        start_timestamp,
65                        interval,
66                    }
67                }
68            }
69        } else {
70            // default to `UpdateCurrent` for backward-compatibility
71            NowMode::UpdateCurrent
72        };
73
74        let state_table = StateTableBuilder::new(node.get_state_table()?, store, None)
75            .enable_preload_all_rows_by_config(&params.config)
76            .build()
77            .await;
78        let progress_ratio = params.config.developer.now_progress_ratio;
79        let fragment_id = params.fragment_id;
80        let exec = NowExecutor::new(
81            params.info.schema.data_types(),
82            mode,
83            params.eval_error_report,
84            barrier_receiver,
85            state_table,
86            progress_ratio,
87            params.executor_stats,
88            fragment_id,
89        );
90        Ok((params.info, exec).into())
91    }
92}
risingwave_stream/from_proto/now.rs

risingwave_stream/from_proto/
now.rs
