risingwave_connector/source/cdc/
mongodb.rs

1// Copyright 2026 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::collections::BTreeMap;
16
17use risingwave_pb::plan_common::{PbEncodeType, PbFormatType};
18use risingwave_pb::secret::PbSecretRef;
19
20use crate::error::ConnectorResult;
21use crate::source::cdc::{
22    CdcSourceDdlValidator, CdcSourceRuntimeFlags, CdcSourceTypeTrait, Mongodb,
23    check_alter_with_options_inner, check_common_with_options, check_format_encode_inner,
24    check_include_column_inner,
25};
26
27// ------- Runtime flags -------
28
29impl CdcSourceRuntimeFlags for Mongodb {}
30
31// ------- DDL validation -------
32
33impl CdcSourceDdlValidator for Mongodb {
34    fn check_format_encode(format: PbFormatType, encode: PbEncodeType) -> ConnectorResult<()> {
35        check_format_encode_inner(
36            Self::CDC_CONNECTOR_NAME,
37            format,
38            encode,
39            &[(PbFormatType::DebeziumMongo, &[PbEncodeType::Json])],
40        )
41    }
42
43    fn check_include_column(additional_column_type: &str) -> ConnectorResult<()> {
44        check_include_column_inner(
45            Self::CDC_CONNECTOR_NAME,
46            additional_column_type,
47            &[
48                "timestamp",
49                "partition",
50                "offset",
51                "database_name",
52                "collection_name",
53            ],
54        )
55    }
56
57    fn check_alter_with_options(
58        props: &BTreeMap<String, String>,
59        secret_refs: &BTreeMap<String, PbSecretRef>,
60    ) -> ConnectorResult<()> {
61        check_common_with_options(props)?;
62        check_alter_with_options_inner(
63            Self::CDC_CONNECTOR_NAME,
64            props,
65            secret_refs,
66            &[
67                "cdc.source.wait.streaming.start.timeout",
68                "debezium.max.queue.size",
69                "debezium.queue.memory.ratio",
70            ],
71        )
72    }
73}
risingwave_connector/source/cdc/mongodb.rs

risingwave_connector/source/cdc/
mongodb.rs
