feldera · mihaibudiu · Jul 9, 2026 · Jul 9, 2026 · mythical-fred · Jul 2, 2026
diff --git a/crates/adapterlib/src/connector_metadata.rs b/crates/adapterlib/src/connector_metadata.rs
@@ -13,12 +13,24 @@ pub struct ConnectorMetadata(BTreeMap<Variant, Variant>);
 
 impl ConnectorMetadata {
     pub fn new() -> Self {
-        Self(BTreeMap::new())
+        Self::default()
     }
 
     pub fn insert(&mut self, name: &str, value: Variant) {
         self.0.insert(Variant::String(SqlString::from(name)), value);
     }
+
+    /// Returns the attribute stored under `key`, or `None` if absent.
+    pub fn get(&self, key: &Variant) -> Option<&Variant> {
+        self.0.get(key)
+    }
+
+    /// Returns the attribute stored under the string key `name`, or `None` if
+    /// absent.  Attributes added with [`insert`](Self::insert) are keyed by
+    /// string.
+    pub fn get_by_name(&self, name: &str) -> Option<&Variant> {
+        self.0.get(&Variant::String(SqlString::from(name)))
+    }
 }
 
 impl From<BTreeMap<Variant, Variant>> for ConnectorMetadata {
@@ -32,3 +44,29 @@ impl From<ConnectorMetadata> for Variant {
         Variant::Map(Arc::new(metadata.0))
     }
 }
