Skip to content

sql: add CREATE SINK ... INTO POSTGRES - #39704

Open
patrickwwbutler wants to merge 1 commit into
mainfrom
pg-sink-sql
Open

patrickwwbutler wants to merge 1 commit into
mainfrom
pg-sink-sql

Conversation

@patrickwwbutler

@patrickwwbutler patrickwwbutler commented Oct 9, 2026 •

Copy link
Copy Markdown
Contributor

Parse, purify and plan Postgres sinks, add the PostgresSinkConnection
storage type, and teach the catalog and mz-deploy about the new sink
connection. The storage dataflow is left as an unimplemented stub.

Tests: adds parser cases for CREATE SINK ... INTO POSTGRES to src/sql-parser/tests/testdata/ddl.

NOTE: no feature flag gates Postgres sinks, and until the rest of the stack lands render_sink is unimplemented!(), so creating one would panic clusterd. Merge this stack together, or gate the sink behind a flag first.

🤖 Generated with Claude Code

Parse, purify and plan Postgres sinks, add the PostgresSinkConnection
storage type, and teach the catalog and mz-deploy about the new sink
connection. The storage dataflow is left as an unimplemented stub.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
@patrickwwbutler
patrickwwbutler added this pull request to stack #39709 October 9, 2026 17:15
@patrickwwbutler
patrickwwbutler marked this pull request as ready for review October 9, 2026 17:18
@patrickwwbutler
patrickwwbutler requested review from a team and sjwiesman as code owners October 9, 2026 17:18

@peterdukelarsen peterdukelarsen 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.

Def would pref a feature flag earlier gating the SQL to make it easier to merge incrementally.

/// on the conservative side and return `true`.
fn redact_value(&self) -> bool {
match self {
PostgresSinkConfigOptionName::Schema | PostgresSinkConfigOptionName::Table => false,

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.

Do we have docs on what's safe/unsafe to log? I assume table names are not considered sensitive even though they're user-entered?

Comment thread src/sql/src/pure.rs
CreateSinkConnection::Postgres { connection, .. } => {
// Only check that the referenced connection is a Postgres one.
// TODO: reach the upstream server here, the way the Kafka arm above
// proves the broker is reachable.

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.

Not-blocking more something to think about medium term: Moritz has described to me a vision where we avoid calls to upstream systems in purification and instead just report errors to connect/validate vie the error stream for the created operation. Ideally we'd be able to set up some sort of syntactic sugar so we can still select a success/failure of connection from the sync (otherwise I think not getting immediate feedback is too big of a regression in behavior to be worth the cleanliness of dropping network calls from purification).

/// The schema holding the target table.
pub schema: String,
/// The table to write into.
pub table: String,

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.

any comment we should make about escaped/not escaped?

@def-

def- commented Oct 9, 2026

Copy link
Copy Markdown
Contributor

QA LLM Review

1. HIGH -- CREATE SINK ... INTO POSTGRES is ungated and its render path is unimplemented!()

src/storage/src/sink/postgres.rs:47

Any role that can create a sink can run CREATE SINK ... INTO POSTGRES once this merges, and the resulting RunSink command reaches PostgresSinkConnection::render_sink, which panics and aborts clusterd. The controller replays the export every time the replica reconnects, so the replica crash-loops and takes every index, materialized view, source and sink on that cluster down with it until someone drops the sink.

Details

The PR description flags the panic and offers two ways out: merge the stack together, or gate the sink first. Neither is in place. None of #39705 to #39708 adds a flag, so merging the whole stack ships Postgres sinks to production ungated. Merging this PR first, which is the normal order for a stack against main, leaves a user-triggerable cluster crash on main until #39707 lands. Nothing on the Postgres path calls require_feature_flag, either in the plan_sink arm (src/sql/src/plan/statement/ddl.rs:3315) or in postgres_sink_builder (ddl.rs:3855).

The preconditions are small. CREATE on a schema and on a cluster is enough to make a Postgres connection, a table and the sink. Neither planning nor purification contacts the upstream server (src/sql/src/pure.rs:683 only checks the connection type).

Fix: add an enable_postgres_sink feature flag in src/sql/src/session/vars/definitions.rs. Default it to false and set enable_for_item_parsing: true, as enable_metric_sink does, so already-created sinks still re-plan at boot. Require it in postgres_sink_builder, following scx.require_feature_flag(&ENABLE_METRIC_SINK) at ddl.rs:4426. Then turn it on in the mzcompose and sqllogictest defaults so the testdrive coverage in #39708 runs with it.

2. LOW -- COMMIT INTERVAL is silently accepted and ignored on Postgres sinks

src/sql/src/plan/statement/ddl.rs:3855

postgres_sink_builder never looks at commit_interval, so WITH (COMMIT INTERVAL = '0s'), or any other value, plans and is persisted in create_sql. Nothing in this PR or the rest of the stack reads it for Postgres sinks, and ALTER SINK ... SET (COMMIT INTERVAL = ...) restarts the sink without changing anything.

Details

Kafka rejects the option (kafka_sink_builder, "COMMIT INTERVAL option is not supported with KAFKA sinks"), and src/adapter/src/catalog/migrate.rs:1245 documents that only Iceberg sinks accept it. That comment stops being true with this PR. Once values are persisted, rejecting or honoring them later needs a catalog rewrite like ast_rewrite_small_commit_intervals. Bailing in postgres_sink_builder now, the way Kafka does, avoids that.

This branch has not been deployed

No deployments
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.

3 participants