risingwave_stream/from_proto/source/
trad_source.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 risingwave_common::catalog::{
16    KAFKA_TIMESTAMP_COLUMN_NAME, default_key_column_name_version_mapping,
17};
18use risingwave_connector::source::filesystem::opendal_source::{OpendalGcs, OpendalS3};
19use risingwave_connector::source::reader::desc::SourceDescBuilder;
20use risingwave_connector::source::should_copy_to_format_encode_options;
21use risingwave_connector::{WithOptionsSecResolved, WithPropertiesExt};
22use risingwave_expr::bail;
23use risingwave_pb::data::data_type::TypeName as PbTypeName;
24use risingwave_pb::plan_common::additional_column::ColumnType as AdditionalColumnType;
25use risingwave_pb::plan_common::{
26    AdditionalColumn, AdditionalColumnKey, AdditionalColumnTimestamp,
27    AdditionalColumnType as LegacyAdditionalColumnType, ColumnDescVersion, FormatType,
28    PbColumnCatalog, PbEncodeType,
29};
30use risingwave_pb::stream_plan::SourceNode;
31
32use super::*;
33use crate::executor::TroublemakerExecutor;
34use crate::executor::source::{
35    BatchAdbcSnowflakeListExecutor, BatchIcebergListExecutor, BatchOpendalFsListExecutor,
36    BatchPosixFsListExecutor, DummySourceExecutor, FsListExecutor, IcebergListExecutor,
37    SourceExecutor, SourceStateTableHandler, StreamSourceCore,
38};
39use crate::from_proto::source::is_full_reload_refresh;
40
41pub struct SourceExecutorBuilder;
42
43pub fn create_source_desc_builder(
44    mut source_columns: Vec<PbColumnCatalog>,
45    params: &ExecutorParams,
46    source_info: PbStreamSourceInfo,
47    row_id_index: Option<u32>,
48    with_properties: WithOptionsSecResolved,
49) -> SourceDescBuilder {
50    {
51        // compatible code: introduced in https://github.com/risingwavelabs/risingwave/pull/13707
52        // for upsert and (avro | protobuf) overwrite the `_rw_key` column's ColumnDesc.additional_column_type to Key
53        if source_info.format() == FormatType::Upsert
54            && (source_info.row_encode() == PbEncodeType::Avro
55                || source_info.row_encode() == PbEncodeType::Protobuf
56                || source_info.row_encode() == PbEncodeType::Json)
57        {
58            for c in &mut source_columns {
59                if let Some(desc) = c.column_desc.as_mut() {
60                    let is_bytea = desc
61                        .get_column_type()
62                        .map(|col_type| col_type.type_name == PbTypeName::Bytea as i32)
63                        .unwrap();
64                    if desc.name == default_key_column_name_version_mapping(
65                        &desc.version()
66                    )
67                        && is_bytea
68                        // the column is from a legacy version (before v1.5.x)
69                        && desc.version == ColumnDescVersion::Unspecified as i32
70                    {
71                        desc.additional_column = Some(AdditionalColumn {
72                            column_type: Some(AdditionalColumnType::Key(AdditionalColumnKey {})),
73                        });
74                    }
75
76                    // the column is from a legacy version (v1.6.x)
77                    // introduced in https://github.com/risingwavelabs/risingwave/pull/15226
78                    if desc.additional_column_type == LegacyAdditionalColumnType::Key as i32 {
79                        desc.additional_column = Some(AdditionalColumn {
80                            column_type: Some(AdditionalColumnType::Key(AdditionalColumnKey {})),
81                        });
82                    }
83                }
84            }
85        }
86    }
87
88    {
89        // compatible code: handle legacy column `_rw_kafka_timestamp`
90        // the column is auto added for all kafka source to empower batch query on source
91        // solution: rewrite the column `additional_column` to Timestamp
92
93        let _ = source_columns.iter_mut().map(|c| {
94            let _ = c.column_desc.as_mut().map(|desc| {
95                let is_timestamp = desc
96                    .get_column_type()
97                    .map(|col_type| col_type.type_name == PbTypeName::Timestamptz as i32)
98                    .unwrap();
99                if desc.name == KAFKA_TIMESTAMP_COLUMN_NAME
100                    && is_timestamp
101                    // the column is from a legacy version
102                    && desc.version == ColumnDescVersion::Unspecified as i32
103                {
104                    desc.additional_column = Some(AdditionalColumn {
105                        column_type: Some(AdditionalColumnType::Timestamp(
106                            AdditionalColumnTimestamp {},
107                        )),
108                    });
109                }
110            });
111        });
112    }
113
114    SourceDescBuilder::new(
115        source_columns.clone(),
116        params.env.source_metrics(),
117        row_id_index.map(|x| x as _),
118        with_properties,
119        source_info,
120        params.config.developer.connector_message_buffer_size,
121        // `pk_indices` is used to ensure that a message will be skipped instead of parsed
122        // with null pk when the pk column is missing.
123        //
124        // Currently pk_indices for source is always empty since pk information is not
125        // passed via `StreamSource` so null pk may be emitted to downstream.
126        //
127        // TODO: use the correct information to fill in pk_dicies.
128        // We should consider add back the "pk_column_ids" field removed by #8841 in
129        // StreamSource
130        params.info.stream_key.clone(),
131    )
132}
133
134impl_stream_node_body!(Source(SourceNode) => SourceExecutorBuilder);
135
136impl ExecutorBuilder for SourceExecutorBuilder {
137    type Node = SourceNode;
138
139    async fn new_boxed_executor(
140        params: ExecutorParams,
141        node: &Self::Node,
142        store: impl StateStore,
143    ) -> StreamResult<Executor> {
144        let barrier_receiver = params
145            .local_barrier_manager
146            .subscribe_barrier(params.actor_context.id);
147        let system_params = params.env.system_params_manager_ref().get_params();
148
149        if let Some(source) = &node.source_inner {
150            let is_full_reload_refresh = is_full_reload_refresh(&source.refresh_mode);
151            let exec = {
152                let source_id = source.source_id;
153                let source_name = source.source_name.clone();
154                let mut source_info = source.get_info()?.clone();
155                let associated_table_id = source.associated_table_id;
156
157                if source_info.format_encode_options.is_empty() {
158                    // compatible code: quick fix for <https://github.com/risingwavelabs/risingwave/issues/14755>,
159                    // will move the logic to FragmentManager::init in release 1.7.
160                    let connector = get_connector_name(&source.with_properties);
161                    source_info.format_encode_options.extend(
162                        source.with_properties.iter().filter_map(|(k, v)| {
163                            should_copy_to_format_encode_options(k, &connector)
164                                .then_some((k.to_owned(), v.to_owned()))
165                        }),
166                    );
167                }
168
169                let with_properties = WithOptionsSecResolved::new(
170                    source.with_properties.clone(),
171                    source.secret_refs.clone(),
172                );
173
174                let source_desc_builder = create_source_desc_builder(
175                    source.columns.clone(),
176                    &params,
177                    source_info,
178                    source.row_id_index,
179                    with_properties,
180                );
181
182                let source_column_ids: Vec<_> = source_desc_builder
183                    .column_catalogs_to_source_column_descs()
184                    .iter()
185                    .map(|column| column.column_id)
186                    .collect();
187
188                let state_table_handler = SourceStateTableHandler::from_table_catalog(
189                    source.state_table.as_ref().unwrap(),
190                    store.clone(),
191                )
192                .await;
193                let stream_source_core = StreamSourceCore::new(
194                    source_id,
195                    source_name,
196                    source_column_ids,
197                    source_desc_builder,
198                    state_table_handler,
199                );
200
201                let is_legacy_fs_connector = source.with_properties.is_legacy_fs_connector();
202                let is_fs_v2_connector = source.with_properties.is_new_fs_connector();
203                let is_s3_connector = source
204                    .with_properties
205                    .get_connector()
206                    .map(|c| {
207                        c.eq_ignore_ascii_case(risingwave_connector::source::OPENDAL_S3_CONNECTOR)
208                    })
209                    .unwrap_or(false);
210                let is_gcs_connector = source
211                    .with_properties
212                    .get_connector()
213                    .map(|c| c.eq_ignore_ascii_case(risingwave_connector::source::GCS_CONNECTOR))
214                    .unwrap_or(false);
215
216                if is_legacy_fs_connector {
217                    // Changed to default since v2.0 https://github.com/risingwavelabs/risingwave/pull/17963
218                    bail!(
219                        "legacy s3 connector is fully deprecated since v2.4.0, please DROP and recreate the s3 source.\nexecutor: {:?}",
220                        params
221                    );
222                } else if is_full_reload_refresh && is_s3_connector {
223                    BatchOpendalFsListExecutor::<_, OpendalS3>::new(
224                        params.actor_context.clone(),
225                        stream_source_core,
226                        params.executor_stats.clone(),
227                        barrier_receiver,
228                        system_params,
229                        source.rate_limit,
230                        params.local_barrier_manager.clone(),
231                        associated_table_id,
232                    )
233                    .boxed()
234                } else if is_full_reload_refresh && is_gcs_connector {
235                    BatchOpendalFsListExecutor::<_, OpendalGcs>::new(
236                        params.actor_context.clone(),
237                        stream_source_core,
238                        params.executor_stats.clone(),
239                        barrier_receiver,
240                        system_params,
241                        source.rate_limit,
242                        params.local_barrier_manager.clone(),
243                        associated_table_id,
244                    )
245                    .boxed()
246                } else if is_fs_v2_connector {
247                    FsListExecutor::new(
248                        params.actor_context.clone(),
249                        stream_source_core,
250                        params.executor_stats.clone(),
251                        barrier_receiver,
252                        system_params,
253                        source.rate_limit,
254                    )
255                    .boxed()
256                } else if source.with_properties.is_iceberg_connector() {
257                    if is_full_reload_refresh {
258                        BatchIcebergListExecutor::new(
259                            params.actor_context.clone(),
260                            stream_source_core,
261                            source
262                                .downstream_columns
263                                .as_ref()
264                                .map(|x| x.columns.clone().into_iter().map(|c| c.into()).collect()),
265                            params.executor_stats.clone(),
266                            barrier_receiver,
267                            params.local_barrier_manager.clone(),
268                            associated_table_id,
269                        )
270                        .boxed()
271                    } else {
272                        IcebergListExecutor::new(
273                            params.actor_context.clone(),
274                            stream_source_core,
275                            source
276                                .downstream_columns
277                                .as_ref()
278                                .map(|x| x.columns.clone().into_iter().map(|c| c.into()).collect()),
279                            params.executor_stats.clone(),
280                            barrier_receiver,
281                            system_params,
282                            source.rate_limit,
283                            params.config.clone(),
284                        )
285                        .boxed()
286                    }
287                } else if source.with_properties.is_batch_connector() {
288                    if source
289                        .with_properties
290                        .get_connector()
291                        .map(|c| {
292                            c.eq_ignore_ascii_case(
293                                risingwave_connector::source::BATCH_POSIX_FS_CONNECTOR,
294                            )
295                        })
296                        .unwrap_or(false)
297                    {
298                        BatchPosixFsListExecutor::new(
299                            params.actor_context.clone(),
300                            stream_source_core,
301                            params.executor_stats.clone(),
302                            barrier_receiver,
303                            system_params,
304                            source.rate_limit,
305                            params.local_barrier_manager.clone(),
306                            associated_table_id,
307                        )
308                        .boxed()
309                    } else if source
310                        .with_properties
311                        .get_connector()
312                        .map(|c| {
313                            c.eq_ignore_ascii_case(
314                                risingwave_connector::source::ADBC_SNOWFLAKE_CONNECTOR,
315                            )
316                        })
317                        .unwrap_or(false)
318                    {
319                        BatchAdbcSnowflakeListExecutor::new(
320                            params.actor_context.clone(),
321                            stream_source_core,
322                            params.executor_stats.clone(),
323                            barrier_receiver,
324                            params.local_barrier_manager.clone(),
325                            associated_table_id,
326                        )
327                        .boxed()
328                    } else {
329                        unreachable!("unknown batch connector");
330                    }
331                } else {
332                    let is_shared = source.info.as_ref().is_some_and(|info| info.is_shared());
333                    SourceExecutor::new(
334                        params.actor_context.clone(),
335                        stream_source_core,
336                        params.executor_stats.clone(),
337                        barrier_receiver,
338                        system_params,
339                        source.rate_limit,
340                        is_shared && !source.with_properties.is_cdc_connector(),
341                        params.local_barrier_manager.clone(),
342                    )
343                    .boxed()
344                }
345            };
346
347            if crate::consistency::insane() {
348                let mut info = params.info.clone();
349                info.identity = format!("{} (troubled)", info.identity);
350                Ok((
351                    params.info,
352                    TroublemakerExecutor::new(
353                        (info, exec).into(),
354                        params.config.developer.chunk_size,
355                    ),
356                )
357                    .into())
358            } else {
359                Ok((params.info, exec).into())
360            }
361        } else {
362            // If there is no external stream source, then no data should be persisted.
363            // Use DummySourceExecutor which only forwards barrier messages.
364            let exec = DummySourceExecutor::new(params.actor_context, barrier_receiver);
365            Ok((params.info, exec).into())
366        }
367    }
368}
risingwave_stream/from_proto/source/trad_source.rs

risingwave_stream/from_proto/source/
trad_source.rs
