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 mongodb;
19pub mod mysql;
20pub mod oracle;
21pub mod postgres;
22pub mod source;
23pub mod split;
24pub mod sql_server;
25
26use std::collections::{BTreeMap, HashMap};
27use std::marker::PhantomData;
28
29pub use enumerator::*;
30use itertools::Itertools;
31use risingwave_common::bail;
32use risingwave_common::id::{ActorId, SourceId};
33use risingwave_common::license::FeatureNotAvailable;
34use risingwave_pb::catalog::PbSource;
35use risingwave_pb::connector_service::{PbSourceType, PbTableSchema, SourceType, TableSchema};
36use risingwave_pb::plan_common::column_desc::GeneratedOrDefaultColumn;
37use risingwave_pb::plan_common::{ExternalTableDesc, PbEncodeType, PbFormatType};
38use risingwave_pb::secret::PbSecretRef;
39use risingwave_pb::source::{PbCdcTableSnapshotSplit, PbCdcTableSnapshotSplitsWithGeneration};
40use risingwave_pb::stream_plan::StreamCdcScanOptions;
41use simd_json::prelude::ArrayTrait;
42pub use source::*;
43
44use crate::enforce_secret::EnforceSecret;
45use crate::error::ConnectorResult;
46use crate::source::{CdcTableSnapshotSplitRaw, SourceProperties, SplitImpl, TryFromBTreeMap};
47use crate::{for_all_classified_sources, impl_cdc_source_type};
48
49pub const CDC_CONNECTOR_NAME_SUFFIX: &str = "-cdc";
50pub const CDC_SNAPSHOT_MODE_KEY: &str = "debezium.snapshot.mode";
51pub const CDC_SNAPSHOT_BACKFILL: &str = "rw_cdc_backfill";
52pub const CDC_SHARING_MODE_KEY: &str = "rw.sharing.mode.enable";
53// User can set snapshot='false' to disable cdc backfill
54pub const CDC_BACKFILL_ENABLE_KEY: &str = "snapshot";
55pub const CDC_BACKFILL_SNAPSHOT_INTERVAL_KEY: &str = "snapshot.interval";
56pub const CDC_BACKFILL_SNAPSHOT_BATCH_SIZE_KEY: &str = "snapshot.batch_size";
57pub const CDC_BACKFILL_PARALLELISM: &str = "backfill.parallelism";
58pub const CDC_BACKFILL_NUM_ROWS_PER_SPLIT: &str = "backfill.num_rows_per_split";
59pub const CDC_BACKFILL_AS_EVEN_SPLITS: &str = "backfill.as_even_splits";
60pub const CDC_BACKFILL_SPLIT_PK_COLUMN_INDEX: &str = "backfill.split_pk_column_index";
61// We enable transaction for shared cdc source by default
62pub const CDC_TRANSACTIONAL_KEY: &str = "transactional";
63pub const CDC_WAIT_FOR_STREAMING_START_TIMEOUT: &str = "cdc.source.wait.streaming.start.timeout";
64pub const CDC_BACKFILL_MAX_PARALLELISM: u32 = 256;
65
66// User can set strong-schema='true' to enable strong schema for mongo cdc source
67pub const CDC_MONGODB_STRONG_SCHEMA_KEY: &str = "strong_schema";
68
69pub const MYSQL_CDC_CONNECTOR: &str = Mysql::CDC_CONNECTOR_NAME;
70pub const POSTGRES_CDC_CONNECTOR: &str = Postgres::CDC_CONNECTOR_NAME;
71pub const MONGODB_CDC_CONNECTOR: &str = Mongodb::CDC_CONNECTOR_NAME;
72pub const SQL_SERVER_CDC_CONNECTOR: &str = SqlServer::CDC_CONNECTOR_NAME;
73pub const ORACLE_CDC_CONNECTOR: &str = Oracle::CDC_CONNECTOR_NAME;
74
75/// Build a unique CDC table identifier from a source ID and external table name
76pub fn build_cdc_table_id(source_id: SourceId, external_table_name: &str) -> String {
77    format!("{}.{}", source_id, external_table_name)
78}
79
80pub fn normalize_simple_postgres_quoted_table_name(table_name: &str) -> Option<String> {
81    let (schema_name, table_name) = table_name.split_once('.')?;
82    let table_name = table_name.strip_prefix('"')?.strip_suffix('"')?;
83    if schema_name.is_empty()
84        || table_name.is_empty()
85        || table_name.contains('.')
86        || table_name.contains('"')
87        || table_name.contains('\\')
88        || schema_name.contains('"')
89    {
90        return None;
91    }
92
93    Some(format!("{schema_name}.{table_name}"))
94}
95
96pub trait CdcSourceTypeTrait: Send + Sync + Clone + std::fmt::Debug + 'static {
97    const CDC_CONNECTOR_NAME: &'static str;
98    fn source_type() -> CdcSourceType;
99}
100
101/// Behavior switches.
102pub trait CdcSourceRuntimeFlags: CdcSourceTypeTrait {
103    /// When true, only shared sources are supported; otherwise only direct sources are supported.
104    const CAN_BACKFILL: bool = false;
105    /// When true, the planner may choose parallelized backfill.
106    const CAN_BACKFILL_IN_PARALLEL: bool = false;
107    /// When true, a shared source sets Debezium's `provide.transaction.metadata`, so that all rows
108    /// of one upstream transaction go into a single chunk.
109    const CAN_USE_TRANSACTION_METADATA: bool = false;
110    /// When true, the source executor waits for a non-empty split offset before reporting the
111    /// initial CDC offset.
112    const NEEDS_NON_EMPTY_INITIAL_OFFSET: bool = false;
113    /// When true, the consumed offset is committed back to the upstream database on each
114    /// checkpoint via a JNI call.
115    const NEEDS_OFFSET_COMMIT_ON_CHECKPOINT: bool = false;
116}
117
118/// Checks the `CREATE` and `ALTER` statements of a CDC source connector.
119pub trait CdcSourceDdlValidator: CdcSourceTypeTrait {
120    /// Checks the `FORMAT ... ENCODE ...` clause of a `CREATE` statement.
121    fn check_format_encode(format: PbFormatType, encode: PbEncodeType) -> ConnectorResult<()> {
122        check_format_encode_inner(
123            Self::CDC_CONNECTOR_NAME,
124            format,
125            encode,
126            &[
127                (PbFormatType::Debezium, &[PbEncodeType::Json]),
128                // support source stream job
129                (PbFormatType::Plain, &[PbEncodeType::Json]),
130            ],
131        )
132    }
133
134    /// Checks an additional column type in the `INCLUDE` clause of a `CREATE` statement, such as
135    /// `timestamp` in `INCLUDE timestamp AS ts`.
136    fn check_include_column(additional_column_type: &str) -> ConnectorResult<()> {
137        bail!(
138            "additional column type {} is not supported for connector {}, which accepts no additional column",
139            additional_column_type,
140            Self::CDC_CONNECTOR_NAME
141        )
142    }
143
144    /// Checks the `WITH (...)` options of a `CREATE` statement, and fills in the options that the
145    /// CDC source generates when they are omitted, such as a random `slot.name` for Postgres.
146    /// The connector node later renders the filled options into the Debezium config through the
147    /// `.properties` template of the CDC source. An override calls [`check_common_with_options`]
148    /// first.
149    fn check_and_fill_create_with_options(
150        props: &mut BTreeMap<String, String>,
151    ) -> ConnectorResult<()> {
152        check_common_with_options(props)
153    }
154
155    /// Checks that the license allows the CDC source.
156    fn check_license() -> Result<(), FeatureNotAvailable> {
157        Ok(())
158    }
159
160    /// Checks that `ALTER SOURCE ... CONNECTOR WITH (...)` only changes options that a running
161    /// CDC source can apply. An override calls [`check_common_with_options`] first.
162    fn check_alter_with_options(
163        props: &BTreeMap<String, String>,
164        _secret_refs: &BTreeMap<String, PbSecretRef>,
165    ) -> ConnectorResult<()> {
166        check_common_with_options(props)
167    }
168}
169
170/// Checks the `WITH (...)` options that every CDC source accepts, on both `CREATE` and `ALTER`.
171pub fn check_common_with_options(props: &BTreeMap<String, String>) -> ConnectorResult<()> {
172    // RisingWave consumes schema-less JSON from Debezium and cannot reconstruct the scale of
173    // binary logical decimals emitted by `precise`. `double` can lose precision, so an explicit
174    // override must retain the common `string` default from `debezium.properties`.
175    if let Some(mode) = props.get("debezium.decimal.handling.mode")
176        && mode != "string"
177    {
178        bail!("'debezium.decimal.handling.mode' must be 'string', got: '{mode}'");
179    }
180    for key in [
181        CDC_WAIT_FOR_STREAMING_START_TIMEOUT,
182        "debezium.max.queue.size",
183    ] {
184        if let Some(value) = props.get(key)
185            && value.parse::<u32>().is_err()
186        {
187            bail!("Invalid value `{value}` for `{key}`");
188        }
189    }
190    // Keep this policy in sync with Java's `SourceValidateHandler.validateHeartbeatInterval`:
191    // match Java's ASCII integer syntax and Debezium's signed 32-bit range.
192    if let Some(value) = props.get("debezium.heartbeat.interval.ms")
193        && !value.parse::<i32>().is_ok_and(|interval| interval > 0)
194    {
195        bail!("'debezium.heartbeat.interval.ms' must be a positive integer, got: '{value}'");
196    }
197    Ok(())
198}
199
200pub fn check_format_encode_inner(
201    connector_name: &str,
202    format: PbFormatType,
203    encode: PbEncodeType,
204    compatible_formats: &[(PbFormatType, &[PbEncodeType])],
205) -> ConnectorResult<()> {
206    let Some((_, compatible_encodes)) = compatible_formats.iter().find(|(f, _)| *f == format)
207    else {
208        bail!(
209            "connector {} does not support format {:?}",
210            connector_name,
211            format
212        );
213    };
214    if !compatible_encodes.contains(&encode) {
215        bail!(
216            "connector {} does not support format {:?} with encode {:?}",
217            connector_name,
218            format,
219            encode
220        );
221    }
222    Ok(())
223}
224
225pub fn check_include_column_inner(
226    connector_name: &str,
227    additional_column_type: &str,
228    compatible_additional_column_types: &[&str],
229) -> ConnectorResult<()> {
230    if !compatible_additional_column_types.contains(&additional_column_type) {
231        bail!(
232            "additional column type {} is not supported for connector {}, acceptable column types: {:?}",
233            additional_column_type,
234            connector_name,
235            compatible_additional_column_types
236        );
237    }
238    Ok(())
239}
240
241pub fn check_alter_with_options_inner(
242    connector_name: &str,
243    props: &BTreeMap<String, String>,
244    secret_refs: &BTreeMap<String, PbSecretRef>,
245    allowed_keys: &[&str],
246) -> ConnectorResult<()> {
247    if let Some(key) = props
248        .keys()
249        .chain(secret_refs.keys())
250        .find(|key| !allowed_keys.contains(&key.as_str()))
251    {
252        bail!(
253            "Field '{}' is not allowed to be altered on the fly for connector: {}",
254            key,
255            connector_name
256        );
257    }
258    Ok(())
259}
260
261for_all_classified_sources!(impl_cdc_source_type);
262
263impl<'a> From<&'a str> for CdcSourceType {
264    fn from(name: &'a str) -> Self {
265        match name {
266            MYSQL_CDC_CONNECTOR => CdcSourceType::Mysql,
267            POSTGRES_CDC_CONNECTOR => CdcSourceType::Postgres,
268            MONGODB_CDC_CONNECTOR => CdcSourceType::Mongodb,
269            SQL_SERVER_CDC_CONNECTOR => CdcSourceType::SqlServer,
270            ORACLE_CDC_CONNECTOR => CdcSourceType::Oracle,
271            _ => CdcSourceType::Unspecified,
272        }
273    }
274}
275
276impl CdcSourceType {
277    pub fn as_str_name(&self) -> &str {
278        match self {
279            CdcSourceType::Mysql => "MySQL",
280            CdcSourceType::Postgres => "Postgres",
281            CdcSourceType::Mongodb => "MongoDB",
282            CdcSourceType::SqlServer => "SQL Server",
283            CdcSourceType::Oracle => "Oracle",
284            CdcSourceType::Unspecified => "Unspecified",
285        }
286    }
287}
288
289#[derive(Clone, Debug, Default)]
290pub struct CdcProperties<T: CdcSourceTypeTrait> {
291    /// Properties specified in the WITH clause by user
292    pub properties: BTreeMap<String, String>,
293
294    /// Schema of the source specified by users
295    pub table_schema: TableSchema,
296
297    /// Whether it is created by a cdc source job
298    pub is_cdc_source_job: bool,
299
300    /// For validation purpose, mark if the table is a backfill cdc table
301    pub is_backfill_table: bool,
302
303    pub _phantom: PhantomData<T>,
304}
305
306pub fn table_schema_exclude_additional_columns(table_schema: &TableSchema) -> TableSchema {
307    TableSchema {
308        columns: table_schema
309            .columns
310            .iter()
311            .filter(|col| {
312                col.additional_column
313                    .as_ref()
314                    .is_some_and(|val| val.column_type.is_none())
315            })
316            .cloned()
317            .collect(),
318        pk_indices: table_schema.pk_indices.clone(),
319    }
320}
321
322impl<T: CdcSourceTypeTrait> TryFromBTreeMap for CdcProperties<T> {
323    fn try_from_btreemap(
324        properties: BTreeMap<String, String>,
325        _deny_unknown_fields: bool,
326    ) -> ConnectorResult<Self> {
327        let is_share_source: bool = properties
328            .get(CDC_SHARING_MODE_KEY)
329            .is_some_and(|v| v == "true");
330        Ok(CdcProperties {
331            properties,
332            table_schema: Default::default(),
333            // TODO(siyuan): use serde to deserialize input hashmap
334            is_cdc_source_job: is_share_source,
335            is_backfill_table: false,
336            _phantom: PhantomData,
337        })
338    }
339}
340
341impl<T: CdcSourceTypeTrait> EnforceSecret for CdcProperties<T> {} // todo: enforce jdbc like properties
342
343impl<T: CdcSourceTypeTrait> SourceProperties for CdcProperties<T>
344where
345    DebeziumCdcSplit<T>: TryFrom<SplitImpl, Error = crate::error::ConnectorError> + Into<SplitImpl>,
346    DebeziumSplitEnumerator<T>: ListCdcSplits<CdcSourceType = T> + enumerator::CdcMonitor,
347{
348    type Split = DebeziumCdcSplit<T>;
349    type SplitEnumerator = DebeziumSplitEnumerator<T>;
350    type SplitReader = CdcSplitReader<T>;
351
352    const SOURCE_NAME: &'static str = T::CDC_CONNECTOR_NAME;
353
354    fn init_from_pb_source(&mut self, source: &PbSource) {
355        let pk_indices = source
356            .pk_column_ids
357            .iter()
358            .map(|&id| {
359                source
360                    .columns
361                    .iter()
362                    .position(|col| col.column_desc.as_ref().unwrap().column_id == id)
363                    .unwrap() as u32
364            })
365            .collect_vec();
366
367        let table_schema = PbTableSchema {
368            columns: source
369                .columns
370                .iter()
371                .flat_map(|col| &col.column_desc)
372                .filter(|col| {
373                    !matches!(
374                        col.generated_or_default_column,
375                        Some(GeneratedOrDefaultColumn::GeneratedColumn(_))
376                    )
377                })
378                .cloned()
379                .collect(),
380            pk_indices,
381        };
382        self.table_schema = table_schema;
383        if let Some(info) = source.info.as_ref() {
384            self.is_cdc_source_job = info.is_shared();
385        }
386    }
387
388    fn init_from_pb_cdc_table_desc(&mut self, table_desc: &ExternalTableDesc) {
389        let table_schema = TableSchema {
390            columns: table_desc
391                .columns
392                .iter()
393                .filter(|col| {
394                    !matches!(
395                        col.generated_or_default_column,
396                        Some(GeneratedOrDefaultColumn::GeneratedColumn(_))
397                    )
398                })
399                .cloned()
400                .collect(),
401            pk_indices: table_desc.stream_key.clone(),
402        };
403
404        self.table_schema = table_schema;
405        self.is_cdc_source_job = false;
406        self.is_backfill_table = true;
407    }
408}
409
410impl<T: CdcSourceTypeTrait> crate::source::UnknownFields for CdcProperties<T> {
411    fn unknown_fields(&self) -> HashMap<String, String> {
412        // FIXME: CDC does not handle unknown fields yet
413        HashMap::new()
414    }
415}
416
417impl<T: CdcSourceTypeTrait> CdcProperties<T> {
418    pub fn get_source_type_pb(&self) -> SourceType {
419        SourceType::from(T::source_type())
420    }
421}
422
423pub const INVALID_CDC_SPLIT_ASSIGNMENT_GENERATION_ID: u64 = 0;
424pub const INITIAL_CDC_SPLIT_ASSIGNMENT_GENERATION_ID: u64 = 1;
425
426#[derive(Clone, Debug, PartialEq, Default)]
427pub struct CdcTableSnapshotSplitAssignmentWithGeneration {
428    pub splits: HashMap<ActorId, (Vec<CdcTableSnapshotSplitRaw>, u64)>,
429}
430
431impl CdcTableSnapshotSplitAssignmentWithGeneration {
432    pub fn new(splits: HashMap<ActorId, (Vec<CdcTableSnapshotSplitRaw>, u64)>) -> Self {
433        Self { splits }
434    }
435
436    pub fn empty() -> Self {
437        Self {
438            splits: HashMap::default(),
439        }
440    }
441}
442
443pub fn build_cdc_table_snapshot_split(s: &CdcTableSnapshotSplitRaw) -> PbCdcTableSnapshotSplit {
444    PbCdcTableSnapshotSplit {
445        split_id: s.split_id,
446        left_bound_inclusive: s.left_bound_inclusive.clone(),
447        right_bound_exclusive: s.right_bound_exclusive.clone(),
448    }
449}
450
451pub fn build_actor_cdc_table_snapshot_splits_with_generation(
452    pb_cdc_table_snapshot_split_assignment: PbCdcTableSnapshotSplitsWithGeneration,
453) -> CdcTableSnapshotSplitAssignmentWithGeneration {
454    let splits = pb_cdc_table_snapshot_split_assignment
455        .splits
456        .into_iter()
457        .map(|(actor_id, splits)| {
458            let generation = splits.generation;
459            let splits = splits
460                .splits
461                .into_iter()
462                .map(|s| CdcTableSnapshotSplitRaw {
463                    split_id: s.split_id,
464                    left_bound_inclusive: s.left_bound_inclusive,
465                    right_bound_exclusive: s.right_bound_exclusive,
466                })
467                .collect();
468            (actor_id, (splits, generation))
469        })
470        .collect();
471    CdcTableSnapshotSplitAssignmentWithGeneration { splits }
472}
473
474#[derive(Debug, Clone, Hash, PartialEq)]
475pub struct CdcScanOptions {
476    /// Used by Used in non-parallel backfill, i.e. backfill V1.
477    pub disable_backfill: bool,
478    /// Used by non-parallelized backfill. The frequency of snapshot read resets for consuming the WAL backlog.
479    pub snapshot_barrier_interval: u32,
480    /// Used by non-parallelized backfill. The number of rows to fetch in a single batch when reading from an external table.
481    pub snapshot_batch_size: u32,
482    /// Used by parallelized backfill, i.e. backfill V2. The initial parallelism of parallel backfill.
483    pub backfill_parallelism: u32,
484    /// Used by parallelized backfill. The estimated number of rows per split used in splits generation.
485    pub backfill_num_rows_per_split: u64,
486    /// Used by parallelized backfill. For supported split column data type, assume an uniform distribution and adopt a much faster splits generation method.
487    pub backfill_as_even_splits: bool,
488    /// Used by parallelized backfill. Specify the index of primary key column to use as split column.
489    pub backfill_split_pk_column_index: u32,
490}
491
492impl Default for CdcScanOptions {
493    fn default() -> Self {
494        Self {
495            disable_backfill: false,
496            snapshot_barrier_interval: 10,
497            snapshot_batch_size: 1000,
498            // 0 means disable backfill v2.
499            backfill_parallelism: 0,
500            backfill_num_rows_per_split: 100_000,
501            backfill_as_even_splits: true,
502            backfill_split_pk_column_index: 0,
503        }
504    }
505}
506
507impl CdcScanOptions {
508    pub fn to_proto(&self) -> StreamCdcScanOptions {
509        StreamCdcScanOptions {
510            disable_backfill: self.disable_backfill,
511            snapshot_barrier_interval: self.snapshot_barrier_interval,
512            snapshot_batch_size: self.snapshot_batch_size,
513            backfill_parallelism: self.backfill_parallelism,
514            backfill_num_rows_per_split: self.backfill_num_rows_per_split,
515            backfill_as_even_splits: self.backfill_as_even_splits,
516            backfill_split_pk_column_index: self.backfill_split_pk_column_index,
517        }
518    }
519
520    pub fn from_proto(proto: &StreamCdcScanOptions) -> Self {
521        Self {
522            disable_backfill: proto.disable_backfill,
523            snapshot_barrier_interval: proto.snapshot_barrier_interval,
524            snapshot_batch_size: proto.snapshot_batch_size,
525            backfill_parallelism: proto.backfill_parallelism,
526            backfill_num_rows_per_split: proto.backfill_num_rows_per_split,
527            backfill_as_even_splits: proto.backfill_as_even_splits,
528            backfill_split_pk_column_index: proto.backfill_split_pk_column_index,
529        }
530    }
531
532    pub fn is_parallelized_backfill(&self) -> bool {
533        !self.disable_backfill
534            && self.backfill_num_rows_per_split > 0
535            && self.backfill_parallelism > 0
536    }
537}
538
539#[cfg(test)]
540mod tests {
541    use super::*;
542
543    #[test]
544    fn test_normalize_simple_postgres_quoted_table_name() {
545        assert_eq!(
546            normalize_simple_postgres_quoted_table_name(r#"public."TableName""#).as_deref(),
547            Some("public.TableName")
548        );
549        assert_eq!(
550            normalize_simple_postgres_quoted_table_name(r#"public."pg_my_table""#).as_deref(),
551            Some("public.pg_my_table")
552        );
553        assert_eq!(
554            normalize_simple_postgres_quoted_table_name("public.table"),
555            None
556        );
557        assert_eq!(
558            normalize_simple_postgres_quoted_table_name(r#""Mixed.Schema"."TableName""#),
559            None
560        );
561        assert_eq!(
562            normalize_simple_postgres_quoted_table_name(r#"public."table.name""#),
563            None
564        );
565        assert_eq!(
566            normalize_simple_postgres_quoted_table_name(r#"public."table\name""#),
567            None
568        );
569    }
570}
risingwave_connector/source/cdc/mod.rs

risingwave_connector/source/cdc/
mod.rs
