replicators: Add an option to skip CREATE PUBLICATION in postgres

Readyset contains a command line option for disabling setting up
the DDL event triggers (--disable-setup-ddl-replication). This is
useful when the database user is not a superuser and triggers are
created manually.

CREATE PUBLICATION also requires superuser privileges. This commit
adds a new command line option, --disable-create-publication for
skipping publication creation. If this option is passed, the
publication has to be created manually before the replication starts.

Release-Note-Core: New option --disable-create-publication for skipping
creating a publication. Used when readyset is not running
as a superuser and the publication has been created manually.

Change-Id: If25d880f8730efd778c7d77d81c49ace428a0eea
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/7705
Tested-by: Buildkite CI
Reviewed-by: Vassili Zarouba <vassili@readyset.io>
Reviewed-by: Jason Brown <jason.b@readyset.io>

Author: sandaru1

database-utils/src/lib.rs

@@ -64,6 +64,13 @@ pub struct UpstreamConfig {
     #[serde(default)]
     pub disable_setup_ddl_replication: bool,
 
+    /// Disable running CREATE PUBLICATION query for PostgreSQL. If this flag is set
+    /// a publication named readyset should be adced for all tables manually on the
+    /// primary server before streaming replication will start.
+    #[arg(long, env = "DISABLE_CREATE_PUBLICATION", hide = true)]
+    #[serde(default)]
+    pub disable_create_publication: bool,
+
     /// Server ID to use when registering as a replication follower with the upstream db
     ///
     /// This can be used to differentiate different ReadySet deployments connected to the same
@@ -244,6 +251,7 @@ impl Default for UpstreamConfig {
             upstream_db_url: Default::default(),
             disable_upstream_ssl_verification: false,
             disable_setup_ddl_replication: false,
+            disable_create_publication: false,
             replication_server_id: Default::default(),
             replicator_restart_timeout: Duration::from_secs(1),
             replication_tables: Default::default(),

replicators/src/postgres_connector/connector.rs

@@ -167,14 +167,18 @@ impl PostgresWalConnector {
             //
             // Note that later on, this means we'll need to make sure we resnapshot *all* tables!
             connector
-                .create_publication_and_slot(repl_slot_name)
+                .create_publication_and_slot(repl_slot_name, config)
                 .await?;
         }
 
         Ok(connector)
     }
 
-    async fn create_publication_and_slot(&mut self, repl_slot_name: &str) -> ReadySetResult<()> {
+    async fn create_publication_and_slot(
+        &mut self,
+        repl_slot_name: &str,
+        config: UpstreamConfig,
+    ) -> ReadySetResult<()> {
         let system = self.identify_system().await?;
         debug!(
             id = %system.id,
@@ -183,20 +187,22 @@ impl PostgresWalConnector {
             dbname = ?system.dbname
         );
 
-        match self.create_publication(PUBLICATION_NAME).await {
-            Ok(()) => {
-                // Created a new publication, everything is good
-            }
-            Err(err)
-                if err.to_string().contains("publication")
-                    && err.to_string().contains("already exists") =>
-            {
-                // This is an existing publication we are going to use
-            }
-            Err(err) if err.to_string().contains("permission denied") => {
-                error!("Insufficient permissions to create publication FOR ALL TABLES");
+        if !config.disable_create_publication {
+            match self.create_publication(PUBLICATION_NAME).await {
+                Ok(()) => {
+                    // Created a new publication, everything is good
+                }
+                Err(err)
+                    if err.to_string().contains("publication")
+                        && err.to_string().contains("already exists") =>
+                {
+                    // This is an existing publication we are going to use
+                }
+                Err(err) if err.to_string().contains("permission denied") => {
+                    error!("Insufficient permissions to create publication FOR ALL TABLES");
+                }
+                Err(err) => return Err(err),
             }
-            Err(err) => return Err(err),
         }
 
         // Drop the existing slot if any