risingwave_connector/source/cdc/
mod.rs

1// Copyright 2022 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
15pub mod enumerator;
16pub mod external;
17pub mod jni_source;
18pub mod source;
19pub mod split;
20
21use std::collections::{BTreeMap, HashMap};
22use std::marker::PhantomData;
23
24pub use enumerator::*;
25use itertools::Itertools;
26use risingwave_common::id::{ActorId, SourceId};
27use risingwave_pb::catalog::PbSource;
28use risingwave_pb::connector_service::{PbSourceType, PbTableSchema, SourceType, TableSchema};
29use risingwave_pb::plan_common::ExternalTableDesc;
30use risingwave_pb::plan_common::column_desc::GeneratedOrDefaultColumn;
31use risingwave_pb::source::{PbCdcTableSnapshotSplit, PbCdcTableSnapshotSplitsWithGeneration};
32use risingwave_pb::stream_plan::StreamCdcScanOptions;
33use simd_json::prelude::ArrayTrait;
34pub use source::*;
35
36use crate::enforce_secret::EnforceSecret;
37use crate::error::ConnectorResult;
38use crate::source::{CdcTableSnapshotSplitRaw, SourceProperties, SplitImpl, TryFromBTreeMap};
39use crate::{for_all_classified_sources, impl_cdc_source_type};
40
41pub const CDC_CONNECTOR_NAME_SUFFIX: &str = "-cdc";
42pub const CDC_SNAPSHOT_MODE_KEY: &str = "debezium.snapshot.mode";
43pub const CDC_SNAPSHOT_BACKFILL: &str = "rw_cdc_backfill";
44pub const CDC_SHARING_MODE_KEY: &str = "rw.sharing.mode.enable";
45// User can set snapshot='false' to disable cdc backfill
46pub const CDC_BACKFILL_ENABLE_KEY: &str = "snapshot";
47pub const CDC_BACKFILL_SNAPSHOT_INTERVAL_KEY: &str = "snapshot.interval";
48pub const CDC_BACKFILL_SNAPSHOT_BATCH_SIZE_KEY: &str = "snapshot.batch_size";
49pub const CDC_BACKFILL_PARALLELISM: &str = "backfill.parallelism";
50pub const CDC_BACKFILL_NUM_ROWS_PER_SPLIT: &str = "backfill.num_rows_per_split";
51pub const CDC_BACKFILL_AS_EVEN_SPLITS: &str = "backfill.as_even_splits";
52pub const CDC_BACKFILL_SPLIT_PK_COLUMN_INDEX: &str = "backfill.split_pk_column_index";
53// We enable transaction for shared cdc source by default
54pub const CDC_TRANSACTIONAL_KEY: &str = "transactional";
55pub const CDC_WAIT_FOR_STREAMING_START_TIMEOUT: &str = "cdc.source.wait.streaming.start.timeout";
56pub const CDC_BACKFILL_MAX_PARALLELISM: u32 = 256;
57
58// User can set strong-schema='true' to enable strong schema for mongo cdc source
59pub const CDC_MONGODB_STRONG_SCHEMA_KEY: &str = "strong_schema";
60
61pub const MYSQL_CDC_CONNECTOR: &str = Mysql::CDC_CONNECTOR_NAME;
62pub const POSTGRES_CDC_CONNECTOR: &str = Postgres::CDC_CONNECTOR_NAME;
63pub const CITUS_CDC_CONNECTOR: &str = Citus::CDC_CONNECTOR_NAME;
64pub const MONGODB_CDC_CONNECTOR: &str = Mongodb::CDC_CONNECTOR_NAME;
65pub const SQL_SERVER_CDC_CONNECTOR: &str = SqlServer::CDC_CONNECTOR_NAME;
66pub const ORACLE_CDC_CONNECTOR: &str = Oracle::CDC_CONNECTOR_NAME;
67
68/// Build a unique CDC table identifier from a source ID and external table name
69pub fn build_cdc_table_id(source_id: SourceId, external_table_name: &str) -> String {
70    format!("{}.{}", source_id, external_table_name)
71}
72
73pub fn normalize_simple_postgres_quoted_table_name(table_name: &str) -> Option<String> {
74    let (schema_name, table_name) = table_name.split_once('.')?;
75    let table_name = table_name.strip_prefix('"')?.strip_suffix('"')?;
76    if schema_name.is_empty()
77        || table_name.is_empty()
78        || table_name.contains('.')
79        || table_name.contains('"')
80        || table_name.contains('\\')
81        || schema_name.contains('"')
82    {
83        return None;
84    }
85
86    Some(format!("{schema_name}.{table_name}"))
87}
88
89pub trait CdcSourceTypeTrait: Send + Sync + Clone + std::fmt::Debug + 'static {
90    const CDC_CONNECTOR_NAME: &'static str;
91    fn source_type() -> CdcSourceType;
92}
93
94for_all_classified_sources!(impl_cdc_source_type);
95
96impl<'a> From<&'a str> for CdcSourceType {
97    fn from(name: &'a str) -> Self {
98        match name {
99            MYSQL_CDC_CONNECTOR => CdcSourceType::Mysql,
100            POSTGRES_CDC_CONNECTOR => CdcSourceType::Postgres,
101            CITUS_CDC_CONNECTOR => CdcSourceType::Citus,
102            MONGODB_CDC_CONNECTOR => CdcSourceType::Mongodb,
103            SQL_SERVER_CDC_CONNECTOR => CdcSourceType::SqlServer,
104            ORACLE_CDC_CONNECTOR => CdcSourceType::Oracle,
105            _ => CdcSourceType::Unspecified,
106        }
107    }
108}
109
110impl CdcSourceType {
111    pub fn as_str_name(&self) -> &str {
112        match self {
113            CdcSourceType::Mysql => "MySQL",
114            CdcSourceType::Postgres => "Postgres",
115            CdcSourceType::Citus => "Citus",
116            CdcSourceType::Mongodb => "MongoDB",
117            CdcSourceType::SqlServer => "SQL Server",
118            CdcSourceType::Oracle => "Oracle",
119            CdcSourceType::Unspecified => "Unspecified",
120        }
121    }
122}
123
124#[derive(Clone, Debug, Default)]
125pub struct CdcProperties<T: CdcSourceTypeTrait> {
126    /// Properties specified in the WITH clause by user
127    pub properties: BTreeMap<String, String>,
128
129    /// Schema of the source specified by users
130    pub table_schema: TableSchema,
131
132    /// Whether it is created by a cdc source job
133    pub is_cdc_source_job: bool,
134
135    /// For validation purpose, mark if the table is a backfill cdc table
136    pub is_backfill_table: bool,
137
138    pub _phantom: PhantomData<T>,
139}
140
141pub fn table_schema_exclude_additional_columns(table_schema: &TableSchema) -> TableSchema {
142    TableSchema {
143        columns: table_schema
144            .columns
145            .iter()
146            .filter(|col| {
147                col.additional_column
148                    .as_ref()
149                    .is_some_and(|val| val.column_type.is_none())
150            })
151            .cloned()
152            .collect(),
153        pk_indices: table_schema.pk_indices.clone(),
154    }
155}
156
157impl<T: CdcSourceTypeTrait> TryFromBTreeMap for CdcProperties<T> {
158    fn try_from_btreemap(
159        properties: BTreeMap<String, String>,
160        _deny_unknown_fields: bool,
161    ) -> ConnectorResult<Self> {
162        let is_share_source: bool = properties
163            .get(CDC_SHARING_MODE_KEY)
164            .is_some_and(|v| v == "true");
165        Ok(CdcProperties {
166            properties,
167            table_schema: Default::default(),
168            // TODO(siyuan): use serde to deserialize input hashmap
169            is_cdc_source_job: is_share_source,
170            is_backfill_table: false,
171            _phantom: PhantomData,
172        })
173    }
174}
175
176impl<T: CdcSourceTypeTrait> EnforceSecret for CdcProperties<T> {} // todo: enforce jdbc like properties
177
178impl<T: CdcSourceTypeTrait> SourceProperties for CdcProperties<T>
179where
180    DebeziumCdcSplit<T>: TryFrom<SplitImpl, Error = crate::error::ConnectorError> + Into<SplitImpl>,
181    DebeziumSplitEnumerator<T>: ListCdcSplits<CdcSourceType = T> + enumerator::CdcMonitor,
182{
183    type Split = DebeziumCdcSplit<T>;
184    type SplitEnumerator = DebeziumSplitEnumerator<T>;
185    type SplitReader = CdcSplitReader<T>;
186
187    const SOURCE_NAME: &'static str = T::CDC_CONNECTOR_NAME;
188
189    fn init_from_pb_source(&mut self, source: &PbSource) {
190        let pk_indices = source
191            .pk_column_ids
192            .iter()
193            .map(|&id| {
194                source
195                    .columns
196                    .iter()
197                    .position(|col| col.column_desc.as_ref().unwrap().column_id == id)
198                    .unwrap() as u32
199            })
200            .collect_vec();
201
202        let table_schema = PbTableSchema {
203            columns: source
204                .columns
205                .iter()
206                .flat_map(|col| &col.column_desc)
207                .filter(|col| {
208                    !matches!(
209                        col.generated_or_default_column,
210                        Some(GeneratedOrDefaultColumn::GeneratedColumn(_))
211                    )
212                })
213                .cloned()
214                .collect(),
215            pk_indices,
216        };
217        self.table_schema = table_schema;
218        if let Some(info) = source.info.as_ref() {
219            self.is_cdc_source_job = info.is_shared();
220        }
221    }
222
223    fn init_from_pb_cdc_table_desc(&mut self, table_desc: &ExternalTableDesc) {
224        let table_schema = TableSchema {
225            columns: table_desc
226                .columns
227                .iter()
228                .filter(|col| {
229                    !matches!(
230                        col.generated_or_default_column,
231                        Some(GeneratedOrDefaultColumn::GeneratedColumn(_))
232                    )
233                })
234                .cloned()
235                .collect(),
236            pk_indices: table_desc.stream_key.clone(),
237        };
238
239        self.table_schema = table_schema;
240        self.is_cdc_source_job = false;
241        self.is_backfill_table = true;
242    }
243}
244
245impl<T: CdcSourceTypeTrait> crate::source::UnknownFields for CdcProperties<T> {
246    fn unknown_fields(&self) -> HashMap<String, String> {
247        // FIXME: CDC does not handle unknown fields yet
248        HashMap::new()
249    }
250}
251
252impl<T: CdcSourceTypeTrait> CdcProperties<T> {
253    pub fn get_source_type_pb(&self) -> SourceType {
254        SourceType::from(T::source_type())
255    }
256}
257
258pub const INVALID_CDC_SPLIT_ASSIGNMENT_GENERATION_ID: u64 = 0;
259pub const INITIAL_CDC_SPLIT_ASSIGNMENT_GENERATION_ID: u64 = 1;
260
261#[derive(Clone, Debug, PartialEq, Default)]
262pub struct CdcTableSnapshotSplitAssignmentWithGeneration {
263    pub splits: HashMap<ActorId, (Vec<CdcTableSnapshotSplitRaw>, u64)>,
264}
265
266impl CdcTableSnapshotSplitAssignmentWithGeneration {
267    pub fn new(splits: HashMap<ActorId, (Vec<CdcTableSnapshotSplitRaw>, u64)>) -> Self {
268        Self { splits }
269    }
270
271    pub fn empty() -> Self {
272        Self {
273            splits: HashMap::default(),
274        }
275    }
276}
277
278pub fn build_cdc_table_snapshot_split(s: &CdcTableSnapshotSplitRaw) -> PbCdcTableSnapshotSplit {
279    PbCdcTableSnapshotSplit {
280        split_id: s.split_id,
281        left_bound_inclusive: s.left_bound_inclusive.clone(),
282        right_bound_exclusive: s.right_bound_exclusive.clone(),
283    }
284}
285
286pub fn build_actor_cdc_table_snapshot_splits_with_generation(
287    pb_cdc_table_snapshot_split_assignment: PbCdcTableSnapshotSplitsWithGeneration,
288) -> CdcTableSnapshotSplitAssignmentWithGeneration {
289    let splits = pb_cdc_table_snapshot_split_assignment
290        .splits
291        .into_iter()
292        .map(|(actor_id, splits)| {
293            let generation = splits.generation;
294            let splits = splits
295                .splits
296                .into_iter()
297                .map(|s| CdcTableSnapshotSplitRaw {
298                    split_id: s.split_id,
299                    left_bound_inclusive: s.left_bound_inclusive,
300                    right_bound_exclusive: s.right_bound_exclusive,
301                })
302                .collect();
303            (actor_id, (splits, generation))
304        })
305        .collect();
306    CdcTableSnapshotSplitAssignmentWithGeneration { splits }
307}
308
309#[derive(Debug, Clone, Hash, PartialEq)]
310pub struct CdcScanOptions {
311    /// Used by Used in non-parallel backfill, i.e. backfill V1.
312    pub disable_backfill: bool,
313    /// Used by non-parallelized backfill. The frequency of snapshot read resets for consuming the WAL backlog.
314    pub snapshot_barrier_interval: u32,
315    /// Used by non-parallelized backfill. The number of rows to fetch in a single batch when reading from an external table.
316    pub snapshot_batch_size: u32,
317    /// Used by parallelized backfill, i.e. backfill V2. The initial parallelism of parallel backfill.
318    pub backfill_parallelism: u32,
319    /// Used by parallelized backfill. The estimated number of rows per split used in splits generation.
320    pub backfill_num_rows_per_split: u64,
321    /// Used by parallelized backfill. For supported split column data type, assume an uniform distribution and adopt a much faster splits generation method.
322    pub backfill_as_even_splits: bool,
323    /// Used by parallelized backfill. Specify the index of primary key column to use as split column.
324    pub backfill_split_pk_column_index: u32,
325}
326
327impl Default for CdcScanOptions {
328    fn default() -> Self {
329        Self {
330            disable_backfill: false,
331            snapshot_barrier_interval: 10,
332            snapshot_batch_size: 1000,
333            // 0 means disable backfill v2.
334            backfill_parallelism: 0,
335            backfill_num_rows_per_split: 100_000,
336            backfill_as_even_splits: true,
337            backfill_split_pk_column_index: 0,
338        }
339    }
340}
341
342impl CdcScanOptions {
343    pub fn to_proto(&self) -> StreamCdcScanOptions {
344        StreamCdcScanOptions {
345            disable_backfill: self.disable_backfill,
346            snapshot_barrier_interval: self.snapshot_barrier_interval,
347            snapshot_batch_size: self.snapshot_batch_size,
348            backfill_parallelism: self.backfill_parallelism,
349            backfill_num_rows_per_split: self.backfill_num_rows_per_split,
350            backfill_as_even_splits: self.backfill_as_even_splits,
351            backfill_split_pk_column_index: self.backfill_split_pk_column_index,
352        }
353    }
354
355    pub fn from_proto(proto: &StreamCdcScanOptions) -> Self {
356        Self {
357            disable_backfill: proto.disable_backfill,
358            snapshot_barrier_interval: proto.snapshot_barrier_interval,
359            snapshot_batch_size: proto.snapshot_batch_size,
360            backfill_parallelism: proto.backfill_parallelism,
361            backfill_num_rows_per_split: proto.backfill_num_rows_per_split,
362            backfill_as_even_splits: proto.backfill_as_even_splits,
363            backfill_split_pk_column_index: proto.backfill_split_pk_column_index,
364        }
365    }
366
367    pub fn is_parallelized_backfill(&self) -> bool {
368        !self.disable_backfill
369            && self.backfill_num_rows_per_split > 0
370            && self.backfill_parallelism > 0
371    }
372}
373
374#[cfg(test)]
375mod tests {
376    use super::*;
377
378    #[test]
379    fn test_normalize_simple_postgres_quoted_table_name() {
380        assert_eq!(
381            normalize_simple_postgres_quoted_table_name(r#"public."TableName""#).as_deref(),
382            Some("public.TableName")
383        );
384        assert_eq!(
385            normalize_simple_postgres_quoted_table_name(r#"public."pg_my_table""#).as_deref(),
386            Some("public.pg_my_table")
387        );
388        assert_eq!(
389            normalize_simple_postgres_quoted_table_name("public.table"),
390            None
391        );
392        assert_eq!(
393            normalize_simple_postgres_quoted_table_name(r#""Mixed.Schema"."TableName""#),
394            None
395        );
396        assert_eq!(
397            normalize_simple_postgres_quoted_table_name(r#"public."table.name""#),
398            None
399        );
400        assert_eq!(
401            normalize_simple_postgres_quoted_table_name(r#"public."table\name""#),
402            None
403        );
404    }
405}
risingwave_connector/source/cdc/mod.rs

risingwave_connector/source/cdc/
mod.rs
