diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 6b5368ce68786..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", 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..0eca98ad31e7f 100644 --- a/src/sql-parser/tests/testdata/ddl +++ b/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'); ---- diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs index 0ff9c69b8bcf1..ae12eb7663ef6 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. @@ -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, }) } 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/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", 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 adb32279f1a0a..0511a8baaa391 100644 --- a/src/storage/src/source/postgres.rs +++ b/src/storage/src/source/postgres.rs @@ -439,15 +439,24 @@ impl From 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) => { diff --git a/src/storage/src/source/postgres/snapshot.rs b/src/storage/src/source/postgres/snapshot.rs index 485d5249d7d9d..05cf2037ec784 100644 --- a/src/storage/src/source/postgres/snapshot.rs +++ b/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 diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py index a5a8f6e6c6576..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 diff --git a/test/pg-cdc/mzcompose.py b/test/pg-cdc/mzcompose.py index 08c7bcffaf41c..f1cf0847aa9dc 100644 --- a/test/pg-cdc/mzcompose.py +++ b/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"]: 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..86b1d01ed6a7d --- /dev/null +++ b/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