feldera_adapterlib/
connector_metadata.rs

1use std::{collections::BTreeMap, sync::Arc};
2
3use feldera_sqllib::{SqlString, Variant};
4
5/// Connector metadata attached to each input record.
6///
7/// Both the transport connector and the parser can add metadata attributes
8/// such as Kafka topic name or Avro schema id. These attributes are passed
9/// to the deserializer along with the actual record, which can use them
10/// to populate some of the table columns.
11#[derive(Default, Clone, Debug, PartialEq, Eq)]
12pub struct ConnectorMetadata(BTreeMap<Variant, Variant>);
13
14impl ConnectorMetadata {
15    pub fn new() -> Self {
16        Self::default()
17    }
18
19    pub fn insert(&mut self, name: &str, value: Variant) {
20        self.0.insert(Variant::String(SqlString::from(name)), value);
21    }
22
23    /// Returns the attribute stored under `key`, or `None` if absent.
24    pub fn get(&self, key: &Variant) -> Option<&Variant> {
25        self.0.get(key)
26    }
27
28    /// Returns the attribute stored under the string key `name`, or `None` if
29    /// absent.  Attributes added with [`insert`](Self::insert) are keyed by
30    /// string.
31    pub fn get_by_name(&self, name: &str) -> Option<&Variant> {
32        self.0.get(&Variant::String(SqlString::from(name)))
33    }
34}
35
36impl From<BTreeMap<Variant, Variant>> for ConnectorMetadata {
37    fn from(metadata: BTreeMap<Variant, Variant>) -> Self {
38        Self(metadata)
39    }
40}
41
42impl From<ConnectorMetadata> for Variant {
43    fn from(metadata: ConnectorMetadata) -> Self {
44        Variant::Map(Arc::new(metadata.0))
45    }
46}
47
48impl From<&ConnectorMetadata> for Variant {
49    fn from(metadata: &ConnectorMetadata) -> Self {
50        Variant::Map(Arc::new(metadata.0.clone()))
51    }
52}
53
54#[cfg(test)]
55mod tests {
56    use super::*;
57
58    /// Both getters must use the same key encoding as `insert`.
59    #[test]
60    fn test_get_uses_insert_key_encoding() {
61        let mut metadata = ConnectorMetadata::new();
62        metadata.insert("topic", Variant::String(SqlString::from("events")));
63
64        let expected = Variant::String(SqlString::from("events"));
65        assert_eq!(metadata.get_by_name("topic"), Some(&expected));
66        assert_eq!(
67            metadata.get(&Variant::String(SqlString::from("topic"))),
68            Some(&expected)
69        );
70        assert_eq!(metadata.get_by_name("missing"), None);
71    }
72}
feldera_adapterlib/connector_metadata.rs

feldera_adapterlib/
connector_metadata.rs
