feat: Custom replication slot name

tap_postgres/client.py

@@ -340,8 +340,11 @@ def get_records(self, context: Context | None) -> Iterable[dict[str, Any]]:
         # even though we still want logs with an LSN == start_lsn.
         logical_replication_cursor.send_feedback(flush_lsn=start_lsn)
 
+        # get the slot name from the configuration or use the default value
+        replication_slot_name = self.config.get("replication_slot_name", "tappostgres")
+
         logical_replication_cursor.start_replication(
-            slot_name="tappostgres",
+            slot_name=replication_slot_name,  # use slot name
             decode=True,
             start_lsn=start_lsn,
             status_interval=status_interval,
@@ -460,13 +463,20 @@ def logical_replication_connection(self):
         Uses a direct psycopg2 implementation rather than through sqlalchemy.
         """
         connection_string = (
-            f"dbname={self.config['database']} user={self.config['user']} password="
-            f"{self.config['password']} host={self.config['host']} port="
-            f"{self.config['port']}"
+            f"dbname={self.config['database']} "
+            f"user={self.config['user']} "
+            f"password{self.config['password']} "
+            f"host={self.config['host']} "
+            f"port={self.config['port']}"
+            "replication=database"
         )
         return psycopg2.connect(
             connection_string,
-            application_name="tap_postgres",
+            # add slot name to application_name
+            application_name=(
+                f"tap_postgres_{self.config.get(
+                        'replication_slot_name', "f"'tappostgres')}"
+            ),
             connection_factory=extras.LogicalReplicationConnection,
         )
 

tap_postgres/tap.py

@@ -50,6 +50,23 @@ def __init__(
         See https://github.com/MeltanoLabs/tap-postgres/issues/141
         """
         super().__init__(*args, **kwargs)
+
+        # Add replication_slot_name validation
+        if self.config.get("replication_method") == "LOG_BASED" and self.config.get(
+            "replication_slot_name"
+        ):
+            slot_name = self.config["replication_slot_name"]
+            assert (
+                slot_name.isalnum() or "_" in slot_name
+            ), "Replication slot name must contain letters, numbers and underscores"
+            max_slot_name_len = 63
+            assert (
+                len(slot_name) <= max_slot_name_len
+            ), "Replication slot name must be less than 63 characters"
+            assert not slot_name.startswith(
+                "pg_"
+            ), "Replication slot name cannot start with 'pg_'"
+
         assert (self.config.get("sqlalchemy_url") is not None) or (
             self.config.get("host") is not None
             and self.config.get("port") is not None
@@ -98,6 +115,16 @@ def __init__(
         )
 
     config_jsonschema = th.PropertiesList(
+        th.Property(
+            "replication_slot_name",
+            th.StringType,
+            default="tappostgres",
+            description=(
+                "Name of the replication slot to use for logical replication. "
+                "Must be unique for parallel extractions. "
+                "Only applicable when replication_method is LOG_BASED."
+            ),
+        ),  # New Property
         th.Property(
             "host",
             th.StringType,
@@ -617,3 +644,20 @@ def discover_streams(self) -> Sequence[Stream]:
                     PostgresStream(self, catalog_entry, connector=self.connector)
                 )
         return streams
+
+
+# Configuration Example for Parallel Replication
+
+config_1 = {
+    "host": "database1.example.com",
+    "port": 5432,
+    "dbname": "example_db_1",
+    "replication_slot_name": "slot_1",
+}
+
+config_2 = {
+    "host": "database2.example.com",
+    "port": 5432,
+    "dbname": "example_db_2",
+    "replication_slot_name": "slot_2",
+}

tests/test_slot_name.py

@@ -0,0 +1,51 @@
+import unittest
+
+from tap_postgres.tap import TapPostgres
+
+
+class TestReplicationSlot(unittest.TestCase):
+    def setUp(self):
+        self.default_config = {
+            "host": "localhost",
+            "port": 5432,
+            "dbname": "test_db",
+        }
+
+    def test_default_slot_name(self):
+        # Test backward compatibility when slot name is not provided.
+        config = self.default_config
+        tap = TapPostgres(config)
+        self.assertEqual(
+            tap.config.get("replication_slot_name", "tappostgres"), "tappostgres"
+        )
+
+    def test_custom_slot_name(self):
+        # Test if the custom slot name is used.
+        config = {**self.default_config, "replication_slot_name": "custom_slot"}
+        tap = TapPostgres(config)
+        self.assertEqual(tap.config["replication_slot_name"], "custom_slot")
+
+    def test_multiple_slots(self):
+        # Simulate using multiple configurations with different slot names.
+        config_1 = {**self.default_config, "replication_slot_name": "slot_1"}
+        config_2 = {**self.default_config, "replication_slot_name": "slot_2"}
+
+        tap_1 = TapPostgres(config_1)
+        tap_2 = TapPostgres(config_2)
+
+        self.assertNotEqual(
+            tap_1.config["replication_slot_name"],
+            tap_2.config["replication_slot_name"],
+        )
+        self.assertEqual(tap_1.config["replication_slot_name"], "slot_1")
+        self.assertEqual(tap_2.config["replication_slot_name"], "slot_2")
+
+    def test_invalid_slot_name(self):
+        # Test validation for invalid slot names (if any validation rules exist).
+        invalid_config = {
+            **self.default_config,
+            "replication_slot_name": "invalid slot name!",
+        }
+
+        with self.assertRaises(ValueError):
+            TapPostgres(invalid_config)