sql: add CREATE SINK ... INTO POSTGRES

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.

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<serde_json::Value, String> {
                         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"

src/catalog/src/memory/objects.rs

@@ -1414,7 +1414,9 @@ impl Sink {
     pub fn combined_format(&self) -> Option<Cow<'_, str>> {
         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,
         }
     }
 

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 {

src/mz-deploy/src/project/resolve/normalize/visitor.rs

@@ -136,11 +136,12 @@ impl<T: NameTransformer> NormalizingVisitor<T> {
 
     /// 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<Raw>) {
         match connection {
-            CreateSinkConnection::Kafka { connection, .. } => {
+            CreateSinkConnection::Kafka { connection, .. }
+            | CreateSinkConnection::Postgres { connection, .. } => {
                 self.normalize_raw_item_name(connection);
             }
             CreateSinkConnection::Iceberg {

src/sql-parser/src/ast/defs/ddl.rs

@@ -1322,6 +1322,43 @@ pub struct IcebergSinkConfigOption<T: AstInfo> {
 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<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
+        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<T: AstInfo> {
+    pub name: PostgresSinkConfigOptionName,
+    pub value: Option<WithOptionValue<T>>,
+}
+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<T: AstInfo> {
         key: Option<SinkKey>,
         options: Vec<IcebergSinkConfigOption<T>>,
     },
+    Postgres {
+        connection: T::ItemName,
+        key: Option<SinkKey>,
+        options: Vec<PostgresSinkConfigOption<T>>,
+    },
 }
 
 impl<T: AstInfo> AstDisplay for CreateSinkConnection<T> {
@@ -1768,6 +1810,23 @@ impl<T: AstInfo> AstDisplay for CreateSinkConnection<T> {
                     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);
+                }
+            }
         }
     }
 }