laminar_connectors/
config.rs

1#![allow(clippy::disallowed_types)] // cold path: connector configuration
2
3use std::collections::HashMap;
4use std::fmt;
5use std::sync::Arc;
6
7use arrow_ipc::writer::{DictionaryTracker, IpcDataGenerator, IpcWriteOptions};
8use arrow_schema::{Schema, SchemaRef};
9
10use crate::error::ConnectorError;
11
12/// Encode an Arrow schema as a hex-encoded IPC flatbuffer string.
13#[must_use]
14pub fn encode_arrow_schema_ipc(schema: &Schema) -> String {
15    let mut dt = DictionaryTracker::new(false);
16    let enc = IpcDataGenerator::default().schema_to_bytes_with_dictionary_tracker(
17        schema,
18        &mut dt,
19        &IpcWriteOptions::default(),
20    );
21    let bytes = &enc.ipc_message;
22    let mut s = String::with_capacity(bytes.len() * 2);
23    for b in bytes {
24        use std::fmt::Write;
25        let _ = write!(s, "{b:02x}");
26    }
27    s
28}
29
30/// Decode a hex-encoded IPC flatbuffer string to an Arrow schema.
31#[must_use]
32pub fn decode_arrow_schema_ipc(hex: &str) -> Option<Schema> {
33    let bytes: Option<Vec<u8>> = (0..hex.len())
34        .step_by(2)
35        .map(|i| {
36            hex.get(i..i + 2)
37                .and_then(|h| u8::from_str_radix(h, 16).ok())
38        })
39        .collect();
40    arrow_ipc::convert::try_schema_from_flatbuffer_bytes(&bytes?).ok()
41}
42
43/// Configuration for a connector instance.
44///
45/// Connectors receive their configuration as a string key-value map,
46/// typically parsed from SQL `WITH (...)` clauses or programmatic config.
47#[derive(Debug, Clone, Default)]
48pub struct ConnectorConfig {
49    /// The connector type identifier (e.g., "kafka", "postgres-cdc").
50    connector_type: String,
51
52    /// Configuration properties.
53    properties: HashMap<String, String>,
54}
55
56impl ConnectorConfig {
57    /// Creates a new connector config with the given type.
58    #[must_use]
59    pub fn new(connector_type: impl Into<String>) -> Self {
60        Self {
61            connector_type: connector_type.into(),
62            properties: HashMap::new(),
63        }
64    }
65
66    /// Creates a config from existing properties.
67    #[must_use]
68    pub fn with_properties(
69        connector_type: impl Into<String>,
70        properties: HashMap<String, String>,
71    ) -> Self {
72        Self {
73            connector_type: connector_type.into(),
74            properties,
75        }
76    }
77
78    /// Returns the connector type identifier.
79    #[must_use]
80    pub fn connector_type(&self) -> &str {
81        &self.connector_type
82    }
83
84    /// Sets a configuration property.
85    pub fn set(&mut self, key: impl Into<String>, value: impl Into<String>) {
86        self.properties.insert(key.into(), value.into());
87    }
88
89    /// Gets a configuration property.
90    #[must_use]
91    pub fn get(&self, key: &str) -> Option<&str> {
92        self.properties.get(key).map(String::as_str)
93    }
94
95    /// Gets a required configuration property, returning an error if missing.
96    ///
97    /// # Errors
98    ///
99    /// Returns `ConnectorError::MissingConfig` if the key is not set.
100    pub fn require(&self, key: &str) -> Result<&str, ConnectorError> {
101        self.get(key)
102            .ok_or_else(|| ConnectorError::missing_config(key.to_string()))
103    }
104
105    /// Gets a property parsed as the given type.
106    ///
107    /// # Errors
108    ///
109    /// Returns `ConnectorError::ConfigurationError` if the value cannot be parsed.
110    pub fn get_parsed<T: std::str::FromStr>(&self, key: &str) -> Result<Option<T>, ConnectorError>
111    where
112        T::Err: fmt::Display,
113    {
114        match self.get(key) {
115            Some(v) => v.parse::<T>().map(Some).map_err(|e| {
116                ConnectorError::ConfigurationError(format!("invalid value for '{key}': {e}"))
117            }),
118            None => Ok(None),
119        }
120    }
121
122    /// Gets a required property parsed as the given type.
123    ///
124    /// # Errors
125    ///
126    /// Returns `ConnectorError::MissingConfig` if the key is missing, or
127    /// `ConnectorError::ConfigurationError` if parsing fails.
128    pub fn require_parsed<T: std::str::FromStr>(&self, key: &str) -> Result<T, ConnectorError>
129    where
130        T::Err: fmt::Display,
131    {
132        let value = self.require(key)?;
133        value.parse::<T>().map_err(|e| {
134            ConnectorError::ConfigurationError(format!("invalid value for '{key}': {e}"))
135        })
136    }
137
138    /// Returns all properties as a reference.
139    #[must_use]
140    pub fn properties(&self) -> &HashMap<String, String> {
141        &self.properties
142    }
143
144    /// Returns properties with a given prefix, with the prefix stripped.
145    #[must_use]
146    pub fn properties_with_prefix(&self, prefix: &str) -> HashMap<String, String> {
147        self.properties
148            .iter()
149            .filter_map(|(k, v)| {
150                k.strip_prefix(prefix)
151                    .map(|stripped| (stripped.to_string(), v.clone()))
152            })
153            .collect()
154    }
155
156    /// Decodes the `_arrow_schema` IPC flatbuffer, or `None` if absent.
157    #[must_use]
158    pub fn arrow_schema(&self) -> Option<SchemaRef> {
159        let s = self.get("_arrow_schema")?;
160        decode_arrow_schema_ipc(s).map(Arc::new)
161    }
162
163    /// Formats all properties for display with secrets redacted.
164    ///
165    /// Uses [`SecretMasker`](crate::storage::SecretMasker) to replace
166    /// values of secret keys (passwords, access keys, tokens) with `"***"`.
167    #[cfg(test)]
168    #[must_use]
169    pub fn display_redacted(&self) -> String {
170        use crate::storage::SecretMasker;
171        SecretMasker::display_map(&self.properties)
172    }
173
174    /// Validates the configuration against a set of key specifications.
175    ///
176    /// # Errors
177    ///
178    /// Returns `ConnectorError::MissingConfig` for missing required keys, or
179    /// `ConnectorError::ConfigurationError` for invalid values.
180    pub fn validate(&self, specs: &[ConfigKeySpec]) -> Result<(), ConnectorError> {
181        for spec in specs {
182            if spec.required && self.get(&spec.key).is_none() {
183                if let Some(ref default) = spec.default {
184                    // Has a default, skip
185                    let _ = default;
186                } else {
187                    return Err(ConnectorError::missing_config(spec.key.clone()));
188                }
189            }
190        }
191        Ok(())
192    }
193
194    /// Rejects properties outside `allowed`.
195    ///
196    /// # Errors
197    ///
198    /// Returns [`ConnectorError::ConfigurationError`] for unknown properties.
199    pub fn reject_unknown_properties(
200        &self,
201        allowed: &[&str],
202        connector: &str,
203    ) -> Result<(), ConnectorError> {
204        let mut unknown: Vec<&str> = self
205            .properties
206            .keys()
207            .map(String::as_str)
208            .filter(|key| !allowed.contains(key))
209            .collect();
210        if unknown.is_empty() {
211            return Ok(());
212        }
213
214        unknown.sort_unstable();
215        Err(ConnectorError::ConfigurationError(format!(
216            "unknown {connector} propert{}: {}",
217            if unknown.len() == 1 { "y" } else { "ies" },
218            unknown.join(", ")
219        )))
220    }
221}
222
223/// Validates that a string field is non-empty.
224///
225/// # Errors
226///
227/// Returns `ConnectorError::ConfigurationError` if the value is empty.
228pub fn require_non_empty(value: &str, field_name: &str) -> Result<(), ConnectorError> {
229    if value.is_empty() {
230        return Err(ConnectorError::ConfigurationError(format!(
231            "{field_name} must not be empty",
232        )));
233    }
234    Ok(())
235}
236
237/// Parses a port string into a `u16`.
238///
239/// # Errors
240///
241/// Returns `ConnectorError::ConfigurationError` if the value is not a valid port number.
242pub fn parse_port(value: &str) -> Result<u16, ConnectorError> {
243    value
244        .parse()
245        .map_err(|_| ConnectorError::ConfigurationError(format!("invalid port: '{value}'")))
246}
247
248/// Specification for a configuration key.
249///
250/// Used by connectors to declare their expected configuration.
251#[derive(Debug, Clone, serde::Serialize)]
252pub struct ConfigKeySpec {
253    /// The configuration key name.
254    pub key: String,
255
256    /// Human-readable description.
257    pub description: String,
258
259    /// Whether this key is required.
260    pub required: bool,
261
262    /// Default value if not provided.
263    pub default: Option<String>,
264}
265
266impl ConfigKeySpec {
267    /// Creates a required configuration key spec.
268    #[must_use]
269    pub fn required(key: impl Into<String>, description: impl Into<String>) -> Self {
270        Self {
271            key: key.into(),
272            description: description.into(),
273            required: true,
274            default: None,
275        }
276    }
277
278    /// Creates an optional configuration key spec with a default value.
279    #[must_use]
280    pub fn optional(
281        key: impl Into<String>,
282        description: impl Into<String>,
283        default: impl Into<String>,
284    ) -> Self {
285        Self {
286            key: key.into(),
287            description: description.into(),
288            required: false,
289            default: Some(default.into()),
290        }
291    }
292}
293
294/// Metadata about a connector implementation.
295#[derive(Debug, Clone, serde::Serialize)]
296pub struct ConnectorInfo {
297    /// Unique connector type name (e.g., "kafka", "postgres-cdc").
298    pub name: String,
299
300    /// Human-readable display name.
301    pub display_name: String,
302
303    /// Version string.
304    pub version: String,
305
306    /// Whether this is a source connector.
307    pub is_source: bool,
308
309    /// Whether this is a sink connector.
310    pub is_sink: bool,
311
312    /// Configuration keys this connector accepts.
313    pub config_keys: Vec<ConfigKeySpec>,
314}
315
316/// Lifecycle state of a running connector.
317#[derive(Debug, Clone, Copy, PartialEq, Eq)]
318pub enum ConnectorState {
319    /// Connector has been created but not yet opened.
320    Created,
321
322    /// Connector is initializing (connecting, schema discovery, etc.).
323    Initializing,
324
325    /// Connector is running and processing data.
326    Running,
327
328    /// Connector is paused (e.g., due to backpressure or manual pause).
329    Paused,
330
331    /// Connector encountered an error and is attempting recovery.
332    Recovering,
333
334    /// Connector has been closed.
335    Closed,
336
337    /// Connector has failed and cannot recover.
338    Failed,
339}
340
341impl fmt::Display for ConnectorState {
342    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
343        match self {
344            ConnectorState::Created => write!(f, "Created"),
345            ConnectorState::Initializing => write!(f, "Initializing"),
346            ConnectorState::Running => write!(f, "Running"),
347            ConnectorState::Paused => write!(f, "Paused"),
348            ConnectorState::Recovering => write!(f, "Recovering"),
349            ConnectorState::Closed => write!(f, "Closed"),
350            ConnectorState::Failed => write!(f, "Failed"),
351        }
352    }
353}
354
355#[cfg(test)]
356mod tests {
357    use super::*;
358    use arrow_schema::Schema;
359
360    #[test]
361    fn test_config_basic_operations() {
362        let mut config = ConnectorConfig::new("kafka");
363        config.set("bootstrap.servers", "localhost:9092");
364        config.set("topic", "events");
365
366        assert_eq!(config.connector_type(), "kafka");
367        assert_eq!(config.get("bootstrap.servers"), Some("localhost:9092"));
368        assert_eq!(config.get("topic"), Some("events"));
369        assert_eq!(config.get("missing"), None);
370    }
371
372    #[test]
373    fn test_config_require() {
374        let mut config = ConnectorConfig::new("kafka");
375        config.set("topic", "events");
376
377        assert!(config.require("topic").is_ok());
378        assert!(config.require("missing").is_err());
379    }
380
381    #[test]
382    fn test_config_parsed() {
383        let mut config = ConnectorConfig::new("kafka");
384        config.set("batch.size", "1000");
385        config.set("bad_number", "not_a_number");
386
387        let size: Option<usize> = config.get_parsed("batch.size").unwrap();
388        assert_eq!(size, Some(1000));
389
390        let missing: Option<usize> = config.get_parsed("missing").unwrap();
391        assert_eq!(missing, None);
392
393        let bad: Result<Option<usize>, _> = config.get_parsed("bad_number");
394        assert!(bad.is_err());
395    }
396
397    #[test]
398    fn test_config_require_parsed() {
399        let mut config = ConnectorConfig::new("test");
400        config.set("port", "8080");
401
402        let port: u16 = config.require_parsed("port").unwrap();
403        assert_eq!(port, 8080);
404
405        let missing: Result<u16, _> = config.require_parsed("missing");
406        assert!(missing.is_err());
407    }
408
409    #[test]
410    fn test_config_prefix_extraction() {
411        let mut config = ConnectorConfig::new("kafka");
412        config.set("kafka.bootstrap.servers", "localhost:9092");
413        config.set("kafka.group.id", "my-group");
414        config.set("topic", "events");
415
416        let kafka_props = config.properties_with_prefix("kafka.");
417        assert_eq!(kafka_props.len(), 2);
418        assert_eq!(
419            kafka_props.get("bootstrap.servers"),
420            Some(&"localhost:9092".to_string())
421        );
422        assert_eq!(kafka_props.get("group.id"), Some(&"my-group".to_string()));
423    }
424
425    #[test]
426    fn test_config_validate() {
427        let specs = vec![
428            ConfigKeySpec::required("topic", "Kafka topic"),
429            ConfigKeySpec::optional("batch.size", "Batch size", "100"),
430        ];
431
432        let mut config = ConnectorConfig::new("kafka");
433        config.set("topic", "events");
434
435        assert!(config.validate(&specs).is_ok());
436
437        let empty_config = ConnectorConfig::new("kafka");
438        assert!(empty_config.validate(&specs).is_err());
439    }
440
441    #[test]
442    fn test_config_with_properties() {
443        let mut props = HashMap::new();
444        props.insert("key1".to_string(), "val1".to_string());
445        props.insert("key2".to_string(), "val2".to_string());
446
447        let config = ConnectorConfig::with_properties("test", props);
448        assert_eq!(config.get("key1"), Some("val1"));
449        assert_eq!(config.get("key2"), Some("val2"));
450    }
451
452    #[test]
453    fn test_connector_state_display() {
454        assert_eq!(ConnectorState::Running.to_string(), "Running");
455        assert_eq!(ConnectorState::Failed.to_string(), "Failed");
456    }
457
458    #[test]
459    fn test_display_redacted_masks_secrets() {
460        let mut config = ConnectorConfig::new("delta-lake");
461        config.set("aws_region", "us-east-1");
462        config.set("aws_secret_access_key", "TOP_SECRET");
463        config.set("aws_access_key_id", "AKID123");
464
465        let display = config.display_redacted();
466        assert!(display.contains("aws_region=us-east-1"));
467        assert!(display.contains("aws_secret_access_key=***"));
468        assert!(display.contains("aws_access_key_id=AKID123"));
469        assert!(!display.contains("TOP_SECRET"));
470    }
471
472    #[test]
473    fn test_display_redacted_empty() {
474        let config = ConnectorConfig::new("test");
475        assert!(config.display_redacted().is_empty());
476    }
477
478    #[test]
479    fn test_arrow_schema_ipc_round_trip() {
480        use arrow_schema::{DataType, Field};
481
482        let original = Schema::new(vec![
483            Field::new("s", DataType::Utf8, true),
484            Field::new("p", DataType::Float64, true),
485            Field::new("q", DataType::Float64, true),
486            Field::new("T", DataType::Int64, true),
487        ]);
488        let hex = encode_arrow_schema_ipc(&original);
489
490        let mut config = ConnectorConfig::new("websocket");
491        config.set("_arrow_schema", hex);
492
493        let schema = config.arrow_schema().expect("should parse");
494        assert_eq!(schema.fields().len(), 4);
495        assert_eq!(schema.field(0).name(), "s");
496        assert_eq!(schema.field(0).data_type(), &DataType::Utf8);
497        assert_eq!(schema.field(1).name(), "p");
498        assert_eq!(schema.field(1).data_type(), &DataType::Float64);
499        assert_eq!(schema.field(3).name(), "T");
500        assert_eq!(schema.field(3).data_type(), &DataType::Int64);
501    }
502
503    #[test]
504    fn test_arrow_schema_ipc_handles_map_type() {
505        use arrow_schema::{DataType, Field, Fields};
506
507        let map_field = Field::new(
508            "entries",
509            DataType::Struct(Fields::from(vec![
510                Field::new("key", DataType::Utf8, false),
511                Field::new("value", DataType::Float64, true),
512            ])),
513            false,
514        );
515        let original = Schema::new(vec![
516            Field::new("sensor_id", DataType::Utf8, true),
517            Field::new("data", DataType::Map(Arc::new(map_field), false), true),
518        ]);
519        let hex = encode_arrow_schema_ipc(&original);
520
521        let mut config = ConnectorConfig::new("kafka");
522        config.set("_arrow_schema", hex);
523
524        let schema = config.arrow_schema().expect("should parse Map type");
525        assert_eq!(schema.fields().len(), 2);
526        assert_eq!(schema.field(0).name(), "sensor_id");
527        assert!(matches!(schema.field(1).data_type(), DataType::Map(_, _)));
528    }
529
530    #[test]
531    fn test_arrow_schema_returns_none_when_absent() {
532        let config = ConnectorConfig::new("kafka");
533        assert!(config.arrow_schema().is_none());
534    }
535
536    #[test]
537    fn test_arrow_schema_returns_none_for_empty_value() {
538        let mut config = ConnectorConfig::new("kafka");
539        config.set("_arrow_schema", "");
540        assert!(config.arrow_schema().is_none());
541    }
542}
laminar_connectors/config.rs

laminar_connectors/
config.rs
