storage: Add a FAILOVER option for PostgreSQL sources

misc/python/materialize/mzcompose/__init__.py

@@ -116,6 +116,7 @@ def get_minimal_system_parameters(
         "enable_lgalloc": "false",
         "enable_load_generator_counter": "true",
         "enable_logical_compaction_window": "true",
+        "enable_postgres_source_failover": "true",
         "enable_mcp_protocol_2026_07_28": "true",
         "enable_mcp_unified": "true",
         "enable_metric_sink": "true",

src/adapter/src/catalog/migrate.rs

@@ -417,7 +417,9 @@ fn rewrite_sources_to_tables(
         match &mut connection {
             CreateSourceConnection::Postgres { options, .. } => {
                 options.retain(|o| match o.name {
-                    PgConfigOptionName::Details | PgConfigOptionName::Publication => true,
+                    PgConfigOptionName::Details
+                    | PgConfigOptionName::Publication
+                    | PgConfigOptionName::Failover => true,
                     PgConfigOptionName::TextColumns | PgConfigOptionName::ExcludeColumns => false,
                 });
             }

src/sql-lexer/src/keywords.txt

@@ -188,6 +188,7 @@ Expressions
 External
 Extract
 Factor
+Failover
 False
 Fast
 Features

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

@@ -1329,6 +1329,8 @@ pub enum PgConfigOptionName {
     Details,
     /// The name of the publication to sync
     Publication,
+    /// Whether to create the replication slot with `FAILOVER`
+    Failover,
     /// Columns whose types you want to unconditionally format as text
     /// NOTE(roshan): This value is kept around to allow round-tripping a
     /// `CREATE SOURCE` statement while we still allow creating implicit
@@ -1350,6 +1352,7 @@ impl AstDisplay for PgConfigOptionName {
         f.write_str(match self {
             PgConfigOptionName::Details => "DETAILS",
             PgConfigOptionName::Publication => "PUBLICATION",
+            PgConfigOptionName::Failover => "FAILOVER",
             PgConfigOptionName::TextColumns => "TEXT COLUMNS",
             PgConfigOptionName::ExcludeColumns => "EXCLUDE COLUMNS",
         })
@@ -1367,6 +1370,7 @@ impl WithOptionName for PgConfigOptionName {
         match self {
             PgConfigOptionName::Details
             | PgConfigOptionName::Publication
+            | PgConfigOptionName::Failover
             | PgConfigOptionName::TextColumns
             | PgConfigOptionName::ExcludeColumns => false,
         }

src/sql-parser/src/parser.rs

@@ -3858,9 +3858,11 @@ impl<'a> Parser<'a> {
     }
 
     fn parse_pg_connection_option(&mut self) -> Result<PgConfigOption<Raw>, ParserError> {
-        let name = match self.expect_one_of_keywords(&[DETAILS, PUBLICATION, TEXT, EXCLUDE])? {
+        let keywords = [DETAILS, PUBLICATION, FAILOVER, TEXT, EXCLUDE];
+        let name = match self.expect_one_of_keywords(&keywords)? {
             DETAILS => PgConfigOptionName::Details,
             PUBLICATION => PgConfigOptionName::Publication,
+            FAILOVER => PgConfigOptionName::Failover,
             TEXT => {
                 self.expect_keyword(COLUMNS)?;
 

src/sql-parser/tests/testdata/ddl

@@ -3329,6 +3329,27 @@ CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION = 'mz_source')
 =>
 CreateSource(CreateSourceStatement { name: UnresolvedItemName([Ident("mz_source")]), in_cluster: None, col_names: [], connection: Postgres { connection: Name(UnresolvedItemName([Ident("pg")])), options: [PgConfigOption { name: Publication, value: Some(Value(String("mz_source"))) }] }, include_metadata: [], format: None, envelope: None, if_not_exists: false, key_constraint: None, with_options: [], external_references: Some(All), progress_subsource: None })
 
+parse-statement
+CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER = true);
+----
+CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION = 'mz_source', FAILOVER = true)
+=>
+CreateSource(CreateSourceStatement { name: UnresolvedItemName([Ident("mz_source")]), in_cluster: None, col_names: [], connection: Postgres { connection: Name(UnresolvedItemName([Ident("pg")])), options: [PgConfigOption { name: Publication, value: Some(Value(String("mz_source"))) }, PgConfigOption { name: Failover, value: Some(Value(Boolean(true))) }] }, include_metadata: [], format: None, envelope: None, if_not_exists: false, key_constraint: None, with_options: [], external_references: None, progress_subsource: None })
+
+parse-statement
+CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER false);
+----
+CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION = 'mz_source', FAILOVER = false)
+=>
+CreateSource(CreateSourceStatement { name: UnresolvedItemName([Ident("mz_source")]), in_cluster: None, col_names: [], connection: Postgres { connection: Name(UnresolvedItemName([Ident("pg")])), options: [PgConfigOption { name: Publication, value: Some(Value(String("mz_source"))) }, PgConfigOption { name: Failover, value: Some(Value(Boolean(false))) }] }, include_metadata: [], format: None, envelope: None, if_not_exists: false, key_constraint: None, with_options: [], external_references: None, progress_subsource: None })
+
+parse-statement
+CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER);
+----
+CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION = 'mz_source', FAILOVER)
+=>
+CreateSource(CreateSourceStatement { name: UnresolvedItemName([Ident("mz_source")]), in_cluster: None, col_names: [], connection: Postgres { connection: Name(UnresolvedItemName([Ident("pg")])), options: [PgConfigOption { name: Publication, value: Some(Value(String("mz_source"))) }, PgConfigOption { name: Failover, value: None }] }, include_metadata: [], format: None, envelope: None, if_not_exists: false, key_constraint: None, with_options: [], external_references: None, progress_subsource: None })
+
 parse-statement
 CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source');
 ----

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

@@ -550,6 +550,7 @@ generate_extracted_config!(
     PgConfigOption,
     (Details, String),
     (Publication, String),
+    (Failover, bool, Default(false)),
     (TextColumns, Vec::<UnresolvedItemName>, Default(vec![])),
     (ExcludeColumns, Vec::<UnresolvedItemName>, Default(vec![]))
 );
@@ -1173,6 +1174,7 @@ fn plan_postgres_source_connection(
     let PgConfigOptionExtracted {
         details,
         publication,
+        failover,
         // text columns are already part of the source-exports and are only included
         // in these options for round-tripping of a `CREATE SOURCE` statement. This should
         // be removed once we drop support for implicitly created subsources.
@@ -1181,8 +1183,11 @@ fn plan_postgres_source_connection(
         // in these options for round-tripping of a `CREATE SOURCE` statement. This should
         // be removed once we drop support for implicitly created subsources.
         exclude_columns: _,
-        seen: _,
+        seen,
     } = options.clone().try_into()?;
+    if seen.contains(&PgConfigOptionName::Failover) {
+        scx.require_feature_flag(&vars::ENABLE_POSTGRES_SOURCE_FAILOVER)?;
+    }
     let details = details
         .as_ref()
         .ok_or_else(|| internal_err!("Postgres source missing details"))?;
@@ -1197,6 +1202,7 @@ fn plan_postgres_source_connection(
         // Validated during purification.
         publication: publication.ok_or_else(|| internal_err!("PUBLICATION option is required"))?,
         publication_details,
+        failover,
     })
 }
 

src/sql/src/pure.rs

@@ -1090,6 +1090,7 @@ async fn execute_create_source_preparation(
         ) => {
             let crate::plan::statement::PgConfigOptionExtracted {
                 publication,
+                failover,
                 text_columns,
                 exclude_columns,
                 details,
@@ -1117,6 +1118,14 @@ async fn execute_create_source_preparation(
             // the latest LSN.
             let is_physical_replica = mz_postgres_util::get_is_in_recovery(&client).await?;
 
+            // PostgreSQL 17 introduced failover slots, and rejects creating one on a standby.
+            if failover
+                && (is_physical_replica
+                    || mz_postgres_util::schemas::get_pg_major_version(&client).await? < 17)
+            {
+                Err(PgSourcePurificationError::FailoverUnsupported)?;
+            }
+
             // Read after the references above, so it bounds the LSN their schemas belong to.
             let initial_lsn = MzOffset::from(
                 mz_postgres_util::fetch_max_lsn(&client, is_physical_replica).await?,

src/sql/src/pure/error.rs

@@ -61,6 +61,8 @@ pub enum PgSourcePurificationError {
     NotPgConnection(FullItemName),
     #[error("CONNECTION must specify PUBLICATION")]
     ConnectionMissingPublication,
+    #[error("FAILOVER requires the upstream to be a PostgreSQL 17+ primary")]
+    FailoverUnsupported,
     #[error(transparent)]
     InvalidConnection(PostgresConnectionValidationError),
 }

src/sql/src/session/vars/definitions.rs

@@ -2184,6 +2184,12 @@ feature_flags!(
         default: false,
         enable_for_item_parsing: true,
     },
+    {
+        name: enable_postgres_source_failover,
+        desc: "the FAILOVER option for PostgreSQL sources",
+        default: false,
+        enable_for_item_parsing: true,
+    },
     {
         name: enable_load_generator_counter,
         desc: "Create a LOAD GENERATOR COUNTER",

src/sql/system-parameters.json

@@ -1543,6 +1543,15 @@
       "userVisible": false,
       "dyncfg": false
     },
+    {
+      "name": "enable_postgres_source_failover",
+      "typeName": "boolean",
+      "default": "off",
+      "description": "Whether the FAILOVER option for PostgreSQL sources is allowed (Materialize).",
+      "scope": "environment",
+      "userVisible": false,
+      "dyncfg": false
+    },
     {
       "name": "enable_primary_key_not_enforced",
       "typeName": "boolean",

src/storage-types/src/sources/postgres.rs

@@ -34,6 +34,8 @@ pub struct PostgresSourceConnection<C: ConnectionAccess = InlinedConnection> {
     pub connection: C::Pg,
     pub publication: String,
     pub publication_details: PostgresSourcePublicationDetails,
+    /// Whether to create the replication slot with `FAILOVER`.
+    pub failover: bool,
 }
 
 impl<R: ConnectionResolver> IntoInlineConnection<PostgresSourceConnection, R>
@@ -45,13 +47,15 @@ impl<R: ConnectionResolver> IntoInlineConnection<PostgresSourceConnection, R>
             connection,
             publication,
             publication_details,
+            failover,
         } = self;
 
         PostgresSourceConnection {
             connection_id,
             connection: r.resolve_connection(connection).unwrap_pg(),
             publication,
             publication_details,
+            failover,
         }
     }
 }
@@ -142,6 +146,7 @@ impl<C: ConnectionAccess> AlterCompatible for PostgresSourceConnection<C> {
             connection,
             publication,
             publication_details,
+            failover,
         } = self;
 
         let compatibility_checks = [
@@ -157,6 +162,7 @@ impl<C: ConnectionAccess> AlterCompatible for PostgresSourceConnection<C> {
                     .is_ok(),
                 "publication_details",
             ),
+            (failover == &other.failover, "failover"),
         ];
 
         for (compatible, field) in compatibility_checks {

src/storage/src/source/postgres.rs

@@ -439,15 +439,24 @@ impl From<DefiniteError> for DataflowError {
     }
 }
 
+/// Creates the replication slot `slot` unless it exists, with `FAILOVER` if `failover` is set.
 async fn ensure_replication_slot(
     client: &Client,
     slot: &str,
+    failover: bool,
     timeout: Duration,
 ) -> Result<(), TransientError> {
-    let query = sql!(
-        "CREATE_REPLICATION_SLOT {} LOGICAL \"pgoutput\" NOEXPORT_SNAPSHOT",
-        Sql::ident(slot)
-    );
+    let query = if failover {
+        sql!(
+            "CREATE_REPLICATION_SLOT {} LOGICAL \"pgoutput\" (SNAPSHOT 'nothing', FAILOVER)",
+            Sql::ident(slot)
+        )
+    } else {
+        sql!(
+            "CREATE_REPLICATION_SLOT {} LOGICAL \"pgoutput\" NOEXPORT_SNAPSHOT",
+            Sql::ident(slot)
+        )
+    };
     match create_replication_slot(client, slot, query, timeout).await {
         Ok(_) => Ok(()),
         Err(TransientError::ReplicationSlotAlreadyExists) => {

src/storage/src/source/postgres/snapshot.rs

@@ -481,7 +481,13 @@ pub(crate) fn render<'scope>(
                 let main_slot = &connection.publication_details.slot;
 
                 tracing::info!(%id, "ensuring replication slot {main_slot} exists");
-                super::ensure_replication_slot(&client, main_slot, slot_creation_timeout).await?;
+                super::ensure_replication_slot(
+                    &client,
+                    main_slot,
+                    connection.failover,
+                    slot_creation_timeout,
+                )
+                .await?;
                 Some(client)
             } else {
                 None

test/launchdarkly-flag-consistency/mzcompose.py

@@ -270,6 +270,7 @@
     enable_load_generator_counter
     enable_load_generator_datums
     enable_managed_cluster_availability_zones
+    enable_postgres_source_failover
     enable_notices_for_equals_null
     enable_notices_for_index_already_exists
     enable_notices_for_index_empty_key

test/pg-cdc/mzcompose.py

@@ -205,6 +205,39 @@ def workflow_slot_creation_timeout(
         c.run_testdrive_files("--no-reset", "override/slot-creation-timeout-part-2.td")
 
 
+def workflow_replication_slot_failover(
+    c: Composition, parser: WorkflowArgumentParser
+) -> None:
+    """
+    Test that the FAILOVER source option decides whether a slot is a failover slot.
+    """
+    pg_version = get_targeted_pg_version(parser)
+    if pg_version is not None and int(pg_version.split(".")[0]) < 17:
+        return
+    with c.override(create_postgres(pg_version=pg_version)):
+        c.up("materialized", "postgres")
+        c.run_testdrive_files("override/replication-slot-failover.td")
+
+        pg_conn = psycopg.connect(
+            host="localhost",
+            user="postgres",
+            password="postgres",
+            port=c.default_port("postgres"),
+        )
+        expected = {"implied": True, "explicit_true": True, "explicit_false": False}
+        slots = c.sql_query("""
+            SELECT s.name, p.replication_slot
+            FROM mz_sources s JOIN mz_internal.mz_postgres_sources p USING (id)
+            """)
+        failover = dict(
+            pg_conn.execute(
+                "SELECT slot_name, failover FROM pg_replication_slots"
+            ).fetchall()
+        )
+        actual = {name: failover.get(slot) for name, slot in slots}
+        assert actual == expected, f"expected {expected}, found {actual}"
+
+
 def workflow_wal_level(c: Composition, parser: WorkflowArgumentParser) -> None:
     pg_version = get_targeted_pg_version(parser)
     for wal_level in ["replica", "minimal"]:

test/pg-cdc/override/replication-slot-failover.td

@@ -0,0 +1,73 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# 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.
+
+> CREATE SECRET pgpass AS 'postgres'
+
+> CREATE CONNECTION pg TO POSTGRES (
+    HOST postgres,
+    DATABASE postgres,
+    USER postgres,
+    PASSWORD SECRET pgpass
+  )
+
+$ postgres-execute connection=postgres://postgres:postgres@postgres
+ALTER USER postgres WITH replication;
+DROP SCHEMA IF EXISTS public CASCADE;
+CREATE SCHEMA public;
+
+DROP PUBLICATION IF EXISTS mz_source;
+CREATE PUBLICATION mz_source FOR ALL TABLES;
+
+CREATE TABLE pk_table (pk INTEGER PRIMARY KEY, f2 TEXT);
+ALTER TABLE pk_table REPLICA IDENTITY FULL;
+INSERT INTO pk_table VALUES (1, 'one');
+
+$ postgres-execute connection=mz_system
+ALTER SYSTEM SET enable_postgres_source_failover = false
+
+! CREATE SOURCE implied
+  FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER);
+contains:the FAILOVER option for PostgreSQL sources is not available
+
+! CREATE SOURCE explicit_false
+  FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER = false);
+contains:the FAILOVER option for PostgreSQL sources is not available
+
+$ postgres-execute connection=mz_system
+ALTER SYSTEM SET enable_postgres_source_failover = true
+
+> CREATE SOURCE implied
+  FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER);
+> CREATE SOURCE explicit_true
+  FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER = true);
+> CREATE SOURCE explicit_false
+  FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER = false);
+
+# A source only creates its slot once it ingests a table.
+> CREATE TABLE implied_t FROM SOURCE implied (REFERENCE pk_table);
+> CREATE TABLE explicit_true_t FROM SOURCE explicit_true (REFERENCE pk_table);
+> CREATE TABLE explicit_false_t FROM SOURCE explicit_false (REFERENCE pk_table);
+
+$ postgres-execute connection=postgres://postgres:postgres@postgres
+INSERT INTO pk_table VALUES (2, 'two');
+
+> SELECT * FROM implied_t
+1 one
+2 two
+
+> SELECT * FROM explicit_true_t
+1 one
+2 two
+
+> SELECT * FROM explicit_false_t
+1 one
+2 two
+
+> SELECT count(*) FROM mz_internal.mz_postgres_sources
+3