support origin param in aiven_extras.pg_create_subscription

aiven · Jun 12, 2024 · ce21ec5 · ce21ec5
1 parent 1504010
commit ce21ec5
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 5 deletions.
diff --git a/sql/aiven_extras--1.1.11--1.1.12.sql b/sql/aiven_extras--1.1.11--1.1.12.sql
@@ -0,0 +1,8 @@
+DROP FUNCTION IF EXISTS aiven_extras.pg_create_subscription(
+    arg_subscription_name TEXT,
+    arg_connection_string TEXT,
+    arg_publication_name TEXT,
+    arg_slot_name TEXT,
+    arg_slot_create BOOLEAN,
+    arg_copy_data BOOLEAN
+);
diff --git a/sql/aiven_extras.sql b/sql/aiven_extras.sql
@@ -100,26 +100,44 @@ END;
 $$;
 
 
-DROP FUNCTION IF EXISTS aiven_extras.pg_create_subscription(TEXT, TEXT, TEXT, TEXT, BOOLEAN, BOOLEAN);
+DROP FUNCTION IF EXISTS aiven_extras.pg_create_subscription(TEXT, TEXT, TEXT, TEXT, BOOLEAN, BOOLEAN, TEXT);
 CREATE FUNCTION aiven_extras.pg_create_subscription(
     arg_subscription_name TEXT,
     arg_connection_string TEXT,
     arg_publication_name TEXT,
     arg_slot_name TEXT,
     arg_slot_create BOOLEAN = FALSE,
-    arg_copy_data BOOLEAN = TRUE
+    arg_copy_data BOOLEAN = TRUE,
+    arg_origin TEXT = 'any'
 )
 RETURNS VOID LANGUAGE plpgsql
 SECURITY DEFINER
 SET search_path = pg_catalog, aiven_extras
 AS $$
+DECLARE
+    pg_version INT;
+    create_subscription_cmd TEXT;
 BEGIN
+    -- Get the PostgreSQL version
+    SELECT current_setting('server_version_num')::INT INTO pg_version;
+
     IF (arg_slot_create IS TRUE) THEN
         PERFORM aiven_extras.dblink_slot_create_or_drop(arg_connection_string, arg_slot_name, 'create');
     END IF;
-    EXECUTE pg_catalog.format(
-        'CREATE SUBSCRIPTION %I connection %L publication %I WITH (slot_name=%L, create_slot=FALSE, copy_data=%s)',
-        arg_subscription_name, arg_connection_string, arg_publication_name, arg_slot_name, arg_copy_data::TEXT);
+
+    -- PG16 and later: Include the origin parameter only if it's 'none', as its default is any
+   IF pg_version >= 160000 AND arg_origin = 'none' THEN
+        create_subscription_cmd := pg_catalog.format(
+            'CREATE SUBSCRIPTION %I CONNECTION %L PUBLICATION %I WITH (slot_name=%L, create_slot=FALSE, copy_data=%s, origin=%L)',
+            arg_subscription_name, arg_connection_string, arg_publication_name, arg_slot_name, arg_copy_data::TEXT, arg_origin);
+    ELSE
+        create_subscription_cmd := pg_catalog.format(
+            'CREATE SUBSCRIPTION %I CONNECTION %L PUBLICATION %I WITH (slot_name=%L, create_slot=FALSE, copy_data=%s)',
+            arg_subscription_name, arg_connection_string, arg_publication_name, arg_slot_name, arg_copy_data::TEXT);
+    END IF;
+
+    -- Execute the CREATE SUBSCRIPTION command
+    EXECUTE create_subscription_cmd;
 END;
 $$;