[SQL][connectors] Support for preprocessors in connectors by mihaibudiu · Pull Request #5681 · feldera/feldera

mihaibudiu · 2026-02-23T23:59:27Z

See the issue for a description of the architecture

Describe Manual Test Plan

I have used the web console to inspect the logs of the new python test, showing that indeed the preprocessor is installed and executed.

Checklist

Unit tests added/updated
Integration tests added/updated
Documentation updated
Changelog updated

Breaking Changes?

None, but this introduces an experimental API

mihaibudiu · 2026-02-24T00:04:31Z

Here are some log snippets from the pipeline manager logs:

[local_test_udps] 2026-02-23T23:26:26.836802Z  INFO feldera_adapterlib::preprocess:  Registering factory for preprocessor 'logger'
[local_test_udps] 2026-02-23T23:26:26.840303Z  INFO dbsp_adapters::controller:  Creating preprocessor for logger
[local_test_udps] 2026-02-23T23:26:26.843035Z  INFO dbsp_adapters::server:  Pipeline initialization complete
[manager] 2026-02-23T23:26:27.832577Z  INFO pipeline_manager::runner::pipeline_automata: Resources transition: provisioning -> provisioned (desired: provisioned) pipeline-id="019c8cd3-4232-7fe0-b9cd-a2eba32e69f2" pipeline-name="local_test_udps"
[manager] 2026-02-23T23:26:27.832608Z  INFO pipeline_manager::runner::pipeline_automata: Runtime transition: (none) -> initializing (desired: paused) pipeline-id="019c8cd3-4232-7fe0-b9cd-a2eba32e69f2" pipeline-name="local_test_udps"
[manager] 2026-02-23T23:26:27.834499Z  INFO pipeline_manager::runner::pipeline_automata: Runtime transition: initializing -> paused (desired: paused) pipeline-id="019c8cd3-4232-7fe0-b9cd-a2eba32e69f2" pipeline-name="local_test_udps"
[local_test_udps] 2026-02-23T23:26:30.884922Z  INFO dbsp_adapters::server:  start: Transitioning from Paused to Running
[manager] 2026-02-23T23:26:30.887240Z  INFO pipeline_manager::runner::pipeline_automata: Runtime transition: paused -> running (desired: running) pipeline-id="019c8cd3-4232-7fe0-b9cd-a2eba32e69f2" pipeline-name="local_test_udps"
[local_test_udps] 2026-02-23T23:26:30.929652Z  INFO feldera_pipe_pipeline_019c8cd3_4232_7fe0_b9cd_a2eba32e69f2_globals::udf:  Processed 1068899 bytes of data log.target="feldera_pipe_pipeline_019c8cd3_4232_7fe0_b9cd_a2eba32e69f2_globals::udf" log.module_path="feldera_pipe_pipeline_019c8cd3_4232_7fe0_b9cd_a2eba32e69f2_globals::udf" log.file="crates/feldera_pipe_pipeline_019c8cd3_4232_7fe0_b9cd_a2eba32e69f2_globals/src/udf.rs" log.line=19
[local_test_udps] 2026-02-23T23:26:30.972396Z  INFO

Copilot

Pull request overview

This pull request adds experimental support for user-defined preprocessors in SQL connectors. Preprocessors are Rust components that transform raw data bytes before they reach the format parser, enabling use cases like decryption, decompression, and protocol-specific framing removal.

Changes:

Introduces new Preprocessor and PreprocessorFactory traits in the adapterlib crate with a global registry mechanism
Adds preprocessor field to connector configuration with validation in the SQL compiler
Implements preprocessor registration code generation in the Rust backend that auto-registers user-defined preprocessors
Removes deprecated --jdbcSource compiler option and fixes several documentation typos

Reviewed changes

Copilot reviewed 33 out of 34 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
crates/adapterlib/src/preprocess.rs	New module defining preprocessor traits, error types, and global registry
crates/adapterlib/src/format.rs	Implements `PreprocessedParser` that chains preprocessor with parser
crates/adapters/src/preprocess.rs	Defines `PassthroughPreprocessor` for testing (not registered)
crates/adapters/src/controller.rs	Creates and integrates preprocessors into input pipelines
crates/feldera-types/src/config.rs	Adds `preprocessor` field to `ConnectorConfig`
sql-to-dbsp-compiler/.../SqlToRelCompiler.java	Validates preprocessor configuration at compile time
sql-to-dbsp-compiler/.../ToRustVisitor.java	Generates preprocessor registration code in compiled Rust output
sql-to-dbsp-compiler/.../ProgramMetadata.java	Collects preprocessor names from connector metadata
sql-to-dbsp-compiler/.../CompilerOptions.java	Removes deprecated `--jdbcSource` option
python/tests/runtime/test_udp.py	Adds integration test for user-defined preprocessors (with bug)
docs.feldera.com/docs/sql/udf.md	Documents preprocessor API, traits, and usage examples
sql-to-dbsp-compiler/.../MetadataTests.java	Adds validation and integration tests for preprocessor config
crates/adapters/src/controller/test.rs	Adds unit test for passthrough preprocessor

