From e5bb182a3c80bb113dcb56bd2ed30bbcb9e852fd Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Thu, 12 Nov 2026 15:01:43 -0500 Subject: [PATCH 1/7] storage: Add a flag to create PostgreSQL replication slots with failover PostgreSQL 17+ synchronizes a logical replication slot to standbys only if the slot was created with FAILOVER. A source's slot was created without it, so the slot did not survive an upstream failover. The new dyncfg `pg_source_replication_slot_failover` (default off) makes a source create its slot with FAILOVER. It is replica-scoped, so it can be targeted at single clusters. Servers that cannot create such a slot (older than PostgreSQL 17, or a standby) get a plain slot as before. Existing slots are not altered. Part of CS-752 Co-Authored-By: Claude Fable 5.2 --- misc/python/materialize/mzcompose/__init__.py | 9 ++++ .../materialize/parallel_workload/action.py | 3 ++ src/storage-types/src/dyncfgs.rs | 13 ++++++ src/storage/src/source/postgres.rs | 34 ++++++++++++-- src/storage/src/source/postgres/snapshot.rs | 12 ++++- .../mzcompose.py | 1 + test/pg-cdc/mzcompose.py | 29 ++++++++++++ .../override/replication-slot-failover.td | 45 +++++++++++++++++++ 8 files changed, 140 insertions(+), 6 deletions(-) create mode 100644 test/pg-cdc/override/replication-slot-failover.td diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 6b5368ce68786..a67a0a9651aab 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -657,6 +657,15 @@ def get_variable_system_parameters( if version >= MzVersion.parse_mz("v26.44.0-dev") else [] ), + *( + [ + VariableSystemParameter( + "pg_source_replication_slot_failover", "true", ["true", "false"] + ) + ] + if version >= MzVersion.parse_mz("v26.47.0-dev") + else [] + ), VariableSystemParameter( "persist_validate_part_bounds_on_read", "false", ["true", "false"] ), diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 25a91e251cd59..422da7605b4e6 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -2945,6 +2945,9 @@ def __init__( ) self.flags_with_values["enable_eager_delta_joins"] = BOOLEAN_FLAG_VALUES self.flags_with_values["kafka_sink_emit_sink_id_header"] = BOOLEAN_FLAG_VALUES + self.flags_with_values["pg_source_replication_slot_failover"] = ( + BOOLEAN_FLAG_VALUES + ) self.flags_with_values["compute_mv_sink_shared_batch_max_bytes"] = [ # 0 never shares, 64 KiB, 1 MiB, 128 MiB "0", diff --git a/src/storage-types/src/dyncfgs.rs b/src/storage-types/src/dyncfgs.rs index 49378c5de15fc..d1cf0a012497c 100644 --- a/src/storage-types/src/dyncfgs.rs +++ b/src/storage-types/src/dyncfgs.rs @@ -321,6 +321,18 @@ pub const PG_SCHEMA_VALIDATION_INTERVAL: Config = Config::new( ParameterScope::Environment, ); +/// Whether PostgreSQL sources create their replication slot with `FAILOVER`, which lets +/// PostgreSQL 17+ synchronize the slot to standbys so that it survives a failover. Slots that +/// already exist are not altered. +/// Replica-scoped because it decides how the upstream slot is created and never what a +/// collection contains. +pub const PG_SOURCE_REPLICATION_SLOT_FAILOVER: Config = Config::new( + "pg_source_replication_slot_failover", + false, + "Whether PostgreSQL sources create their replication slot with failover enabled.", + ParameterScope::Replica, +); + /// Controls behavior of PG Source when the upstream DB timeline changes. The default behavior /// is to emit a definite error forcing source recreation. In cases of HA, the upstream DB may /// provide guarantees of failover without loss of data (e.g. CloudSQL maintenance). Changing this @@ -624,6 +636,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&PG_FETCH_SLOT_RESUME_LSN_INTERVAL) .add(&PG_SCHEMA_VALIDATION_INTERVAL) .add(&PG_SOURCE_REPLICATION_SLOT_CREATION_TIMEOUT) + .add(&PG_SOURCE_REPLICATION_SLOT_FAILOVER) .add(&PG_SOURCE_VALIDATE_TIMELINE) .add(&REPLICA_METRICS_HISTORY_RETENTION_INTERVAL) .add(&SINK_ENSURE_TOPIC_CONFIG) diff --git a/src/storage/src/source/postgres.rs b/src/storage/src/source/postgres.rs index adb32279f1a0a..55710cbf688b3 100644 --- a/src/storage/src/source/postgres.rs +++ b/src/storage/src/source/postgres.rs @@ -439,15 +439,27 @@ impl From for DataflowError { } } +/// Creates the replication slot `slot` unless it exists. +/// +/// `failover` asks for a slot that PostgreSQL synchronizes to standbys. It only takes effect on a +/// server that can create one. 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 && supports_failover_slots(client).await? { + 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) => { @@ -514,6 +526,20 @@ async fn set_timeout( Ok(()) } +/// Whether the server accepts `FAILOVER` when creating a logical replication slot. +async fn supports_failover_slots(client: &Client) -> Result { + // PostgreSQL 17 introduced the option, and rejects it on a standby. + let row = simple_query_opt( + client, + sql!( + "SELECT pg_catalog.current_setting('server_version_num')::int >= 170000 \ + AND NOT pg_catalog.pg_is_in_recovery() AS supported" + ), + ) + .await?; + Ok(row.is_some_and(|row| row.get("supported") == Some("t"))) +} + /// The state of a replication slot. struct SlotMetadata { /// The process ID of the session using this slot if the slot is currently actively being used. diff --git a/src/storage/src/source/postgres/snapshot.rs b/src/storage/src/source/postgres/snapshot.rs index 485d5249d7d9d..a98798fd0098d 100644 --- a/src/storage/src/source/postgres/snapshot.rs +++ b/src/storage/src/source/postgres/snapshot.rs @@ -178,7 +178,7 @@ use mz_repr::{Datum, DatumVec, Diff, Row}; use mz_storage_types::connections::ConnectionContext; use mz_storage_types::dyncfgs::{ PG_FETCH_SLOT_RESUME_LSN_INTERVAL, PG_SOURCE_REPLICATION_SLOT_CREATION_TIMEOUT, - STORAGE_PERSIST_SINK_DESCRIPTION_LOOKAHEAD, + PG_SOURCE_REPLICATION_SLOT_FAILOVER, STORAGE_PERSIST_SINK_DESCRIPTION_LOOKAHEAD, }; use mz_storage_types::errors::DataflowError; use mz_storage_types::parameters::PgSourceSnapshotConfig; @@ -481,7 +481,15 @@ 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?; + let failover = + PG_SOURCE_REPLICATION_SLOT_FAILOVER.get(config.config.config_set()); + super::ensure_replication_slot( + &client, + main_slot, + failover, + slot_creation_timeout, + ) + .await?; Some(client) } else { None diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py index a5a8f6e6c6576..0cc8bed6268ad 100644 --- a/test/launchdarkly-flag-consistency/mzcompose.py +++ b/test/launchdarkly-flag-consistency/mzcompose.py @@ -384,6 +384,7 @@ persist_use_critical_since_txn persist_write_combine_inline_writes pg_source_connect_timeout + pg_source_replication_slot_failover pg_source_snapshot_statement_timeout pg_source_tcp_configure_server pg_source_tcp_keepalives_idle diff --git a/test/pg-cdc/mzcompose.py b/test/pg-cdc/mzcompose.py index 08c7bcffaf41c..47463edd55d02 100644 --- a/test/pg-cdc/mzcompose.py +++ b/test/pg-cdc/mzcompose.py @@ -205,6 +205,35 @@ 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 pg_source_replication_slot_failover creates the replication slot + with failover enabled on PostgreSQL 17+, and that the source still works on + older versions, which have no such slots. + """ + pg_version = get_targeted_pg_version(parser) + 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"), + ) + with pg_conn.cursor() as cur: + cur.execute("SELECT current_setting('server_version_num')::int >= 170000") + if cur.fetchall()[0][0]: + cur.execute("SELECT failover FROM pg_replication_slots") + failover = cur.fetchall() + assert failover == [ + (True,) + ], f"expected one failover slot, found {failover}" + + def workflow_wal_level(c: Composition, parser: WorkflowArgumentParser) -> None: pg_version = get_targeted_pg_version(parser) for wal_level in ["replica", "minimal"]: diff --git a/test/pg-cdc/override/replication-slot-failover.td b/test/pg-cdc/override/replication-slot-failover.td new file mode 100644 index 0000000000000..f14f428c4cfbc --- /dev/null +++ b/test/pg-cdc/override/replication-slot-failover.td @@ -0,0 +1,45 @@ +# 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. + +$ postgres-execute connection=mz_system +ALTER SYSTEM SET pg_source_replication_slot_failover = true + +> 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); +INSERT INTO pk_table VALUES (1, 'one'); + +> CREATE SOURCE pg_source + FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source'); +> CREATE TABLE pk_table FROM SOURCE pg_source (REFERENCE pk_table); + +> SELECT * FROM pk_table +1 one + +$ postgres-execute connection=postgres://postgres:postgres@postgres +INSERT INTO pk_table VALUES (2, 'two'); + +> SELECT * FROM pk_table +1 one +2 two From 43c87b815d14aeefdb7e370df337b4427f16a6bc Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Thu, 12 Nov 2026 15:19:21 -0500 Subject: [PATCH 2/7] human review Co-Authored-By: Claude Fable 5.2 --- misc/python/materialize/mzcompose/__init__.py | 2 +- src/storage-types/src/dyncfgs.rs | 7 ++----- src/storage/src/source/postgres.rs | 7 ++----- test/pg-cdc/mzcompose.py | 4 +--- 4 files changed, 6 insertions(+), 14 deletions(-) diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index a67a0a9651aab..f874ff80268ee 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -663,7 +663,7 @@ def get_variable_system_parameters( "pg_source_replication_slot_failover", "true", ["true", "false"] ) ] - if version >= MzVersion.parse_mz("v26.47.0-dev") + if version >= MzVersion.parse_mz("v26.51.0-dev") else [] ), VariableSystemParameter( diff --git a/src/storage-types/src/dyncfgs.rs b/src/storage-types/src/dyncfgs.rs index d1cf0a012497c..d652c636d670b 100644 --- a/src/storage-types/src/dyncfgs.rs +++ b/src/storage-types/src/dyncfgs.rs @@ -321,11 +321,8 @@ pub const PG_SCHEMA_VALIDATION_INTERVAL: Config = Config::new( ParameterScope::Environment, ); -/// Whether PostgreSQL sources create their replication slot with `FAILOVER`, which lets -/// PostgreSQL 17+ synchronize the slot to standbys so that it survives a failover. Slots that -/// already exist are not altered. -/// Replica-scoped because it decides how the upstream slot is created and never what a -/// collection contains. +/// Whether PostgreSQL sources create their replication slot with `FAILOVER`, so that PostgreSQL +/// 17+ can synchronize it to standbys. pub const PG_SOURCE_REPLICATION_SLOT_FAILOVER: Config = Config::new( "pg_source_replication_slot_failover", false, diff --git a/src/storage/src/source/postgres.rs b/src/storage/src/source/postgres.rs index 55710cbf688b3..827e2d913aca2 100644 --- a/src/storage/src/source/postgres.rs +++ b/src/storage/src/source/postgres.rs @@ -439,10 +439,8 @@ impl From for DataflowError { } } -/// Creates the replication slot `slot` unless it exists. -/// -/// `failover` asks for a slot that PostgreSQL synchronizes to standbys. It only takes effect on a -/// server that can create one. +/// Creates the replication slot `slot` unless it exists, with `FAILOVER` if `failover` is set +/// and the server supports it. async fn ensure_replication_slot( client: &Client, slot: &str, @@ -526,7 +524,6 @@ async fn set_timeout( Ok(()) } -/// Whether the server accepts `FAILOVER` when creating a logical replication slot. async fn supports_failover_slots(client: &Client) -> Result { // PostgreSQL 17 introduced the option, and rejects it on a standby. let row = simple_query_opt( diff --git a/test/pg-cdc/mzcompose.py b/test/pg-cdc/mzcompose.py index 47463edd55d02..f46c8c8b78900 100644 --- a/test/pg-cdc/mzcompose.py +++ b/test/pg-cdc/mzcompose.py @@ -209,9 +209,7 @@ def workflow_replication_slot_failover( c: Composition, parser: WorkflowArgumentParser ) -> None: """ - Test that pg_source_replication_slot_failover creates the replication slot - with failover enabled on PostgreSQL 17+, and that the source still works on - older versions, which have no such slots. + Test that pg_source_replication_slot_failover creates a failover slot on PostgreSQL 17+. """ pg_version = get_targeted_pg_version(parser) with c.override(create_postgres(pg_version=pg_version)): From e5eb133d75764226ed4632855ba04d7d1d7e4d81 Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Thu, 12 Nov 2026 15:30:50 -0500 Subject: [PATCH 3/7] Replace the dyncfg with a FAILOVER source option behind a feature flag Co-Authored-By: Claude Fable 5.2 --- misc/python/materialize/mzcompose/__init__.py | 10 +--------- .../materialize/parallel_workload/action.py | 3 --- src/adapter/src/catalog/migrate.rs | 4 +++- src/sql-lexer/src/keywords.txt | 1 + src/sql-parser/src/ast/defs/ddl.rs | 4 ++++ src/sql-parser/src/parser.rs | 4 +++- src/sql-parser/tests/testdata/ddl | 7 +++++++ src/sql/src/plan/statement/ddl.rs | 6 ++++++ src/sql/src/pure.rs | 9 +++++++++ src/sql/src/pure/error.rs | 2 ++ src/sql/src/session/vars/definitions.rs | 6 ++++++ src/storage-types/src/dyncfgs.rs | 10 ---------- src/storage-types/src/sources/postgres.rs | 6 ++++++ src/storage/src/source/postgres.rs | 18 ++---------------- src/storage/src/source/postgres/snapshot.rs | 6 ++---- .../launchdarkly-flag-consistency/mzcompose.py | 2 +- test/pg-cdc/mzcompose.py | 16 +++++++--------- .../override/replication-slot-failover.td | 15 +++++++++++---- 18 files changed, 71 insertions(+), 58 deletions(-) diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index f874ff80268ee..938d4f348346e 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/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", @@ -657,15 +658,6 @@ def get_variable_system_parameters( if version >= MzVersion.parse_mz("v26.44.0-dev") else [] ), - *( - [ - VariableSystemParameter( - "pg_source_replication_slot_failover", "true", ["true", "false"] - ) - ] - if version >= MzVersion.parse_mz("v26.51.0-dev") - else [] - ), VariableSystemParameter( "persist_validate_part_bounds_on_read", "false", ["true", "false"] ), diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 422da7605b4e6..25a91e251cd59 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -2945,9 +2945,6 @@ def __init__( ) self.flags_with_values["enable_eager_delta_joins"] = BOOLEAN_FLAG_VALUES self.flags_with_values["kafka_sink_emit_sink_id_header"] = BOOLEAN_FLAG_VALUES - self.flags_with_values["pg_source_replication_slot_failover"] = ( - BOOLEAN_FLAG_VALUES - ) self.flags_with_values["compute_mv_sink_shared_batch_max_bytes"] = [ # 0 never shares, 64 KiB, 1 MiB, 128 MiB "0", diff --git a/src/adapter/src/catalog/migrate.rs b/src/adapter/src/catalog/migrate.rs index f1192c1146910..4890d5405c94c 100644 --- a/src/adapter/src/catalog/migrate.rs +++ b/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, }); } diff --git a/src/sql-lexer/src/keywords.txt b/src/sql-lexer/src/keywords.txt index 1e3d34cc8b060..9400cf540101b 100644 --- a/src/sql-lexer/src/keywords.txt +++ b/src/sql-lexer/src/keywords.txt @@ -188,6 +188,7 @@ Expressions External Extract Factor +Failover False Fast Features diff --git a/src/sql-parser/src/ast/defs/ddl.rs b/src/sql-parser/src/ast/defs/ddl.rs index c3626f9631ffa..192d1c11dc76f 100644 --- a/src/sql-parser/src/ast/defs/ddl.rs +++ b/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, } diff --git a/src/sql-parser/src/parser.rs b/src/sql-parser/src/parser.rs index dfb903a57bd99..184a5b1bc6401 100644 --- a/src/sql-parser/src/parser.rs +++ b/src/sql-parser/src/parser.rs @@ -3858,9 +3858,11 @@ impl<'a> Parser<'a> { } fn parse_pg_connection_option(&mut self) -> Result, 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)?; diff --git a/src/sql-parser/tests/testdata/ddl b/src/sql-parser/tests/testdata/ddl index 8fb9e4453e4c2..d996bc352aa31 100644 --- a/src/sql-parser/tests/testdata/ddl +++ b/src/sql-parser/tests/testdata/ddl @@ -3329,6 +3329,13 @@ 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); +---- +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'); ---- diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs index 0ff9c69b8bcf1..f8892d3a1bf76 100644 --- a/src/sql/src/plan/statement/ddl.rs +++ b/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::, Default(vec![])), (ExcludeColumns, Vec::, 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. @@ -1183,6 +1185,9 @@ fn plan_postgres_source_connection( exclude_columns: _, seen: _, } = options.clone().try_into()?; + if 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, }) } diff --git a/src/sql/src/pure.rs b/src/sql/src/pure.rs index eb8b86a7684fc..3d25da86297c0 100644 --- a/src/sql/src/pure.rs +++ b/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?, diff --git a/src/sql/src/pure/error.rs b/src/sql/src/pure/error.rs index ec6182bca247b..fb796a892e7cd 100644 --- a/src/sql/src/pure/error.rs +++ b/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), } diff --git a/src/sql/src/session/vars/definitions.rs b/src/sql/src/session/vars/definitions.rs index 4f4d50d54face..0771b7e93c4bb 100644 --- a/src/sql/src/session/vars/definitions.rs +++ b/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", diff --git a/src/storage-types/src/dyncfgs.rs b/src/storage-types/src/dyncfgs.rs index d652c636d670b..49378c5de15fc 100644 --- a/src/storage-types/src/dyncfgs.rs +++ b/src/storage-types/src/dyncfgs.rs @@ -321,15 +321,6 @@ pub const PG_SCHEMA_VALIDATION_INTERVAL: Config = Config::new( ParameterScope::Environment, ); -/// Whether PostgreSQL sources create their replication slot with `FAILOVER`, so that PostgreSQL -/// 17+ can synchronize it to standbys. -pub const PG_SOURCE_REPLICATION_SLOT_FAILOVER: Config = Config::new( - "pg_source_replication_slot_failover", - false, - "Whether PostgreSQL sources create their replication slot with failover enabled.", - ParameterScope::Replica, -); - /// Controls behavior of PG Source when the upstream DB timeline changes. The default behavior /// is to emit a definite error forcing source recreation. In cases of HA, the upstream DB may /// provide guarantees of failover without loss of data (e.g. CloudSQL maintenance). Changing this @@ -633,7 +624,6 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&PG_FETCH_SLOT_RESUME_LSN_INTERVAL) .add(&PG_SCHEMA_VALIDATION_INTERVAL) .add(&PG_SOURCE_REPLICATION_SLOT_CREATION_TIMEOUT) - .add(&PG_SOURCE_REPLICATION_SLOT_FAILOVER) .add(&PG_SOURCE_VALIDATE_TIMELINE) .add(&REPLICA_METRICS_HISTORY_RETENTION_INTERVAL) .add(&SINK_ENSURE_TOPIC_CONFIG) diff --git a/src/storage-types/src/sources/postgres.rs b/src/storage-types/src/sources/postgres.rs index 4cb8f2db33206..a7e01edff9008 100644 --- a/src/storage-types/src/sources/postgres.rs +++ b/src/storage-types/src/sources/postgres.rs @@ -34,6 +34,8 @@ pub struct PostgresSourceConnection { pub connection: C::Pg, pub publication: String, pub publication_details: PostgresSourcePublicationDetails, + /// Whether to create the replication slot with `FAILOVER`. + pub failover: bool, } impl IntoInlineConnection @@ -45,6 +47,7 @@ impl IntoInlineConnection connection, publication, publication_details, + failover, } = self; PostgresSourceConnection { @@ -52,6 +55,7 @@ impl IntoInlineConnection connection: r.resolve_connection(connection).unwrap_pg(), publication, publication_details, + failover, } } } @@ -142,6 +146,7 @@ impl AlterCompatible for PostgresSourceConnection { connection, publication, publication_details, + failover, } = self; let compatibility_checks = [ @@ -157,6 +162,7 @@ impl AlterCompatible for PostgresSourceConnection { .is_ok(), "publication_details", ), + (failover == &other.failover, "failover"), ]; for (compatible, field) in compatibility_checks { diff --git a/src/storage/src/source/postgres.rs b/src/storage/src/source/postgres.rs index 827e2d913aca2..0511a8baaa391 100644 --- a/src/storage/src/source/postgres.rs +++ b/src/storage/src/source/postgres.rs @@ -439,15 +439,14 @@ impl From for DataflowError { } } -/// Creates the replication slot `slot` unless it exists, with `FAILOVER` if `failover` is set -/// and the server supports it. +/// 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 = if failover && supports_failover_slots(client).await? { + let query = if failover { sql!( "CREATE_REPLICATION_SLOT {} LOGICAL \"pgoutput\" (SNAPSHOT 'nothing', FAILOVER)", Sql::ident(slot) @@ -524,19 +523,6 @@ async fn set_timeout( Ok(()) } -async fn supports_failover_slots(client: &Client) -> Result { - // PostgreSQL 17 introduced the option, and rejects it on a standby. - let row = simple_query_opt( - client, - sql!( - "SELECT pg_catalog.current_setting('server_version_num')::int >= 170000 \ - AND NOT pg_catalog.pg_is_in_recovery() AS supported" - ), - ) - .await?; - Ok(row.is_some_and(|row| row.get("supported") == Some("t"))) -} - /// The state of a replication slot. struct SlotMetadata { /// The process ID of the session using this slot if the slot is currently actively being used. diff --git a/src/storage/src/source/postgres/snapshot.rs b/src/storage/src/source/postgres/snapshot.rs index a98798fd0098d..05cf2037ec784 100644 --- a/src/storage/src/source/postgres/snapshot.rs +++ b/src/storage/src/source/postgres/snapshot.rs @@ -178,7 +178,7 @@ use mz_repr::{Datum, DatumVec, Diff, Row}; use mz_storage_types::connections::ConnectionContext; use mz_storage_types::dyncfgs::{ PG_FETCH_SLOT_RESUME_LSN_INTERVAL, PG_SOURCE_REPLICATION_SLOT_CREATION_TIMEOUT, - PG_SOURCE_REPLICATION_SLOT_FAILOVER, STORAGE_PERSIST_SINK_DESCRIPTION_LOOKAHEAD, + STORAGE_PERSIST_SINK_DESCRIPTION_LOOKAHEAD, }; use mz_storage_types::errors::DataflowError; use mz_storage_types::parameters::PgSourceSnapshotConfig; @@ -481,12 +481,10 @@ pub(crate) fn render<'scope>( let main_slot = &connection.publication_details.slot; tracing::info!(%id, "ensuring replication slot {main_slot} exists"); - let failover = - PG_SOURCE_REPLICATION_SLOT_FAILOVER.get(config.config.config_set()); super::ensure_replication_slot( &client, main_slot, - failover, + connection.failover, slot_creation_timeout, ) .await?; diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py index 0cc8bed6268ad..9d2d829d6d638 100644 --- a/test/launchdarkly-flag-consistency/mzcompose.py +++ b/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 @@ -384,7 +385,6 @@ persist_use_critical_since_txn persist_write_combine_inline_writes pg_source_connect_timeout - pg_source_replication_slot_failover pg_source_snapshot_statement_timeout pg_source_tcp_configure_server pg_source_tcp_keepalives_idle diff --git a/test/pg-cdc/mzcompose.py b/test/pg-cdc/mzcompose.py index f46c8c8b78900..338bc46464158 100644 --- a/test/pg-cdc/mzcompose.py +++ b/test/pg-cdc/mzcompose.py @@ -209,9 +209,11 @@ def workflow_replication_slot_failover( c: Composition, parser: WorkflowArgumentParser ) -> None: """ - Test that pg_source_replication_slot_failover creates a failover slot on PostgreSQL 17+. + Test that the FAILOVER source option creates a failover slot on PostgreSQL 17+. """ 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") @@ -222,14 +224,10 @@ def workflow_replication_slot_failover( password="postgres", port=c.default_port("postgres"), ) - with pg_conn.cursor() as cur: - cur.execute("SELECT current_setting('server_version_num')::int >= 170000") - if cur.fetchall()[0][0]: - cur.execute("SELECT failover FROM pg_replication_slots") - failover = cur.fetchall() - assert failover == [ - (True,) - ], f"expected one failover slot, found {failover}" + failover = pg_conn.execute( + "SELECT failover FROM pg_replication_slots" + ).fetchall() + assert failover == [(True,)], f"expected one failover slot, found {failover}" def workflow_wal_level(c: Composition, parser: WorkflowArgumentParser) -> None: diff --git a/test/pg-cdc/override/replication-slot-failover.td b/test/pg-cdc/override/replication-slot-failover.td index f14f428c4cfbc..48e66ecb4e86e 100644 --- a/test/pg-cdc/override/replication-slot-failover.td +++ b/test/pg-cdc/override/replication-slot-failover.td @@ -7,9 +7,6 @@ # the Business Source License, use of this software will be governed # by the Apache License, Version 2.0. -$ postgres-execute connection=mz_system -ALTER SYSTEM SET pg_source_replication_slot_failover = true - > CREATE SECRET pgpass AS 'postgres' > CREATE CONNECTION pg TO POSTGRES ( @@ -30,8 +27,18 @@ CREATE PUBLICATION mz_source FOR ALL TABLES; CREATE TABLE pk_table (pk INTEGER PRIMARY KEY, f2 TEXT); INSERT INTO pk_table VALUES (1, 'one'); +$ postgres-execute connection=mz_system +ALTER SYSTEM SET enable_postgres_source_failover = false + +! CREATE SOURCE pg_source + FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER); +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 pg_source - FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source'); + FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER); > CREATE TABLE pk_table FROM SOURCE pg_source (REFERENCE pk_table); > SELECT * FROM pk_table From 1e834d46666b7d4c64e7e9511cf8159409d2c923 Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Thu, 12 Nov 2026 15:46:23 -0500 Subject: [PATCH 4/7] Gate FAILOVER on its presence and test all three forms Co-Authored-By: Claude Fable 5.2 --- src/sql-parser/tests/testdata/ddl | 14 ++++++++++++ src/sql/src/plan/statement/ddl.rs | 4 ++-- test/pg-cdc/mzcompose.py | 22 ++++++++++++++----- .../override/replication-slot-failover.td | 17 +++++++++++--- 4 files changed, 47 insertions(+), 10 deletions(-) diff --git a/src/sql-parser/tests/testdata/ddl b/src/sql-parser/tests/testdata/ddl index d996bc352aa31..0eca98ad31e7f 100644 --- a/src/sql-parser/tests/testdata/ddl +++ b/src/sql-parser/tests/testdata/ddl @@ -3329,6 +3329,20 @@ 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); ---- diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs index f8892d3a1bf76..ae12eb7663ef6 100644 --- a/src/sql/src/plan/statement/ddl.rs +++ b/src/sql/src/plan/statement/ddl.rs @@ -1183,9 +1183,9 @@ 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 failover { + if seen.contains(&PgConfigOptionName::Failover) { scx.require_feature_flag(&vars::ENABLE_POSTGRES_SOURCE_FAILOVER)?; } let details = details diff --git a/test/pg-cdc/mzcompose.py b/test/pg-cdc/mzcompose.py index 338bc46464158..e5e60d82fcd99 100644 --- a/test/pg-cdc/mzcompose.py +++ b/test/pg-cdc/mzcompose.py @@ -209,7 +209,7 @@ def workflow_replication_slot_failover( c: Composition, parser: WorkflowArgumentParser ) -> None: """ - Test that the FAILOVER source option creates a failover slot on PostgreSQL 17+. + 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: @@ -224,10 +224,22 @@ def workflow_replication_slot_failover( password="postgres", port=c.default_port("postgres"), ) - failover = pg_conn.execute( - "SELECT failover FROM pg_replication_slots" - ).fetchall() - assert failover == [(True,)], f"expected one failover slot, found {failover}" + 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) + """) + for _ in range(60): + failover = dict( + pg_conn.execute( + "SELECT slot_name, failover FROM pg_replication_slots" + ).fetchall() + ) + if len(failover) == len(expected): + break + time.sleep(1) + 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: diff --git a/test/pg-cdc/override/replication-slot-failover.td b/test/pg-cdc/override/replication-slot-failover.td index 48e66ecb4e86e..ea8e5d309be50 100644 --- a/test/pg-cdc/override/replication-slot-failover.td +++ b/test/pg-cdc/override/replication-slot-failover.td @@ -30,16 +30,27 @@ INSERT INTO pk_table VALUES (1, 'one'); $ postgres-execute connection=mz_system ALTER SYSTEM SET enable_postgres_source_failover = false -! CREATE SOURCE pg_source +! 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 pg_source +> CREATE SOURCE implied FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER); -> CREATE TABLE pk_table FROM SOURCE pg_source (REFERENCE pk_table); +> 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); +> CREATE TABLE pk_table FROM SOURCE implied (REFERENCE pk_table); + +> SELECT count(*) FROM mz_internal.mz_postgres_sources +3 > SELECT * FROM pk_table 1 one From 655449f8db6f6854e9727e1b97354653f1a09f70 Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Thu, 12 Nov 2026 16:09:13 -0500 Subject: [PATCH 5/7] Set REPLICA IDENTITY FULL on the failover test table Co-Authored-By: Claude Fable 5.2 --- test/pg-cdc/override/replication-slot-failover.td | 1 + 1 file changed, 1 insertion(+) diff --git a/test/pg-cdc/override/replication-slot-failover.td b/test/pg-cdc/override/replication-slot-failover.td index ea8e5d309be50..497d721ce59d9 100644 --- a/test/pg-cdc/override/replication-slot-failover.td +++ b/test/pg-cdc/override/replication-slot-failover.td @@ -25,6 +25,7 @@ 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 From c49cfec9539eb3c581bf2922535c16f25c2eae61 Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Thu, 12 Nov 2026 16:14:26 -0500 Subject: [PATCH 6/7] Give each failover test source a table so it creates its slot Co-Authored-By: Claude Fable 5.2 --- test/pg-cdc/mzcompose.py | 14 ++++------- .../override/replication-slot-failover.td | 23 +++++++++++++------ 2 files changed, 21 insertions(+), 16 deletions(-) diff --git a/test/pg-cdc/mzcompose.py b/test/pg-cdc/mzcompose.py index e5e60d82fcd99..f1cf0847aa9dc 100644 --- a/test/pg-cdc/mzcompose.py +++ b/test/pg-cdc/mzcompose.py @@ -229,15 +229,11 @@ def workflow_replication_slot_failover( SELECT s.name, p.replication_slot FROM mz_sources s JOIN mz_internal.mz_postgres_sources p USING (id) """) - for _ in range(60): - failover = dict( - pg_conn.execute( - "SELECT slot_name, failover FROM pg_replication_slots" - ).fetchall() - ) - if len(failover) == len(expected): - break - time.sleep(1) + 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}" diff --git a/test/pg-cdc/override/replication-slot-failover.td b/test/pg-cdc/override/replication-slot-failover.td index 497d721ce59d9..86b1d01ed6a7d 100644 --- a/test/pg-cdc/override/replication-slot-failover.td +++ b/test/pg-cdc/override/replication-slot-failover.td @@ -48,17 +48,26 @@ ALTER SYSTEM SET enable_postgres_source_failover = true FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER = true); > CREATE SOURCE explicit_false FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source', FAILOVER = false); -> CREATE TABLE pk_table FROM SOURCE implied (REFERENCE pk_table); -> SELECT count(*) FROM mz_internal.mz_postgres_sources -3 - -> SELECT * FROM pk_table -1 one +# 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 pk_table +> 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 From 4efab9891cb853c67310b967b8d9f2c5531c569a Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Fri, 13 Nov 2026 10:06:52 -0500 Subject: [PATCH 7/7] Regenerate the system parameter catalog for the failover flag The checked-in src/sql/system-parameters.json lacked enable_postgres_source_failover, which failed the checked_in_catalog_is_current test in mz-system-parameter-catalog. Co-Authored-By: Claude Fable 5.2 --- src/sql/system-parameters.json | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/sql/system-parameters.json b/src/sql/system-parameters.json index 16320e8c932fe..46ca7409c9733 100644 --- a/src/sql/system-parameters.json +++ b/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",