risingwave_stream/from_proto/
stream_cdc_scan.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 anyhow::Context;
18use risingwave_common::catalog::{CdcKeyComparison, Schema};
19use risingwave_common::hash::VnodeBitmapExt;
20use risingwave_common::util::sort_util::OrderType;
21use risingwave_connector::source::cdc::CdcScanOptions;
22use risingwave_connector::source::cdc::external::{
23    ExternalCdcTableType, ExternalTableConfig, SchemaTableName,
24};
25use risingwave_pb::data::data_type::TypeName;
26use risingwave_pb::plan_common::ExternalTableDesc;
27use risingwave_pb::stream_plan::StreamCdcScanNode;
28
29use super::*;
30use crate::common::table::state_table::StateTableBuilder;
31use crate::executor::{CdcBackfillExecutor, ExternalStorageTable, ParallelizedCdcBackfillExecutor};
32use crate::task::cdc_progress::CdcProgressReporter;
33
34pub struct StreamCdcScanExecutorBuilder;
35
36impl_stream_node_body!(StreamCdcScan(StreamCdcScanNode) => StreamCdcScanExecutorBuilder);
37
38type DecodedTablePk = (Vec<OrderType>, Option<Vec<CdcKeyComparison>>, Vec<usize>);
39
40fn decode_table_pk(
41    table_desc: &ExternalTableDesc,
42    table_type: &ExternalCdcTableType,
43) -> StreamResult<DecodedTablePk> {
44    if let Some(table_pk) = &table_desc.pk_ordering {
45        let order_types = table_pk
46            .columns
47            .iter()
48            .map(|_| OrderType::ascending())
49            .collect_vec();
50        let comparisons = table_pk
51            .columns
52            .iter()
53            .map(|column| column.get_comparison().map(CdcKeyComparison::from_protobuf))
54            .try_collect()?;
55        let indices = table_pk
56            .columns
57            .iter()
58            .map(|column| column.pk_col_idx as usize)
59            .collect_vec();
60        Ok((order_types, Some(comparisons), indices))
61    } else {
62        let order_types = table_desc
63            .pk
64            .iter()
65            .map(|column| OrderType::from_protobuf(column.get_order_type().unwrap()))
66            .collect_vec();
67        let indices = table_desc
68            .pk
69            .iter()
70            .map(|column| column.column_index as usize)
71            .collect_vec();
72        // Legacy graphs do not retain MySQL signedness: signed and unsigned BIGINT stored as
73        // Int64 are indistinguishable until the reader checks upstream. Only Int64 PK columns
74        // can need unsigned reinterpretation; other types (including Decimal) use native ordering.
75        let mut needs_reader_comparisons = false;
76        if *table_type == ExternalCdcTableType::MySql {
77            for &idx in &indices {
78                if table_desc.columns[idx].get_column_type()?.get_type_name()? == TypeName::Int64 {
79                    needs_reader_comparisons = true;
80                    break;
81                }
82            }
83        }
84        let comparisons = (!needs_reader_comparisons)
85            .then(|| vec![CdcKeyComparison::Native; table_desc.pk.len()]);
86        Ok((order_types, comparisons, indices))
87    }
88}
89
90impl ExecutorBuilder for StreamCdcScanExecutorBuilder {
91    type Node = StreamCdcScanNode;
92
93    async fn new_boxed_executor(
94        params: ExecutorParams,
95        node: &Self::Node,
96        state_store: impl StateStore,
97    ) -> StreamResult<Executor> {
98        let [upstream]: [_; 1] = params.input.try_into().unwrap();
99
100        let output_indices = node
101            .output_indices
102            .iter()
103            .map(|&i| i as usize)
104            .collect_vec();
105
106        let table_desc: &ExternalTableDesc = node.get_cdc_table_desc()?;
107
108        let output_schema: Schema = table_desc.columns.iter().map(Into::into).collect();
109        assert_eq!(output_indices, (0..output_schema.len()).collect_vec());
110        assert_eq!(output_schema.data_types(), params.info.schema.data_types());
111
112        let properties = table_desc.connect_properties.clone();
113        let table_type = ExternalCdcTableType::from_properties(&properties);
114        let (table_pk_order_types, table_pk_comparisons, table_pk_indices) =
115            decode_table_pk(table_desc, &table_type)?;
116
117        let scan_options = node
118            .options
119            .as_ref()
120            .map(CdcScanOptions::from_proto)
121            .unwrap_or(CdcScanOptions {
122                disable_backfill: node.disable_backfill,
123                ..Default::default()
124            });
125        // Filter out additional columns to construct the external table schema
126        let table_schema: Schema = table_desc
127            .columns
128            .iter()
129            .filter(|col| {
130                col.additional_column
131                    .as_ref()
132                    .is_none_or(|a_col| a_col.column_type.is_none())
133            })
134            .map(Into::into)
135            .collect();
136
137        let schema_table_name = SchemaTableName::from_properties(&properties);
138        let table_config = ExternalTableConfig::try_from_btreemap(
139            properties.clone(),
140            table_desc.secret_refs.clone(),
141        )
142        .context("failed to parse external table config")?;
143
144        let database_name = table_config.database.clone();
145
146        let external_table = ExternalStorageTable::new(
147            table_desc.table_id,
148            schema_table_name,
149            database_name,
150            table_config,
151            table_type,
152            table_schema,
153            table_pk_order_types,
154            table_pk_comparisons,
155            table_pk_indices,
156        );
157
158        let output_columns = table_desc.columns.iter().map(Into::into).collect_vec();
159        if scan_options.is_parallelized_backfill() {
160            // Set state table's vnodes to None to allow splits to be assigned to any actors, without following vnode constraints.
161            let vnodes = None;
162            let state_table = StateTableBuilder::new(node.get_state_table()?, state_store, vnodes)
163                .enable_preload_all_rows_by_config(&params.config)
164                .build()
165                .await;
166            let progress = CdcProgressReporter::new(params.local_barrier_manager.clone());
167            let exec = ParallelizedCdcBackfillExecutor::new(
168                params.actor_context.clone(),
169                external_table,
170                upstream,
171                output_indices,
172                output_columns,
173                params.executor_stats,
174                state_table,
175                node.rate_limit,
176                scan_options,
177                properties,
178                Some(progress),
179            );
180            Ok((params.info, exec).into())
181        } else {
182            assert!(
183                params.vnode_bitmap.is_singleton(),
184                "cdc backfill should be singleton"
185            );
186            let vnodes = Some(Arc::new(params.vnode_bitmap));
187            let state_table = StateTableBuilder::new(node.get_state_table()?, state_store, vnodes)
188                .enable_preload_all_rows_by_config(&params.config)
189                .build()
190                .await;
191            let exec = CdcBackfillExecutor::new(
192                params.actor_context.clone(),
193                external_table,
194                upstream,
195                output_indices,
196                output_columns,
197                None,
198                params.executor_stats,
199                state_table,
200                node.rate_limit,
201                scan_options,
202                properties,
203            );
204            Ok((params.info, exec).into())
205        }
206    }
207}
208
209#[cfg(test)]
210mod tests {
211    use risingwave_common::catalog::{CdcKeyComparison, ColumnDesc};
212    use risingwave_common::types::DataType;
213    use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
214    use risingwave_connector::source::cdc::external::ExternalCdcTableType;
215    use risingwave_pb::plan_common::cdc_key_ordering::{Column, Comparison};
216    use risingwave_pb::plan_common::{CdcKeyOrdering, ExternalTableDesc};
217
218    use super::decode_table_pk;
219
220    fn legacy_desc(pk_type: DataType) -> ExternalTableDesc {
221        ExternalTableDesc {
222            columns: vec![
223                ColumnDesc::unnamed(0.into(), DataType::Varchar).to_protobuf(),
224                ColumnDesc::unnamed(1.into(), DataType::Int64).to_protobuf(),
225                ColumnDesc::unnamed(2.into(), pk_type).to_protobuf(),
226            ],
227            pk: vec![
228                ColumnOrder::new(2, OrderType::descending()).to_protobuf(),
229                ColumnOrder::new(0, OrderType::ascending()).to_protobuf(),
230            ],
231            ..Default::default()
232        }
233    }
234
235    #[test]
236    fn test_decode_legacy_mysql_int64_pk_comparisons_as_unknown() {
237        let (order_types, comparisons, indices) =
238            decode_table_pk(&legacy_desc(DataType::Int64), &ExternalCdcTableType::MySql).unwrap();
239
240        assert_eq!(
241            order_types,
242            vec![OrderType::descending(), OrderType::ascending()]
243        );
244        assert_eq!(comparisons, None);
245        assert_eq!(indices, vec![2, 0]);
246    }
247
248    #[test]
249    fn test_decode_legacy_mysql_non_int64_pk_comparisons_as_native() {
250        // An Int64 non-PK column must not require recovering comparison metadata.
251        for pk_type in [
252            DataType::Int16,
253            DataType::Int32,
254            DataType::Decimal,
255            DataType::Float32,
256            DataType::Float64,
257            DataType::Varchar,
258        ] {
259            let (order_types, comparisons, indices) =
260                decode_table_pk(&legacy_desc(pk_type), &ExternalCdcTableType::MySql).unwrap();
261            assert_eq!(
262                order_types,
263                vec![OrderType::descending(), OrderType::ascending()]
264            );
265            assert_eq!(
266                comparisons,
267                Some(vec![CdcKeyComparison::Native, CdcKeyComparison::Native])
268            );
269            assert_eq!(indices, vec![2, 0]);
270        }
271    }
272
273    #[test]
274    fn test_decode_legacy_non_mysql_pk_comparisons_as_native() {
275        for table_type in [
276            ExternalCdcTableType::Postgres,
277            ExternalCdcTableType::SqlServer,
278            ExternalCdcTableType::Mock,
279        ] {
280            let (_, comparisons, _) =
281                decode_table_pk(&legacy_desc(DataType::Int64), &table_type).unwrap();
282            assert_eq!(
283                comparisons,
284                Some(vec![CdcKeyComparison::Native, CdcKeyComparison::Native])
285            );
286        }
287    }
288
289    #[test]
290    fn test_decode_new_pk_comparisons_and_column_order() {
291        let desc = ExternalTableDesc {
292            pk_ordering: Some(CdcKeyOrdering {
293                columns: vec![
294                    Column {
295                        pk_col_idx: 3,
296                        comparison: Comparison::UnsignedInt64 as i32,
297                    },
298                    Column {
299                        pk_col_idx: 1,
300                        comparison: Comparison::Native as i32,
301                    },
302                ],
303            }),
304            ..Default::default()
305        };
306
307        let (order_types, comparisons, indices) =
308            decode_table_pk(&desc, &ExternalCdcTableType::MySql).unwrap();
309        assert_eq!(
310            order_types,
311            vec![OrderType::ascending(), OrderType::ascending()]
312        );
313        assert_eq!(
314            comparisons,
315            Some(vec![
316                CdcKeyComparison::UnsignedInt64,
317                CdcKeyComparison::Native,
318            ])
319        );
320        assert_eq!(indices, vec![3, 1]);
321    }
322}
risingwave_stream/from_proto/stream_cdc_scan.rs

risingwave_stream/from_proto/
stream_cdc_scan.rs
