Skip to content

issues/6124: feat(postgres-cdc): configurable batch + memory_backpressure - #6847

Open
dumitru-nicolae-marasoiu wants to merge 2 commits into
feldera:mainfrom
dumitru-nicolae-marasoiu:postgres-cdc/mem-vs-throughput-config
Open

issues/6124: feat(postgres-cdc): configurable batch + memory_backpressure#6847
dumitru-nicolae-marasoiu wants to merge 2 commits into
feldera:mainfrom
dumitru-nicolae-marasoiu:postgres-cdc/mem-vs-throughput-config

Conversation

@dumitru-nicolae-marasoiu

Copy link
Copy Markdown

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

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

Breaking Changes?

Mark if you think the answer is yes for any of these components:

Describe Incompatible Changes

…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).
// this is sound despite the `f32` fields.
impl Eq for PostgresCdcMemoryBackpressureConfig {}

impl PostgresCdcMemoryBackpressureConfig {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 mihaibudiu left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@mihaibudiu

Copy link
Copy Markdown
Contributor

Please squash the commits to a single commit and use a descriptive and easy-to-read commit message.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants