Use wal2json non-row messages to detect that wal has moved on

Author: louis-pie

CHANGELOG.md

@@ -1,8 +1,14 @@
+0.75.0 (2026-04-07)
+-------------------
+- `pipelinewise-tap-postgres` from `2.1.0` to `2.2.0`
+    - Use wal2json non-row messages to detect that wal has moved on.
+
 0.74.4 (2026-03-30)
 -------------------
 - Better handling of CSV bulk upload to Snowflake during Full and Partial sync
 
 0.74.3 (2026-02-26)
+-------------------
 - `pipelinewise-target-snowflake` from `2.5.1` to `2.5.2`
     - Support creating new Iceberg tables for pure Singer replications
 

dev-project/entrypoint.sh

@@ -2,15 +2,32 @@
 
 set -e
 
+# Retry wrapper for apt-get to handle transient mirror errors
+apt_retry() {
+  local max_attempts=3
+  local attempt=1
+  while [ $attempt -le $max_attempts ]; do
+    if "$@"; then
+      return 0
+    fi
+    echo "apt command failed (attempt $attempt/$max_attempts), retrying in 5s..."
+    attempt=$((attempt + 1))
+    sleep 5
+    apt-get update
+  done
+  echo "apt command failed after $max_attempts attempts"
+  return 1
+}
+
 apt-get update
-apt-get install -y software-properties-common python3-apt apt-utils
+apt_retry apt-get install -y software-properties-common python3-apt apt-utils
 
 add-apt-repository ppa:deadsnakes/ppa
 apt-get update
 
 echo 'debconf debconf/frontend select Noninteractive' | debconf-set-selections
 
-apt-get install -y --no-install-recommends \
+apt_retry apt-get install -y --no-install-recommends \
   wget \
   gnupg \
   git \

setup.py

