risingwave_stream/from_proto/
sync_log_store.rs

1// Copyright 2025 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 risingwave_pb::stream_plan::SyncLogStoreNode;
16use risingwave_storage::StateStore;
17use tokio::time::Duration;
18
19use crate::common::log_store_impl::kv_log_store::KV_LOG_STORE_V2_INFO;
20use crate::common::log_store_impl::kv_log_store::serde::LogStoreRowSerde;
21use crate::error::StreamResult;
22use crate::executor::{Executor, SyncedKvLogStoreExecutor, SyncedKvLogStoreMetrics};
23use crate::from_proto::ExecutorBuilder;
24use crate::task::ExecutorParams;
25
26pub struct SyncLogStoreExecutorBuilder;
27
28impl_stream_node_body!(SyncLogStore(SyncLogStoreNode) => SyncLogStoreExecutorBuilder);
29
30impl ExecutorBuilder for SyncLogStoreExecutorBuilder {
31    type Node = SyncLogStoreNode;
32
33    async fn new_boxed_executor(
34        params: ExecutorParams,
35        node: &Self::Node,
36        store: impl StateStore,
37    ) -> StreamResult<Executor> {
38        let actor_context = params.actor_context.clone();
39        let table = node.log_store_table.as_ref().unwrap().clone();
40        let table_id = table.id;
41
42        let metrics = {
43            let streaming_metrics = actor_context.streaming_metrics.as_ref();
44            let actor_id = actor_context.id;
45            let fragment_id = params.fragment_id;
46            let name = "sync_log_store";
47            let target = "unaligned_hash_join";
48            SyncedKvLogStoreMetrics::new(streaming_metrics, actor_id, fragment_id, name, target)
49        };
50
51        let serde =
52            LogStoreRowSerde::new(&table, params.vnode_bitmap.into(), &KV_LOG_STORE_V2_INFO);
53        let [upstream] = params.input.try_into().unwrap();
54
55        // Previously, these configs are persisted in the plan node.
56        // Now it's always `None` and we should refer to the job's config override.
57        #[expect(deprecated)]
58        let pause_duration_ms = node.pause_duration_ms.map_or(
59            params.config.developer.sync_log_store_pause_duration_ms,
60            |v| v as usize,
61        );
62        #[expect(deprecated)]
63        let buffer_max_size = node
64            .buffer_size
65            .map_or(params.config.developer.sync_log_store_buffer_size, |v| {
66                v as usize
67            });
68
69        let chunk_size = actor_context.config.developer.chunk_size;
70
71        let executor = SyncedKvLogStoreExecutor::new(
72            actor_context,
73            table_id,
74            metrics,
75            serde,
76            store,
77            buffer_max_size,
78            chunk_size,
79            upstream,
80            Duration::from_millis(pause_duration_ms as _),
81            node.aligned,
82        );
83        Ok((params.info, executor).into())
84    }
85}
risingwave_stream/from_proto/sync_log_store.rs

risingwave_stream/from_proto/
sync_log_store.rs
