tap-diffgen

tap-diffgen is a generic Singer tap for the "diffgen" pattern: an application-level audit-log table that journals every INSERT/UPDATE/DELETE made to other tables in the same database (columns: id, tablename, table_id, op, ts_created). Instead of reading a target table directly, the tap replays that log since the last bookmark and, per event:

• i/u (insert/update): repulls the affected row fresh from the target table and emits it with is_deleted: false.
• d (delete): emits a tombstone -- just the primary key, is_deleted: true, every other business column null.

Every emitted record also carries _sdc_diffgen_ts_created (the diffgen event's own timestamp), regardless of op. This is what a plain timestamp-filtered incremental extraction can never represent on its own: a row that no longer exists doesn't show up in a WHERE updated_at > ... query, so deletes silently vanish downstream. Built with the Meltano Singer SDK, supporting PostgreSQL and MySQL.

Installation

pip install plinxore-tap-diffgen

Configuration

Connection settings are global to the tap -- one tap instance targets one database, shared by every configured table (a single connector is built once per run, not one per table).

tables

Each entry is an object:

{
  "sql_dialect": "postgres",
  "host": "db.example.com",
  "port": 5432,
  "user": "diffgen_reader",
  "password": "...",
  "database": "app",
  "tables": [
    { "table_name": "invoices", "pk_column": "id", "mode": "full" },
    { "table_name": "orders", "pk_column": "id", "mode": "tombstones_only" }
  ]
}

The stream's JSON Schema is derived by introspecting the target table's own columns (information_schema.columns), plus is_deleted (boolean) and _sdc_diffgen_ts_created (date-time) -- never anything from the diffgen table itself (op, tablename, ...).

Configuration via environment variables

Copy .env.example to .env and fill in the real values (never committed). Meltano convention: TAP_DIFFGEN_<SETTING_NAME> in uppercase, e.g. TAP_DIFFGEN_PASSWORD.

The full list of settings is available via:

tap-diffgen --about

Usage

Direct CLI (without Meltano)

tap-diffgen --config config.json --discover > catalog.json
tap-diffgen --config config.json --catalog catalog.json --state state.json

Via Meltano (recommended)

# Install the Meltano CLI (if not already done)
pipx install meltano

# Install the plugins declared in meltano.yml
meltano install

# Check the config
meltano config tap-diffgen list
meltano config tap-diffgen test

# Run the pipeline
meltano run tap-diffgen target-jsonl

Meltano automatically manages state (bookmarks) between runs via its own system database -- no need to manually handle a state.json file.

Development

Tests

Unit tests (fake connector, no database) run standalone:

uv run pytest tests/test_client.py tests/test_tap.py tests/test_connectors_base.py

Connector and end-to-end tests need real Postgres/MySQL instances -- they skip cleanly (not a failure) if none is reachable:

docker compose -f tests/fixtures/docker-compose.yml up --wait
uv run pytest
docker compose -f tests/fixtures/docker-compose.yml down -v

Notable implementation details

• Bookmark: the replication cursor is diffgen's own id, which never appears on an emitted record (records mirror the target table's columns, not diffgen's) -- so the SDK's usual per-record auto-advance is neutralized (DiffgenStream._increment_stream_state is a no-op) and the bookmark is instead advanced manually, once per processed page, to that page's max diffgen id. Batches are always committed in ascending id order, so an interrupted run's last-committed bookmark is always a safe resume point.
• Dedup: within one page, the same row can appear more than once in diffgen (e.g. two updates, or an insert followed by a delete) -- only the latest event per table_id drives a repull/tombstone.
• Decimal precision: NUMERIC/DECIMAL columns are mapped to a JSON Schema multipleOf derived from the column's scale, the same convention target-s3 uses for its Parquet decimal128 mapping -- a value introspected here and later written to Parquet round-trips exact, not through a float64 approximation.
• MySQL BOOLEAN: stored as TINYINT(1), indistinguishable from a genuine small integer by data_type alone -- schema introspection checks column_type too, to avoid misclassifying every boolean column as an integer.
• Read-only connections use autocommit: found the hard way, via a real end-to-end test against a real, long-lived Postgres instance -- without it, every SELECT left an open transaction that could block DDL on the source database indefinitely.

See the code in tap_diffgen/client.py (stream logic) and tap_diffgen/connectors/ (Postgres/MySQL implementations) for details.

SDK Dev Guide

See the Meltano Singer SDK dev guide for more information.