issues/6124: feat(postgres-cdc): configurable batch + memory_backpressure - #6847
Open
dumitru-nicolae-marasoiu wants to merge 2 commits into
Open
Conversation
…sure crates/feldera-types/src/transport/postgres.rs — added PostgresCdcBatchConfig (max_fill_ms, memory_budget_ratio, max_bytes) and PostgresCdcMemoryBackpressureConfig (activate_threshold, resume_threshold), each mirroring etl's own field shapes/defaults, with their own validate(). Wired into PostgresCdcReaderConfig as two new #[serde(default)] fields, so both are fully optional and back-compatible. 15 tests (up from 5). crates/adapters/src/integrated/postgres/cdc_input.rs — replaced the hardcoded BatchConfig::default() / MemoryBackpressureConfig::default() in PipelineConfig construction with values derived from the connector config, via two small conversion functions. 4 new tests (55 total, up from 51).
mihaibudiu
requested changes
Aug 12, 2026
| // this is sound despite the `f32` fields. | ||
| impl Eq for PostgresCdcMemoryBackpressureConfig {} | ||
|
|
||
| impl PostgresCdcMemoryBackpressureConfig { |
Contributor
There was a problem hiding this comment.
This is fine here, but we are duplicating connector validation in the SQL compiler as well (the Java code) - the feedback is much faster. Can you ask Claude to duplicate these checks in the SQL compiler? If it can't figure it out I can give you some pointers, but it's almost the same code.
There was a problem hiding this comment.
yeah "we" hope the duplication is in place now for fail fast purposes, can u pls check?
Two connector-validation paths exist in this codebase, and they run at very different points in the user's workflow:
1. Rust-side (crates/feldera-types/src/transport/postgres.rs:189, called from cdc_input.rs:63) — only runs when the connector actually starts, i.e. after a pipeline is deployed and Feldera tries to connect and begin replication. That's a slow feedback loop: write SQL → deploy pipeline → wait for connector startup → get an error string back.
2. Java-side (ConnectorValidator.java → each config class's validate(), e.g. PostgresCdcReaderConfig.validate()) — runs during SQL compilation of the CREATE TABLE ... WITH ('connectors' = ...) statement, before anything is ever deployed. It's also able to map the bad field back to its exact line/column in the original SQL source (via the JSON-pointer → source-position machinery in ConfigReporter), so the user gets a precise inline warning instead of an opaque runtime error.
Every other connector in this file already follows that pattern (Kafka, S3, Nats, the plain Postgres reader/writer, etc.) — each has matching validation logic on both sides. Mihai's point was that this PR added new fields (batch, memory_backpressure) with range checks only on the Rust runtime side, so a user with e.g. resume_threshold >= activate_threshold wouldn't find out until they deploy and the connector fails to start, instead of getting a compile-time warning immediately. Duplicating the checks closes that gap and keeps the new fields consistent with how the rest of the connector configs behave.
mihaibudiu
approved these changes
Aug 13, 2026
mihaibudiu
left a comment
Contributor
There was a problem hiding this comment.
This looks fine, but I haven't tested it, I hope you have verified that it does solve problems with resource allocation you have in practice.
Contributor
|
Please squash the commits to a single commit and use a descriptive and easy-to-read commit message. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR is only about offering a configuration option on batch & memory backpressure, since they relate to the same tradeoff.
crates/feldera-types/src/transport/postgres.rs — added PostgresCdcBatchConfig (max_fill_ms, memory_budget_ratio, max_bytes) and PostgresCdcMemoryBackpressureConfig (activate_threshold, resume_threshold), each mirroring etl's own field shapes/defaults, with their own validate(). Wired into PostgresCdcReaderConfig as two new #[serde(default)] fields, so both are fully optional and back-compatible. 15 tests (up from 5).
crates/adapters/src/integrated/postgres/cdc_input.rs — replaced the hardcoded BatchConfig::default() / MemoryBackpressureConfig::default() in PipelineConfig construction with values derived from the connector config, via two small conversion functions. 4 new tests (55 total, up from 51).
Describe Manual Test Plan
Checklist
Breaking Changes?
Mark if you think the answer is yes for any of these components:
Describe Incompatible Changes