crates/adapters/src/preprocess.rs

python/tests/runtime/test_udp.py

Copilot · 2026-02-24T00:08:19Z

docs.feldera.com/docs/sql/udf.md

+### Example: passthrough preprocessor
+
+The following example shows a minimal preprocessor that returns its
+input unchanged but logs a message for every 1M data bytes processd.


Typo in comment: "processd" should be "processed".

Suggested change

input unchanged but logs a message for every 1M data bytes processd.

input unchanged but logs a message for every 1M data bytes processed.

crates/adapters/src/controller.rs

crates/adapterlib/src/format.rs

...ler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java

python/tests/runtime/test_udp.py

mythical-fred

Two blocking issues; see inline comments.

crates/adapters/src/controller/test.rs

crates/adapters/src/controller.rs

crates/feldera-types/src/config.rs

mythical-fred · 2026-02-24T01:51:09Z

...SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/IHasSchema.java

+                        }
+                    }
+                } catch (JsonProcessingException e) {
+                    // Ignore


Silent // Ignore on JsonProcessingException is a broken window. If the connector JSON is malformed, preprocessors silently fail to register — the pipeline starts, runs without the preprocessor, and the user gets no diagnostic.

At minimum log a warning. Better: propagate the error so the compiler can reject malformed connector config at compile time rather than silently producing wrong runtime behaviour.

...ler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java

python/tests/runtime/test_udp.py

gz

why is this the right design? why does this need to be part of the pipeline?

python/tests/runtime/test_udp.py

gz · 2026-02-24T00:09:52Z

python/tests/runtime/test_udp.py

+from feldera.testutils import FELDERA_TEST_NUM_WORKERS, FELDERA_TEST_NUM_HOSTS
+
+
+# TEst user-defined preprocessor


Suggested change

# TEst user-defined preprocessor

# Test user-defined preprocessor

crates/feldera-types/src/config.rs

python/tests/runtime/test_udp.py

crates/adapterlib/src/preprocess.rs

crates/adapters/README.md

docs.feldera.com/docs/sql/udf.md

python/tests/runtime/test_udp.py

ryzhyk

Thanks! Looks nice and simple. I'm not sure this covers all use cases, e.g., can a preprocessor have checkpointable state? But starting simple is probably a good idea.

Important question: will secret resolution work for this new config field? cc @snkas

ryzhyk · 2026-02-24T07:36:18Z

crates/adapterlib/src/errors/controller.rs

+            } => {
+                write!(
+                    f,
+                    "Error creating preprocessor for endpoint {endpoint_name}: {error}"


Does the preprocessor itself have a name too?

I don't really see the need for a preprocessor name, unless somehow you could stack multiple copies of the same preprocessor

crates/adapterlib/src/format.rs

crates/adapterlib/src/preprocess.rs

ryzhyk · 2026-02-24T07:52:54Z

crates/adapterlib/src/preprocess.rs

+
+    /// Get the unique instance of the PreprocessorRegistry
+    fn get_instance() -> &'static Mutex<PreprocessorRegistry> {
+        static INSTANCE: OnceLock<Mutex<PreprocessorRegistry>> = OnceLock::new();


That's a good point, but this would require creating the registry inside the runtime. I haven't read the rest of the PR yet, so not sure how it's done yet.

crates/adapters/src/controller/test.rs

crates/adapters/src/controller.rs

ryzhyk · 2026-02-24T08:28:40Z

why is this the right design? why does this need to be part of the pipeline?

There are folks who don't want to build sidecars or have any other kind of a middle man between the data stream and the pipeline.

crates/feldera-types/src/config.rs

crates/adapterlib/src/format.rs

mihaibudiu · 2026-02-25T02:00:57Z

I have addressed all Rust-related comments, except moving the registry to the Runtime (or Controller). I will need some advice to do that.

I haven't addressed two Java comments because I want to rebase on #5683 to improve the JSON-related errors.

mythical-fred

Good progress. Blocker #1 (passthrough factory not registered) is fixed.

Blocker #2 still open — type coverage.
The Python integration test still uses CREATE TABLE t (i INT). Per the rules for connector changes, the integration test needs to cover the full type matrix: integer variants, float, decimal, boolean, varchar, varbinary, date/time/timestamp, nullable. A preprocessor is data-type-agnostic in principle, but connector integration tests serve as a regression net for the entire pipeline. A single INT column does not provide that coverage.

Understand the Java comments are pending the #5683 rebase — that is fine, not blocking anything here.

snkas · 2026-02-25T16:38:23Z

What is the user flow in terms of defining preprocessors? It is part of the connector configuration I see, but then PreprocessorConfig has only a name and an opaque Value.

@ryzhyk Per https://docs.feldera.com/connectors/secret-references secrets are resolved for transport.config and format.config, as such preprocessor would not be resolved out of the box.

mihaibudiu · 2026-02-25T16:40:49Z

There is a python test case which shows the intended user flow. Besides connector configuration you implement two or three traits in udf.rs. I don't have a test with secrets.

snkas · 2026-02-25T16:55:06Z

Will the template also show up in udf_stubs ? It seems the user-defined prepreprocessor also has a lot of boiler plate that might make it more difficult to remember (e.g., LoggerPreprocessorFactory).

In the Python tests, it's configured as:

   "preprocessor": [{
      "name": "logger",
      "config": {}
   }]

Can the preprocessor configuration become strongly typed? Or if there is currently nothing to configure, then it can be left out?

mihaibudiu · 2026-02-25T16:59:18Z

This part of the schema is fixed. Both name and config are designed to be mandatory.
The configuration contents really depends on the type of preprocessor; each of them can have very different needs, user-defined, so it cannot be strongly-typed.

mihaibudiu · 2026-02-26T02:23:17Z

I have pushed one more commit which fixes most of the remaining comments.

The only thing I don't know how to do is to put the registry in the Runtime or Controller. I would appreciate a little help in figuring this out.

mihaibudiu · 2026-02-26T02:24:08Z

I plan to squash the commits before I merge, but I left them this way to show what's new

mythical-fred

Both blockers resolved: passthrough factory registered, and Python test now covers the full type matrix (INT, TINYINT, SMALLINT, BIGINT, REAL, DOUBLE, VARBINARY, DATE, TIME, TIMESTAMP, ARRAY, BOOLEAN, DECIMAL, VARCHAR). The JSON pointer error location is a nice improvement.

If using rebase merges: "Address most comments from first round of reviews" should be folded in before landing.

mihaibudiu · 2026-02-26T18:14:57Z

I think in general you may not be able to use the parser splitter for the preprocessor. There should be a way to say: no splitter, and don't use the parser splitter.

ryzhyk · 2026-02-27T17:58:21Z

I think in general you may not be able to use the parser splitter for the preprocessor. There should be a way to say: no splitter, and don't use the parser splitter.

The preprocessor exposes its own splitter to the transport connector. Internally, it uses the parser's splitter, which respects the parser's framing.

mihaibudiu · 2026-02-28T06:47:16Z

crates/adapterlib/src/format.rs

    fn clear(&mut self);
 }

+/// Helper for breaking a stream of data into groups of records using a


This code was moved unchanged from the adapters crate

mihaibudiu · 2026-02-28T06:50:00Z

@ryzhyk maybe you can review the last commit for the splitter implementation in PreprocessedParser.

Note that all tests with decryption were produced by Claude.
They do not really cover the code completely, since there is no splitting, but I plan to produce some tests that do as well.

gz · 2026-03-01T01:54:36Z

docs.feldera.com/docs/sql/udf.md

+   count: Arc<Mutex<u64>>,
+}
+
+impl Preprocessor for LoggerPreprocessor {


a better example than just couting bytes might be to use snappy library to decompress the data that comes in because it's an example that actually transforms the input

crates/adapters/src/preprocess.rs

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

blp · 2026-03-02T22:58:46Z

crates/adapterlib/src/format.rs

+    fn len(&self) -> BufferSize {
+        let mut size = BufferSize::empty();
+        for v in self.iter() {
+            size.add_assign(v.len());


Suggested change

size.add_assign(v.len());

size += v.len();

blp · 2026-03-02T23:03:30Z

crates/adapterlib/src/format.rs

+        for v in self.iter_mut() {
+            if remaining <= 0 {
+                break;
+            }
+            let buf = v.take_some(remaining);
+            if let Some(ib) = buf {
+                let len = ib.len().records;
+                remaining = remaining.saturating_sub(len);
+                result.push(ib);
+            }
+        }


Most of the time, this loop is going to reduce some (or all) of the inner input buffers to empty, and I would argue that in that case it should remove them entirely from the vector.

crates/adapterlib/src/format.rs

blp · 2026-03-02T23:25:13Z

crates/adapters/src/transport/http/input.rs

                    for chunk in chunks {
                        let (mut buffer, errors) = details.parser.parse(&chunk, None);
-                        details.consumer.buffered(buffer.len());
+                        let len = InputBuffer::len(&buffer);


Suggested change

let len = InputBuffer::len(&buffer);

let len = buffer.len();

blp · 2026-03-02T23:25:42Z

crates/adapters/src/transport/adhoc.rs

                for chunk in chunks {
                    let (mut buffer, errors) = details.parser.parse(&chunk, None);
-                    details.consumer.buffered(buffer.len());
+                    let len = InputBuffer::len(&buffer);


Suggested change

let len = InputBuffer::len(&buffer);

let len = buffer.len();

unless there's something weird going on that I don't recognize

blp · 2026-03-02T23:30:47Z

...bsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/MetadataTests.java

-                    --jdbcSource
-                      Connection string to a database that contains table metadata
-                      Default: <empty string>


This seems unrelated to the PR.

…tagram preprocessors are supported with fault-tolerance Signed-off-by: Mihai Budiu <mbudiu@feldera.com>

mihaibudiu force-pushed the preprocessor branch from 280ee98 to 7963bc4 Compare February 24, 2026 00:00

mihaibudiu requested review from Copilot and gz February 24, 2026 00:02

Copilot started reviewing on behalf of mihaibudiu February 24, 2026 00:02 View session

Copilot AI reviewed Feb 24, 2026

View reviewed changes

mihaibudiu force-pushed the preprocessor branch from eb78a3f to 66b52d7 Compare February 24, 2026 00:15

mythical-fred suggested changes Feb 24, 2026

View reviewed changes

gz reviewed Feb 24, 2026

View reviewed changes

ryzhyk reviewed Feb 24, 2026

View reviewed changes

crates/feldera-types/src/config.rs Outdated Show resolved Hide resolved

ryzhyk reviewed Feb 24, 2026

View reviewed changes

crates/adapterlib/src/format.rs Show resolved Hide resolved

mihaibudiu force-pushed the preprocessor branch from a01379f to 3ef39cf Compare February 25, 2026 01:58

mythical-fred suggested changes Feb 25, 2026

View reviewed changes

mihaibudiu force-pushed the preprocessor branch from 1bd1b28 to a2db4af Compare February 26, 2026 02:22

mythical-fred approved these changes Feb 26, 2026

View reviewed changes

mihaibudiu force-pushed the preprocessor branch from a2db4af to 544aeac Compare February 28, 2026 06:46

mihaibudiu commented Feb 28, 2026

View reviewed changes

mihaibudiu mentioned this pull request Mar 1, 2026

[connectors] Change the signature of Parse::parse to return a (Vec<Box<dyn InputBuffer>>, ...) #5725

Closed

gz reviewed Mar 1, 2026

View reviewed changes

crates/adapters/src/preprocess.rs Show resolved Hide resolved

mihaibudiu added 4 commits March 2, 2026 13:21

[SQL][connectors] Support for preprocessors in connectors

c3d23cd

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

Address most comments from first round of reviews

1ae7e04

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

Improve JSON errors and Python test coverage

582ab6b

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

Move preprocessor_factory_registry into the CircuitCatalog

2d1e40f

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

mihaibudiu force-pushed the preprocessor branch from 544aeac to 0979b18 Compare March 2, 2026 21:23

Use the parser's splitter in PreprocessedParser

d82bf4e

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

mihaibudiu force-pushed the preprocessor branch from 0979b18 to d82bf4e Compare March 2, 2026 21:30

mihaibudiu requested review from blp and ryzhyk March 2, 2026 21:30

blp reviewed Mar 2, 2026

View reviewed changes

mihaibudiu force-pushed the preprocessor branch 2 times, most recently from 3e94576 to 5ac34a1 Compare March 3, 2026 05:18

Introduce two kinds of preprocessors: datagram and streaming. Only da…

549d58f

…tagram preprocessors are supported with fault-tolerance Signed-off-by: Mihai Budiu <mbudiu@feldera.com>

mihaibudiu force-pushed the preprocessor branch from 5e1d2db to 549d58f Compare March 3, 2026 17:04

	input unchanged but logs a message for every 1M data bytes processd.
	input unchanged but logs a message for every 1M data bytes processed.

		from feldera.testutils import FELDERA_TEST_NUM_WORKERS, FELDERA_TEST_NUM_HOSTS


		# TEst user-defined preprocessor

Conversation

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

Uh oh!

Describe Manual Test Plan

Checklist

Breaking Changes?

Uh oh!

mihaibudiu commented Feb 24, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Copilot AI Feb 24, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mythical-fred left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

gz left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

ryzhyk 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!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

ryzhyk commented Feb 24, 2026

Uh oh!

Uh oh!

Uh oh!

mihaibudiu commented Feb 25, 2026

Uh oh!

mythical-fred left a comment

Choose a reason for hiding this comment

Uh oh!

snkas commented Feb 25, 2026

Uh oh!

mihaibudiu commented Feb 25, 2026

Uh oh!

snkas commented Feb 25, 2026

Uh oh!

mihaibudiu commented Feb 25, 2026

Uh oh!

mihaibudiu commented Feb 23, 2026 •

edited

Loading