1use std::collections::{HashMap, HashSet};
19use std::sync::LazyLock;
20use crate::error::ConnectorError;
21
22macro_rules! use_source_properties {
23 ({ $({ $variant_name:ident, $prop_name:ty, $split:ty }),* }) => {
24 $(
25 #[allow(unused_imports)]
26 pub(super) use $prop_name;
27 )*
28 };
29}
30
31mod source_properties {
32 use crate::for_all_sources;
33 use crate::source::base::SourceProperties;
34
35 for_all_sources!(use_source_properties);
36
37 macro_rules! impl_source_name_to_prop_type_name_inner {
43 ({ $({$variant:ident, $prop_name:ty, $split:ty}),* }) => {
44 pub fn source_name_to_prop_type_name(source_name: &str) -> Option<&'static str> {
45 match source_name {
46 $(
47 <$prop_name>::SOURCE_NAME => Some(std::any::type_name::<$prop_name>()),
48 )*
49 _ => None,
50 }
51 }
52 };
53 }
54
55 macro_rules! impl_source_name_to_prop_type_name {
56 () => {
57 $crate::for_all_sources! { impl_source_name_to_prop_type_name_inner }
58 };
59 }
60
61 impl_source_name_to_prop_type_name!();
62}
63
64mod sink_properties {
65 use crate::use_all_sink_configs;
66 use crate::sink::Sink;
67 use crate::sink::file_sink::fs::FsSink;
68
69 use_all_sink_configs!();
70
71 macro_rules! impl_sink_name_to_config_type_name_inner {
72 ({ $({ $variant_name:ident, $sink_type:ty, $config_type:ty }),* }) => {
73 pub fn sink_name_to_config_type_name(sink_name: &str) -> Option<&'static str> {
74 match sink_name {
75 $(
76 <$sink_type>::SINK_NAME => Some(std::any::type_name::<$config_type>()),
77 )*
78 _ => None,
79 }
80 }
81 };
82 }
83
84 macro_rules! impl_sink_name_to_config_type_name {
85 () => {
86 $crate::for_all_sinks! { impl_sink_name_to_config_type_name_inner }
87 };
88 }
89
90 impl_sink_name_to_config_type_name!();
91}
92
93pub static SOURCE_ALLOW_ALTER_ON_FLY_FIELDS: LazyLock<HashMap<String, HashSet<String>>> = LazyLock::new(|| {
95 use source_properties::*;
96 let mut map = HashMap::new();
97 map.try_insert(
99 std::any::type_name::<KafkaProperties>().to_owned(),
100 [
101 "group.id.prefix".to_owned(),
102 "properties.sync.call.timeout".to_owned(),
103 "properties.security.protocol".to_owned(),
104 "properties.ssl.endpoint.identification.algorithm".to_owned(),
105 "properties.sasl.mechanism".to_owned(),
106 "properties.sasl.username".to_owned(),
107 "properties.sasl.password".to_owned(),
108 "properties.message.max.bytes".to_owned(),
109 "properties.receive.message.max.bytes".to_owned(),
110 "properties.statistics.interval.ms".to_owned(),
111 "properties.client.id".to_owned(),
112 "properties.enable.ssl.certificate.verification".to_owned(),
113 "properties.queued.min.messages".to_owned(),
114 "properties.queued.max.messages.kbytes".to_owned(),
115 "properties.fetch.wait.max.ms".to_owned(),
116 "properties.fetch.queue.backoff.ms".to_owned(),
117 "properties.fetch.max.bytes".to_owned(),
118 "properties.enable.auto.commit".to_owned(),
119 ].into_iter().collect(),
120 ).unwrap();
121 map
122});
123
124pub static SINK_ALLOW_ALTER_ON_FLY_FIELDS: LazyLock<HashMap<String, HashSet<String>>> = LazyLock::new(|| {
126 use sink_properties::*;
127 let mut map = HashMap::new();
128 map.try_insert(
130 std::any::type_name::<ClickHouseConfig>().to_owned(),
131 [
132 "commit_checkpoint_interval".to_owned(),
133 ].into_iter().collect(),
134 ).unwrap();
135 map.try_insert(
137 std::any::type_name::<DeltaLakeConfig>().to_owned(),
138 [
139 "commit_checkpoint_interval".to_owned(),
140 ].into_iter().collect(),
141 ).unwrap();
142 map.try_insert(
144 std::any::type_name::<IcebergConfig>().to_owned(),
145 [
146 "commit_checkpoint_interval".to_owned(),
147 "enable_compaction".to_owned(),
148 "compaction_interval_sec".to_owned(),
149 "enable_snapshot_expiration".to_owned(),
150 ].into_iter().collect(),
151 ).unwrap();
152 map.try_insert(
154 std::any::type_name::<KafkaConfig>().to_owned(),
155 [
156 "properties.sync.call.timeout".to_owned(),
157 "properties.security.protocol".to_owned(),
158 "properties.ssl.endpoint.identification.algorithm".to_owned(),
159 "properties.sasl.mechanism".to_owned(),
160 "properties.sasl.username".to_owned(),
161 "properties.sasl.password".to_owned(),
162 "properties.message.max.bytes".to_owned(),
163 "properties.receive.message.max.bytes".to_owned(),
164 "properties.statistics.interval.ms".to_owned(),
165 "properties.client.id".to_owned(),
166 "properties.enable.ssl.certificate.verification".to_owned(),
167 "properties.allow.auto.create.topics".to_owned(),
168 "properties.queue.buffering.max.messages".to_owned(),
169 "properties.queue.buffering.max.kbytes".to_owned(),
170 "properties.queue.buffering.max.ms".to_owned(),
171 "properties.enable.idempotence".to_owned(),
172 "properties.message.send.max.retries".to_owned(),
173 "properties.retry.backoff.ms".to_owned(),
174 "properties.batch.num.messages".to_owned(),
175 "properties.batch.size".to_owned(),
176 "properties.message.timeout.ms".to_owned(),
177 "properties.max.in.flight.requests.per.connection".to_owned(),
178 "properties.request.required.acks".to_owned(),
179 ].into_iter().collect(),
180 ).unwrap();
181 map.try_insert(
183 std::any::type_name::<StarrocksConfig>().to_owned(),
184 [
185 "commit_checkpoint_interval".to_owned(),
186 ].into_iter().collect(),
187 ).unwrap();
188 map
189});
190
191pub static CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS: LazyLock<HashMap<String, HashSet<String>>> = LazyLock::new(|| {
193 use crate::connector_common::*;
194 let mut map = HashMap::new();
195 map.try_insert(
197 std::any::type_name::<KafkaConnection>().to_owned(),
198 [
199 "properties.security.protocol".to_owned(),
200 "properties.ssl.endpoint.identification.algorithm".to_owned(),
201 "properties.sasl.mechanism".to_owned(),
202 "properties.sasl.username".to_owned(),
203 "properties.sasl.password".to_owned(),
204 ].into_iter().collect(),
205 ).unwrap();
206 map
207});
208
209pub fn get_source_connectors_with_allow_alter_on_fly_fields() -> Vec<&'static str> {
211 SOURCE_ALLOW_ALTER_ON_FLY_FIELDS.keys().map(|s| s.as_str()).collect()
212}
213
214pub fn get_sink_connectors_with_allow_alter_on_fly_fields() -> Vec<&'static str> {
216 SINK_ALLOW_ALTER_ON_FLY_FIELDS.keys().map(|s| s.as_str()).collect()
217}
218
219pub fn get_connection_names_with_allow_alter_on_fly_fields() -> Vec<&'static str> {
221 CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS.keys().map(|s| s.as_str()).collect()
222}
223
224pub fn check_source_allow_alter_on_fly_fields(
227 connector_name: &str,
228 fields: &[String],
229) -> crate::error::ConnectorResult<()> {
230 let Some(type_name) = source_properties::source_name_to_prop_type_name(connector_name) else {
232 return Err(ConnectorError::from(anyhow::anyhow!(
233 "Unknown source connector: {connector_name}"
234 )));
235 };
236 let Some(allowed_fields) = SOURCE_ALLOW_ALTER_ON_FLY_FIELDS.get(type_name) else {
237 return Err(ConnectorError::from(anyhow::anyhow!(
238 "No allow_alter_on_fly fields registered for connector: {connector_name}"
239 )));
240 };
241 for field in fields {
242 if !allowed_fields.contains(field) {
243 return Err(ConnectorError::from(anyhow::anyhow!(
244 "Field '{field}' is not allowed to be altered on the fly for connector: {connector_name}"
245 )));
246 }
247 }
248 Ok(())
249}
250
251pub fn check_connection_allow_alter_on_fly_fields(
252 connection_name: &str,
253 fields: &[String],
254) -> crate::error::ConnectorResult<()> {
255 use crate::source::connection_name_to_prop_type_name;
256
257 let Some(type_name) = connection_name_to_prop_type_name(connection_name) else {
259 return Err(ConnectorError::from(anyhow::anyhow!(
260 "Unknown connection: {connection_name}"
261 )));
262 };
263 let Some(allowed_fields) = CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS.get(type_name) else {
264 return Err(ConnectorError::from(anyhow::anyhow!(
265 "No allow_alter_on_fly fields registered for connection: {connection_name}"
266 )));
267 };
268 for field in fields {
269 if !allowed_fields.contains(field) {
270 return Err(ConnectorError::from(anyhow::anyhow!(
271 "Field '{field}' is not allowed to be altered on the fly for connection: {connection_name}"
272 )));
273 }
274 }
275 Ok(())
276}
277
278pub fn check_sink_allow_alter_on_fly_fields(
281 sink_name: &str,
282 fields: &[String],
283) -> crate::error::ConnectorResult<()> {
284 let Some(type_name) = sink_properties::sink_name_to_config_type_name(sink_name) else {
286 return Err(ConnectorError::from(anyhow::anyhow!(
287 "Unknown sink connector: {sink_name}"
288 )));
289 };
290 let Some(allowed_fields) = SINK_ALLOW_ALTER_ON_FLY_FIELDS.get(type_name) else {
291 return Err(ConnectorError::from(anyhow::anyhow!(
292 "No allow_alter_on_fly fields registered for sink: {sink_name}"
293 )));
294 };
295 for field in fields {
296 if !allowed_fields.contains(field) {
297 return Err(ConnectorError::from(anyhow::anyhow!(
298 "Field '{field}' is not allowed to be altered on the fly for sink: {sink_name}"
299 )));
300 }
301 }
302 Ok(())
303}
304