risingwave_connector/source/cdc/
sql_server.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_common::license::{Feature, FeatureNotAvailable};
18use risingwave_pb::secret::PbSecretRef;
19
20use crate::error::ConnectorResult;
21use crate::source::cdc::{
22    CdcSourceDdlValidator, CdcSourceRuntimeFlags, CdcSourceTypeTrait, SqlServer,
23    check_alter_with_options_inner, check_common_with_options,
24};
25
26// ------- Runtime flags -------
27
28impl CdcSourceRuntimeFlags for SqlServer {
29    const CAN_BACKFILL: bool = true;
30    const NEEDS_NON_EMPTY_INITIAL_OFFSET: bool = true;
31    // `CAN_USE_TRANSACTION_METADATA` stays false: Debezium only sends a transaction's `END` event
32    // when the next transaction starts, which delays the newest events.
33    // See https://debezium.io/documentation/reference/2.6/connectors/sqlserver.html#sqlserver-transaction-metadata
34}
35
36// ------- DDL validation -------
37
38impl CdcSourceDdlValidator for SqlServer {
39    fn check_and_fill_create_with_options(
40        props: &mut BTreeMap<String, String>,
41    ) -> ConnectorResult<()> {
42        check_common_with_options(props)?;
43        if !props.contains_key("schema.name") {
44            // Default schema name is "dbo"
45            props.insert("schema.name".into(), "dbo".into());
46        }
47        Ok(())
48    }
49
50    fn check_license() -> Result<(), FeatureNotAvailable> {
51        Feature::SqlServerCdcSource.check_available()
52    }
53
54    fn check_alter_with_options(
55        props: &BTreeMap<String, String>,
56        secret_refs: &BTreeMap<String, PbSecretRef>,
57    ) -> ConnectorResult<()> {
58        check_common_with_options(props)?;
59        check_alter_with_options_inner(
60            Self::CDC_CONNECTOR_NAME,
61            props,
62            secret_refs,
63            &[
64                "cdc.source.wait.streaming.start.timeout",
65                "debezium.max.queue.size",
66                "debezium.queue.memory.ratio",
67                "hostname",
68                "port",
69                "password",
70            ],
71        )
72    }
73}
risingwave_connector/source/cdc/sql_server.rs

risingwave_connector/source/cdc/
sql_server.rs
