[connectors] Add preprocessor API with ConnectorMetadata by mihaibudiu · Pull Request #6580 · feldera/feldera

mihaibudiu · 2026-07-02T20:24:12Z

Two commits:

pass ConnectorMetadata by reference to parsers. I think this was an API design bug
implement new preprocessor method

Adds a new method to the Preprocessor::process_with_metadata trait which receives the connector metadata as well.
This change should be backwards-compatible.

Checklist

Unit tests added/updated
Documentation updated

mythical-fred

LGTM

mythical-fred · 2026-07-02T22:00:46Z

+        &mut self,
+        data: &[u8],
+        metadata: Option<ConnectorMetadata>,
+    ) -> (Vec<u8>, Vec<ParseError>);


Nit: showing process_with_metadata with a bare ; makes it look like a required method. Since it has a default impl that delegates to process, consider showing the body so users know they only need to override it when they actually want the metadata:

fn process_with_metadata( &mut self, data: &[u8], metadata: Option<ConnectorMetadata>, ) -> (Vec<u8>, Vec<ParseError>) { self.process(data) }

mythical-fred · 2026-07-02T22:00:46Z

+    ///
+    /// # Arguments
+    /// * `data` - Raw input data bytes
+    /// * `_metadata` - Connector metadata


Nit: the doc comment says _metadata — drop the underscore prefix in documentation (it's an implementation detail for suppressing the unused-variable warning in the default impl).

mythical-fred

Backwards-compatible extension done the right way — new process_with_metadata with a default that delegates to process, so no existing impl breaks. Tests exercise both the "metadata-aware" and "no metadata" paths through the trait's public conversion into Variant, which is the right seam for user-defined preprocessors.

A couple of small things worth considering:

metadata.clone() per chunk in the streaming path. In StreamingPreprocessedParser::parse, metadata is cloned once for process_with_metadata and then again inside the while let Some(chunk) loop for every parser call. ConnectorMetadata is a BTreeMap<Variant, Variant>, and Variant clones aren't free (some variants carry Arcs, others own their bytes). For high-throughput sources with many attributes this becomes measurable. The typical fix is to hand the parser an &Option<ConnectorMetadata> (or wrap the metadata in Arc at the top of the pipeline) so the leaves clone only when they actually need to keep it. Not blocking for this PR, but worth a follow-up.
The trait in udf.md is now shown without the default body, which reads as "you must implement process_with_metadata". Since the whole point of the new signature is that it defaults to process, the doc snippet should either show the default ({ self.process(data) }) or add a one-liner immediately below the code block saying so. Otherwise UDF authors will wonder why the compiler doesn't complain when they omit it.
Minor: the doc paragraph If the connector does not provide metadata, its value is None. — recommend `None` in code font, and the sentence just above says "an optional argument" — you can drop one of them.

The preprocess → preprocessor field-name doc fix is a nice catch, thanks.

snkas

If I understand correctly, it's backward compatible in that there is a default implementation for the process_with_metadata variant that drops the metadata?

What's the difference between process(data) and process_with_metadata(data, None)? If a user implements both, then the former is never used?

Could you add an end-to-end integration test for it?

snkas · 2026-07-03T16:22:36Z

+
+    impl Preprocessor for TopicPrefixPreprocessor {
+        fn process(&mut self, data: &[u8]) -> (Vec<u8>, Vec<ParseError>) {
+            (data.to_vec(), vec![])


This seems backward, in that shouldn't process call process_with_metadata(data, None) rather than the other way around?

I did it this way for backwards compatibility - since we have users who have already implemented process but didn't know about the other method.

Apparently it's possible to have two default implementations which call each other, so maybe I will do that instead.

snkas · 2026-07-03T17:16:41Z

-        let (pre_data, mut pre_errors) = self.preprocessor.process(data);
+        let (pre_data, mut pre_errors) = self
+            .preprocessor
+            .process_with_metadata(data, metadata.clone());


Why is metadata passed by value instead of an immutable ref?

I think this is a design mistake in the original parser API, I will submit a commit to change this.

mihaibudiu · 2026-07-03T18:12:07Z

I have force-pushed 2 commits.

mythical-fred

Rebase LGTM. Arc-based sharing is correct: Arc::make_mut gives copy-on-write semantics on insert, and the two From<...> for Variant impls correctly distinguish the owned (move) and borrowed (cheap Arc::clone) paths. Two tiny nits inline; none blocking.

mythical-fred · 2026-07-03T19:04:46Z

+the pipeline with a overflow.
+
+:::
+


Nit: "crash the pipeline with a overflow" -> "with a stack overflow" (or "with an infinite recursion"). The Rust source doc comment on process_with_metadata already says "stack overflow" correctly; docs should match.

mythical-fred · 2026-07-03T19:04:46Z

+    /// # Arguments
+    /// * `data` - Raw input data bytes
+    /// * `_metadata` - Connector metadata
+    ///


Nit (carried from prior review): the Arguments doc still names the param _metadata. The underscore is only there to silence the unused-var warning in the default body; readers of the rustdoc shouldn't see it. Prefer * metadata - Connector metadata. Cosmetic.

gz · 2026-07-07T00:12:22Z

 /// to populate some of the table columns.
 #[derive(Default, Clone, Debug, PartialEq, Eq)]
-pub struct ConnectorMetadata(BTreeMap<Variant, Variant>);
+pub struct ConnectorMetadata(Arc<BTreeMap<Variant, Variant>>);


I realize this is pre-existing but why are we using our sql types here in something unrelated to sql types

I wondered the same, these should probably be just strings. @ryzhyk do you know?

Connector metadata exists so we can put it in SQL tables. Which is why we use SQL types.

snkas

Is the preprocessor only used if the trait is implemented? As the default implementation would cause a stack overflow.

mythical-fred

The added process_with_metadata API and its tests look good — the default-forwarding pair is a clean way to keep the trait backward-compatible, and the TopicPrefixPreprocessor test exercises the metadata path end-to-end through the public Variant conversion. Small doc nit inline; my prior APPROVE stands.

mythical-fred · 2026-07-08T01:01:06Z

+
+A preprocessor that implements neither processing method compiles
+without warnings, but the two default implementations will crash
+the pipeline with a overflow.


Typo — should be a stack overflow, matching the trait docstring in preprocess.rs (which says "panic at runtime with a stack overflow"). Consistency helps readers who bounce between docs and source.

Signed-off-by: Mihai Budiu <mbudiu@feldera.com>

mythical-fred

Nice follow-up on tip 04ad9018:

ConnectorMetadata gains a proper public read API (get, get_by_name) with an insert/get key-encoding invariant test — much cleaner than round-tripping through Variant::from(metadata.clone()) and pattern-matching on the resulting Variant::Map.
The topic typo I flagged ("a overflow" → "a stack overflow") is fixed in the trait doc.
Docs updated to teach get_by_name as the common lookup, with a copy-pasteable Kafka-topic example.
Test coverage for both getters and for the metadata-carrying preprocessor path.

APPROVE stands.

mythical-fred approved these changes Jul 2, 2026

View reviewed changes

mythical-fred approved these changes Jul 3, 2026

View reviewed changes

mihaibudiu requested a review from snkas July 3, 2026 16:16

snkas reviewed Jul 3, 2026

View reviewed changes

mihaibudiu force-pushed the issue6578 branch from ace434c to ac751a5 Compare July 3, 2026 18:11

mythical-fred approved these changes Jul 3, 2026

View reviewed changes

mihaibudiu requested a review from snkas July 3, 2026 20:13

gz approved these changes Jul 7, 2026

View reviewed changes

snkas approved these changes Jul 7, 2026

View reviewed changes

Comment thread docs.feldera.com/docs/sql/udf.md Outdated

mihaibudiu force-pushed the issue6578 branch from ac751a5 to 29f93bb Compare July 7, 2026 22:37

mythical-fred approved these changes Jul 8, 2026

View reviewed changes

mihaibudiu force-pushed the issue6578 branch 2 times, most recently from 088acd0 to e655a1a Compare July 9, 2026 20:55

[connectors] Add preprocessor API with ConnectorMetadata

04ad901

Signed-off-by: Mihai Budiu <mbudiu@feldera.com>

mihaibudiu force-pushed the issue6578 branch from e655a1a to 04ad901 Compare July 9, 2026 20:57

mihaibudiu enabled auto-merge July 9, 2026 20:57

mihaibudiu added this pull request to the merge queue Jul 9, 2026

github-merge-queue Bot removed this pull request from the merge queue due to failed status checks Jul 9, 2026

mihaibudiu added this pull request to the merge queue Jul 9, 2026

mythical-fred approved these changes Jul 9, 2026

View reviewed changes

Merged via the queue into feldera:main with commit 6371a13 Jul 9, 2026
1 check passed

mihaibudiu deleted the issue6578 branch July 9, 2026 22:27

Uh oh!

Conversation

mihaibudiu commented Jul 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Checklist

Uh oh!

mythical-fred left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mythical-fred left a comment

Choose a reason for hiding this comment

Uh oh!

snkas left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mihaibudiu Jul 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mihaibudiu commented Jul 3, 2026

Uh oh!

mythical-fred left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

snkas left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mythical-fred left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mythical-fred left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

mihaibudiu commented Jul 2, 2026 •

edited

Loading

mihaibudiu Jul 3, 2026 •

edited

Loading