Perform logical replication after initial sync (continuation)

Dockerfile

@@ -1,14 +1,17 @@
-FROM docker.io/bitnami/postgresql:12
+FROM docker.io/bitnami/postgresql:12-debian-11
 
 USER root
 
-RUN apt-get update \
-    && apt-get -y install git build-essential \
-    && git clone --depth 1 --branch wal2json_2_3 https://github.com/eulerto/wal2json.git \
-    && cd /wal2json \
-    && make && make install \
-    && cd / \
-    && rm -rf wal2json \
-    && rm -r /var/lib/apt/lists /var/cache/apt/archives
+# Git SHA of v2.4
+ENV WAL2JSON_COMMIT_ID=36fbee6cbb7e4bc1bf9ee4f55842ab51393e3ac0
+
+# Compile the plugins from sources and install
+RUN install_packages ca-certificates curl gcc git gnupg make pkgconf && \
+    curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | gpg --dearmor | tee /etc/apt/trusted.gpg.d/apt.postgresql.org.gpg >/dev/null && \
+    echo "deb http://apt.postgresql.org/pub/repos/apt bullseye-pgdg main" > /etc/apt/sources.list.d/pgdg.list && \
+    install_packages postgresql-server-dev-12 && \
+    git clone https://github.com/eulerto/wal2json -b master --single-branch && \
+    (cd /wal2json && git checkout $WAL2JSON_COMMIT_ID && make && make install) && \
+    rm -rf wal2json
 
 USER 1001

docker-compose.yml

@@ -2,20 +2,17 @@ version: "3.3"
 
 services:
   db_primary:
+    build: .
     container_name: "primary"
-    build:
-      context: .
-      dockerfile: Dockerfile
     ports:
       - "5432:5432"
     environment:
       - POSTGRESQL_REPLICATION_MODE=master
-      - POSTGRESQL_REPLICATION_USER=repl_user
-      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
-      - POSTGRES_USER=test_user
-      - POSTGRES_PASSWORD=my-secret-passwd
+      - POSTGRESQL_REPLICATION_USER=test_user
+      - POSTGRESQL_REPLICATION_PASSWORD=my-secret-passwd
       - POSTGRESQL_POSTGRES_PASSWORD=my-secret-passwd
       - POSTGRESQL_DATABASE=tap_postgres_test
+      - POSTGRESQL_WAL_LEVEL=logical
       - ALLOW_EMPTY_PASSWORD=yes
     volumes:
       -  ./db_setup/config:/bitnami/postgresql/conf
@@ -29,8 +26,8 @@ services:
       - db_primary
     environment:
       - POSTGRESQL_REPLICATION_MODE=slave
-      - POSTGRESQL_REPLICATION_USER=repl_user
-      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
+      - POSTGRESQL_REPLICATION_USER=test_user
+      - POSTGRESQL_REPLICATION_PASSWORD=my-secret-passwd
       - POSTGRESQL_MASTER_HOST=db_primary
       - POSTGRESQL_MASTER_PORT_NUMBER=5432
       - ALLOW_EMPTY_PASSWORD=yes

tap_postgres/__init__.py

@@ -131,6 +131,8 @@ def sync_method_for_streams(streams, state, default_replication_method):
             # finishing previously interrupted full-table (first stage of logical replication)
             lookup[stream['tap_stream_id']] = 'logical_initial_interrupted'
             traditional_steams.append(stream)
+            # do any required logical replication after inital sync is complete
+            logical_streams.append(stream)
 
         # inconsistent state
         elif get_bookmark(state, stream['tap_stream_id'], 'xmin') and \
@@ -142,6 +144,8 @@ def sync_method_for_streams(streams, state, default_replication_method):
             # initial full-table phase of logical replication
             lookup[stream['tap_stream_id']] = 'logical_initial'
             traditional_steams.append(stream)
+            # do any required logical replication after inital sync is complete
+            logical_streams.append(stream)
 
         else:  # no xmin but we have an lsn
             # initial stage of logical replication(full-table) has been completed. moving onto pure logical replication

tap_postgres/discovery_utils.py

@@ -23,6 +23,7 @@
 INTEGER_TYPES = {'integer', 'smallint', 'bigint'}
 FLOAT_TYPES = {'real', 'double precision'}
 JSON_TYPES = {'json', 'jsonb'}
