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 assembled
46/// from SQL connector and format clauses or supplied programmatically.
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    /// Trusted catalog binding; deliberately not a user-settable string property.
55    schema_binding: Option<Arc<crate::schema::resolution::SchemaBinding>>,
56}
57
58impl ConnectorConfig {
59    /// Creates a new connector config with the given type.
60    #[must_use]
61    pub fn new(connector_type: impl Into<String>) -> Self {
62        Self {
63            connector_type: connector_type.into(),
64            properties: HashMap::new(),
65            schema_binding: None,
66        }
67    }
68
69    /// Creates a config from existing properties.
70    #[must_use]
71    pub fn with_properties(
72        connector_type: impl Into<String>,
73        properties: HashMap<String, String>,
74    ) -> Self {
75        Self {
76            connector_type: connector_type.into(),
77            properties,
78            schema_binding: None,
79        }
80    }
81
82    /// Returns the connector type identifier.
83    #[must_use]
84    pub fn connector_type(&self) -> &str {
85        &self.connector_type
86    }
87
88    /// Sets a configuration property.
89    pub fn set(&mut self, key: impl Into<String>, value: impl Into<String>) {
90        self.properties.insert(key.into(), value.into());
91    }
92
93    /// Gets a configuration property.
94    #[must_use]
95    pub fn get(&self, key: &str) -> Option<&str> {
96        self.properties.get(key).map(String::as_str)
97    }
98
99    /// Gets a required configuration property, returning an error if missing.
100    ///
101    /// # Errors
102    ///
103    /// Returns `ConnectorError::MissingConfig` if the key is not set.
104    pub fn require(&self, key: &str) -> Result<&str, ConnectorError> {
105        self.get(key)
106            .ok_or_else(|| ConnectorError::missing_config(key.to_string()))
107    }
108
109    /// Gets a property parsed as the given type.
110    ///
111    /// # Errors
112    ///
113    /// Returns `ConnectorError::ConfigurationError` if the value cannot be parsed.
114    pub fn get_parsed<T: std::str::FromStr>(&self, key: &str) -> Result<Option<T>, ConnectorError>
115    where
116        T::Err: fmt::Display,
117    {
118        match self.get(key) {
119            Some(v) => v.parse::<T>().map(Some).map_err(|e| {
120                ConnectorError::ConfigurationError(format!("invalid value for '{key}': {e}"))
121            }),
122            None => Ok(None),
123        }
124    }
125
126    /// Gets a required property parsed as the given type.
127    ///
128    /// # Errors
129    ///
130    /// Returns `ConnectorError::MissingConfig` if the key is missing, or
131    /// `ConnectorError::ConfigurationError` if parsing fails.
132    pub fn require_parsed<T: std::str::FromStr>(&self, key: &str) -> Result<T, ConnectorError>
133    where
134        T::Err: fmt::Display,
135    {
136        let value = self.require(key)?;
137        value.parse::<T>().map_err(|e| {
138            ConnectorError::ConfigurationError(format!("invalid value for '{key}': {e}"))
139        })
140    }
141
142    /// Gets a trimmed property that must not be empty when present.
143    ///
144    /// # Errors
145    ///
146    /// Returns `ConnectorError::ConfigurationError` for a present but blank value.
147    pub fn get_non_empty(&self, key: &str) -> Result<Option<String>, ConnectorError> {
148        match self.get(key).map(str::trim) {
149            None => Ok(None),
150            Some("") => Err(ConnectorError::ConfigurationError(format!(
151                "'{key}' must not be empty"
152            ))),
153            Some(value) => Ok(Some(value.to_string())),
154        }
155    }
156
157    /// Gets a comma-separated property as trimmed, non-empty items; absent is empty.
158    #[must_use]
159    pub fn get_list(&self, key: &str) -> Vec<String> {
160        self.get(key)
161            .into_iter()
162            .flat_map(|value| value.split(','))
163            .map(str::trim)
164            .filter(|item| !item.is_empty())
165            .map(str::to_string)
166            .collect()
167    }
168
169    /// Gets a positive count, or `default` when absent.
170    ///
171    /// # Errors
172    ///
173    /// Returns `ConnectorError::ConfigurationError` for zero or an unparseable value.
174    pub fn get_positive(&self, key: &str, default: usize) -> Result<usize, ConnectorError> {
175        match self.get_parsed::<usize>(key)? {
176            Some(0) => Err(ConnectorError::ConfigurationError(format!(
177                "'{key}' must be positive"
178            ))),
179            Some(value) => Ok(value),
180            None => Ok(default),
181        }
182    }
183
184    /// Gets a millisecond duration in `1..=max`, or `default` when absent.
185    ///
186    /// # Errors
187    ///
188    /// Returns `ConnectorError::ConfigurationError` for an unparseable or out-of-range value.
189    pub fn get_millis(
190        &self,
191        key: &str,
192        default: std::time::Duration,
193        max: std::time::Duration,
194    ) -> Result<std::time::Duration, ConnectorError> {
195        let Some(millis) = self.get_parsed::<u64>(key)? else {
196            return Ok(default);
197        };
198        let value = std::time::Duration::from_millis(millis);
199        if value.is_zero() || value > max {
200            return Err(ConnectorError::ConfigurationError(format!(
201                "'{key}' must be between 1 and {} milliseconds",
202                max.as_millis()
203            )));
204        }
205        Ok(value)
206    }
207
208    /// Install a validated committed reader/writer contract for connector activation.
209    ///
210    /// # Errors
211    /// Rejects malformed contracts and a different connector type.
212    pub fn set_schema_binding(
213        &mut self,
214        binding: crate::schema::resolution::SchemaBinding,
215    ) -> Result<(), ConnectorError> {
216        binding
217            .canonical_bytes()
218            .map_err(crate::schema::resolution::binding_error)?;
219        if binding.connector != self.connector_type {
220            return Err(ConnectorError::SchemaMismatch(
221                "schema contract belongs to another connector".into(),
222            ));
223        }
224        self.schema_binding = Some(Arc::new(binding));
225        Ok(())
226    }
227
228    /// Read the prepared immutable contract without re-resolving external metadata.
229    #[must_use]
230    pub fn schema_binding(&self) -> Option<&crate::schema::resolution::SchemaBinding> {
231        self.schema_binding.as_deref()
232    }
233
234    /// Returns all properties as a reference.
235    #[must_use]
236    pub fn properties(&self) -> &HashMap<String, String> {
237        &self.properties
238    }
239
240    /// Returns properties with a given prefix, with the prefix stripped.
241    #[must_use]
242    pub fn properties_with_prefix(&self, prefix: &str) -> HashMap<String, String> {
243        self.properties
244            .iter()
245            .filter_map(|(k, v)| {
246                k.strip_prefix(prefix)
247                    .map(|stripped| (stripped.to_string(), v.clone()))
248            })
249            .collect()
250    }
251
252    /// Decodes the `_arrow_schema` IPC flatbuffer, or `None` if absent.
253    #[must_use]
254    pub fn arrow_schema(&self) -> Option<SchemaRef> {
255        let s = self.get("_arrow_schema")?;
256        decode_arrow_schema_ipc(s).map(Arc::new)
257    }
258
259    /// Formats all properties for display with secrets redacted.
260    ///
261    /// Uses [`SecretMasker`](crate::storage::SecretMasker) to replace
262    /// values of secret keys (passwords, access keys, tokens) with `"***"`.
263    #[cfg(test)]
264    #[must_use]
265    pub fn display_redacted(&self) -> String {
266        use crate::storage::SecretMasker;
267        SecretMasker::display_map(&self.properties)
268    }
269
270    /// Validates the configuration against a set of key specifications.
271    ///
272    /// # Errors
273    ///
274    /// Returns `ConnectorError::MissingConfig` for missing required keys, or
275    /// `ConnectorError::ConfigurationError` for invalid values.
276    pub fn validate(&self, specs: &[ConfigKeySpec]) -> Result<(), ConnectorError> {
277        for spec in specs {
278            if spec.required && self.get(&spec.key).is_none() {
279                if let Some(ref default) = spec.default {
280                    // Has a default, skip
281                    let _ = default;
282                } else {
283                    return Err(ConnectorError::missing_config(spec.key.clone()));
284                }
285            }
286        }
287        Ok(())
288    }
289
290    /// Rejects properties outside `allowed`.
291    ///
292    /// # Errors
293    ///
294    /// Returns [`ConnectorError::ConfigurationError`] for unknown properties.
295    pub fn reject_unknown_properties(
296        &self,
297        allowed: &[&str],
298        connector: &str,
299    ) -> Result<(), ConnectorError> {
300        let mut unknown: Vec<&str> = self
301            .properties
302            .keys()
303            .map(String::as_str)
304            .filter(|key| !allowed.contains(key))
305            .collect();
306        if unknown.is_empty() {
307            return Ok(());
308        }
309
310        unknown.sort_unstable();
311        Err(ConnectorError::ConfigurationError(format!(
312            "unknown {connector} propert{}: {}",
313            if unknown.len() == 1 { "y" } else { "ies" },
314            unknown.join(", ")
315        )))
316    }
317}
318
319/// Validates that a string field is non-empty.
320///
321/// # Errors
322///
323/// Returns `ConnectorError::ConfigurationError` if the value is empty.
324pub fn require_non_empty(value: &str, field_name: &str) -> Result<(), ConnectorError> {
325    if value.is_empty() {
326        return Err(ConnectorError::ConfigurationError(format!(
327            "{field_name} must not be empty",
328        )));
329    }
330    Ok(())
331}
332
333/// Parses a port string into a `u16`.
334///
335/// # Errors
336///
337/// Returns `ConnectorError::ConfigurationError` if the value is not a valid port number.
338pub fn parse_port(value: &str) -> Result<u16, ConnectorError> {
339    value
340        .parse()
341        .map_err(|_| ConnectorError::ConfigurationError(format!("invalid port: '{value}'")))
342}
343
344/// Specification for a configuration key.
345///
346/// Used by connectors to declare their expected configuration.
347#[derive(Debug, Clone, serde::Serialize)]
348pub struct ConfigKeySpec {
349    /// The configuration key name.
350    pub key: String,
351
352    /// Human-readable description.
353    pub description: String,
354
355    /// Whether this key is required.
356    pub required: bool,
357
358    /// Default value if not provided.
359    pub default: Option<String>,
360}
361
362impl ConfigKeySpec {
363    /// Creates a required configuration key spec.
364    #[must_use]
365    pub fn required(key: impl Into<String>, description: impl Into<String>) -> Self {
366        Self {
367            key: key.into(),
368            description: description.into(),
369            required: true,
370            default: None,
371        }
372    }
373
374    /// Creates an optional configuration key spec with a default value.
375    #[must_use]
376    pub fn optional(
377        key: impl Into<String>,
378        description: impl Into<String>,
379        default: impl Into<String>,
380    ) -> Self {
381        Self {
382            key: key.into(),
383            description: description.into(),
384            required: false,
385            default: Some(default.into()),
386        }
387    }
388}
389
390/// Metadata about a connector implementation.
391#[derive(Debug, Clone, serde::Serialize)]
392pub struct ConnectorInfo {
393    /// Direction- and format-specific schema authority and explicit preparation policy.
394    pub schema_capabilities: crate::schema::resolution::SchemaCapabilities,
395    /// Unique connector type name (e.g., "kafka", "postgres-cdc").
396    pub name: String,
397
398    /// Human-readable display name.
399    pub display_name: String,
400
401    /// Version string.
402    pub version: String,
403
404    /// Whether this is a source connector.
405    pub is_source: bool,
406
407    /// Whether this is a sink connector.
408    pub is_sink: bool,
409
410    /// Configuration keys this connector accepts.
411    pub config_keys: Vec<ConfigKeySpec>,
412}
413
414/// Lifecycle state of a running connector.
415#[derive(Debug, Clone, Copy, PartialEq, Eq)]
416pub enum ConnectorState {
417    /// Connector has been created but not yet opened.
418    Created,
419
420    /// Connector is initializing (connecting, schema discovery, etc.).
421    Initializing,
422
423    /// Connector is running and processing data.
424    Running,
425
426    /// Connector is paused (e.g., due to backpressure or manual pause).
427    Paused,
428
429    /// Connector encountered an error and is attempting recovery.
430    Recovering,
431
432    /// Connector has been closed.
433    Closed,
434
435    /// Connector has failed and cannot recover.
436    Failed,
437}
438
439impl fmt::Display for ConnectorState {
440    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
441        match self {
442            ConnectorState::Created => write!(f, "Created"),
443            ConnectorState::Initializing => write!(f, "Initializing"),
444            ConnectorState::Running => write!(f, "Running"),
445            ConnectorState::Paused => write!(f, "Paused"),
446            ConnectorState::Recovering => write!(f, "Recovering"),
447            ConnectorState::Closed => write!(f, "Closed"),
448            ConnectorState::Failed => write!(f, "Failed"),
449        }
450    }
451}
452
453#[cfg(test)]
454mod tests {
455    use super::*;
456    use arrow_schema::Schema;
457
458    #[test]
459    fn test_config_basic_operations() {
460        let mut config = ConnectorConfig::new("kafka");
461        config.set("bootstrap.servers", "localhost:9092");
462        config.set("topic", "events");
463
464        assert_eq!(config.connector_type(), "kafka");
465        assert_eq!(config.get("bootstrap.servers"), Some("localhost:9092"));
466        assert_eq!(config.get("topic"), Some("events"));
467        assert_eq!(config.get("missing"), None);
468    }
469
470    #[test]
471    fn test_config_require() {
472        let mut config = ConnectorConfig::new("kafka");
473        config.set("topic", "events");
474
475        assert!(config.require("topic").is_ok());
476        assert!(config.require("missing").is_err());
477    }
478
479    #[test]
480    fn test_config_parsed() {
481        let mut config = ConnectorConfig::new("kafka");
482        config.set("batch.size", "1000");
483        config.set("bad_number", "not_a_number");
484
485        let size: Option<usize> = config.get_parsed("batch.size").unwrap();
486        assert_eq!(size, Some(1000));
487
488        let missing: Option<usize> = config.get_parsed("missing").unwrap();
489        assert_eq!(missing, None);
490
491        let bad: Result<Option<usize>, _> = config.get_parsed("bad_number");
492        assert!(bad.is_err());
493    }
494
495    #[test]
496    fn test_config_require_parsed() {
497        let mut config = ConnectorConfig::new("test");
498        config.set("port", "8080");
499
500        let port: u16 = config.require_parsed("port").unwrap();
501        assert_eq!(port, 8080);
502
503        let missing: Result<u16, _> = config.require_parsed("missing");
504        assert!(missing.is_err());
505    }
506
507    #[test]
508    fn test_config_prefix_extraction() {
509        let mut config = ConnectorConfig::new("kafka");
510        config.set("kafka.bootstrap.servers", "localhost:9092");
511        config.set("kafka.group.id", "my-group");
512        config.set("topic", "events");
513
514        let kafka_props = config.properties_with_prefix("kafka.");
515        assert_eq!(kafka_props.len(), 2);
516        assert_eq!(
517            kafka_props.get("bootstrap.servers"),
518            Some(&"localhost:9092".to_string())
519        );
520        assert_eq!(kafka_props.get("group.id"), Some(&"my-group".to_string()));
521    }
522
523    #[test]
524    fn test_config_validate() {
525        let specs = vec![
526            ConfigKeySpec::required("topic", "Kafka topic"),
527            ConfigKeySpec::optional("batch.size", "Batch size", "100"),
528        ];
529
530        let mut config = ConnectorConfig::new("kafka");
531        config.set("topic", "events");
532
533        assert!(config.validate(&specs).is_ok());
534
535        let empty_config = ConnectorConfig::new("kafka");
536        assert!(empty_config.validate(&specs).is_err());
537    }
538
539    #[test]
540    fn test_config_with_properties() {
541        let mut props = HashMap::new();
542        props.insert("key1".to_string(), "val1".to_string());
543        props.insert("key2".to_string(), "val2".to_string());
544
545        let config = ConnectorConfig::with_properties("test", props);
546        assert_eq!(config.get("key1"), Some("val1"));
547        assert_eq!(config.get("key2"), Some("val2"));
548    }
549
550    #[test]
551    fn test_connector_state_display() {
552        assert_eq!(ConnectorState::Running.to_string(), "Running");
553        assert_eq!(ConnectorState::Failed.to_string(), "Failed");
554    }
555
556    #[test]
557    fn test_display_redacted_masks_secrets() {
558        let mut config = ConnectorConfig::new("delta-lake");
559        config.set("aws_region", "us-east-1");
560        config.set("aws_secret_access_key", "TOP_SECRET");
561        config.set("aws_access_key_id", "AKID123");
562
563        let display = config.display_redacted();
564        assert!(display.contains("aws_region=us-east-1"));
565        assert!(display.contains("aws_secret_access_key=***"));
566        assert!(display.contains("aws_access_key_id=***"));
567        assert!(!display.contains("AKID123"));
568        assert!(!display.contains("TOP_SECRET"));
569    }
570
571    #[test]
572    fn test_display_redacted_empty() {
573        let config = ConnectorConfig::new("test");
574        assert!(config.display_redacted().is_empty());
575    }
576
577    #[test]
578    fn test_arrow_schema_ipc_round_trip() {
579        use arrow_schema::{DataType, Field};
580
581        let original = Schema::new(vec![
582            Field::new("s", DataType::Utf8, true),
583            Field::new("p", DataType::Float64, true),
584            Field::new("q", DataType::Float64, true),
585            Field::new("T", DataType::Int64, true),
586        ]);
587        let hex = encode_arrow_schema_ipc(&original);
588
589        let mut config = ConnectorConfig::new("websocket");
590        config.set("_arrow_schema", hex);
591
592        let schema = config.arrow_schema().expect("should parse");
593        assert_eq!(schema.fields().len(), 4);
594        assert_eq!(schema.field(0).name(), "s");
595        assert_eq!(schema.field(0).data_type(), &DataType::Utf8);
596        assert_eq!(schema.field(1).name(), "p");
597        assert_eq!(schema.field(1).data_type(), &DataType::Float64);
598        assert_eq!(schema.field(3).name(), "T");
599        assert_eq!(schema.field(3).data_type(), &DataType::Int64);
600    }
601
602    #[test]
603    fn test_arrow_schema_ipc_handles_map_type() {
604        use arrow_schema::{DataType, Field, Fields};
605
606        let map_field = Field::new(
607            "entries",
608            DataType::Struct(Fields::from(vec![
609                Field::new("key", DataType::Utf8, false),
610                Field::new("value", DataType::Float64, true),
611            ])),
612            false,
613        );
614        let original = Schema::new(vec![
615            Field::new("sensor_id", DataType::Utf8, true),
616            Field::new("data", DataType::Map(Arc::new(map_field), false), true),
617        ]);
618        let hex = encode_arrow_schema_ipc(&original);
619
620        let mut config = ConnectorConfig::new("kafka");
621        config.set("_arrow_schema", hex);
622
623        let schema = config.arrow_schema().expect("should parse Map type");
624        assert_eq!(schema.fields().len(), 2);
625        assert_eq!(schema.field(0).name(), "sensor_id");
626        assert!(matches!(schema.field(1).data_type(), DataType::Map(_, _)));
627    }
628
629    #[test]
630    fn test_arrow_schema_returns_none_when_absent() {
631        let config = ConnectorConfig::new("kafka");
632        assert!(config.arrow_schema().is_none());
633    }
634
635    #[test]
636    fn test_arrow_schema_returns_none_for_empty_value() {
637        let mut config = ConnectorConfig::new("kafka");
638        config.set("_arrow_schema", "");
639        assert!(config.arrow_schema().is_none());
640    }
641}
laminar_connectors/config.rs

laminar_connectors/
config.rs
