risingwave_connector/
allow_alter_on_fly_fields.rs

1// Copyright 2025 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
15// THIS FILE IS AUTO_GENERATED. DO NOT EDIT
16// UPDATE WITH: ./risedev generate-with-options
17// This file is rewritten by `tests::test_allow_alter_on_fly_fields_rust_up_to_date` with
18// `UPDATE_EXPECT=1`.
19// To update content, change source/sink/connection WITH options definitions (for example,
20// `#[with_option(allow_alter_on_fly)]` on struct fields), then run `./risedev generate-with-options`.
21// `./risedev generate-with-options` runs two UPDATE_EXPECT tests:
22// 1) refresh `with_options_{source,sink,connection}.yaml`;
23// 2) regenerate this file from those YAML files.
24
25#![rustfmt::skip]
26
27use std::collections::{HashMap, HashSet};
28use std::sync::LazyLock;
29use crate::error::ConnectorError;
30use crate::sink::remote::JdbcSink;
31use crate::sink::Sink;
32
33macro_rules! use_source_properties {
34    ({ $({ $variant_name:ident, $prop_name:ty, $split:ty }),* }) => {
35        $(
36            #[allow(unused_imports)]
37            pub(super) use $prop_name;
38        )*
39    };
40}
41
42mod source_properties {
43    use crate::for_all_sources;
44    use crate::source::base::SourceProperties;
45
46    for_all_sources!(use_source_properties);
47
48    /// Implements a function that maps a source name string to the Rust type name of the corresponding property type.
49    /// Usage: `impl_source_name_to_prop_type_name!();` will generate:
50    /// ```ignore
51    /// pub fn source_name_to_prop_type_name(source_name: &str) -> Option<&'static str>
52    /// ```
53    macro_rules! impl_source_name_to_prop_type_name_inner {
54        ({ $({$variant:ident, $prop_name:ty, $split:ty}),* }) => {
55            pub fn source_name_to_prop_type_name(source_name: &str) -> Option<&'static str> {
56                match source_name {
57                    $(
58                        <$prop_name>::SOURCE_NAME => Some(std::any::type_name::<$prop_name>()),
59                    )*
60                    _ => None,
61                }
62            }
63        };
64    }
65
66    macro_rules! impl_source_name_to_prop_type_name {
67        () => {
68            $crate::for_all_sources! { impl_source_name_to_prop_type_name_inner }
69        };
70    }
71
72    impl_source_name_to_prop_type_name!();
73}
74
75mod sink_properties {
76    use crate::use_all_sink_configs;
77    use crate::sink::Sink;
78    use crate::sink::file_sink::fs::FsSink;
79
80    use_all_sink_configs!();
81
82    macro_rules! impl_sink_name_to_config_type_name_inner {
83        ({ $({ $variant_name:ident, $sink_type:ty, $config_type:ty }),* }) => {
84            pub fn sink_name_to_config_type_name(sink_name: &str) -> Option<&'static str> {
85                match sink_name {
86                $(
87                    <$sink_type>::SINK_NAME => Some(std::any::type_name::<$config_type>()),
88                )*
89                    _ => None,
90                }
91            }
92        };
93    }
94
95    macro_rules! impl_sink_name_to_config_type_name {
96        () => {
97            $crate::for_all_sinks! { impl_sink_name_to_config_type_name_inner }
98        };
99    }
100
101    impl_sink_name_to_config_type_name!();
102}
103
104/// Map of source connector names to their `allow_alter_on_fly` field names
105pub static SOURCE_ALLOW_ALTER_ON_FLY_FIELDS: LazyLock<HashMap<String, HashSet<String>>> = LazyLock::new(|| {
106    use source_properties::*;
107    let mut map = HashMap::new();
108    // CDC Properties - added for schema.change.failure.policy
109    map.try_insert(
110        std::any::type_name::<MysqlCdcProperties>().to_owned(),
111        [
112            "cdc.source.wait.streaming.start.timeout".to_owned(),
113            "debezium.max.queue.size".to_owned(),
114            "debezium.queue.memory.ratio".to_owned(),
115            "hostname".to_owned(),
116            "port".to_owned(),
117            "password".to_owned(),
118        ].into_iter().collect(),
119    ).unwrap();
120    map.try_insert(
121        std::any::type_name::<PostgresCdcProperties>().to_owned(),
122        [
123            "cdc.source.wait.streaming.start.timeout".to_owned(),
124            "debezium.max.queue.size".to_owned(),
125            "debezium.queue.memory.ratio".to_owned(),
126            "debezium.heartbeat.interval.ms".to_owned(),
127            "hostname".to_owned(),
128            "port".to_owned(),
129            "password".to_owned(),
130        ].into_iter().collect(),
131    ).unwrap();
132    map.try_insert(
133        std::any::type_name::<SqlServerCdcProperties>().to_owned(),
134        [
135            "cdc.source.wait.streaming.start.timeout".to_owned(),
136            "debezium.max.queue.size".to_owned(),
137            "debezium.queue.memory.ratio".to_owned(),
138            "hostname".to_owned(),
139            "port".to_owned(),
140            "password".to_owned(),
141        ].into_iter().collect(),
142    ).unwrap();
143
144    map.try_insert(
145        std::any::type_name::<MongodbCdcProperties>().to_owned(),
146        [
147            "cdc.source.wait.streaming.start.timeout".to_owned(),
148            "debezium.max.queue.size".to_owned(),
149            "debezium.queue.memory.ratio".to_owned(),
150        ].into_iter().collect(),
151    ).unwrap();
152    // KafkaProperties
153    map.try_insert(
154        std::any::type_name::<KafkaProperties>().to_owned(),
155        [
156            "group.id.prefix".to_owned(),
157            "properties.sync.call.timeout".to_owned(),
158            "properties.security.protocol".to_owned(),
159            "properties.ssl.endpoint.identification.algorithm".to_owned(),
160            "properties.ssl.ca.location".to_owned(),
161            "properties.ssl.ca.pem".to_owned(),
162            "properties.ssl.certificate.location".to_owned(),
163            "properties.ssl.certificate.pem".to_owned(),
164            "properties.ssl.key.location".to_owned(),
165            "properties.ssl.key.pem".to_owned(),
166            "properties.ssl.key.password".to_owned(),
167            "properties.sasl.mechanism".to_owned(),
168            "properties.sasl.username".to_owned(),
169            "properties.sasl.password".to_owned(),
170            "properties.message.max.bytes".to_owned(),
171            "properties.receive.message.max.bytes".to_owned(),
172            "properties.statistics.interval.ms".to_owned(),
173            "properties.client.id".to_owned(),
174            "properties.enable.ssl.certificate.verification".to_owned(),
175            "properties.reconnect.backoff.ms".to_owned(),
176            "properties.reconnect.backoff.max.ms".to_owned(),
177            "properties.socket.connection.setup.timeout.ms".to_owned(),
178            "properties.retry.backoff.ms".to_owned(),
179            "properties.retry.backoff.max.ms".to_owned(),
180            "properties.queued.min.messages".to_owned(),
181            "properties.queued.max.messages.kbytes".to_owned(),
182            "properties.fetch.wait.max.ms".to_owned(),
183            "properties.fetch.queue.backoff.ms".to_owned(),
184            "properties.fetch.max.bytes".to_owned(),
185            "properties.enable.auto.commit".to_owned(),
186            "properties.auto.commit.interval.ms".to_owned(),
187        ].into_iter().collect(),
188    ).unwrap();
189    // PubsubProperties
190    map.try_insert(
191        std::any::type_name::<PubsubProperties>().to_owned(),
192        [
193            "pubsub.ack_deadline_seconds".to_owned(),
194        ].into_iter().collect(),
195    ).unwrap();
196    // PulsarProperties
197    map.try_insert(
198        std::any::type_name::<PulsarProperties>().to_owned(),
199        [
200            "pulsar.operation.retry.max.retries".to_owned(),
201            "pulsar.operation.retry.delay".to_owned(),
202        ].into_iter().collect(),
203    ).unwrap();
204    map
205});
206
207/// Map of sink connector names to their `allow_alter_on_fly` field names
208pub static SINK_ALLOW_ALTER_ON_FLY_FIELDS: LazyLock<HashMap<String, HashSet<String>>> = LazyLock::new(|| {
209    use sink_properties::*;
210    let mut map = HashMap::new();
211    // ClickHouseConfig
212    map.try_insert(
213        std::any::type_name::<ClickHouseConfig>().to_owned(),
214        [
215            "commit_checkpoint_interval".to_owned(),
216        ].into_iter().collect(),
217    ).unwrap();
218    // DeltaLakeConfig
219    map.try_insert(
220        std::any::type_name::<DeltaLakeConfig>().to_owned(),
221        [
222            "commit_checkpoint_interval".to_owned(),
223        ].into_iter().collect(),
224    ).unwrap();
225    // DorisConfig
226    map.try_insert(
227        std::any::type_name::<DorisConfig>().to_owned(),
228        [
229            "doris.stream_load.http.timeout.ms".to_owned(),
230        ].into_iter().collect(),
231    ).unwrap();
232    // IcebergConfig
233    map.try_insert(
234        std::any::type_name::<IcebergConfig>().to_owned(),
235        [
236            "commit_checkpoint_interval".to_owned(),
237            "enable_compaction".to_owned(),
238            "compaction_interval_sec".to_owned(),
239            "enable_snapshot_expiration".to_owned(),
240            "snapshot_expiration_max_age_millis".to_owned(),
241            "snapshot_expiration_retain_last".to_owned(),
242            "snapshot_expiration_clear_expired_files".to_owned(),
243            "snapshot_expiration_clear_expired_meta_data".to_owned(),
244            "compaction.max_snapshots_num".to_owned(),
245            "compaction.small_files_threshold_mb".to_owned(),
246            "compaction.delete_files_count_threshold".to_owned(),
247            "compaction.trigger_snapshot_count".to_owned(),
248            "compaction.target_file_size_mb".to_owned(),
249            "compaction.type".to_owned(),
250            "compaction.write_parquet_compression".to_owned(),
251            "compaction.write_parquet_max_row_group_rows".to_owned(),
252            "compaction.write_parquet_max_row_group_bytes".to_owned(),
253        ].into_iter().collect(),
254    ).unwrap();
255    // KafkaConfig
256    map.try_insert(
257        std::any::type_name::<KafkaConfig>().to_owned(),
258        [
259            "properties.sync.call.timeout".to_owned(),
260            "properties.security.protocol".to_owned(),
261            "properties.ssl.endpoint.identification.algorithm".to_owned(),
262            "properties.ssl.ca.location".to_owned(),
263            "properties.ssl.ca.pem".to_owned(),
264            "properties.ssl.certificate.location".to_owned(),
265            "properties.ssl.certificate.pem".to_owned(),
266            "properties.ssl.key.location".to_owned(),
267            "properties.ssl.key.pem".to_owned(),
268            "properties.ssl.key.password".to_owned(),
269            "properties.sasl.mechanism".to_owned(),
270            "properties.sasl.username".to_owned(),
271            "properties.sasl.password".to_owned(),
272            "properties.message.max.bytes".to_owned(),
273            "properties.receive.message.max.bytes".to_owned(),
274            "properties.statistics.interval.ms".to_owned(),
275            "properties.client.id".to_owned(),
276            "properties.enable.ssl.certificate.verification".to_owned(),
277            "properties.reconnect.backoff.ms".to_owned(),
278            "properties.reconnect.backoff.max.ms".to_owned(),
279            "properties.socket.connection.setup.timeout.ms".to_owned(),
280            "properties.retry.backoff.ms".to_owned(),
281            "properties.retry.backoff.max.ms".to_owned(),
282            "properties.allow.auto.create.topics".to_owned(),
283            "properties.queue.buffering.max.messages".to_owned(),
284            "properties.queue.buffering.max.kbytes".to_owned(),
285            "properties.queue.buffering.max.ms".to_owned(),
286            "properties.enable.idempotence".to_owned(),
287            "properties.message.send.max.retries".to_owned(),
288            "properties.batch.num.messages".to_owned(),
289            "properties.batch.size".to_owned(),
290            "properties.message.timeout.ms".to_owned(),
291            "properties.max.in.flight.requests.per.connection".to_owned(),
292            "properties.request.required.acks".to_owned(),
293        ].into_iter().collect(),
294    ).unwrap();
295    // PulsarConfig
296    map.try_insert(
297        std::any::type_name::<PulsarConfig>().to_owned(),
298        [
299            "properties.routing.mode".to_owned(),
300            "properties.routing_mode".to_owned(),
301            "routing_mode".to_owned(),
302            "pulsar.routing_mode".to_owned(),
303            "pulsar.routing.mode".to_owned(),
304            "pulsar.properties.routing.mode".to_owned(),
305            "pulsar.properties.routing_mode".to_owned(),
306        ].into_iter().collect(),
307    ).unwrap();
308    // SnowflakeV2Config
309    map.try_insert(
310        std::any::type_name::<SnowflakeV2Config>().to_owned(),
311        [
312            "commit_checkpoint_interval".to_owned(),
313        ].into_iter().collect(),
314    ).unwrap();
315    // StarrocksConfig
316    map.try_insert(
317        std::any::type_name::<StarrocksConfig>().to_owned(),
318        [
319            "starrocks.stream_load.http.timeout.ms".to_owned(),
320            "commit_checkpoint_interval".to_owned(),
321            "starrocks.max_batch_size_bytes".to_owned(),
322        ].into_iter().collect(),
323    ).unwrap();
324    // TurbopufferConfig
325    map.try_insert(
326        std::any::type_name::<TurbopufferConfig>().to_owned(),
327        [
328            "write_batch_size".to_owned(),
329            "max_linger_second".to_owned(),
330        ].into_iter().collect(),
331    ).unwrap();
332    // Jdbc
333    map.try_insert(
334        JdbcSink::SINK_NAME.to_owned(),
335        [
336            "jdbc.url".to_owned(),
337            "user".to_owned(),
338            "password".to_owned(),
339        ].into_iter().collect(),
340    ).unwrap();
341    map
342});
343
344/// Map of connection names to their `allow_alter_on_fly` field names
345pub static CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS: LazyLock<HashMap<String, HashSet<String>>> = LazyLock::new(|| {
346    use crate::connector_common::*;
347    let mut map = HashMap::new();
348    // KafkaConnection
349    map.try_insert(
350        std::any::type_name::<KafkaConnection>().to_owned(),
351        [
352            "properties.security.protocol".to_owned(),
353            "properties.ssl.endpoint.identification.algorithm".to_owned(),
354            "properties.ssl.ca.location".to_owned(),
355            "properties.ssl.ca.pem".to_owned(),
356            "properties.ssl.certificate.location".to_owned(),
357            "properties.ssl.certificate.pem".to_owned(),
358            "properties.ssl.key.location".to_owned(),
359            "properties.ssl.key.pem".to_owned(),
360            "properties.ssl.key.password".to_owned(),
361            "properties.sasl.mechanism".to_owned(),
362            "properties.sasl.username".to_owned(),
363            "properties.sasl.password".to_owned(),
364        ].into_iter().collect(),
365    ).unwrap();
366    // Jdbc
367    map.try_insert(
368        JdbcSink::SINK_NAME.to_owned(),
369        [
370            "jdbc.url".to_owned(),
371            "user".to_owned(),
372            "password".to_owned(),
373        ].into_iter().collect(),
374    ).unwrap();
375    map
376});
377
378/// Get all source connector names that have `allow_alter_on_fly` fields
379pub fn get_source_connectors_with_allow_alter_on_fly_fields() -> Vec<&'static str> {
380    SOURCE_ALLOW_ALTER_ON_FLY_FIELDS.keys().map(|s| s.as_str()).collect()
381}
382
383/// Get all sink connector names that have `allow_alter_on_fly` fields
384pub fn get_sink_connectors_with_allow_alter_on_fly_fields() -> Vec<&'static str> {
385    SINK_ALLOW_ALTER_ON_FLY_FIELDS.keys().map(|s| s.as_str()).collect()
386}
387
388/// Get all connection names that have `allow_alter_on_fly` fields
389pub fn get_connection_names_with_allow_alter_on_fly_fields() -> Vec<&'static str> {
390    CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS.keys().map(|s| s.as_str()).collect()
391}
392
393/// Checks if all given fields are allowed to be altered on the fly for the specified source connector.
394/// Returns Ok(()) if all fields are allowed, otherwise returns a `ConnectorError`.
395pub fn check_source_allow_alter_on_fly_fields(
396    connector_name: &str,
397    fields: &[String],
398) -> crate::error::ConnectorResult<()> {
399    // Convert connector name to the type name key
400    let Some(type_name) = source_properties::source_name_to_prop_type_name(connector_name) else {
401        return Err(ConnectorError::from(anyhow::anyhow!(
402            "Unknown source connector: {connector_name}"
403        )));
404    };
405    let Some(allowed_fields) = SOURCE_ALLOW_ALTER_ON_FLY_FIELDS.get(type_name) else {
406    return Err(ConnectorError::from(anyhow::anyhow!(
407        "No allow_alter_on_fly fields registered for connector: {connector_name}"
408    )));
409    };
410    for field in fields {
411        if !allowed_fields.contains(field) {
412            return Err(ConnectorError::from(anyhow::anyhow!(
413                "Field '{field}' is not allowed to be altered on the fly for connector: {connector_name}"
414            )));
415        }
416    }
417    Ok(())
418}
419
420pub fn check_connection_allow_alter_on_fly_fields(
421    connection_name: &str,
422    fields: &[String],
423) -> crate::error::ConnectorResult<()> {
424    use crate::source::connection_name_to_prop_type_name;
425
426    // Convert connection name to the type name key
427    let Some(type_name) = connection_name_to_prop_type_name(connection_name) else {
428        return Err(ConnectorError::from(anyhow::anyhow!(
429            "Unknown connection: {connection_name}"
430        )));
431    };
432    let Some(allowed_fields) = CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS.get(type_name) else {
433        return Err(ConnectorError::from(anyhow::anyhow!(
434            "No allow_alter_on_fly fields registered for connection: {connection_name}"
435        )));
436    };
437    for field in fields {
438        if !allowed_fields.contains(field) {
439            return Err(ConnectorError::from(anyhow::anyhow!(
440                "Field '{field}' is not allowed to be altered on the fly for connection: {connection_name}"
441            )));
442        }
443    }
444    Ok(())
445}
446
447/// Checks if all given fields are allowed to be altered on the fly for the specified sink connector.
448/// Returns Ok(()) if all fields are allowed, otherwise returns a `ConnectorError`.
449pub fn check_sink_allow_alter_on_fly_fields(
450    sink_name: &str,
451    fields: &[String],
452) -> crate::error::ConnectorResult<()> {
453    // TODO(#24846): JDBC sink currently uses `()` as sink config type in `for_all_sinks!`,
454    // so it cannot have an isolated key in `SINK_ALLOW_ALTER_ON_FLY_FIELDS`.
455    // Reuse the JDBC entry in `CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS` for now.
456    // TODO(#24846): remove this special case after JDBC sink has a dedicated config type
457    // and allow-alter fields are generated directly into `SINK_ALLOW_ALTER_ON_FLY_FIELDS`.
458    let allowed_fields = if sink_name == JdbcSink::SINK_NAME {
459        CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS.get(JdbcSink::SINK_NAME)
460    } else {
461        // Convert sink name to the type name key
462        let Some(type_name) = sink_properties::sink_name_to_config_type_name(sink_name) else {
463            return Err(ConnectorError::from(anyhow::anyhow!(
464                "Unknown sink connector: {sink_name}"
465            )));
466        };
467        SINK_ALLOW_ALTER_ON_FLY_FIELDS.get(type_name)
468    };
469    let Some(allowed_fields) = allowed_fields else {
470        return Err(ConnectorError::from(anyhow::anyhow!(
471            "No allow_alter_on_fly fields registered for sink: {sink_name}"
472        )));
473    };
474    for field in fields {
475        if !allowed_fields.contains(field) {
476            return Err(ConnectorError::from(anyhow::anyhow!(
477                "Field '{field}' is not allowed to be altered on the fly for sink: {sink_name}"
478            )));
479        }
480    }
481    Ok(())
482}
risingwave_connector/allow_alter_on_fly_fields.rs

risingwave_connector/
allow_alter_on_fly_fields.rs
