risingwave_stream/from_proto/source/
fs_fetch.rs

1// Copyright 2023 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 std::sync::Arc;
16
17use risingwave_connector::WithOptionsSecResolved;
18use risingwave_connector::source::ConnectorProperties;
19use risingwave_connector::source::filesystem::opendal_source::{
20    OpendalAzblob, OpendalGcs, OpendalPosixFs, OpendalS3,
21};
22use risingwave_connector::source::reader::desc::SourceDescBuilder;
23use risingwave_pb::stream_plan::StreamFsFetchNode;
24use risingwave_storage::StateStore;
25
26use crate::error::StreamResult;
27use crate::executor::source::{
28    BatchAdbcSnowflakeFetchExecutor, BatchIcebergFetchExecutor, BatchOpendalFsFetchExecutor,
29    BatchPosixFsFetchExecutor, FsFetchExecutor, IcebergFetchExecutor, SourceStateTableHandler,
30    StreamSourceCore,
31};
32use crate::executor::{Execute, Executor};
33use crate::from_proto::ExecutorBuilder;
34use crate::from_proto::source::is_full_reload_refresh;
35use crate::task::ExecutorParams;
36
37pub struct FsFetchExecutorBuilder;
38
39impl_stream_node_body!(StreamFsFetch(StreamFsFetchNode) => FsFetchExecutorBuilder);
40
41impl ExecutorBuilder for FsFetchExecutorBuilder {
42    type Node = StreamFsFetchNode;
43
44    async fn new_boxed_executor(
45        params: ExecutorParams,
46        node: &Self::Node,
47        store: impl StateStore,
48    ) -> StreamResult<Executor> {
49        let [upstream]: [_; 1] = params.input.try_into().unwrap();
50
51        let source = node.node_inner.as_ref().unwrap();
52        let is_full_reload_refresh = is_full_reload_refresh(&source.refresh_mode);
53
54        let source_id = source.source_id;
55        let source_name = source.source_name.clone();
56        let source_info = source.get_info()?;
57        let source_options_with_secret =
58            WithOptionsSecResolved::new(source.with_properties.clone(), source.secret_refs.clone());
59        let properties = ConnectorProperties::extract(source_options_with_secret.clone(), false)?;
60        let source_desc_builder = SourceDescBuilder::new(
61            source.columns.clone(),
62            params.env.source_metrics(),
63            source.row_id_index.map(|x| x as _),
64            source_options_with_secret,
65            source_info.clone(),
66            params.config.developer.connector_message_buffer_size,
67            params.info.stream_key.clone(),
68        );
69
70        let source_column_ids: Vec<_> = source_desc_builder
71            .column_catalogs_to_source_column_descs()
72            .iter()
73            .map(|column| column.column_id)
74            .collect();
75
76        let vnodes = Some(Arc::new(
77            params
78                .vnode_bitmap
79                .expect("vnodes not set for fetch executor"),
80        ));
81        let state_table_handler = SourceStateTableHandler::from_table_catalog_with_vnodes(
82            source.state_table.as_ref().unwrap(),
83            store.clone(),
84            vnodes,
85        )
86        .await;
87        let stream_source_core = StreamSourceCore::new(
88            source_id,
89            source_name,
90            source_column_ids,
91            source_desc_builder,
92            state_table_handler,
93        );
94
95        let exec = match properties {
96            risingwave_connector::source::ConnectorProperties::Gcs(_) => {
97                if is_full_reload_refresh {
98                    BatchOpendalFsFetchExecutor::<_, OpendalGcs>::new(
99                        params.actor_context.clone(),
100                        stream_source_core,
101                        upstream,
102                        source.rate_limit,
103                        params.local_barrier_manager.clone(),
104                        source.associated_table_id,
105                    )
106                    .boxed()
107                } else {
108                    FsFetchExecutor::<_, OpendalGcs>::new(
109                        params.actor_context.clone(),
110                        stream_source_core,
111                        upstream,
112                        source.rate_limit,
113                    )
114                    .boxed()
115                }
116            }
117            risingwave_connector::source::ConnectorProperties::OpendalS3(_) => {
118                if is_full_reload_refresh {
119                    BatchOpendalFsFetchExecutor::<_, OpendalS3>::new(
120                        params.actor_context.clone(),
121                        stream_source_core,
122                        upstream,
123                        source.rate_limit,
124                        params.local_barrier_manager.clone(),
125                        source.associated_table_id,
126                    )
127                    .boxed()
128                } else {
129                    FsFetchExecutor::<_, OpendalS3>::new(
130                        params.actor_context.clone(),
131                        stream_source_core,
132                        upstream,
133                        source.rate_limit,
134                    )
135                    .boxed()
136                }
137            }
138            risingwave_connector::source::ConnectorProperties::Iceberg(_) => {
139                if is_full_reload_refresh {
140                    BatchIcebergFetchExecutor::new(
141                        params.actor_context.clone(),
142                        stream_source_core,
143                        upstream,
144                        params.local_barrier_manager.clone(),
145                        params.config.clone(),
146                        source.associated_table_id,
147                    )
148                    .boxed()
149                } else {
150                    IcebergFetchExecutor::new(
151                        params.actor_context.clone(),
152                        stream_source_core,
153                        upstream,
154                        source.rate_limit,
155                        params.config.clone(),
156                    )
157                    .boxed()
158                }
159            }
160            risingwave_connector::source::ConnectorProperties::Azblob(_) => {
161                FsFetchExecutor::<_, OpendalAzblob>::new(
162                    params.actor_context.clone(),
163                    stream_source_core,
164                    upstream,
165                    source.rate_limit,
166                )
167                .boxed()
168            }
169            risingwave_connector::source::ConnectorProperties::PosixFs(_) => {
170                FsFetchExecutor::<_, OpendalPosixFs>::new(
171                    params.actor_context.clone(),
172                    stream_source_core,
173                    upstream,
174                    source.rate_limit,
175                )
176                .boxed()
177            }
178            risingwave_connector::source::ConnectorProperties::BatchPosixFs(_) => {
179                BatchPosixFsFetchExecutor::new(
180                    params.actor_context.clone(),
181                    stream_source_core,
182                    upstream,
183                    source.rate_limit,
184                    params.local_barrier_manager.clone(),
185                    source.associated_table_id,
186                )
187                .boxed()
188            }
189            risingwave_connector::source::ConnectorProperties::AdbcSnowflake(_) => {
190                if is_full_reload_refresh {
191                    BatchAdbcSnowflakeFetchExecutor::new(
192                        params.actor_context.clone(),
193                        stream_source_core,
194                        upstream,
195                        params.local_barrier_manager.clone(),
196                        source.associated_table_id,
197                    )
198                    .boxed()
199                } else {
200                    unreachable!("AdbcSnowflake connector only supports FULL_RELOAD refresh mode")
201                }
202            }
203            _ => unreachable!(),
204        };
205        Ok((params.info, exec).into())
206    }
207}
risingwave_stream/from_proto/source/fs_fetch.rs

risingwave_stream/from_proto/source/
fs_fetch.rs
