From 7eb4c254bc4066abdd6909360c645ddefd9b7f8f Mon Sep 17 00:00:00 2001 From: Patrick Butler Date: Fri, 9 Oct 2026 13:04:31 -0400 Subject: [PATCH 1/3] sql: add CREATE SINK ... INTO POSTGRES 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) --- src/catalog-decode/src/create_sql.rs | 42 ++++++- src/catalog/src/memory/objects.rs | 6 +- src/mz-deploy/src/client/validation.rs | 7 ++ .../src/project/resolve/normalize/visitor.rs | 7 +- src/sql-parser/src/ast/defs/ddl.rs | 59 +++++++++ src/sql-parser/src/parser.rs | 90 +++++++++++++- src/sql-parser/tests/testdata/ddl | 21 ++++ src/sql/src/lib.rs | 1 + src/sql/src/plan/statement/ddl.rs | 90 ++++++++++++-- src/sql/src/postgres.rs | 18 +++ src/sql/src/pure.rs | 14 +++ src/storage-types/src/sinks.rs | 112 ++++++++++++++++++ src/storage/src/render/sinks.rs | 1 + src/storage/src/sink.rs | 1 + src/storage/src/sink/postgres.rs | 49 ++++++++ 15 files changed, 497 insertions(+), 21 deletions(-) create mode 100644 src/sql/src/postgres.rs create mode 100644 src/storage/src/sink/postgres.rs diff --git a/src/catalog-decode/src/create_sql.rs b/src/catalog-decode/src/create_sql.rs index 7b6f924737d13..9de465825ba83 100644 --- a/src/catalog-decode/src/create_sql.rs +++ b/src/catalog-decode/src/create_sql.rs @@ -24,10 +24,10 @@ use mz_sql_parser::ast::{ CreateConnectionType, CreateSinkConnection, CreateSourceOptionName, CreateSubsourceOptionName, Expr, Format, FormatSpecifier, IcebergSinkConfigOptionName, IcebergSinkMode, IndexOptionName, KafkaSinkConfigOptionName, KafkaSourceConfigOptionName, MaterializedViewOptionName, - PgConfigOptionName, ProtobufSchema, Raw, RawClusterName, RawItemName, RefreshAtOptionValue, - RefreshEveryOptionValue, RefreshOptionValue, SinkEnvelope, SourceEnvelope, SourceErrorPolicy, - Statement, TableFromSourceOptionName, TableOptionName, UnresolvedItemName, Value, - WithOptionValue, + PgConfigOptionName, PostgresSinkConfigOptionName, ProtobufSchema, Raw, RawClusterName, + RawItemName, RefreshAtOptionValue, RefreshEveryOptionValue, RefreshOptionValue, SinkEnvelope, + SourceEnvelope, SourceErrorPolicy, Statement, TableFromSourceOptionName, TableOptionName, + UnresolvedItemName, Value, WithOptionValue, }; use prost::Message as _; use serde_json::json; @@ -525,6 +525,40 @@ pub fn item_details(a: &str) -> Result { info.insert("envelope_type", json!(envelope_type)); } } + CreateSinkConnection::Postgres { + connection, + options, + .. + } => { + info.insert("sink_type", json!("postgres")); + info.insert("connection_id", json!(item_id(connection)?)); + + let mut schema = None; + let mut table = None; + for option in options { + match option.name { + PostgresSinkConfigOptionName::Schema => { + schema = option.value.as_ref().and_then(option_string) + } + PostgresSinkConfigOptionName::Table => { + table = option.value.as_ref().and_then(option_string) + } + } + } + // SCHEMA is optional and defaults to `public` at plan time, + // so report that default rather than bailing. + info.insert("schema", json!(schema.unwrap_or_else(|| "public".into()))); + info.insert("table", json!(table.ok_or("postgres sink missing TABLE")?)); + + // Rows are written as SQL, so there are no format columns. + if let Some(envelope) = stmt.envelope { + let envelope_type = match envelope { + SinkEnvelope::Upsert => "upsert", + SinkEnvelope::Debezium => "debezium", + }; + info.insert("envelope_type", json!(envelope_type)); + } + } } "sink" diff --git a/src/catalog/src/memory/objects.rs b/src/catalog/src/memory/objects.rs index c10059143a261..3b1d89eae10d1 100644 --- a/src/catalog/src/memory/objects.rs +++ b/src/catalog/src/memory/objects.rs @@ -1414,7 +1414,9 @@ impl Sink { pub fn combined_format(&self) -> Option> { match &self.connection { StorageSinkConnection::Kafka(connection) => Some(connection.format.get_format_name()), - StorageSinkConnection::Iceberg(_) => None, + // Neither writes an encoded payload: Iceberg writes Parquet and + // Postgres writes SQL, so there is no format to report. + StorageSinkConnection::Iceberg(_) | StorageSinkConnection::Postgres(_) => None, } } @@ -1434,7 +1436,7 @@ impl Sink { let value_format = connection.format.value_format.get_format_name(); Some((key_format, value_format)) } - StorageSinkConnection::Iceberg(_) => None, + StorageSinkConnection::Iceberg(_) | StorageSinkConnection::Postgres(_) => None, } } diff --git a/src/mz-deploy/src/client/validation.rs b/src/mz-deploy/src/client/validation.rs index c7a163f608227..5076efe55c8d9 100644 --- a/src/mz-deploy/src/client/validation.rs +++ b/src/mz-deploy/src/client/validation.rs @@ -820,6 +820,13 @@ pub(crate) async fn validate_sink_connections_exist_impl( } ids } + CreateSinkConnection::Postgres { connection, .. } => { + vec![ObjectId::from_raw_item_name( + connection, + obj.id.expect_database(), + obj.id.schema(), + )] + } }; for conn_id in connection_ids { diff --git a/src/mz-deploy/src/project/resolve/normalize/visitor.rs b/src/mz-deploy/src/project/resolve/normalize/visitor.rs index 0e2bd95494ffa..1aee709ba9388 100644 --- a/src/mz-deploy/src/project/resolve/normalize/visitor.rs +++ b/src/mz-deploy/src/project/resolve/normalize/visitor.rs @@ -136,11 +136,12 @@ impl NormalizingVisitor { /// Normalize connection references in CREATE SINK statements. /// - /// Handles both Kafka and Iceberg sink types, ensuring their connection - /// references are normalized. + /// Handles every sink type, ensuring their connection references are + /// normalized. pub fn normalize_sink_connection(&self, connection: &mut CreateSinkConnection) { match connection { - CreateSinkConnection::Kafka { connection, .. } => { + CreateSinkConnection::Kafka { connection, .. } + | CreateSinkConnection::Postgres { connection, .. } => { self.normalize_raw_item_name(connection); } CreateSinkConnection::Iceberg { diff --git a/src/sql-parser/src/ast/defs/ddl.rs b/src/sql-parser/src/ast/defs/ddl.rs index 192d1c11dc76f..1822143d78da8 100644 --- a/src/sql-parser/src/ast/defs/ddl.rs +++ b/src/sql-parser/src/ast/defs/ddl.rs @@ -1322,6 +1322,43 @@ pub struct IcebergSinkConfigOption { impl_display_for_with_option!(IcebergSinkConfigOption); impl_display_t!(IcebergSinkConfigOption); +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum PostgresSinkConfigOptionName { + Schema, + Table, +} + +impl AstDisplay for PostgresSinkConfigOptionName { + fn fmt(&self, f: &mut AstFormatter) { + f.write_str(match self { + PostgresSinkConfigOptionName::Schema => "SCHEMA", + PostgresSinkConfigOptionName::Table => "TABLE", + }) + } +} +impl_display!(PostgresSinkConfigOptionName); + +impl WithOptionName for PostgresSinkConfigOptionName { + /// # WARNING + /// + /// Whenever implementing this trait consider very carefully whether or not + /// this value could contain sensitive user data. If you're uncertain, err + /// on the conservative side and return `true`. + fn redact_value(&self) -> bool { + match self { + PostgresSinkConfigOptionName::Schema | PostgresSinkConfigOptionName::Table => false, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct PostgresSinkConfigOption { + pub name: PostgresSinkConfigOptionName, + pub value: Option>, +} +impl_display_for_with_option!(PostgresSinkConfigOption); +impl_display_t!(PostgresSinkConfigOption); + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub enum PgConfigOptionName { /// Hex encoded string of binary serialization of @@ -1719,6 +1756,11 @@ pub enum CreateSinkConnection { key: Option, options: Vec>, }, + Postgres { + connection: T::ItemName, + key: Option, + options: Vec>, + }, } impl AstDisplay for CreateSinkConnection { @@ -1768,6 +1810,23 @@ impl AstDisplay for CreateSinkConnection { f.write_node(key); } } + CreateSinkConnection::Postgres { + connection, + key, + options, + } => { + f.write_str("POSTGRES CONNECTION "); + f.write_node(connection); + if !options.is_empty() { + f.write_str(" ("); + f.write_node(&display::comma_separated(options)); + f.write_str(")"); + } + if let Some(key) = key.as_ref() { + f.write_str(" "); + f.write_node(key); + } + } } } } diff --git a/src/sql-parser/src/parser.rs b/src/sql-parser/src/parser.rs index 184a5b1bc6401..8e8d3431b0f7a 100644 --- a/src/sql-parser/src/parser.rs +++ b/src/sql-parser/src/parser.rs @@ -2922,6 +2922,20 @@ impl<'a> Parser<'a> { }) } + fn parse_postgres_sink_config_option( + &mut self, + ) -> Result, ParserError> { + let name = match self.expect_one_of_keywords(&[SCHEMA, TABLE])? { + SCHEMA => PostgresSinkConfigOptionName::Schema, + TABLE => PostgresSinkConfigOptionName::Table, + _ => unreachable!(), + }; + Ok(PostgresSinkConfigOption { + name, + value: self.parse_optional_option_value()?, + }) + } + fn parse_kafka_sink_config_option( &mut self, ) -> Result, ParserError> { @@ -3603,6 +3617,43 @@ impl<'a> Parser<'a> { }) } + fn parse_create_postgres_sink( + &mut self, + name: Option, + in_cluster: Option, + from: RawItemName, + if_not_exists: bool, + connection: CreateSinkConnection, + ) -> Result, ParserError> { + let envelope = if self.parse_keyword(ENVELOPE) { + Some(self.parse_sink_envelope()?) + } else { + None + }; + + let with_options = if self.parse_keyword(WITH) { + self.expect_token(&Token::LParen)?; + let options = self.parse_comma_separated(Parser::parse_create_sink_option)?; + self.expect_token(&Token::RParen)?; + options + } else { + vec![] + }; + + Ok(CreateSinkStatement { + name, + in_cluster, + from, + connection, + // Rows are written as SQL, so there is no wire format to choose. + format: None, + envelope, + mode: None, + if_not_exists, + with_options, + }) + } + fn parse_create_kafka_sink( &mut self, name: Option, @@ -3685,6 +3736,9 @@ impl<'a> Parser<'a> { conn @ CreateSinkConnection::Iceberg { .. } => { self.parse_create_iceberg_sink(name, in_cluster, from, if_not_exists, conn) } + conn @ CreateSinkConnection::Postgres { .. } => { + self.parse_create_postgres_sink(name, in_cluster, from, if_not_exists, conn) + } }?; Ok(Statement::CreateSink(statement)) @@ -4175,13 +4229,47 @@ impl<'a> Parser<'a> { }) } + fn parse_create_postgres_sink_connection( + &mut self, + ) -> Result, ParserError> { + self.expect_keyword(CONNECTION)?; + let connection = self.parse_raw_name()?; + + let options = if self.consume_token(&Token::LParen) { + let options = self.parse_comma_separated(Parser::parse_postgres_sink_config_option)?; + self.expect_token(&Token::RParen)?; + options + } else { + vec![] + }; + + let key = if self.parse_keyword(KEY) { + let key_columns = self.parse_parenthesized_column_list(Mandatory)?; + + let not_enforced = self.parse_keywords(&[NOT, ENFORCED]); + Some(SinkKey { + key_columns, + not_enforced, + }) + } else { + None + }; + + Ok(CreateSinkConnection::Postgres { + connection, + key, + options, + }) + } + fn parse_create_sink_connection(&mut self) -> Result, ParserError> { - match self.expect_one_of_keywords(&[KAFKA, ICEBERG])? { + match self.expect_one_of_keywords(&[KAFKA, ICEBERG, POSTGRES])? { KAFKA => self.parse_create_kafka_sink_connection(), ICEBERG => { self.expect_keyword(CATALOG)?; self.parse_create_iceberg_sink_connection() } + POSTGRES => self.parse_create_postgres_sink_connection(), _ => unreachable!(), } } diff --git a/src/sql-parser/tests/testdata/ddl b/src/sql-parser/tests/testdata/ddl index 0eca98ad31e7f..c7f98f1b43b8d 100644 --- a/src/sql-parser/tests/testdata/ddl +++ b/src/sql-parser/tests/testdata/ddl @@ -1187,6 +1187,27 @@ CREATE SINK bar FROM foo INTO ICEBERG CATALOG CONNECTION s3tables (NAMESPACE = ' => CreateSink(CreateSinkStatement { name: Some(UnresolvedItemName([Ident("bar")])), in_cluster: None, if_not_exists: false, from: Name(UnresolvedItemName([Ident("foo")])), connection: Iceberg { catalog_connection: Name(UnresolvedItemName([Ident("s3tables")])), aws_connection: Some(Name(UnresolvedItemName([Ident("aws_conn")]))), key: None, options: [IcebergSinkConfigOption { name: Namespace, value: Some(Value(String("testnamespace"))) }, IcebergSinkConfigOption { name: Table, value: Some(Value(String("daily_sales"))) }] }, format: None, envelope: None, mode: Some(Append), with_options: [] }) +parse-statement +CREATE SINK bar FROM foo INTO POSTGRES CONNECTION pg (SCHEMA = 'analytics', TABLE = 'daily_sales') KEY (a) NOT ENFORCED; +---- +CREATE SINK bar FROM foo INTO POSTGRES CONNECTION pg (SCHEMA = 'analytics', TABLE = 'daily_sales') KEY (a) NOT ENFORCED +=> +CreateSink(CreateSinkStatement { name: Some(UnresolvedItemName([Ident("bar")])), in_cluster: None, if_not_exists: false, from: Name(UnresolvedItemName([Ident("foo")])), connection: Postgres { connection: Name(UnresolvedItemName([Ident("pg")])), key: Some(SinkKey { key_columns: [Ident("a")], not_enforced: true }), options: [PostgresSinkConfigOption { name: Schema, value: Some(Value(String("analytics"))) }, PostgresSinkConfigOption { name: Table, value: Some(Value(String("daily_sales"))) }] }, format: None, envelope: None, mode: None, with_options: [] }) + +parse-statement +CREATE SINK bar FROM foo INTO POSTGRES CONNECTION pg (TABLE = 'daily_sales'); +---- +CREATE SINK bar FROM foo INTO POSTGRES CONNECTION pg (TABLE = 'daily_sales') +=> +CreateSink(CreateSinkStatement { name: Some(UnresolvedItemName([Ident("bar")])), in_cluster: None, if_not_exists: false, from: Name(UnresolvedItemName([Ident("foo")])), connection: Postgres { connection: Name(UnresolvedItemName([Ident("pg")])), key: None, options: [PostgresSinkConfigOption { name: Table, value: Some(Value(String("daily_sales"))) }] }, format: None, envelope: None, mode: None, with_options: [] }) + +parse-statement +CREATE SINK bar FROM foo INTO POSTGRES CONNECTION pg (BLAH = 'boo!'); +---- +error: Expected one of SCHEMA or TABLE, found identifier "blah" +CREATE SINK bar FROM foo INTO POSTGRES CONNECTION pg (BLAH = 'boo!'); + ^ + parse-statement CREATE METRIC SINK m IN CLUSTER c FROM v ---- diff --git a/src/sql/src/lib.rs b/src/sql/src/lib.rs index 010bb05996f7d..b760998991a43 100644 --- a/src/sql/src/lib.rs +++ b/src/sql/src/lib.rs @@ -157,6 +157,7 @@ pub mod iceberg; pub use mz_compute_types::optimizer_metrics; pub mod parse; pub mod plan; +pub mod postgres; pub mod pure; pub mod rbac; pub mod session; diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs index ae12eb7663ef6..3553172fa394c 100644 --- a/src/sql/src/plan/statement/ddl.rs +++ b/src/sql/src/plan/statement/ddl.rs @@ -77,13 +77,13 @@ use mz_sql_parser::ast::{ MaterializedViewOptionName, MySqlConfigOption, MySqlConfigOptionName, NetworkPolicyOption, NetworkPolicyOptionName, NetworkPolicyRuleDefinition, NetworkPolicyRuleOption, NetworkPolicyRuleOptionName, OnHydrationOptionValue, PgConfigOption, PgConfigOptionName, - ProtobufSchema, QualifiedReplica, RefreshAtOptionValue, RefreshEveryOptionValue, - RefreshOptionValue, ReplicaDefinition, ReplicaOption, ReplicaOptionName, RoleAttribute, - SetRoleVar, SourceErrorPolicy, SourceIncludeMetadata, SqlServerConfigOption, - SqlServerConfigOptionName, Statement, TableConstraint, TableFromSourceColumns, - TableFromSourceOption, TableFromSourceOptionName, TableOption, TableOptionName, - UnresolvedDatabaseName, UnresolvedItemName, UnresolvedObjectName, UnresolvedSchemaName, Value, - ViewDefinition, WithOptionValue, + PostgresSinkConfigOption, ProtobufSchema, QualifiedReplica, RefreshAtOptionValue, + RefreshEveryOptionValue, RefreshOptionValue, ReplicaDefinition, ReplicaOption, + ReplicaOptionName, RoleAttribute, SetRoleVar, SourceErrorPolicy, SourceIncludeMetadata, + SqlServerConfigOption, SqlServerConfigOptionName, Statement, TableConstraint, + TableFromSourceColumns, TableFromSourceOption, TableFromSourceOptionName, TableOption, + TableOptionName, UnresolvedDatabaseName, UnresolvedItemName, UnresolvedObjectName, + UnresolvedSchemaName, Value, ViewDefinition, WithOptionValue, }; use mz_sql_parser::ident; use mz_sql_parser::parser::StatementParseResult; @@ -91,7 +91,7 @@ use mz_storage_types::connections::inline::ReferencedConnection; use mz_storage_types::connections::{Connection, KafkaTopicOptions}; use mz_storage_types::sinks::{ IcebergSinkConnection, KafkaIdStyle, KafkaSinkConnection, KafkaSinkFormat, KafkaSinkFormatType, - SinkEnvelope, StorageSinkConnection, iceberg_type_overrides, + PostgresSinkConnection, SinkEnvelope, StorageSinkConnection, iceberg_type_overrides, }; use mz_storage_types::sources::encoding::{ AvroEncoding, ColumnSpec, CsvEncoding, DataEncoding, ProtobufEncoding, RegexEncoding, @@ -168,6 +168,7 @@ use crate::plan::{ TableDataSource, Type, VariableValue, View, WebhookBodyFormat, WebhookHeaderFilters, WebhookHeaders, WebhookValidation, literal, plan_utils, query, transform_ast, }; +use crate::postgres::PostgresSinkConfigOptionExtracted; use crate::session::vars::{ self, ENABLE_AUTO_SCALING_STRATEGY, ENABLE_CLUSTER_SCHEDULE_REFRESH, ENABLE_COLLECTION_PARTITION_BY, ENABLE_CREATE_TABLE_FROM_SOURCE, ENABLE_KAFKA_SINK_HEADERS, @@ -3321,6 +3322,16 @@ fn plan_sink( (CreateSinkConnection::Iceberg { .. }, Some(_), _) => { sql_bail!("ENVELOPE is not supported for Iceberg sinks, use MODE instead") } + // Postgres sinks have no envelope. They mirror the collection into the + // target table, and the KEY clause alone decides whether that table is + // keyed. The value stored here is inert for them. + (CreateSinkConnection::Postgres { .. }, None, None) => SinkEnvelope::Upsert, + (CreateSinkConnection::Postgres { .. }, Some(_), _) => { + sql_bail!("ENVELOPE is not supported for Postgres sinks") + } + (CreateSinkConnection::Postgres { .. }, _, Some(_)) => { + sql_bail!("MODE is not supported for Postgres sinks") + } }; let from_name = &from; @@ -3357,7 +3368,8 @@ fn plan_sink( .ok_or_else(|| sql_err!("item does not have a relation description"))?; let key_indices = match &connection { CreateSinkConnection::Kafka { key: Some(key), .. } - | CreateSinkConnection::Iceberg { key: Some(key), .. } => { + | CreateSinkConnection::Iceberg { key: Some(key), .. } + | CreateSinkConnection::Postgres { key: Some(key), .. } => { let key_columns = key .key_columns .clone() @@ -3476,7 +3488,8 @@ fn plan_sink( Some(indices) } CreateSinkConnection::Kafka { key: None, .. } - | CreateSinkConnection::Iceberg { key: None, .. } => None, + | CreateSinkConnection::Iceberg { key: None, .. } + | CreateSinkConnection::Postgres { key: None, .. } => None, }; if key_indices.is_some() && envelope == SinkEnvelope::Append { @@ -3553,7 +3566,12 @@ fn plan_sink( (RelationDesc::new(typ, names), key_indices) }); - if key_desc_and_indices.is_none() && envelope == SinkEnvelope::Upsert { + // Postgres sinks are exempt: having no envelope, they have nothing that + // demands a key. Without one the target table holds a bag of rows whose + // multiplicities the sink maintains exactly. + let requires_key = envelope == SinkEnvelope::Upsert + && !matches!(connection, CreateSinkConnection::Postgres { .. }); + if key_desc_and_indices.is_none() && requires_key { return Err(PlanError::UpsertSinkWithoutKey); } @@ -3598,6 +3616,17 @@ fn plan_sink( commit_interval, &desc, )?, + CreateSinkConnection::Postgres { + connection, + options, + .. + } => postgres_sink_builder( + scx, + connection, + options, + relation_key_indices, + key_desc_and_indices, + )?, }; // WITH SNAPSHOT defaults to true @@ -3840,6 +3869,45 @@ fn iceberg_sink_builder( })) } +fn postgres_sink_builder( + scx: &StatementContext, + connection: ResolvedItemName, + options: Vec>, + relation_key_indices: Option>, + key_desc_and_indices: Option<(RelationDesc, Vec)>, +) -> Result, PlanError> { + let connection_item = scx.get_item_by_resolved_name(&connection)?; + let connection_id = connection_item.id(); + if !matches!(connection_item.connection()?, Connection::Postgres(_)) { + sql_bail!( + "{} is not a postgres connection", + scx.catalog + .resolve_full_name(connection_item.name()) + .to_string() + .quoted() + ); + }; + + let PostgresSinkConfigOptionExtracted { + schema, + table, + seen: _, + }: PostgresSinkConfigOptionExtracted = options.try_into()?; + + let Some(table) = table else { + sql_bail!("Postgres sink must specify TABLE"); + }; + + Ok(StorageSinkConnection::Postgres(PostgresSinkConnection { + connection_id, + connection: connection_id, + schema, + table, + relation_key_indices, + key_desc_and_indices, + })) +} + fn kafka_sink_builder( scx: &StatementContext, connection: ResolvedItemName, diff --git a/src/sql/src/postgres.rs b/src/sql/src/postgres.rs new file mode 100644 index 0000000000000..2129847558b8f --- /dev/null +++ b/src/sql/src/postgres.rs @@ -0,0 +1,18 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +use crate::names::Aug; +use mz_sql_parser::ast::display::AstDisplay; +use mz_sql_parser::ast::{PostgresSinkConfigOption, PostgresSinkConfigOptionName}; + +generate_extracted_config!( + PostgresSinkConfigOption, + (Table, String), + (Schema, String, Default("public".to_string())) +); diff --git a/src/sql/src/pure.rs b/src/sql/src/pure.rs index 3d25da86297c0..d3f5212bf72e2 100644 --- a/src/sql/src/pure.rs +++ b/src/sql/src/pure.rs @@ -680,6 +680,20 @@ async fn purify_create_sink( .await .map_err(|e| IcebergSinkPurificationError::CatalogError(Arc::new(e)))?; } + 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. + let scx = StatementContext::new(None, &catalog); + let item = scx.get_item_by_resolved_name(connection)?; + match item.connection()? { + Connection::Postgres(_) => {} + _ => sql_bail!( + "{} is not a postgres connection", + scx.catalog.resolve_full_name(item.name()) + ), + } + } } let mut csr_connection_ids = BTreeSet::new(); diff --git a/src/storage-types/src/sinks.rs b/src/storage-types/src/sinks.rs index d5c3dd4c8a5a0..a74fa75a56699 100644 --- a/src/storage-types/src/sinks.rs +++ b/src/storage-types/src/sinks.rs @@ -137,6 +137,7 @@ pub enum SinkEnvelope { pub enum StorageSinkConnection { Kafka(KafkaSinkConnection), Iceberg(IcebergSinkConnection), + Postgres(PostgresSinkConnection), } impl StorageSinkConnection { @@ -159,6 +160,9 @@ impl StorageSinkConnection { (StorageSinkConnection::Iceberg(s), StorageSinkConnection::Iceberg(o)) => { s.alter_compatible(id, o)? } + (StorageSinkConnection::Postgres(s), StorageSinkConnection::Postgres(o)) => { + s.alter_compatible(id, o)? + } _ => { tracing::warn!( "StorageSinkConnection incompatible:\nself:\n{:#?}\n\nother\n{:#?}", @@ -180,6 +184,7 @@ impl IntoInlineConnection match self { Self::Kafka(conn) => StorageSinkConnection::Kafka(conn.into_inline_connection(r)), Self::Iceberg(conn) => StorageSinkConnection::Iceberg(conn.into_inline_connection(r)), + Self::Postgres(conn) => StorageSinkConnection::Postgres(conn.into_inline_connection(r)), } } } @@ -197,6 +202,7 @@ impl StorageSinkConnection { catalog_connection_id: connection_id, .. }) => Some(*connection_id), + Postgres(PostgresSinkConnection { connection_id, .. }) => Some(*connection_id), } } @@ -209,6 +215,7 @@ impl StorageSinkConnection { match self { Kafka(_) => "kafka", Iceberg(_) => "iceberg", + Postgres(_) => "postgres", } } } @@ -838,3 +845,108 @@ impl IntoInlineConnection } } } + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))] +pub struct PostgresSinkConnection { + pub connection_id: CatalogItemId, + pub connection: C::Pg, + /// The schema holding the target table. + pub schema: String, + /// The table to write into. + pub table: String, + pub relation_key_indices: Option>, + pub key_desc_and_indices: Option<(RelationDesc, Vec)>, +} + +impl PostgresSinkConnection { + /// Schema in the target database that holds Materialize's bookkeeping + /// tables. + pub const MZ_SCHEMA: &str = "materialize"; + /// Progress table shared by every sink writing into the target database, + /// one row per sink. + pub const PROGRESS_TABLE: &str = "sink_progress"; + + /// Name of the staging table that sits next to the target table in + /// `self.schema`. + /// + /// NOTE: PostgreSQL truncates identifiers to 63 bytes, so a very long + /// target table name can make two sinks' staging tables collide. + pub fn staging_table_name(&self, sink_id: GlobalId) -> String { + format!("{}_mz_staging_{}", self.table, sink_id) + } +} + +impl PostgresSinkConnection { + pub fn alter_compatible( + &self, + id: GlobalId, + other: &PostgresSinkConnection, + ) -> Result<(), AlterError> { + if self == other { + return Ok(()); + } + let PostgresSinkConnection { + connection_id, + connection, + schema, + table, + relation_key_indices, + key_desc_and_indices, + } = self; + + let compatibility_checks = [ + (connection_id == &other.connection_id, "connection_id"), + ( + connection.alter_compatible(id, &other.connection).is_ok(), + "connection", + ), + (schema == &other.schema, "schema"), + (table == &other.table, "table"), + ( + relation_key_indices == &other.relation_key_indices, + "relation_key_indices", + ), + ( + key_desc_and_indices == &other.key_desc_and_indices, + "key_desc_and_indices", + ), + ]; + for (compatible, field) in compatibility_checks { + if !compatible { + tracing::warn!( + "PostgresSinkConnection incompatible at {field}:\nself:\n{:#?}\n\nother\n{:#?}", + self, + other + ); + + return Err(AlterError { id }); + } + } + + Ok(()) + } +} + +impl IntoInlineConnection + for PostgresSinkConnection +{ + fn into_inline_connection(self, r: R) -> PostgresSinkConnection { + let PostgresSinkConnection { + connection_id, + connection, + schema, + table, + relation_key_indices, + key_desc_and_indices, + } = self; + PostgresSinkConnection { + connection_id, + connection: r.resolve_connection(connection).unwrap_pg(), + schema, + table, + relation_key_indices, + key_desc_and_indices, + } + } +} diff --git a/src/storage/src/render/sinks.rs b/src/storage/src/render/sinks.rs index 006cc25593655..cf26bfdf05930 100644 --- a/src/storage/src/render/sinks.rs +++ b/src/storage/src/render/sinks.rs @@ -260,5 +260,6 @@ fn get_sink_render_for<'scope>(connection: &StorageSinkConnection) -> Box Box::new(connection.clone()), StorageSinkConnection::Iceberg(connection) => Box::new(connection.clone()), + StorageSinkConnection::Postgres(connection) => Box::new(connection.clone()), } } diff --git a/src/storage/src/sink.rs b/src/storage/src/sink.rs index 12a32d0c35e36..532b7f5ef04b1 100644 --- a/src/storage/src/sink.rs +++ b/src/storage/src/sink.rs @@ -14,6 +14,7 @@ use timely::progress::Antichain; mod iceberg; mod kafka; +mod postgres; /// Formats the error reported when a sink's input has been compacted past the /// upper the sink last committed, so the sink can never resume. `remedy` tells diff --git a/src/storage/src/sink/postgres.rs b/src/storage/src/sink/postgres.rs new file mode 100644 index 0000000000000..22350d7acda13 --- /dev/null +++ b/src/storage/src/sink/postgres.rs @@ -0,0 +1,49 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +//! Postgres sink. + +use differential_dataflow::VecCollection; +use mz_repr::{Diff, GlobalId, Timestamp}; +use mz_storage_types::controller::CollectionMetadata; +use mz_storage_types::errors::DataflowError; +use mz_storage_types::sinks::{PostgresSinkConnection, StorageSinkDesc}; +use mz_timely_util::builder_async::PressOnDropButton; +use timely::dataflow::StreamVec; + +use crate::healthcheck::HealthStatusMessage; +use crate::render::sinks::{SinkBatchStream, SinkRender}; +use crate::storage_state::StorageState; + +impl<'scope> SinkRender<'scope> for PostgresSinkConnection { + fn get_key_indices(&self) -> Option<&[usize]> { + self.key_desc_and_indices + .as_ref() + .map(|(_, indices)| indices.as_slice()) + } + + fn get_relation_key_indices(&self) -> Option<&[usize]> { + self.relation_key_indices.as_deref() + } + + fn render_sink( + &self, + _storage_state: &mut StorageState, + _sink: &StorageSinkDesc, + _sink_id: GlobalId, + _batches: SinkBatchStream<'scope>, + _key_is_synthetic: bool, + _err_collection: VecCollection<'scope, Timestamp, DataflowError, Diff>, + ) -> ( + StreamVec<'scope, Timestamp, HealthStatusMessage>, + Vec, + ) { + unimplemented!("Postgres sink dataflow") + } +} From 95c25cad2a58b9d828cfef166bb36e47516e9f01 Mon Sep 17 00:00:00 2001 From: Patrick Butler Date: Fri, 13 Nov 2026 14:27:16 -0500 Subject: [PATCH 2/3] storage-types: drop the Postgres sink staging table name The sink now stages each window in a session temporary table, so no staging table sits next to the target table and nothing needs its name. Co-Authored-By: Claude Opus 5.5 --- src/storage-types/src/sinks.rs | 9 --------- 1 file changed, 9 deletions(-) diff --git a/src/storage-types/src/sinks.rs b/src/storage-types/src/sinks.rs index a74fa75a56699..8615a33da997e 100644 --- a/src/storage-types/src/sinks.rs +++ b/src/storage-types/src/sinks.rs @@ -866,15 +866,6 @@ impl PostgresSinkConnection { /// Progress table shared by every sink writing into the target database, /// one row per sink. pub const PROGRESS_TABLE: &str = "sink_progress"; - - /// Name of the staging table that sits next to the target table in - /// `self.schema`. - /// - /// NOTE: PostgreSQL truncates identifiers to 63 bytes, so a very long - /// target table name can make two sinks' staging tables collide. - pub fn staging_table_name(&self, sink_id: GlobalId) -> String { - format!("{}_mz_staging_{}", self.table, sink_id) - } } impl PostgresSinkConnection { From 216701f5f4ef4d523aac9ff73526688697cee1c0 Mon Sep 17 00:00:00 2001 From: Patrick Butler Date: Fri, 13 Nov 2026 14:27:16 -0500 Subject: [PATCH 3/3] doc: add a design doc for Postgres sinks Co-Authored-By: Claude Opus 5.5 --- .../design/20261016_postgres_sink.md | 138 ++++++++++++++++++ 1 file changed, 138 insertions(+) create mode 100644 doc/developer/design/20261016_postgres_sink.md diff --git a/doc/developer/design/20261016_postgres_sink.md b/doc/developer/design/20261016_postgres_sink.md new file mode 100644 index 0000000000000..bec794095886d --- /dev/null +++ b/doc/developer/design/20261016_postgres_sink.md @@ -0,0 +1,138 @@ +# Postgres sink + +- Associated: [#39163](https://github.com/MaterializeInc/materialize/pull/39163) + +## The problem + +Users want a materialized view kept up to date in a PostgreSQL table, with +the same exactly-once guarantees the Kafka sink gives. The first cut of the +sink used three operators, a persistent staging table per sink, and +`pg_terminate_backend` to fence earlier incarnations. This document records +the simpler design that replaced it. + +## Success criteria + +- A `CREATE SINK ... INTO POSTGRES CONNECTION` keeps the target table equal to + the sinked relation, applying each Materialize timestamp exactly once across + restarts, replica changes and `ALTER SINK`. +- A superseded incarnation of the sink (a zombie replica, or two replicas + running by mistake) can never commit to the target. +- The sink leaves nothing in the target database beyond the target table and + one bookkeeping row. + +## Solution proposal + +The dataflow has two operators, matching the Kafka sink: + +1. `encode` runs on every worker and turns each `(row, time, diff)` into a + `COPY ... BINARY` staging row (the source columns, `mz_timestamp`, + `mz_diff = ±1`). Encoding is the CPU-heavy part, so it is sharded. +2. `write` runs on one worker, chosen by hashing the sink id, and holds the + only connection to the target database. + +The write operator is the whole protocol. At startup it: + +- creates `materialize.sink_progress` and the target table if needed. The + sink's `KEY` becomes the primary key, so the per-window delete is an index + lookup. A key with a nullable column gets a plain index instead, and a + keyless sink's target gets neither. +- claims the sink's progress row: `UPDATE ... SET epoch = epoch + 1, + sink_version = $v WHERE sink_id = $id AND sink_version <= $v RETURNING + epoch, mz_frontier`. No row returned means a newer version of the sink owns + the table, and the operator halts. +- reconciles the persist progress shard with the recorded frontier. The row is + the source of truth because it moves in the same transaction as the data. + A shard behind the row is pulled forward. A shard ahead of the row means the + target lost committed writes, and the sink halts with that message. +- creates a session temporary table `mz_sink_stage ... ON COMMIT DELETE + ROWS`. + +Then, for each window `[lower, upper)` the input frontier closes, in one +transaction: + +``` +BEGIN ISOLATION LEVEL SERIALIZABLE +SELECT epoch, mz_frontier FROM materialize.sink_progress WHERE sink_id = $id + -- halt unless epoch is ours and mz_frontier = lower +COPY pg_temp.mz_sink_stage FROM STDIN (FORMAT BINARY) + -- streamed as rows arrive; rows exactly at `lower` go straight in, + -- which is what streams a snapshot without buffering it +WITH latest AS (SELECT DISTINCT ON (key) ... FROM pg_temp.mz_sink_stage + ORDER BY key, mz_timestamp DESC, mz_diff DESC), + deleted AS (DELETE FROM target t USING latest l WHERE t.key = l.key), + inserted AS (INSERT INTO target SELECT ... FROM latest WHERE mz_diff > 0) +SELECT counts +UPDATE materialize.sink_progress SET mz_frontier = upper + WHERE sink_id = $id AND epoch = $epoch -- halt unless one row +COMMIT +``` + +A sink without a key replaces the `WITH latest ...` statement with an +`INSERT` of the staged insertions, then a `DELETE` of as many copies of each +retracted row as were retracted, picked out by `ctid`. + +After the commit, and only then, the persist progress shard is advanced to +`upper`. + +### Why this is correct + +- **Exactly once.** A window is applied and its frontier recorded in one + transaction, so a crash leaves either both or neither. On restart the + recorded frontier is the resume point, and the as-of handling mirrors the + Kafka sink's: progress is only recorded once strictly beyond the as-of. +- **Fencing.** Every incarnation bumps the epoch before writing, and every + window checks the epoch at `BEGIN` and again in the final `UPDATE`. Under + `SERIALIZABLE` a zombie whose window straddles a newer claim fails at + commit, and a zombie that starts a window after the claim fails at + `BEGIN`. `sink_version` fences the same way across `ALTER SINK`, as the + Kafka and Iceberg sinks do. +- **Multiple timestamps per key in one window.** `DISTINCT ON` keeps the + latest event for each key, and at that timestamp the insertion over the + retraction, so an update lands as its new value and a delete as nothing. + Every touched key is deleted and the survivors reinserted. This is why a + key must be unique: with duplicates a key-matched retraction cannot say + which row to remove. The planner enforces it exactly as it does for Kafka + upsert sinks, `KEY (...)` must be a unique key of the relation unless + `NOT ENFORCED` is given. +- **Duplicates without a key.** A keyless target may hold identical rows, so + a retraction deletes exactly its count of matching rows by `ctid`, and the + window halts if fewer are found. The match is on every column and cannot + use an index, which is the cost of a sink without a key. +- **Nothing persistent to clean.** The staging table is session temporary. + A dropped connection drops it and rolls back the open window. + +### Comparison with the other sinks + +| | Kafka | Iceberg | Postgres | +| --- | --- | --- | --- | +| Encoding | all workers | all workers | all workers | +| Writing | one worker, transactional producer | all workers write files, one commits | one worker, one connection | +| Progress store | progress topic record | snapshot property | `materialize.sink_progress` row | +| Fence | producer epoch via `init_transactions` + `sink_version` | `sink_version` in snapshot | `epoch` + `sink_version` in progress row | +| Key required | yes for `ENVELOPE UPSERT` | yes for upsert mode | no | + +The single writer bounds throughput by one connection's `COPY` rate, the +same bound the Kafka sink has on one producer. PostgreSQL executes a `COPY` +and the apply statement single-threaded, so a very large snapshot is +applied by one backend. Parallel writers would need per-writer temporary +tables and a two-phase commit, which is not worth it until a workload shows +the bound. + +## Alternatives + +- **Persistent staging table and a truncate on restart.** Needs + `pg_terminate_backend` to clear out a zombie's open `COPY` before the + truncate, and leaves a table and two indexes per sink in the user's schema. +- **Row locks instead of `SERIALIZABLE`.** `SELECT ... FOR UPDATE` on the + progress row gives the same compare-and-set and avoids serialization + failures from the user's own serializable transactions on the target table. + Kept as the fallback if those failures turn up in practice. +- **Requiring a key.** Would halve the apply logic and keep every retraction + an index lookup, but rejects relations that have no unique key and whose + users want duplicates kept. + +## Open questions + +- Whether to record progress on every input frontier advance (as now, and as + Kafka does) or on a `COMMIT INTERVAL` like Iceberg, to reduce write traffic + to the user's database when the sink is idle.