+
+impl From<&ConnectorMetadata> for Variant {
+    fn from(metadata: &ConnectorMetadata) -> Self {
+        Variant::Map(Arc::new(metadata.0.clone()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Both getters must use the same key encoding as `insert`.
+    #[test]
+    fn test_get_uses_insert_key_encoding() {
+        let mut metadata = ConnectorMetadata::new();
+        metadata.insert("topic", Variant::String(SqlString::from("events")));
+
+        let expected = Variant::String(SqlString::from("events"));
+        assert_eq!(metadata.get_by_name("topic"), Some(&expected));
+        assert_eq!(
+            metadata.get(&Variant::String(SqlString::from("topic"))),
+            Some(&expected)
+        );
+        assert_eq!(metadata.get_by_name("missing"), None);
+    }
+}
diff --git a/crates/adapterlib/src/format.rs b/crates/adapterlib/src/format.rs
@@ -393,7 +393,9 @@ impl Parser for StreamingPreprocessedParser {
         data: &[u8],
         metadata: Option<ConnectorMetadata>,
     ) -> (Option<Box<dyn InputBuffer>>, Vec<ParseError>) {
-        let (pre_data, mut pre_errors) = self.preprocessor.process(data);
+        let (pre_data, mut pre_errors) = self
+            .preprocessor
+            .process_with_metadata(data, metadata.as_ref());
         self.stream_splitter.append(&pre_data);
         let mut parsed: Vec<Box<dyn InputBuffer>> = Vec::new();
         while let Some(chunk) = self.stream_splitter.next(true) {
@@ -451,7 +453,9 @@ impl Parser for MessageOrientedPreprocessedParser {
         data: &[u8],
         metadata: Option<ConnectorMetadata>,
     ) -> (Option<Box<dyn InputBuffer>>, Vec<ParseError>) {
-        let (pre_data, mut pre_errors) = self.preprocessor.process(data);
+        let (pre_data, mut pre_errors) = self
+            .preprocessor
+            .process_with_metadata(data, metadata.as_ref());
         let mut parser_splitter = self.parser.splitter();
         let mut parsed: Vec<Box<dyn InputBuffer>> = Vec::new();
         let mut remaining = pre_data.as_slice();

diff --git a/crates/adapterlib/src/preprocess.rs b/crates/adapterlib/src/preprocess.rs
@@ -9,6 +9,7 @@
 //! Transport → Preprocessor → Parser → Circuit
 //! ```
 
+use crate::ConnectorMetadata;
 use crate::format::{ParseError, Splitter};
 use feldera_types::preprocess::PreprocessorConfig;
 use std::collections::BTreeMap;
@@ -47,12 +48,39 @@ impl std::error::Error for PreprocessorCreateError {}
 pub trait Preprocessor: Send + Sync {
     /// Process raw input data and return transformed data.
     ///
+    /// The default implementation forwards to `process_with_metadata` with no
+    /// metadata.
+    ///
     /// # Arguments
     /// * `data` - Raw input data bytes
     ///
     /// # Returns
-    /// A `PreprocessResult` containing the transformed data or errors.
-    fn process(&mut self, data: &[u8]) -> (Vec<u8>, Vec<ParseError>);
+    /// The transformed data and any errors that occurred.
+    fn process(&mut self, data: &[u8]) -> (Vec<u8>, Vec<ParseError>) {
+        self.process_with_metadata(data, None)
+    }
+
+    /// Process raw input data and return transformed data using connector metadata.
+    ///
+    /// Users should implement exactly one of the two processing methods: `process` or
+    /// `process_with_metadata`.
+    ///
+    /// WARNING: a preprocessor that implements neither processing method
+    /// compiles, but will panic at runtime with a stack overflow.
+    ///
+    /// # Arguments
+    /// * `data` - Raw input data bytes
+    /// * `_metadata` - Connector metadata
+    ///
+    /// # Returns
+    /// The transformed data and any errors that occurred.
+    fn process_with_metadata(
+        &mut self,
+        data: &[u8],
+        _metadata: Option<&ConnectorMetadata>,
+    ) -> (Vec<u8>, Vec<ParseError>) {
+        self.process(data)
+    }
 
     /// Create a new preprocessor with the same configuration as `self`.
     ///
@@ -101,3 +129,129 @@ impl PreprocessorRegistry {
         self.registered.get(name).cloned()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use feldera_sqllib::{SqlString, Variant};
+    use serde_json::json;
+
+    /// Preprocessor whose output depends on connector metadata: it prepends
+    /// the value of the `topic` attribute to each record.  Reads the attribute
+    /// through `ConnectorMetadata::get_by_name`, the same path a user-defined
+    /// preprocessor must use.
+    struct TopicPrefixPreprocessor;
+
+    impl Preprocessor for TopicPrefixPreprocessor {
+        fn process_with_metadata(
+            &mut self,
+            data: &[u8],
+            metadata: Option<&ConnectorMetadata>,
+        ) -> (Vec<u8>, Vec<ParseError>) {
+            let Some(metadata) = metadata else {
+                // No metadata attached: pass the record through unchanged.
+                return (data.to_vec(), vec![]);
+            };
+            let mut output = Vec::new();
+            if let Some(Variant::String(topic)) = metadata.get_by_name("topic") {
+                output.extend_from_slice(topic.str().as_bytes());
+                output.push(b':');
+            }
+            output.extend_from_slice(data);
+            (output, vec![])
+        }
+
+        fn fork(&self) -> Box<dyn Preprocessor> {
+            Box::new(TopicPrefixPreprocessor)
+        }
+
+        fn splitter(&self) -> Option<Box<dyn Splitter>> {
+            None
+        }
+    }
+
+    /// Preprocessor with an observable transformation, for registry tests.
+    struct UppercasePreprocessor;
+
+    impl Preprocessor for UppercasePreprocessor {
+        fn process(&mut self, data: &[u8]) -> (Vec<u8>, Vec<ParseError>) {
+            (data.to_ascii_uppercase(), vec![])
+        }
+
+        fn fork(&self) -> Box<dyn Preprocessor> {
+            Box::new(UppercasePreprocessor)
+        }
+
+        fn splitter(&self) -> Option<Box<dyn Splitter>> {
+            None
+        }
+    }
+
+    struct UppercasePreprocessorFactory;
+
+    impl PreprocessorFactory for UppercasePreprocessorFactory {
+        fn create(
+            &self,
+            _config: &PreprocessorConfig,
+        ) -> Result<Box<dyn Preprocessor>, PreprocessorCreateError> {
+            Ok(Box::new(UppercasePreprocessor))
+        }
+    }
+
+    fn make_config(name: &str) -> PreprocessorConfig {
+        PreprocessorConfig {
+            name: name.to_string(),
+            message_oriented: false,
+            config: json!({}),
+        }
+    }
+
+    fn make_metadata() -> ConnectorMetadata {
+        let mut metadata = ConnectorMetadata::new();
+        metadata.insert("topic", Variant::String(SqlString::from("events")));
+        metadata
+    }
+
+    #[test]
+    fn test_process_metadata_transforms_using_metadata() {
+        let mut preprocessor = TopicPrefixPreprocessor;
+
+        // `make_metadata` sets the `topic` attribute to `events`.
+        let metadata = make_metadata();
+        let (output, errors) = preprocessor.process_with_metadata(b"payload", Some(&metadata));
+        assert!(errors.is_empty(), "unexpected errors: {errors:?}");
+        assert_eq!(output, b"events:payload");
+
+        // Without metadata the record passes through unchanged.
+        let (output, errors) = preprocessor.process_with_metadata(b"payload", None);
+        assert!(errors.is_empty(), "unexpected errors: {errors:?}");
+        assert_eq!(output, b"payload");
+    }
+
+    /// A preprocessor that implements only `process_with_metadata` is still
+    /// callable through `process`, whose default implementation forwards with
+    /// no metadata.  This test compiles only because `process` has a default;
+    /// it pins the backward-compatible surface of the trait.
+    #[test]
+    fn test_process_defaults_to_process_with_metadata() {
+        let mut preprocessor = TopicPrefixPreprocessor;
+
+        let (output, errors) = preprocessor.process(b"payload");
+        assert!(errors.is_empty(), "unexpected errors: {errors:?}");
+        assert_eq!(output, b"payload");
+    }
+
+    #[test]
+    fn test_registry_register_and_get() {
+        let mut registry = PreprocessorRegistry::new();
+        registry.register("upper", Box::new(UppercasePreprocessorFactory));
+
+        let factory = registry.get("upper").expect("factory must be registered");
+        let mut preprocessor = factory.create(&make_config("upper")).unwrap();
+        let (output, errors) = preprocessor.process(b"xyz");
+        assert!(errors.is_empty(), "unexpected errors: {errors:?}");
+        assert_eq!(output, b"XYZ");
+
+        assert!(registry.get("missing").is_none());
+    }
+}
diff --git a/docs.feldera.com/docs/sql/udf.md b/docs.feldera.com/docs/sql/udf.md
@@ -730,11 +730,34 @@ pub struct PreprocessorConfig {
 
 ```rust
 pub trait Preprocessor: Send + Sync {
-    /// Transform raw input bytes.
+    /// Process raw input data and return transformed data.
+    ///
+    /// # Arguments
+    /// * `data` - Raw input data bytes
     ///
     /// Returns the transformed bytes together with any non-fatal parse errors
     /// encountered during transformation.
-    fn process(&mut self, data: &[u8]) -> (Vec<u8>, Vec<ParseError>);
+    /// The default implementation calls `process_with_metadata` without metadata.
+    fn process(&mut self, data: &[u8]) -> (Vec<u8>, Vec<ParseError>) {
+        self.process_with_metadata(data, None)
+    }
+
+    /// Process raw input data and return transformed data using connector metadata.
+    ///
+    /// # Arguments
+    /// * `data` - Raw input data bytes
+    /// * `metadata` - Connector metadata
+    ///
+    /// # Returns
+    /// The transformed data and any parse errors that occurred.
+    /// The default implementation just calls `process` ignoring the metadata.
+    fn process_with_metadata(
+        &mut self,
+        data: &[u8],
+        metadata: Option<&ConnectorMetadata>,
+    ) -> (Vec<u8>, Vec<ParseError>) {
+        self.process(data)
+    }
 
     /// Create an independent copy of this preprocessor with the same configuration.
     ///
@@ -749,6 +772,26 @@ pub trait Preprocessor: Send + Sync {
 }
 ```
 
+Each processing method has a default implementation that forwards to the
+other, so a preprocessor needs to implement exactly one of them:
+
+- Implement `process` when the transformation does not use connector
+  metadata.
+
+- Implement `process_with_metadata` when the transformation depends on
+  metadata.
+
+The pipeline only invokes `process_with_metadata`, which, by default,
+discards the metadata and invokes `process`.
+
+:::danger
+
+A preprocessor that implements neither processing method compiles
+without warnings, but the two default implementations will crash
+the pipeline with a stack overflow.
+
+:::
+
 There are two kinds of preprocessors: streaming and message-oriented,
 depending on the "message_oriented" configuration field value.
 
@@ -763,6 +806,62 @@ records from the preprocessor to produce even one record.
 Currently only "message-oriented" preprocessors are supported with
 fault-tolerance.
 
+#### `ConnectorMetadata`
+
+The `process_with_metadata` method receives an optional argument with type
+`feldera_adapterlib::ConnectorMetadata`.  Connector metadata contains
+key-value attributes that the input transport endpoint attaches to each
+record it produces.  A preprocessor can
+use these attributes to guide its transformation, for example to select
+a decryption key based on the topic a record arrived on.
+
+```rust
+/// Connector metadata attached to each input record.
+pub struct ConnectorMetadata(BTreeMap<Variant, Variant>);
+```
+
+Attribute names are `Variant::String` keys; values are
+[`Variant`](/sql/json/#the-variant-type)s.
+For example, the Kafka connector may attach an attribute `kafka_timestamp` with a
+type `feldera_sqllib::Timestamp`.
+
+If the connector does not provide metadata, its value is `None`.
+
+A preprocessor reads attributes from the `ConnectorMetadata` through two getters:
+
+```rust
+impl ConnectorMetadata {
+    /// Returns the attribute stored under `key`, or `None` if absent.
+    pub fn get(&self, key: &Variant) -> Option<&Variant>;
+
+    /// Returns the attribute stored under the string key `name`, or `None`
+    /// if absent.
+    pub fn get_by_name(&self, name: &str) -> Option<&Variant>;
+}
+```
+
+Because connectors key their attributes by string, `get_by_name` is the
+common lookup; `get` accepts an arbitrary `Variant` key.  For example, a
+preprocessor can select its behavior based on the Kafka topic a record
+arrived on:
+
+```rust
+if let Some(Variant::String(topic)) = metadata.get_by_name("kafka_topic") {
+    // Choose a transformation based on the topic name.
+}
+```
+
+Here is an example extracting a string value:
+
+```rust
+// Retrieve the topic name as a string; use "" when the attribute is
+// absent or is not a string.
+let topic: &str = match metadata.get_by_name("kafka_topic") {
+    Some(Variant::String(topic)) => topic.str(),
+    _ => "",
+};
+```
+
 #### `PreprocessorFactory` trait
 
 A factory is responsible for constructing `Preprocessor` instances from a JSON configuration
@@ -876,7 +975,7 @@ tracing = { version = "0.1.40" }
 
 ### Configuring a preprocessor in a connector
 
-Preprocessors are enabled by adding a `preprocess` field to the connector configuration.
+Preprocessors are enabled by adding a `preprocessor` field to the connector configuration.
 Each entry has
 - a string `name` (matching a registered factory)
 - a boolean `message_oriented` property