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, StreamSource};
31
32use super::*;
33use crate::executor::TroublemakerExecutor;
34use crate::executor::source::{
35    BatchAdbcSnowflakeListExecutor, BatchIcebergListExecutor, BatchOpendalFsListExecutor,
36    BatchPosixFsListExecutor, DummySourceExecutor, FsListExecutor, IcebergListExecutor,
37    IcebergUpdateListExecutor, SourceExecutor, SourceStateTableHandler, StreamSourceCore,
38};
39use crate::from_proto::source::{iceberg_update_mode, 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
136fn iceberg_update_list_mode(source: &StreamSource, params: &ExecutorParams) -> StreamResult<bool> {
137    let enabled = iceberg_update_mode(
138        &source.with_properties,
139        &source.refresh_mode,
140        source.rate_limit,
141        source.row_id_index,
142        params.info.stream_kind,
143    )?;
144    if source.iceberg_fetch_state_table.is_some() != enabled {
145        bail!("Iceberg update List requires an opted-in source with a dedicated Fetch state table");
146    }
147    Ok(enabled)
148}
149
150impl ExecutorBuilder for SourceExecutorBuilder {
151    type Node = SourceNode;
152
153    async fn new_boxed_executor(
154        params: ExecutorParams,
155        node: &Self::Node,
156        store: impl StateStore,
157    ) -> StreamResult<Executor> {
158        let barrier_receiver = params
159            .local_barrier_manager
160            .subscribe_barrier(params.actor_context.id);
161        let system_params = params.env.system_params_manager_ref().get_params();
162
163        if let Some(source) = &node.source_inner {
164            let is_full_reload_refresh = is_full_reload_refresh(&source.refresh_mode);
165            let exec = {
166                let source_id = source.source_id;
167                let source_name = source.source_name.clone();
168                let mut source_info = source.get_info()?.clone();
169                let associated_table_id = source.associated_table_id;
170
171                if source_info.format_encode_options.is_empty() {
172                    // compatible code: quick fix for <https://github.com/risingwavelabs/risingwave/issues/14755>,
173                    // will move the logic to FragmentManager::init in release 1.7.
174                    let connector = get_connector_name(&source.with_properties);
175                    source_info.format_encode_options.extend(
176                        source.with_properties.iter().filter_map(|(k, v)| {
177                            should_copy_to_format_encode_options(k, &connector)
178                                .then_some((k.to_owned(), v.to_owned()))
179                        }),
180                    );
181                }
182
183                let with_properties = WithOptionsSecResolved::new(
184                    source.with_properties.clone(),
185                    source.secret_refs.clone(),
186                );
187
188                let source_desc_builder = create_source_desc_builder(
189                    source.columns.clone(),
190                    &params,
191                    source_info,
192                    source.row_id_index,
193                    with_properties,
194                );
195
196                let source_column_ids: Vec<_> = source_desc_builder
197                    .column_catalogs_to_source_column_descs()
198                    .iter()
199                    .map(|column| column.column_id)
200                    .collect();
201
202                let is_iceberg_update = iceberg_update_list_mode(source, &params)?;
203                if is_iceberg_update {
204                    crate::executor::source::validate_update_state_table(
205                        source.get_state_table()?,
206                        true,
207                    )?;
208                    if params.info.schema.data_types()
209                        != vec![
210                            risingwave_common::types::DataType::Int64,
211                            risingwave_common::types::DataType::Int64,
212                            risingwave_common::types::DataType::Jsonb,
213                        ]
214                    {
215                        bail!(
216                            "Iceberg update List requires a (route, seq, assignment) output schema"
217                        );
218                    }
219                }
220                let state_table_handler = SourceStateTableHandler::from_table_catalog(
221                    source.state_table.as_ref().unwrap(),
222                    store.clone(),
223                )
224                .await;
225                let stream_source_core = StreamSourceCore::new(
226                    source_id,
227                    source_name,
228                    source_column_ids,
229                    source_desc_builder,
230                    state_table_handler,
231                );
232
233                let is_legacy_fs_connector = source.with_properties.is_legacy_fs_connector();
234                let is_fs_v2_connector = source.with_properties.is_new_fs_connector();
235                let is_s3_connector = source
236                    .with_properties
237                    .get_connector()
238                    .map(|c| {
239                        c.eq_ignore_ascii_case(risingwave_connector::source::OPENDAL_S3_CONNECTOR)
240                    })
241                    .unwrap_or(false);
242                let is_gcs_connector = source
243                    .with_properties
244                    .get_connector()
245                    .map(|c| c.eq_ignore_ascii_case(risingwave_connector::source::GCS_CONNECTOR))
246                    .unwrap_or(false);
247
248                if is_legacy_fs_connector {
249                    // Changed to default since v2.0 https://github.com/risingwavelabs/risingwave/pull/17963
250                    bail!(
251                        "legacy s3 connector is fully deprecated since v2.4.0, please DROP and recreate the s3 source.\nexecutor: {:?}",
252                        params
253                    );
254                } else if is_full_reload_refresh && is_s3_connector {
255                    BatchOpendalFsListExecutor::<_, OpendalS3>::new(
256                        params.actor_context.clone(),
257                        stream_source_core,
258                        params.executor_stats.clone(),
259                        barrier_receiver,
260                        system_params,
261                        source.rate_limit,
262                        params.local_barrier_manager.clone(),
263                        associated_table_id,
264                    )
265                    .boxed()
266                } else if is_full_reload_refresh && is_gcs_connector {
267                    BatchOpendalFsListExecutor::<_, OpendalGcs>::new(
268                        params.actor_context.clone(),
269                        stream_source_core,
270                        params.executor_stats.clone(),
271                        barrier_receiver,
272                        system_params,
273                        source.rate_limit,
274                        params.local_barrier_manager.clone(),
275                        associated_table_id,
276                    )
277                    .boxed()
278                } else if is_fs_v2_connector {
279                    FsListExecutor::new(
280                        params.actor_context.clone(),
281                        stream_source_core,
282                        params.executor_stats.clone(),
283                        barrier_receiver,
284                        system_params,
285                        source.rate_limit,
286                    )
287                    .boxed()
288                } else if source.with_properties.is_iceberg_connector() {
289                    if is_iceberg_update {
290                        IcebergUpdateListExecutor::new(
291                            stream_source_core,
292                            source
293                                .get_downstream_columns()?
294                                .columns
295                                .iter()
296                                .cloned()
297                                .map(Into::into)
298                                .collect(),
299                            source.get_iceberg_fetch_state_table()?.clone(),
300                            barrier_receiver,
301                            params.config.clone(),
302                        )
303                        .boxed()
304                    } else if is_full_reload_refresh {
305                        BatchIcebergListExecutor::new(
306                            params.actor_context.clone(),
307                            stream_source_core,
308                            source
309                                .downstream_columns
310                                .as_ref()
311                                .map(|x| x.columns.clone().into_iter().map(|c| c.into()).collect()),
312                            params.executor_stats.clone(),
313                            barrier_receiver,
314                            params.local_barrier_manager.clone(),
315                            associated_table_id,
316                        )
317                        .boxed()
318                    } else {
319                        IcebergListExecutor::new(
320                            params.actor_context.clone(),
321                            stream_source_core,
322                            source
323                                .downstream_columns
324                                .as_ref()
325                                .map(|x| x.columns.clone().into_iter().map(|c| c.into()).collect()),
326                            params.executor_stats.clone(),
327                            barrier_receiver,
328                            system_params,
329                            source.rate_limit,
330                            params.config.clone(),
331                        )
332                        .boxed()
333                    }
334                } else if source.with_properties.is_batch_connector() {
335                    if source
336                        .with_properties
337                        .get_connector()
338                        .map(|c| {
339                            c.eq_ignore_ascii_case(
340                                risingwave_connector::source::BATCH_POSIX_FS_CONNECTOR,
341                            )
342                        })
343                        .unwrap_or(false)
344                    {
345                        BatchPosixFsListExecutor::new(
346                            params.actor_context.clone(),
347                            stream_source_core,
348                            params.executor_stats.clone(),
349                            barrier_receiver,
350                            system_params,
351                            source.rate_limit,
352                            params.local_barrier_manager.clone(),
353                            associated_table_id,
354                        )
355                        .boxed()
356                    } else if source
357                        .with_properties
358                        .get_connector()
359                        .map(|c| {
360                            c.eq_ignore_ascii_case(
361                                risingwave_connector::source::ADBC_SNOWFLAKE_CONNECTOR,
362                            )
363                        })
364                        .unwrap_or(false)
365                    {
366                        BatchAdbcSnowflakeListExecutor::new(
367                            params.actor_context.clone(),
368                            stream_source_core,
369                            params.executor_stats.clone(),
370                            barrier_receiver,
371                            params.local_barrier_manager.clone(),
372                            associated_table_id,
373                        )
374                        .boxed()
375                    } else {
376                        unreachable!("unknown batch connector");
377                    }
378                } else {
379                    let is_shared = source.info.as_ref().is_some_and(|info| info.is_shared());
380                    SourceExecutor::new(
381                        params.actor_context.clone(),
382                        stream_source_core,
383                        params.executor_stats.clone(),
384                        barrier_receiver,
385                        system_params,
386                        source.rate_limit,
387                        is_shared && !source.with_properties.is_cdc_connector(),
388                        params.local_barrier_manager.clone(),
389                    )
390                    .boxed()
391                }
392            };
393
394            if crate::consistency::insane() {
395                let mut info = params.info.clone();
396                info.identity = format!("{} (troubled)", info.identity);
397                Ok((
398                    params.info,
399                    TroublemakerExecutor::new(
400                        (info, exec).into(),
401                        params.config.developer.chunk_size,
402                    ),
403                )
404                    .into())
405            } else {
406                Ok((params.info, exec).into())
407            }
408        } else {
409            // If there is no external stream source, then no data should be persisted.
410            // Use DummySourceExecutor which only forwards barrier messages.
411            let exec = DummySourceExecutor::new(params.actor_context, barrier_receiver);
412            Ok((params.info, exec).into())
413        }
414    }
415}
risingwave_stream/from_proto/source/trad_source.rs

risingwave_stream/from_proto/source/
trad_source.rs