+RANGE_TYPES = {'int4range', 'int8range', 'numrange', 'tsrange', 'tstzrange', 'daterange'}
 BASE_RECURSIVE_SCHEMAS = {
     'sdc_recursive_integer_array': {'type': ['null', 'integer', 'array'],
                                     'items': {'$ref': '#/definitions/sdc_recursive_integer_array'}},
@@ -280,6 +281,9 @@ def schema_for_column_datatype(col):
         schema['type'] = nullable_column('string', col.is_primary_key)
         return schema
 
+    if data_type in RANGE_TYPES:
+        schema['type'] = nullable_column('string', col.is_primary_key)
+
     return schema
 
 

tap_postgres/sync_strategies/logical_replication.py

@@ -10,7 +10,7 @@
 
 from select import select
 from psycopg2 import sql
-from singer import metadata, utils, get_bookmark
+from singer import metadata, get_bookmark
 from dateutil.parser import parse, UnknownTimezoneWarning, ParserError
 from functools import reduce
 
@@ -377,12 +377,14 @@ def row_to_singer_message(stream, row, version, columns, time_extracted, md_map,
 
 
 # pylint: disable=unused-argument,too-many-locals
-def consume_message(streams, state, msg, time_extracted, conn_info):
+def consume_message(streams, state, msg, conn_info):
     try:
         payload = json.loads(msg.payload)
     except Exception:
         return state
 
+    time_extracted = singer.utils.strptime_to_utc(payload['timestamp'])
+
     lsn = msg.data_start
 
     streams_lookup = {s['tap_stream_id']: s for s in streams}
@@ -575,7 +577,6 @@ def sync_tables(conn_info, logical_streams, state, end_lsn, state_file):
     lsn_comitted = min([get_bookmark(state_comitted, s['tap_stream_id'], 'lsn') for s in logical_streams])
     start_lsn = lsn_comitted
     lsn_to_flush = None
-    time_extracted = utils.now()
     slot = locate_replication_slot(conn_info)
     lsn_last_processed = None
     lsn_currently_processing = None
@@ -606,7 +607,6 @@ def sync_tables(conn_info, logical_streams, state, end_lsn, state_file):
                     int_to_lsn(start_lsn),
                     int_to_lsn(end_lsn),
                     slot)
-        # psycopg2 2.8.4 will send a keep-alive message to postgres every status_interval
         cur.start_replication(slot_name=slot,
                               decode=True,
                               start_lsn=start_lsn,
@@ -652,7 +652,7 @@ def sync_tables(conn_info, logical_streams, state, end_lsn, state_file):
                                 int_to_lsn(end_lsn))
                     break
 
-                state = consume_message(logical_streams, state, msg, time_extracted, conn_info)
+                state = consume_message(logical_streams, state, msg, conn_info)
 
                 # When using wal2json with write-in-chunks, multiple messages can have the same lsn
                 # This is to ensure we only flush to lsn that has completed entirely

tests/integration/test_unsupported_pk.py

@@ -28,7 +28,6 @@ def setUp(self):
                                   {"name": "circle_col",      "type": "circle"},
                                   {"name": "xml_col",      "type": "xml"},
                                   {"name": "composite_col",      "type": "person_composite"},
-                                  {"name": "int_range_col",      "type": "int4range"},
         ],
                       "name": Unsupported.table_name}
         with get_test_connection() as conn:
@@ -53,7 +52,6 @@ def test_catalog(self):
                           ('properties', 'bit_string_col'):     {'sql-datatype': 'bit(5)', 'selected-by-default': False, 'inclusion': 'unsupported'},
                           ('properties', 'line_col'):           {'sql-datatype': 'line', 'selected-by-default': False, 'inclusion': 'unsupported'},
                           ('properties', 'xml_col'):            {'sql-datatype': 'xml', 'selected-by-default': False, 'inclusion': 'unsupported'},
-                          ('properties', 'int_range_col'):      {'sql-datatype': 'int4range', 'selected-by-default': False, 'inclusion': 'unsupported'},
                           ('properties', 'circle_col'):         {'sql-datatype': 'circle', 'selected-by-default': False, 'inclusion': 'unsupported'},
                           ('properties', 'polygon_col'):        {'sql-datatype': 'polygon', 'selected-by-default': False, 'inclusion': 'unsupported'},
                           ('properties', 'box_col'):            {'sql-datatype': 'box', 'selected-by-default': False, 'inclusion': 'unsupported'},