@@ -261,6 +261,19 @@ Log-based replication will modify the schemas output by the tap. Specifically, a
 
 Note that changing what streams are selected after already beginning log-based replication can have unexpected consequences. To ensure consistent output, it is best to keep selected streams the same across invocations of the tap.
 
+#### Adding a new stream to an existing LOG_BASED set
+
+When a stream has no prior bookmark (i.e. it is new to log-based replication), the tap starts reading the replication slot from LSN 0 — the oldest WAL position the slot still retains. This means the tap will walk through all retained WAL for every registered table before reaching the current tip, which can be very slow if the slot has accumulated a large backlog.
+
+Other CDC tools (e.g. [Debezium](https://debezium.io/blog/2021/10/07/incremental-snapshots/), [Airbyte](https://docs.airbyte.com/platform/understanding-airbyte/cdc#adding-new-schemascolumns)) handle this by taking an initial snapshot of the new table at a known LSN and then starting WAL streaming from that point. This tap does not implement that pattern.
+
+The recommended workaround is:
+
+1. Sync the new table once using `FULL_TABLE` or `INCREMENTAL` replication to establish its baseline state.
+2. Switch that stream's replication method to `LOG_BASED` for subsequent syncs.
+
+This ensures the new stream has a valid starting LSN bookmark and does not force the other streams to re-read old WAL.
+
 Note also that using log-based replication will cause the replication key for all streams to be set to "_sdc_lsn", which is the Postgres LSN for the record in question.
 
 ### How to Set Up Log-Based Replication
 
@@ -0,0 +1,138 @@
+"""Helper functions for LOG_BASED replication."""
+
+from __future__ import annotations
+
+import json
+import re
+import typing as t
+
+import psycopg2
+
+if t.TYPE_CHECKING:
+    import logging
+
+    from psycopg2 import extras
+
+
+# wal2json emits enum type names with unescaped double quotes, producing invalid JSON
+# like "type":""EnumName"" ... strip the extra quotes via regex and re-parse
+_WAL2JSON_ENUM_QUOTE_RE = re.compile(r'"type":""([^"]+)""')
+
+
+def normalize_fqn(schema: str, table: str) -> str:
+    """Generate canonical, fully-qualified name for dispatch.
+
+    This is the source of truth for matching a WAL message to a registered stream.
+    Both sides -- the tap (when registering streams with the reader) and the WAL reader
+    (when dispatching a parsed payload) -- *must* call this function with the raw schema
+    and table name strings.
+
+    wal2json's format-version=2 output includes ``"schema"`` and ``"table"`` fields
+    as the raw, unquoted identifier names (wal2json reports whatever Postgres has stored).
+    Therefore, use the raw names joined by a single dot, with no quoting and no case folding.
+
+    Do *not* use ``SQLStream.fully_qualified_name`` for dispatch.
+    """
+    return f"{schema}.{table}"
+
+
+def escape_for_add_tables(identifier: str) -> str:
+    """Escape a schema or table name for use in wal2json's ``add-tables``.
+
+    wal2json's ``add-tables`` option takes a comma-separated list of "schema.table" entries.
+    Backslash is the escape character; comma and dot within an identifier must be escaped,
+    and backslash itself must be doubled.
+
+    References:
+        - https://github.com/eulerto/wal2json#parameters
+    """
+    return identifier.replace("\\", "\\\\").replace(",", "\\,").replace(".", "\\.")
+
+
+def build_add_tables_option(fqn_pairs: list[tuple[str, str]]) -> str:
+    """Build the wal2json ``add-tables`` option from a list of (schema, table).
+
+    Each identifier is escaped with ``escape_for_add_tables`` and joined with
+    the appropriate separators. For example::
+
+        >>> build_add_tables_option([("public", "users"), ("public", "orders")])
+        'public.users,public.orders'
+    """
+    return ",".join(
+        f"{escape_for_add_tables(schema)}.{escape_for_add_tables(table)}"
+        for schema, table in fqn_pairs
+    )
+
+
+def parse_wal_message(raw_payload: str, cursor: extras.ReplicationCursor | None) -> dict | None:
+    """Parse a raw wal2json JSON payload into a Python dict.
+
+    Handles the known wal2json enum-quoting bug via one retry after regex repair.
+    Returns None if the payload can't be decoded even after repair, in which case
+    the caller should log and skip.
+
+    When ``cursor`` is provided, pre-parses ``text[]`` column values into Python lists
+    using psycopg2's ``STRINGARRAY`` type caster. This must be done while the cursor is alive,
+    since ``STRINGARRAY`` reads connection-level encoding info from it.
+    """
+    try:
+        payload = json.loads(raw_payload)
+    except json.JSONDecodeError:
+        try:
+            payload = json.loads(fix_wal2json_enum_quotes(raw_payload))
+        except json.JSONDecodeError:
+            return None
+
+    if cursor is not None:
+        pre_parse_text_arrays(payload, cursor)
+
+    return payload
+
+
+def fix_wal2json_enum_quotes(payload: str) -> str:
+    """Repair the wal2json enum-quoting bug in a raw JSON payload.
+
+    wal2json outputs enum type names with unescaped double quotes (e.g. "type":""EnumName""),
+    which is invalid JSON. Normalize this to "type":"EnumName" so a second ``json.loads``
+    attempt succeeds.
+    """
+    return _WAL2JSON_ENUM_QUOTE_RE.sub(r'"type":"\1"', payload)
+
+
+def pre_parse_text_arrays(payload: dict, cursor: extras.ReplicationCursor) -> None:
+    """Pre-parse ``text[]`` column values in a wal2json payload, in place.
+
+    wal2json returns ``text[]`` values as Postgres's array literal string (e.g. '{a,b,c}').
+    Converting to a Python list requires ``psycopg2.extensions.STRINGARRAY``, which needs
+    a live cursor for encoding context. Calling this during the WAL read means downstream code
+    -- ``consume()``, etc. -- can operate on plain Python lists with no cursor dependency.
+    """
+    for key in ("columns", "identity"):
+        for column in payload.get(key, ()) or ():
+            if column.get("type") == "text[]" and column.get("value") is not None:
+                column["value"] = psycopg2.extensions.STRINGARRAY(column["value"], cursor)
+
+
+def query_current_wal_lsn(dsn: str, logger: logging.Logger) -> int | None:
+    """Query ``pg_current_wal_flush_lsn()`` on a non-replication connection."""
+    try:
+        conn = psycopg2.connect(dsn)
+        try:
+            conn.autocommit = True
+            with conn.cursor() as cur:
+                cur.execute("SELECT pg_current_wal_flush_lsn()")
+                row = cur.fetchone()
+                if row is None:
+                    return None
+                return _lsn_str_to_int(row[0])
+        finally:
+            conn.close()
+    except Exception as exc:
+        logger.warning("Could not query current WAL LSN: %s", exc)
+        return None
+
+
+def _lsn_str_to_int(lsn_str: str) -> int:
+    """Convert a Postgres LSN string (e.g. '6/4A3B2C10') to an int."""
+    hi, lo = lsn_str.split("/")
+    return (int(hi, 16) << 32) + int(lo, 16)