Add 'skip_last_n_seconds' config parameter for incremental replication

tap_postgres/__init__.py

@@ -393,6 +393,7 @@ def main_impl():
     args = parse_args(REQUIRED_CONFIG_KEYS)
 
     limit = args.config.get('limit')
+    skip_last_n_seconds = args.config.get('skip_last_n_seconds')
     conn_config = {
         # Required config keys
         'host': args.config['host'],
@@ -409,7 +410,8 @@ def main_impl():
         'break_at_end_lsn': args.config.get('break_at_end_lsn', True),
         'logical_poll_total_seconds': float(args.config.get('logical_poll_total_seconds', 0)),
         'use_secondary': args.config.get('use_secondary', False),
-        'limit': int(limit) if limit else None
+        'limit': int(limit) if limit else None,
+        'skip_last_n_seconds': int(skip_last_n_seconds) if skip_last_n_seconds else None
     }
 
     if conn_config['use_secondary']:

tap_postgres/sync_strategies/incremental.py

@@ -87,7 +87,8 @@ def sync_table(conn_info, stream, state, desired_columns, md_map):
                                               "replication_key_value": replication_key_value,
                                               "schema_name": schema_name,
                                               "table_name": stream['table_name'],
-                                              "limit": conn_info['limit']
+                                              "limit": conn_info['limit'],
+                                              "skip_last_n_seconds": conn_info['skip_last_n_seconds']
                                               })
                 LOGGER.info('select statement: %s with itersize %s', select_sql, cur.itersize)
                 cur.execute(select_sql)
@@ -130,9 +131,16 @@ def _get_select_sql(params):
     table_name = params['table_name']
 
     limit_statement = f'LIMIT {params["limit"]}' if params["limit"] else ''
-    where_statement = f"WHERE {replication_key} >= '{replication_key_value}'::{replication_key_sql_datatype}" \
+
+    where_incr = f"{replication_key} >= '{replication_key_value}'::{replication_key_sql_datatype}" \
         if replication_key_value else ""
 
+    where_skip = f"{replication_key} <= NOW() - interval '{params['skip_last_n_seconds']} seconds'" \
+        if params["skip_last_n_seconds"] and replication_key_sql_datatype.startswith("timestamp") else ""
+
+    where_statement = f"WHERE {where_incr}{' AND ' if where_incr and where_skip else ''}{where_skip}" \
+        if where_incr or where_skip else ""
+
     select_sql = f"""
     SELECT {','.join(escaped_columns)}
     FROM (

tests/unit/test_incremental.py

@@ -30,7 +30,8 @@ def setUp(self) -> None:
             'password': 'foo_pass',
             'port': 12345,
             'use_secondary': False,
-            'limit': None
+            'limit': None,
+            'skip_last_n_seconds': None
         }
         self.stream = {'tap_stream_id': 5, 'stream': 'bar', 'table_name': 'pg_tbl'}
         self.md_map = {