@@ -7,7 +7,7 @@
 
 setup(name='pipelinewise',
       python_requires='==3.10.*',
-      version='0.74.4',
+      version='0.75.0',
       description='PipelineWise',
       long_description=LONG_DESCRIPTION,
       long_description_content_type='text/markdown',

singer-connectors/tap-postgres/CHANGELOG.md

@@ -1,10 +1,15 @@
+2.2.0 (2026-04-07)
+-------------------
+**Changes**
+- LOG_BASED: Use wal2json non-row messages to detect that wal has moved on.
+
 2.1.0 (2023-03-30)
 -------------------
 **Changes**
-- INCREMENTAL: An optional config `limit` to be appended to incremental queries to limit their runtime.  
+- INCREMENTAL: An optional config `limit` to be appended to incremental queries to limit their runtime.
 
 **Fixes**
-- INCREMENTAL: `ORDER BY` added back to query in case replication key value is None. 
+- INCREMENTAL: `ORDER BY` added back to query in case replication key value is None.
 
 2.0.0 (2022-11-02)
 -------------------

singer-connectors/tap-postgres/setup.py

@@ -6,7 +6,7 @@
     long_description = f.read()
 
 setup(name='pipelinewise-tap-postgres',
-      version='2.1.0',
+      version='2.2.0',
       description='Singer.io tap for extracting data from PostgresSQL - PipelineWise compatible',
       long_description=long_description,
       long_description_content_type='text/markdown',

singer-connectors/tap-postgres/tap_postgres/sync_strategies/logical_replication.py

@@ -385,6 +385,23 @@ def consume_message(streams, state, msg, time_extracted, conn_info):
 
     lsn = msg.data_start
 
+    action = payload.get('action')
+    # Action Types:
+    # I = Insert
+    # U = Update
+    # D = Delete
+    # B = Begin Transaction
+    # C = Commit Transaction
+    # M = Message
+    # T = Truncate
+
+    # Advance the slot LSN for non-row actions without doing any processing
+    # This avoids the slot growing when the source has very busy tables that are NOT selected for replication
+    if action not in {'I', 'U', 'D'}:
+        LOGGER.debug('Skipping non-row wal2json message: action=%s, lsn=%s', action,
+                     int_to_lsn(lsn) if isinstance(lsn, int) else lsn)
+        return state
+
     streams_lookup = {s['tap_stream_id']: s for s in streams}
 
     tap_stream_id = post_db.compute_tap_stream_id(payload['schema'], payload['table'])
@@ -415,19 +432,6 @@ def consume_message(streams, state, msg, time_extracted, conn_info):
     #   ]
     # }
 
-    # Action Types:
-    # I = Insert
-    # U = Update
-    # D = Delete
-    # B = Begin Transaction
-    # C = Commit Transaction
-    # M = Message
-    # T = Truncate
-    action = payload['action']
-
-    if action not in {'I', 'U', 'D'}:
-        raise UnsupportedPayloadKindError(f"unrecognized replication operation: {action}")
-
     # Get the additional fields in payload that are not in schema properties:
     # only inserts and updates have the list of columns that can be used to detect any different in columns
     diff = set()
@@ -613,7 +617,7 @@ def sync_tables(conn_info, logical_streams, state, end_lsn, state_file):
                               status_interval=poll_interval,
                               options={
                                   'format-version': 2,
-                                  'include-transaction': False,
+                                  'include-transaction': True,
                                   'include-timestamp': True,
                                   'include-types': False,
                                   'actions': 'insert,update,delete',

singer-connectors/tap-postgres/tests/integration/test_logical_replication.py

@@ -206,3 +206,100 @@ def test_logical_replication(self):
         })
         self.assertEqual(messages[4]['type'], 'STATE')
         self.assertDictEqual(state, messages[4]['value'])
+
+
+class TestUnselectedTableSlotAdvancement(unittest.TestCase):
+    """Test that WAL slot LSN advances even when only non-selected tables have activity.
+
+    This verifies the include-transaction: true fix — B/C markers from transactions
+    on unselected tables cause the slot to advance, preventing unbounded slot growth.
+    """
+    selected_table = None
+    unselected_table = None
+    maxDiff = None
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.selected_table = 'selected_table'
+        cls.unselected_table = 'unselected_table'
+
+        selected_spec = {
+            "columns": [
+                {"name": "id", "type": "serial", "primary_key": True},
+                {"name": "val", "type": "character varying"},
+            ],
+            "name": cls.selected_table,
+        }
+        unselected_spec = {
+            "columns": [
+                {"name": "id", "type": "serial", "primary_key": True},
+                {"name": "val", "type": "character varying"},
+            ],
+            "name": cls.unselected_table,
+        }
+
+        ensure_test_table(selected_spec)
+        ensure_test_table(unselected_spec)
+        create_replication_slot()
+
+        cls.config = get_test_connection_config()
+        tap_postgres.dump_catalog = lambda catalog: True
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        drop_replication_slot()
+        drop_table(cls.selected_table)
+        drop_table(cls.unselected_table)
+
+    def test_slot_advances_with_only_unselected_table_activity(self):
+        """Slot LSN must advance when only the unselected table receives writes."""
+
+        # Discover streams, select only `selected_table` for LOG_BASED
+        streams = tap_postgres.do_discovery(self.config)
+        selected_stream = [s for s in streams if s['tap_stream_id'] == f'public-{self.selected_table}'][0]
+        selected_stream = set_replication_method_for_stream(selected_stream, 'LOG_BASED')
+
+        # Insert a row into the selected table so initial sync has something to process
+        conn = get_test_connection()
+        try:
+            with conn.cursor() as cur:
+                insert_record(cur, self.selected_table, {'val': 'seed'})
+        finally:
+            conn.close()
+
+        # Initial sync to establish bookmarks
+        state = {}
+        my_stdout = io.StringIO()
+        with contextlib.redirect_stdout(my_stdout):
+            state = tap_postgres.do_sync(self.config, {'streams': [selected_stream]}, 'LOG_BASED', state, None)
+
+        # Capture LSN after initial sync
+        initial_lsn = state['bookmarks'][f'public-{self.selected_table}']['lsn']
+        self.assertIsNotNone(initial_lsn, "Initial sync should set an LSN bookmark")
+
+        # Now insert rows ONLY into the unselected table
+        conn = get_test_connection()
+        try:
+            with conn.cursor() as cur:
+                for i in range(5):
+                    insert_record(cur, self.unselected_table, {'val': f'noise_{i}'})
+        finally:
+            conn.close()
+
+        # Run sync again — no rows from selected table, but B/C markers should advance the slot
+        my_stdout.seek(0)
+        my_stdout.truncate()
+        with contextlib.redirect_stdout(my_stdout):
+            state = tap_postgres.do_sync(self.config, {'streams': [selected_stream]}, 'LOG_BASED', state, None)
+
+        # Assert that the LSN bookmark has advanced past the unselected-table activity
+        new_lsn = state['bookmarks'][f'public-{self.selected_table}']['lsn']
+        self.assertGreater(new_lsn, initial_lsn,
+                           "LSN bookmark should advance when unselected tables have WAL activity "
+                           "(B/C markers from include-transaction: true)")
+
+        # Verify no RECORD messages were emitted (only the unselected table had activity)
+        messages = [json.loads(msg) for msg in my_stdout.getvalue().splitlines()]
+        record_messages = [m for m in messages if m['type'] == 'RECORD']
+        self.assertEqual(len(record_messages), 0,
+                         "No RECORD messages should be emitted for unselected table activity")

singer-connectors/tap-postgres/tests/unit/test_logical_replication.py

@@ -11,7 +11,6 @@
 from dateutil.tz import tzoffset
 
 from tap_postgres.sync_strategies import logical_replication
-from tap_postgres.sync_strategies.logical_replication import UnsupportedPayloadKindError
 
 
 class PostgresCurReplicationSlotMock:
@@ -177,16 +176,44 @@ def test_consume_with_message_stream_in_payload_is_not_selected_expect_same_stat
 
         self.assertDictEqual({}, output)
 
-    def test_consume_with_payload_kind_is_not_supported_expect_exception(self):
-        with self.assertRaises(UnsupportedPayloadKindError):
-            logical_replication.consume_message(
-                [{'tap_stream_id': 'myschema-mytable'}],
-                {},
-                self.WalMessage(payload='{"action":"truncate", "schema": "myschema", "table": "mytable"}',
-                                data_start='some lsn'),
-                None,
-                {}
-            )
+    def test_consume_with_truncate_action_returns_state_unchanged(self):
+        """Truncate (T) actions with schema/table keys are silently skipped"""
+        output = logical_replication.consume_message(
+            [{'tap_stream_id': 'myschema-mytable'}],
+            {},
+            self.WalMessage(payload='{"action":"T", "schema": "myschema", "table": "mytable"}',
+                            data_start='some lsn'),
+            None,
+            {}
+        )
+
+        self.assertDictEqual({}, output)
+
+    def test_consume_with_begin_action_without_schema_table_returns_state_unchanged(self):
+        """Begin (B) messages from wal2json don't include schema/table keys — must not KeyError"""
+        output = logical_replication.consume_message(
+            [{'tap_stream_id': 'myschema-mytable'}],
+            {},
+            self.WalMessage(payload='{"action":"B","xid":12345,"timestamp":"2026-04-06 12:00:00.000000+00"}',
+                            data_start='some lsn'),
+            None,
+            {}
+        )
+
+        self.assertDictEqual({}, output)
+
+    def test_consume_with_commit_action_without_schema_table_returns_state_unchanged(self):
+        """Commit (C) messages from wal2json don't include schema/table keys — must not KeyError"""
+        output = logical_replication.consume_message(
+            [{'tap_stream_id': 'myschema-mytable'}],
+            {},
+            self.WalMessage(payload='{"action":"C","xid":12345,"timestamp":"2026-04-06 12:00:00.000000+00"}',
+                            data_start='some lsn'),
+            None,
+            {}
+        )
+
+        self.assertDictEqual({}, output)
 
     @patch('tap_postgres.logical_replication.singer.write_message')
     @patch('tap_postgres.logical_replication.sync_common.send_schema_message')
@@ -1104,7 +1131,7 @@ def test_sync_tables_if_poll_duration_greater_than_logical_poll_total_seconds(se
             status_interval=10,
             options={
                 'format-version': 2,
-                'include-transaction': False,
+                'include-transaction': True,
                 'include-timestamp': True,
                 'include-types': False,
                 'actions': 'insert,update,delete',
@@ -1145,7 +1172,7 @@ def test_sync_tables_if_reached_max_run_seconds(self,
             status_interval=10,
             options={
                 'format-version': 2,
-                'include-transaction': False,
+                'include-transaction': True,
                 'include-timestamp': True,
                 'include-types': False,
                 'actions': 'insert,update,delete',

tests/db/tap_postgres_data_logical.sql

@@ -19,6 +19,13 @@ CREATE TABLE logical1.logical1_table2(
     PRIMARY KEY (cid)
 );
 
+CREATE TABLE logical1.logical1_not_selected(
+    cid serial NOT NULL,
+    cvarchar varchar,
+    PRIMARY KEY (cid)
+);
+
+
 CREATE TABLE logical1.logical1_edgydata (LIKE public.edgydata INCLUDING INDEXES);
 
 
@@ -106,6 +113,10 @@ INSERT INTO logical2.logical2_table1 (cvarchar) VALUES ('inserted row');
 
 UPDATE logical2.logical2_table1 SET cvarchar = 'updated row';
 
+INSERT INTO logical1.logical1_not_selected (cvarchar) VALUES ('inserted row');
+INSERT INTO logical1.logical1_not_selected (cvarchar) VALUES ('inserted row');
+INSERT INTO logical1.logical1_not_selected (cvarchar) VALUES ('inserted row');
+
 INSERT INTO logical3.logical3_table1 (cvarchar) VALUES ('inserted row');
 INSERT INTO logical3.logical3_table1 (cvarchar) VALUES ('inserted row');
 INSERT INTO logical3.logical3_table1 (cvarchar) VALUES ('inserted row');