laminar_connectors/schema/
types.rs

1//! Schema types used across the connector framework.
2//!
3//! Defines the core data structures for schema inference, resolution,
4//! and schema annotations:
5//!
6//! - [`RawRecord`]: A raw record with key, value, timestamp, and headers
7//! - [`SourceMetadata`]: Type-erased metadata from a source connector
8//! - [`FieldMeta`]: Per-field metadata for schema annotations
9
10#![allow(clippy::disallowed_types)] // cold path: schema management
11
12use std::any::Any;
13use std::collections::HashMap;
14
15/// A raw record read from a source before schema application.
16///
17/// Carries the key, value, optional timestamp, headers, and arbitrary
18/// source-specific metadata.
19#[derive(Debug, Clone)]
20pub struct RawRecord {
21    /// Optional record key (e.g., Kafka message key).
22    pub key: Option<Vec<u8>>,
23
24    /// Record value (payload bytes).
25    pub value: Vec<u8>,
26
27    /// Optional event-time timestamp in milliseconds since epoch.
28    pub timestamp: Option<i64>,
29
30    /// Optional key-value headers (e.g., Kafka headers).
31    pub headers: HashMap<String, Vec<u8>>,
32
33    /// Source-specific metadata (e.g., partition, offset, topic).
34    pub metadata: SourceMetadata,
35}
36
37impl RawRecord {
38    /// Creates a new raw record with only a value.
39    #[must_use]
40    pub fn new(value: Vec<u8>) -> Self {
41        Self {
42            key: None,
43            value,
44            timestamp: None,
45            headers: HashMap::new(),
46            metadata: SourceMetadata::empty(),
47        }
48    }
49
50    /// Sets the record key.
51    #[must_use]
52    pub fn with_key(mut self, key: Vec<u8>) -> Self {
53        self.key = Some(key);
54        self
55    }
56
57    /// Sets the event-time timestamp.
58    #[must_use]
59    pub fn with_timestamp(mut self, ts: i64) -> Self {
60        self.timestamp = Some(ts);
61        self
62    }
63
64    /// Adds a header.
65    #[must_use]
66    pub fn with_header(mut self, key: impl Into<String>, value: Vec<u8>) -> Self {
67        self.headers.insert(key.into(), value);
68        self
69    }
70
71    /// Sets the source metadata.
72    #[must_use]
73    pub fn with_metadata(mut self, metadata: SourceMetadata) -> Self {
74        self.metadata = metadata;
75        self
76    }
77}
78
79/// Type-erased metadata from a source connector.
80///
81/// Wraps a `Box<dyn Any + Send + Sync>` to allow connectors to attach
82/// arbitrary metadata (e.g., Kafka offset, CDC LSN) to raw records.
83pub struct SourceMetadata {
84    inner: Option<Box<dyn Any + Send + Sync>>,
85}
86
87impl SourceMetadata {
88    /// Creates empty metadata.
89    #[must_use]
90    pub fn empty() -> Self {
91        Self { inner: None }
92    }
93
94    /// Creates metadata from a typed value.
95    pub fn new<T: Any + Send + Sync>(value: T) -> Self {
96        Self {
97            inner: Some(Box::new(value)),
98        }
99    }
100
101    /// Returns `true` if no metadata is present.
102    #[must_use]
103    pub fn is_empty(&self) -> bool {
104        self.inner.is_none()
105    }
106
107    /// Attempts to downcast the metadata to a concrete type.
108    #[must_use]
109    pub fn downcast_ref<T: Any>(&self) -> Option<&T> {
110        self.inner.as_ref()?.downcast_ref::<T>()
111    }
112}
113
114impl std::fmt::Debug for SourceMetadata {
115    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
116        if self.inner.is_some() {
117            write!(f, "SourceMetadata(<opaque>)")
118        } else {
119            write!(f, "SourceMetadata(empty)")
120        }
121    }
122}
123
124impl Clone for SourceMetadata {
125    fn clone(&self) -> Self {
126        // Metadata is not cloneable in general; cloning produces empty metadata.
127        Self::empty()
128    }
129}
130
131/// Per-field metadata for schema annotations.
132///
133/// Provides additional information about a field beyond what Arrow's
134/// `Field` captures (description, original source type, etc.).
135#[derive(Debug, Clone, Default)]
136pub struct FieldMeta {
137    /// Optional stable field identifier (for evolution tracking).
138    pub field_id: Option<u32>,
139
140    /// Human-readable description of the field.
141    pub description: Option<String>,
142
143    /// Original type name in the source system (e.g., `"VARCHAR(255)"`).
144    pub source_type: Option<String>,
145
146    /// Default expression if the field is missing (e.g., `"0"`, `"now()"`).
147    pub default_expr: Option<String>,
148
149    /// Arbitrary key-value properties.
150    pub properties: HashMap<String, String>,
151}
152
153impl FieldMeta {
154    /// Creates empty field metadata.
155    #[must_use]
156    pub fn new() -> Self {
157        Self::default()
158    }
159
160    /// Sets the field ID.
161    #[must_use]
162    pub fn with_field_id(mut self, id: u32) -> Self {
163        self.field_id = Some(id);
164        self
165    }
166
167    /// Sets the description.
168    #[must_use]
169    pub fn with_description(mut self, desc: impl Into<String>) -> Self {
170        self.description = Some(desc.into());
171        self
172    }
173
174    /// Sets the original source type.
175    #[must_use]
176    pub fn with_source_type(mut self, src_type: impl Into<String>) -> Self {
177        self.source_type = Some(src_type.into());
178        self
179    }
180
181    /// Sets the default expression.
182    #[must_use]
183    pub fn with_default(mut self, expr: impl Into<String>) -> Self {
184        self.default_expr = Some(expr.into());
185        self
186    }
187
188    /// Sets an arbitrary property.
189    #[must_use]
190    pub fn with_property(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
191        self.properties.insert(key.into(), value.into());
192        self
193    }
194}
195
196#[cfg(test)]
197mod tests {
198    use super::*;
199
200    #[test]
201    fn test_raw_record_builder() {
202        let record = RawRecord::new(b"hello".to_vec())
203            .with_key(b"k1".to_vec())
204            .with_timestamp(1000)
205            .with_header("content-type", b"application/json".to_vec());
206
207        assert_eq!(record.key.as_deref(), Some(b"k1".as_slice()));
208        assert_eq!(record.value, b"hello");
209        assert_eq!(record.timestamp, Some(1000));
210        assert!(record.headers.contains_key("content-type"));
211    }
212
213    #[test]
214    fn test_source_metadata_empty() {
215        let meta = SourceMetadata::empty();
216        assert!(meta.is_empty());
217        assert!(meta.downcast_ref::<String>().is_none());
218    }
219
220    #[test]
221    fn test_source_metadata_typed() {
222        let meta = SourceMetadata::new(42u64);
223        assert!(!meta.is_empty());
224        assert_eq!(meta.downcast_ref::<u64>(), Some(&42u64));
225        assert!(meta.downcast_ref::<String>().is_none());
226    }
227
228    #[test]
229    fn test_source_metadata_debug() {
230        let empty = SourceMetadata::empty();
231        assert!(format!("{empty:?}").contains("empty"));
232
233        let full = SourceMetadata::new("data");
234        assert!(format!("{full:?}").contains("opaque"));
235    }
236
237    #[test]
238    fn test_field_meta_builder() {
239        let meta = FieldMeta::new()
240            .with_field_id(1)
241            .with_description("User ID")
242            .with_source_type("BIGINT")
243            .with_default("0")
244            .with_property("pii", "true");
245
246        assert_eq!(meta.field_id, Some(1));
247        assert_eq!(meta.description.as_deref(), Some("User ID"));
248        assert_eq!(meta.source_type.as_deref(), Some("BIGINT"));
249        assert_eq!(meta.default_expr.as_deref(), Some("0"));
250        assert_eq!(meta.properties.get("pii").map(String::as_str), Some("true"));
251    }
252}
laminar_connectors/schema/types.rs

laminar_connectors/schema/
types.rs
