Mastering Meltano: Meltano Incremental Replication Is One State Record Away From a Full Reload

Meltano runs Singer taps (extractors) that send rows to targets (loaders). In INCREMENTAL mode it saves the newest updated_at it loaded as a bookmark in its state store, and the next run reads rows at or after it. Lose the bookmark and the next run reads the whole table, logs No state was found, complete import., and exits 0.

What the lab found:

A wiped volume isn’t the only way to lose state. Renaming the extractor or running in another environment starts a fresh state record with no bookmark.
A stream with no key properties gains a duplicate row on an ordinary second run. The documented key-properties metadata fixes it; setting the loader’s load_method doesn’t.
--full-refresh repairs changed rows and leaves deleted ones in the warehouse.
Key-based replication misses hard deletes, and rows from long transactions that commit after a newer row was loaded.
A common alert wrapper reports a crashed run as success. A corrected one is below.

Every log excerpt and count comes from the lab scripts listed under Reproduce it, run on 2026-10-13, except one error string marked as coming from an earlier lab run. The stack was Meltano 4.4.1 on Python 3.12.15, tap-postgres 0.10.0 and target-postgres 0.8.0 (both on Meltano software development kit (SDK) 0.54.8), and PostgreSQL 17.11, with test tables of 1,000 and 100 rows. Part 1 and part 2 of this series get a pipeline to run once.

How the bookmark drives each run

Each table is a stream named schema-table, so public.customers becomes public-customers. Each plugin here is the meltanolabs variant, one of several published builds under the same name. Key-based replication is Meltano’s INCREMENTAL method. The stream metadata names a replication key, the column Meltano’s docs say is used to “identify new and updated data”.^[1]

    metadata:
      public-customers:
        replication-method: INCREMENTAL
        replication-key: updated_at

After each run Meltano saves the newest key value it loaded and passes that state to the extractor next time.^[1:1] Five runs against customers:

Run	State at start	Records read	Warehouse rows after
1	none	1,000	1,000
2	bookmark `21:56:13.408600`	25	1,020
3	cleared with `meltano state clear --force`	1,020	1,020
4	bookmark `21:58:05.438692`	25	1,020 (source: 1,017)
5	`meltano run --full-refresh`	1,017	1,020 (source: 1,017)

Run 3 is a lost state record. Same configuration, no bookmark, every row:

2026-10-13T21:58:39.442570Z [info     ] meltano      Using systemdb state backend
2026-10-13T21:58:39.442980Z [info     ] meltano      Reading state from Database
2026-10-13T21:58:39.445280Z [warning  ] meltano      No state was found, complete import.
2026-10-13T21:58:40.315456Z [info     ] tap-postgres tap-postgres v0.10.0, Meltano SDK v0.54.8
2026-10-13T21:58:40.315978Z [info     ] tap-postgres Skipping parse of env var settings...
2026-10-13T21:58:40.318224Z [info     ] tap-postgres Beginning sync of 'public-customers' in incremental mode
2026-10-13T21:58:40.318715Z [info     ] tap-postgres Starting incremental sync of 'public-customers' with bookmark value: None
2026-10-13T21:58:40.405356Z [info     ] target-postgres target-postgres v0.8.0, Meltano SDK v0.54.8
2026-10-13T21:58:40.405759Z [info     ] target-postgres Skipping parse of env var settings...
2026-10-13T21:58:40.463822Z [info     ] tap-postgres METRIC                         metric_name=sync_duration metric_value=0.14496421813964844
2026-10-13T21:58:40.464337Z [info     ] tap-postgres METRIC                         metric_name=record_count metric_value=1020

meltano state list prints the saved records and meltano state get <id> prints one on a single line.^[2] Before run 1, select count(*), max(updated_at) from customers returned 1000|2026-10-13 21:56:13.4086+00, and after it the record held that same max(updated_at):

{"singer_state":{"bookmarks":{"public-customers":{"replication_key":"updated_at","replication_key_value":"2026-10-13T21:56:13.408600+00:00"}}}}

The bookmark is a value from your data. Run 1 started at 21:57:42, more than a minute later.

Where state lives, and how to keep it

Each record is filed under a state ID, <environment>:<extractor>-to-<loader>, here dev:tap-postgres-to-target-postgres. By default it sits in Meltano’s system database. The Settings reference calls that “a SQLite database stored inside the .meltano directory at .meltano/meltano.db”.^[3] The lab printed:

database_uri [env: MELTANO_DATABASE_URI] current value: 'sqlite:////project/lab/.meltano/meltano.db' (default: 'sqlite:///$MELTANO_SYS_DIR_ROOT/meltano.db')
state_backend.uri [env: MELTANO_STATE_BACKEND_URI] current value: 'systemdb' (default)

In this Compose file .meltano is a named volume, so docker compose down -v deleted the state: meltano state list then printed No state IDs found., and the next run logged No state was found, complete import. and read all 1,000 rows. Part 1’s output shows the same warning on its first container run, and each docker run there starts a fresh container.

A new state ID has no bookmark either. An extractor inherited from tap-postgres under a new name, tap-shop, read the same table from scratch:

2026-10-13T22:37:50.244056Z [info     ] meltano      Running job with name 'dev:tap-shop-to-target-postgres' and run ID '01a1376a-35f4-7e50-94fc-6d9c99f0904f'
2026-10-13T22:37:53.806879Z [info     ] meltano      Using systemdb state backend
2026-10-13T22:37:53.807375Z [info     ] meltano      Reading state from Database
2026-10-13T22:37:53.809852Z [warning  ] meltano      No state was found, complete import.

It read 1,000 records. meltano --environment=prod run tap-postgres target-postgres did the same under prod:tap-postgres-to-target-postgres while the dev record still held its bookmark, and meltano state list showed all four IDs whatever the environment. When a production run reloads everything, check for a renamed plugin or a deployment that sets a different environment before blaming the volume.

Move the store off the container. We’d use the Postgres system database if you already run and back up a Postgres server, and the file backend on a host directory for a single-host Compose deployment. For Postgres, the image needs the meltano[postgres] extra (this lab’s Dockerfile has it) and one environment variable:

services:
  meltano:
    environment:
      MELTANO_DATABASE_URI: postgresql+psycopg://postgres:lab@dst:5432/meltano_state

Meltano doesn’t create that database. Before create database meltano_state, meltano state list exited 1:

2026-10-13T22:10:02.606580Z [error    ] meltano      Could not connect to the database after 3 attempts. Max retries exceeded.
2026-10-13T22:10:02.607658Z [error    ] meltano      (psycopg.OperationalError) connection failed: connection to server at "172.18.0.2", port 5432 failed: FATAL:  database "meltano_state" does not exist

Once it existed, the bookmark survived a new, empty .meltano volume: the next run started from bookmark value: 2026-10-13T22:08:27.868012+00:00 and read 1 record. It was lost on a full down -v, because the destination Postgres here keeps its data on an anonymous volume, so give that database real storage.

The file backend survived down -v, and the next run started from bookmark value: 2026-10-13T22:10:04.855393+00:00. An s3:// backend wasn’t tested.

services:
  meltano:
    environment:
      MELTANO_STATE_BACKEND_URI: file:///state-store
    volumes: [ "./state-store:/state-store" ]

Alert on a full reload. meltano run exits 0 after a full reload, so wrap it:

set -o pipefail; meltano run tap-postgres target-postgres 2>&1 | tee run.log; rc=$?; if grep -q 'No state was found, complete import.' run.log; then echo "FULL RELOAD" >&2; exit 2; fi; exit "$rc"

Situation	Without `rc=$?`	This wrapper
Brand-new pipeline, first run	not run	`FULL RELOAD`, exit 2
State present, run succeeds	exit 0	exit 0
State cleared, run succeeds	`FULL RELOAD`, exit 2	`FULL RELOAD`, exit 2
Destination Postgres stopped	exit 0	exit 1

The version without rc=$? ends on an if that takes no branch, so it reported a crashed load as success. With the destination stopped, this wrapper passed the failure through:

2026-10-13T22:36:52.632969Z [error    ] target-postgres (psycopg.OperationalError) failed to resolve host 'dst': [Errno -2] Name or service not known plugin_exc_message="(psycopg.OperationalError) failed to resolve host 'dst': [Errno -2] Name or service not known\n(Background on this error at: https://sqlalche.me/e/21/e3q8)" plugin_exc_type=OperationalError
2026-10-13T22:36:52.829921Z [error    ] meltano      Loader failed
2026-10-13T22:36:52.830283Z [error    ] meltano      Block run completed
2026-10-13T22:36:52.830512Z [info     ] meltano      Run completed
FIXED wrapper exit code: 1

A brand-new pipeline’s first run trips the alert too, so run the initial load without it. A crash on a run that also logs the no-state warning wasn’t tested; reading the script, it returns 2 and the reload alert hides the crash.

Why a no-change run re-sends rows

Run 4 read 25 records although no source row was newer than the bookmark. To see what the next run will send, run the tap by hand from the saved state (singer_output.py is listed under Reproduce it):

meltano-container$ meltano state get dev:tap-postgres-to-target-postgres 2>/dev/null > /tmp/state-full.json
meltano-container$ python - state > /tmp/state.json < singer_output.py
meltano-container$ cat /tmp/state.json
{"bookmarks": {"public-customers": {"replication_key": "updated_at", "replication_key_value": "2026-10-13T21:58:05.438692+00:00"}}}
meltano-container$ meltano invoke tap-postgres --state /tmp/state.json > /tmp/tap-out.jsonl

meltano-container$ python - records < singer_output.py
record_count 25
ids [1, 2, 3, 4, 5, 1001, 1002, 1003, 1004, 1005, 1006, 1007, 1008, 1009, 1010, 1011, 1012, 1013, 1014, 1015, 1016, 1017, 1018, 1019, 1020]
distinct updated_at ['2026-10-13T21:58:05.438692+00:00']
shop> select count(*), min(id), max(id) from customers where updated_at = (select max(updated_at) from customers)
25|1|1020
shop> select count(*) from customers where updated_at > (select max(updated_at) from customers)
0

These are the 20 inserts and 5 renames from one source change. psql runs a multi-statement -c string as one transaction,^[4] and now() returns the transaction’s start time,^[5] so all 25 share one updated_at, which became the bookmark. The Meltano SDK documentation says “According to the Singer spec, bookmark comparisons are performed on the basis of greater than or equal to logic”,^[6] which SDK 0.54.8 implements as column >= start_val (singer_sdk/sql/stream.py, line 206).

Our reasoning for why that’s the safe choice: with a strict >, a row committed later with exactly the bookmark’s timestamp would never be read. Run 2 read 25, where you might expect 26, because the row at the old bookmark (id 1) was one of the five renamed. Re-sends do no harm when your loader upserts, that is, updates the row whose key matches and inserts the rest. Whether yours does depends on the stream’s keys.

Streams without key properties duplicate rows

Key properties are the columns a stream declares as its primary key. Meltano builds a catalog, the stream list it hands to the tap, and the tap repeats each stream’s keys in its SCHEMA message, the Singer message that describes a stream before its records. target-postgres 0.8.0 upserts on those keys. With none, it appends. The events table has no primary key:

CREATE TABLE events (
  id         int,
  note       text,
  updated_at timestamptz DEFAULT now()
);
INSERT INTO events (id, note, updated_at)
SELECT g, 'event ' || g, now() - (g || ' minutes')::interval FROM generate_series(1, 100) g;

Counts are rows, distinct IDs and duplicates in raw.events:

Extractor setup	First run	Second run, state kept	State cleared, run again	`key_properties` in SCHEMA
No key	100 / 100 / 0	101 / 100 / 1	201 / 100 / 101	`[]`
No key, `TARGET_POSTGRES_LOAD_METHOD=upsert`	100 / 100 / 0	101 / 100 / 1	201 / 100 / 101	`[]`
Metadata `table-key-properties: [id]`	100 / 100 / 0	101 / 100 / 1	201 / 100 / 101	`[]`
Metadata `key-properties: [id]`	100 / 100 / 0	100 / 100 / 0	100 / 100 / 0	`["id"]`
Stream map `__key_properties__: [id]`	100 / 100 / 0	100 / 100 / 0	100 / 100 / 0	`["id"]`

The keyless stream duplicated with state intact: its second run re-sent the newest row, and the target appended it. One extra row per run, every run.

warehouse> select count(*), count(distinct id), count(*) - count(distinct id) as duplicate_rows from raw.events
101|100|1

The fix is the documented metadata key. Meltano’s guide says “You can also use the metadata extra to override the key-properties (the columns that uniquely identify a record)”,^[1:2] and the generated catalog then carried key_properties ['id']:

  - name: tap-postgres-events-keyprops
    inherit_from: tap-postgres
    select:
    - public-events.*
    metadata:
      public-events:
        replication-method: INCREMENTAL
        replication-key: updated_at
        key-properties: [id]

The warehouse table got "events_pkey" PRIMARY KEY, btree (id). table-key-properties isn’t a Meltano metadata key, and the tap ignored it. A stream map, the SDK’s per-stream transformation setting, works too:

  - name: tap-postgres-events-mapkey
    inherit_from: tap-postgres
    config:
      stream_maps:
        public-events:
          __key_properties__: [id]
    select:
    - public-events.*
    metadata:
      public-events:
        replication-method: INCREMENTAL
        replication-key: updated_at

Check the key in a normal invoke’s SCHEMA line, where the plain stream shows "key_properties":[]:

meltano-container$ meltano invoke tap-postgres-events-keyprops 2>/dev/null | grep '"type":"SCHEMA"'
{"type":"SCHEMA","stream":"public-events","schema":{"properties":{"id":{"type":["integer","null"]},"note":{"type":["string","null"]},"updated_at":{"format":"date-time","type":["string","null"]}},"type":"object","$schema":"https://json-schema.org/draft/2020-12/schema"},"key_properties":["id"],"bookmark_properties":["updated_at"]}

meltano invoke <extractor> --test=schema can’t do this check. It printed [] for the working key-properties stream and failed on the stream map with StreamMapConfigError: Invalid key properties for 'public-events': [id]. Property 'id' was not detected in schema.

load_method doesn’t change the outcome. target-postgres 0.8.0 replaces the SDK’s table setup (connector.py line 369) and sets self.append_only = not self.key_properties (sinks.py line 55), so by the code keyed streams upsert and keyless ones append. The lab measured the keyless stream under the default append-only and under upsert (counts above), and the keyed customers stream under append-only and overwrite, each ending at 1020|1020|0; a keyless stream under overwrite wasn’t run.

What incremental replication misses

Before run 4, customer 500 was renamed with updated_at left alone, and IDs 900 to 902 were hard-deleted, meaning removed from the table, where a soft delete only marks a row with an update. After run 4:

warehouse> select name from raw.customers where id=500
customer 500
warehouse> select count(*) from raw.customers
1020
warehouse> select count(*) from raw.customers where id in (900,901,902)
3

--full-refresh repairs updates and keeps deletes. The flag ignores saved state for one run.^[2:1] Run 5 logged Performing full refresh, ignoring state left behind by any previous runs., read 1,017 rows and corrected row 500 to silently changed. IDs 900 to 902 stayed, since upserts never delete, and the state was written again with the same 21:58:05.438692.

Meltano’s guide says key-based replication captures records “inserted or updated (but not deleted)”. Log-based replication, it says, identifies records “inserted, updated, and deleted”.^[1:3]

Silent updates: a trigger moves the key. If you own the source, stamp every update:

CREATE FUNCTION set_updated_at() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
  NEW.updated_at := now();
  RETURN NEW;
END
$$;
CREATE TRIGGER customers_set_updated_at
  BEFORE UPDATE ON customers
  FOR EACH ROW EXECUTE FUNCTION set_updated_at();

Without it, a later rename to silent, no trigger was still missing from the warehouse after the next run. With it:

shop> update customers set name='silent, with trigger' where id=500
UPDATE 1
shop> select id, name, updated_at from customers where id=500
500|silent, with trigger|2026-10-13 22:02:03.715819+00

The next run read 26 records, one more than the 25 at the bookmark, and the warehouse row became 500|silent, with trigger|2026-10-13 22:02:03.715819.

Deletes: reconcile keys with an anti-join. A soft delete turns a delete into an update the trigger stamps, if your application can use one. Otherwise, copy the source’s IDs into the warehouse and anti-join, returning warehouse rows with no matching ID. The copy is a FULL_TABLE stream, which rereads the whole table every run, from a second extractor:

  - name: tap-postgres-keys
    inherit_from: tap-postgres
    config:
      emit_activate_version_messages: true
      stream_maps:
        public-customers:
          __alias__: customer_keys
    select:
    - public-customers.id
    metadata:
      public-customers:
        replication-method: FULL_TABLE
        replication-key: ''

emit_activate_version_messages: true makes the tap send ACTIVATE_VERSION, the message that tells the target to remove rows a full load didn’t send, and hard_delete: true on the keys loader makes it delete them:

  - name: target-postgres-keys
    inherit_from: target-postgres
    config:
      hard_delete: true

Without the first setting, the keys table kept all 1,000 IDs after the deletes and the anti-join returned nothing. With both, after meltano run tap-postgres-keys target-postgres-keys:

warehouse> select w.id from raw.customers w left join raw.customer_keys k using (id) where k.id is null order by w.id
900
901
902

Log-based replication. tap-postgres 0.10.0 supports LOG_BASED, which reads changes from the write-ahead log (WAL) through a replication slot, a named position the server keeps for one consumer. Its README lists four requirements.^[7] You need the wal2json plugin; wal_level = logical, max_replication_slots and max_wal_senders, then a restart; a slot from pg_create_logical_replication_slot('tappostgres', 'wal2json'); and replication_key: _sdc_lsn in the stream metadata. The stock postgres:17.11 image ships only pgoutput.so, so this lab didn’t run it. Whether the role needs the REPLICATION attribute is untested too.

Long transactions: the gap a trigger can’t close. now() is the transaction’s start time, so a row updated in a long transaction is stamped before it commits. Two 20-second transactions, trigger in place:

Case	Long transaction’s `now()`	Other commit while open	Bookmark after a run during it	Row loaded after its commit?
Alone	22:02:19.090986	none	22:02:03.715819	yes, next run
With a concurrent commit	22:02:50.077291	`id` 21 at 22:02:52.695407	22:02:52.695407	no, after 2 more runs

A row is lost when another commit moves the bookmark past its transaction’s start time while that transaction is still open. In the second case:

shop> select id, name, updated_at from customers where id in (20,21) order by id
20|late commit B|2026-10-13 22:02:50.077291+00
21|concurrent commit|2026-10-13 22:02:52.695407+00

Two runs later the warehouse still held 20|customer 20|2026-10-13 21:37:13.4086. As reasoning, the chance that a run lands inside a given long transaction is roughly its duration divided by your run interval, and the row is lost only if another commit also lands first. clock_timestamp() in the trigger narrows the gap but still stamps before the commit (untested). What holds is reconciliation that compares id and updated_at between source and warehouse; the lab ran only the ID comparison.

Troubleshooting Meltano state

No state was found, complete import. on a run that isn’t the first. The state store had no record under that run’s state ID: it was cleared, its volume was recreated, or the ID changed. Use the wrapper from the state section so the run exits 2.

A job name you don’t recognize. Compare Running job with name '...' in the run log with meltano state list. A renamed extractor or a different --environment gives a new state ID, such as dev:tap-shop-to-target-postgres or prod:tap-postgres-to-target-postgres, with no bookmark.

Aborted! from meltano state clear in a script. Without a terminal the confirmation prompt reads end-of-input, nothing is cleared, and the command exits 1, so pass --force:

Aborted!
This will clear state for the job(s). Continue? [y/N]: exit code: 1

FATAL: database "meltano\_state" does not exist. MELTANO_DATABASE_URI names a database nobody created. Create it; Meltano adds its tables (alembic_version, plugin_settings, runs, state) on the next command.

ModuleNotFoundError: No module named 'psycopg' with a postgresql+psycopg:// system database. Plain meltano lacks the driver; install meltano[postgres]==4.4.1. Error text from an earlier lab run.

KeyError: 'updated\_at' from an ID-only child extractor. An extractor with inherit_from also inherits replication-key: updated_at, and a stream that selects only id has no such column. Removing the key or setting it to null still failed; replication-key: '' worked:

2026-10-13T22:13:05.239362Z [error    ] meltano      updated_at                     plugin_exc_message="'updated_at'" plugin_exc_type=KeyError
exit code: 1

Wrong counts when comparing warehouse times to a bookmark. target-postgres stores updated_at as timestamp without time zone holding the Coordinated Universal Time (UTC) wall-clock value, and Postgres reads that as session-local time against a timestamptz. In an America/Chicago session, updated_at >= timestamptz '2026-10-13 11:57:13.4086+00' counted 301 rows where the right answer was 1. updated_at at time zone 'UTC' >= ... returned 1 in both sessions.

Reproduce it

Dockerfile. It installs no system packages: every Postgres driver here ships as a binary wheel, and git matters only for plugins installed from a git repository.

FROM python:3.12.15-slim
RUN pip install --no-cache-dir "meltano[postgres]==4.4.1"
WORKDIR /project/lab
COPY meltano.yml .
RUN meltano install

compose.yaml:

services:
  src:
    image: postgres:17.11
    environment: { POSTGRES_PASSWORD: lab, POSTGRES_DB: shop }
    volumes: [ "./init:/docker-entrypoint-initdb.d:ro" ]
  dst:
    image: postgres:17.11
    environment: { POSTGRES_PASSWORD: lab, POSTGRES_DB: warehouse }
  meltano:
    build: .
    depends_on: [ src, dst ]
    volumes: [ "meltano_state:/project/lab/.meltano" ]
    command: [ "sleep", "infinity" ]
volumes:
  meltano_state:

init/01-seed.sql, 1,000 customers with updated_at one minute apart:

CREATE TABLE customers (
  id         serial PRIMARY KEY,
  name       text NOT NULL,
  updated_at timestamptz NOT NULL DEFAULT now()
);
INSERT INTO customers (name, updated_at)
SELECT 'customer ' || g, now() - (g || ' minutes')::interval FROM generate_series(1, 1000) g;

init/02-events.sql, the keyless table:

CREATE TABLE events (
  id         int,
  note       text,
  updated_at timestamptz DEFAULT now()
);
INSERT INTO events (id, note, updated_at)
SELECT g, 'event ' || g, now() - (g || ' minutes')::interval FROM generate_series(1, 100) g;

meltano.yml, as used by drill.sh, state-backends.sh, keys-stream-variants.sh and source-reads.sh:

default_environment: dev
environments:
- name: dev
send_anonymous_usage_stats: false
plugins:
  extractors:
  - name: tap-postgres
    variant: meltanolabs
    pip_url: meltanolabs-tap-postgres==0.10.0
    config:
      host: src
      port: 5432
      user: postgres
      password: lab
      database: shop
      filter_schemas: [public]
    select:
    - public-customers.*
    metadata:
      public-customers:
        replication-method: INCREMENTAL
        replication-key: updated_at
  - name: tap-postgres-keys
    inherit_from: tap-postgres
    config:
      emit_activate_version_messages: true
      stream_maps:
        public-customers:
          __alias__: customer_keys
    select:
    - public-customers.id
    metadata:
      public-customers:
        replication-method: FULL_TABLE
        replication-key: ''
  - name: tap-postgres-events
    inherit_from: tap-postgres
    select:
    - public-events.*
    metadata:
      public-events:
        replication-method: INCREMENTAL
        replication-key: updated_at
  - name: tap-postgres-events-keyed
    inherit_from: tap-postgres
    select:
    - public-events.*
    metadata:
      public-events:
        replication-method: INCREMENTAL
        replication-key: updated_at
        table-key-properties: [id]
  - name: tap-postgres-events-mapkey
    inherit_from: tap-postgres
    config:
      stream_maps:
        public-events:
          __key_properties__: [id]
    select:
    - public-events.*
    metadata:
      public-events:
        replication-method: INCREMENTAL
        replication-key: updated_at
  loaders:
  - name: target-postgres
    variant: meltanolabs
    pip_url: meltanolabs-target-postgres==0.8.0
    config:
      host: dst
      port: 5432
      user: postgres
      password: lab
      database: warehouse
      default_target_schema: raw
  - name: target-postgres-keys
    inherit_from: target-postgres
    config:
      hard_delete: true

extras.sh ran with the same meltano.yml plus two additions. The environments list gained prod:

environments:
- name: dev
- name: prod

and the extractors list gained tap-postgres-events-keyprops, the block shown in the key properties section.

compose.pgstate.yaml, the Postgres system database:

services:
  meltano:
    environment:
      MELTANO_DATABASE_URI: postgresql+psycopg://postgres:lab@dst:5432/meltano_state

compose.freshvol.yaml, an empty .meltano volume:

services:
  meltano:
    volumes: [ "meltano_state_fresh:/project/lab/.meltano" ]
volumes:
  meltano_state_fresh:

compose.filestate.yaml, the file state backend:

services:
  meltano:
    environment:
      MELTANO_STATE_BACKEND_URI: file:///state-store
    volumes: [ "./state-store:/state-store" ]

singer_output.py, which summarizes captured tap output:

"""Read Singer output inside the Meltano container.

Usage (script on stdin):
  python - state   < singer_output.py   # meltano state get JSON on /tmp/state-full.json -> tap state on stdout
  python - records < singer_output.py   # RECORD summary of /tmp/tap-out.jsonl
  python - schema  < singer_output.py   # SCHEMA key_properties in /tmp/tap-out.jsonl
"""
import json
import sys

STATE_FULL = "/tmp/state-full.json"
TAP_OUT = "/tmp/tap-out.jsonl"


def messages():
    with open(TAP_OUT) as f:
        return [json.loads(line) for line in f if line.startswith("{")]


mode = sys.argv[1]
if mode == "state":
    with open(STATE_FULL) as f:
        print(json.dumps(json.load(f)["singer_state"]))
elif mode == "records":
    records = [m for m in messages() if m["type"] == "RECORD"]
    print("record_count", len(records))
    print("ids", sorted(m["record"]["id"] for m in records))
    print("distinct updated_at", sorted({m["record"]["updated_at"] for m in records}))
elif mode == "schema":
    for m in messages():
        if m["type"] == "SCHEMA":
            print("SCHEMA", m["stream"], "key_properties", m["key_properties"])

drill.sh, runs 1 to 5 and the customers, events, trigger, transaction, load method and time zone tests:

#!/usr/bin/env bash
# Usage: bash drill.sh 2>&1 | tee drill.log
set -eu
C="docker compose -p mlab4"
ID=dev:tap-postgres-to-target-postgres

src() { echo "shop> $1"; $C exec -T src psql -U postgres -d shop -At -c "$1"; }
dst() { echo "warehouse> $1"; $C exec -T dst psql -U postgres -d warehouse -At -c "$1"; }
mel() { echo "\$ meltano $*"; $C exec -T meltano meltano "$@"; }
in_meltano() { echo "meltano-container\$ $1"; $C exec -T meltano sh -c "$1"; }
singer_output() { echo "meltano-container\$ python - $1 < singer_output.py"; $C exec -T meltano python - "$1" < singer_output.py; }
mel_lm() {
  local method=$1; shift
  echo "\$ TARGET_POSTGRES_LOAD_METHOD=$method meltano $*"
  $C exec -T -e TARGET_POSTGRES_LOAD_METHOD="$method" meltano meltano "$@"
}
events_counts() { dst "select count(*), count(distinct id), count(*) - count(distinct id) as duplicate_rows from raw.events"; }
customers_counts() {
  src "select count(*) from customers"
  dst "select count(*), count(distinct id), count(*) - count(distinct id) as duplicate_rows from raw.customers"
}

echo "== clean start"
$C down -v
$C up -d --build
until $C exec -T src pg_isready -h 127.0.0.1 -U postgres -q \
   && $C exec -T dst pg_isready -h 127.0.0.1 -U postgres -q; do sleep 2; done

echo "== versions"
mel --version
$C exec -T meltano python --version
mel invoke tap-postgres --about
mel invoke target-postgres --about
src "select version()"
in_meltano "grep -n -B2 pip_url meltano.yml"

echo "== run 1: no state"
src "select count(*), max(updated_at) from customers"
mel run tap-postgres target-postgres
dst "select count(*) from raw.customers"
mel state list
mel state get $ID

echo "== key stream: first load (FULL_TABLE, ids only)"
mel run tap-postgres-keys target-postgres-keys
dst "select count(*) from raw.customer_keys"

echo "== change the source: 20 inserts, 5 renames (one -c string, three statements, one transaction)"
src "insert into customers(name) select 'new customer '||g from generate_series(1,20) g;
     update customers set name = name || ' (renamed)', updated_at = now() where id between 1 and 5;
     select count(*) from customers"
src "select updated_at, count(*), min(id), max(id) from customers group by updated_at order by updated_at desc limit 3"

echo "== run 2: state present"
mel run tap-postgres target-postgres
dst "select count(*) from raw.customers"
dst "select count(*) from raw.customers where name like '%(renamed)'"
mel state get $ID

echo "== state clear without --force, no stdin"
rc=0
mel state clear $ID || rc=$?
echo "exit code: $rc"
mel state list

echo "== lose the state, run 3"
mel state clear $ID --force
mel state list
rc=0
mel run tap-postgres target-postgres || rc=$?
echo "exit code: $rc"
dst "select count(*) from raw.customers"
mel state get $ID

echo "== change the source: silent rename, 3 hard deletes"
src "update customers set name='silently changed' where id=500;
     delete from customers where id in (900,901,902);
     select count(*) from customers"

echo "== run 4: state present"
mel run tap-postgres target-postgres
dst "select name from raw.customers where id=500"
dst "select count(*) from raw.customers"
dst "select count(*) from raw.customers where id in (900,901,902)"
mel state get $ID

echo "== which rows a no-change run re-sends"
in_meltano "meltano state get $ID 2>/dev/null > /tmp/state-full.json"
echo "meltano-container\$ python - state > /tmp/state.json < singer_output.py"
$C exec -T meltano sh -c "python - state > /tmp/state.json" < singer_output.py
in_meltano "cat /tmp/state.json"
in_meltano "meltano invoke tap-postgres --state /tmp/state.json > /tmp/tap-out.jsonl"
singer_output records
src "select count(*), min(id), max(id) from customers where updated_at = (select max(updated_at) from customers)"
src "select count(*) from customers where updated_at > (select max(updated_at) from customers)"

echo "== run 5: --full-refresh with state present"
mel state get $ID
mel run --full-refresh tap-postgres target-postgres
dst "select name from raw.customers where id=500"
dst "select count(*) from raw.customers"
dst "select count(*) from raw.customers where id in (900,901,902)"
mel state get $ID

SNIPPET='set -o pipefail; meltano run tap-postgres target-postgres 2>&1 | tee run.log; if grep -q '"'"'No state was found, complete import.'"'"' run.log; then echo "FULL RELOAD" >&2; exit 2; fi'
echo "== alert snippet, state present"
echo "meltano-container\$ bash -c '$SNIPPET'"
rc=0
$C exec -T meltano bash -c "$SNIPPET" || rc=$?
echo "exit code: $rc"
echo "== alert snippet, state absent"
mel state clear $ID --force
echo "meltano-container\$ bash -c '$SNIPPET'"
rc=0
$C exec -T meltano bash -c "$SNIPPET" || rc=$?
echo "exit code: $rc"

echo "== key stream: second load, then reconcile deletes"
mel run tap-postgres-keys target-postgres-keys
dst "select count(*) from raw.customer_keys"
dst "select w.id from raw.customers w left join raw.customer_keys k using (id) where k.id is null order by w.id"

echo "== keyless stream: events, no primary key"
src "select count(*), count(distinct id) from events"
mel run tap-postgres-events target-postgres
events_counts
mel run tap-postgres-events target-postgres
events_counts
mel state clear dev:tap-postgres-events-to-target-postgres --force
mel run tap-postgres-events target-postgres
events_counts
in_meltano "meltano invoke tap-postgres-events > /tmp/tap-out.jsonl 2>/dev/null"
singer_output schema

echo "== keyless stream with metadata table-key-properties: [id]"
dst "drop table raw.events"
mel run tap-postgres-events-keyed target-postgres
events_counts
mel run tap-postgres-events-keyed target-postgres
events_counts
mel state clear dev:tap-postgres-events-keyed-to-target-postgres --force
mel run tap-postgres-events-keyed target-postgres
events_counts
in_meltano "meltano invoke tap-postgres-events-keyed > /tmp/tap-out.jsonl 2>/dev/null"
singer_output schema

echo "== keyless stream with stream map __key_properties__: [id]"
dst "drop table raw.events"
mel run tap-postgres-events-mapkey target-postgres
events_counts
mel run tap-postgres-events-mapkey target-postgres
events_counts
mel state clear dev:tap-postgres-events-mapkey-to-target-postgres --force
mel run tap-postgres-events-mapkey target-postgres
events_counts
in_meltano "meltano invoke tap-postgres-events-mapkey > /tmp/tap-out.jsonl 2>/dev/null"
singer_output schema
$C exec -T dst psql -U postgres -d warehouse -c '\d raw.events'

echo "== silent update, no trigger"
src "update customers set name='silent, no trigger' where id=500"
mel run tap-postgres target-postgres
src "select id, name, updated_at from customers where id=500"
dst "select id, name, updated_at from raw.customers where id=500"

echo "== BEFORE UPDATE trigger"
$C exec -T src psql -U postgres -d shop -a <<'SQL'
CREATE FUNCTION set_updated_at() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
  NEW.updated_at := now();
  RETURN NEW;
END
$$;
CREATE TRIGGER customers_set_updated_at
  BEFORE UPDATE ON customers
  FOR EACH ROW EXECUTE FUNCTION set_updated_at();
SQL
src "update customers set name='silent, with trigger' where id=500"
src "select id, name, updated_at from customers where id=500"
mel state get $ID
mel run tap-postgres target-postgres
dst "select id, name, updated_at from raw.customers where id=500"
mel state get $ID

echo "== long transaction, alone"
$C exec -T src psql -U postgres -d shop -At -c "begin; update customers set name='late commit' where id=10; select 'txn now()='||now(); select pg_sleep(20); select 'commit at '||clock_timestamp(); commit;" &
BG=$!
sleep 3
src "select state, xact_start, left(query, 60) from pg_stat_activity where datname='shop' and xact_start is not null and pid <> pg_backend_pid()"
mel run tap-postgres target-postgres
dst "select id, name, updated_at from raw.customers where id=10"
mel state get $ID
wait $BG
src "select id, name, updated_at from customers where id=10"
mel run tap-postgres target-postgres
dst "select id, name, updated_at from raw.customers where id=10"
mel state get $ID

echo "== long transaction, with a concurrent commit"
$C exec -T src psql -U postgres -d shop -At -c "begin; update customers set name='late commit B' where id=20; select 'txn now()='||now(); select pg_sleep(20); select 'commit at '||clock_timestamp(); commit;" &
BG=$!
sleep 3
src "update customers set name='concurrent commit' where id=21"
src "select id, name, updated_at from customers where id in (20,21) order by id"
mel run tap-postgres target-postgres
dst "select id, name, updated_at from raw.customers where id in (20,21) order by id"
mel state get $ID
wait $BG
src "select id, name, updated_at from customers where id in (20,21) order by id"
mel run tap-postgres target-postgres
dst "select id, name, updated_at from raw.customers where id in (20,21) order by id"
mel run tap-postgres target-postgres
dst "select id, name, updated_at from raw.customers where id in (20,21) order by id"
mel state get $ID

echo "== load_method"
$C exec -T dst psql -U postgres -d warehouse -c '\d raw.customers'
in_meltano "TARGET_POSTGRES_LOAD_METHOD=overwrite meltano config print target-postgres 2>/dev/null | grep load_method"
customers_counts
mel state clear $ID --force
mel_lm append-only run tap-postgres target-postgres
customers_counts
mel state clear $ID --force
mel_lm overwrite run tap-postgres target-postgres
customers_counts

echo "== time zones"
src "show timezone"
dst "show timezone"
src "select id, updated_at from customers where id=600"
dst "select id, updated_at from raw.customers where id=600"
src "set timezone='America/Chicago'; select id, updated_at from customers where id=600"
dst "set timezone='America/Chicago'; select id, updated_at from raw.customers where id=600"
BM=$($C exec -T src psql -U postgres -d shop -At -c "select max(updated_at) from customers where id between 600 and 1000")
echo "BM=$BM"
dst "select count(*) from raw.customers where id between 600 and 1000 and updated_at >= timestamptz '$BM'"
dst "select count(*) from raw.customers where id between 600 and 1000 and updated_at at time zone 'UTC' >= timestamptz '$BM'"
dst "set timezone='America/Chicago'; select count(*) from raw.customers where id between 600 and 1000 and updated_at >= timestamptz '$BM'"
dst "set timezone='America/Chicago'; select count(*) from raw.customers where id between 600 and 1000 and updated_at at time zone 'UTC' >= timestamptz '$BM'"

echo "== where Meltano keeps state"
in_meltano "meltano config list meltano --all 2>/dev/null | grep -E '^(database_uri|state_backend.uri) '"
in_meltano "ls -la .meltano"
echo "== drill done"

state-backends.sh, the state backend tests:

#!/usr/bin/env bash
# Usage: bash state-backends.sh 2>&1 | tee state-backends.log
set -eu
BASE="docker compose -p mlab4"
PG="docker compose -p mlab4 -f compose.yaml -f compose.pgstate.yaml"
PGFRESH="docker compose -p mlab4 -f compose.yaml -f compose.pgstate.yaml -f compose.freshvol.yaml"
FILE="docker compose -p mlab4 -f compose.yaml -f compose.filestate.yaml"
ID=dev:tap-postgres-to-target-postgres
C=$BASE

mel() { echo "\$ meltano $*"; $C exec -T meltano meltano "$@"; }
in_meltano() { echo "meltano-container\$ $1"; $C exec -T meltano sh -c "$1"; }
dst_db() { local db=$1; shift; echo "$db> $1"; $C exec -T dst psql -U postgres -d "$db" -At -c "$1"; }
wait_pg() { until $C exec -T src pg_isready -h 127.0.0.1 -U postgres -q \
              && $C exec -T dst pg_isready -h 127.0.0.1 -U postgres -q; do sleep 2; done; }
state_settings() { in_meltano "meltano config list meltano --all 2>/dev/null | grep -E '^(database_uri|state_backend.uri) '"; }
mounts() { echo "\$ docker inspect $1 (mounts)"; docker inspect "$1" --format '{{range .Mounts}}{{.Type}} {{.Name}}{{.Source}} -> {{.Destination}}{{println}}{{end}}'; }

echo "== clean start"
$BASE down -v
$BASE up -d --build
wait_pg

echo "== default: SQLite system database on the named volume"
state_settings
in_meltano "ls -la .meltano"
mel run tap-postgres target-postgres
mel state get $ID
echo "== docker compose down -v, then up"
$BASE down -v
$BASE up -d
wait_pg
in_meltano "ls -la .meltano"
mel state list
mel run tap-postgres target-postgres

echo "== Postgres system database: before the database exists"
cat compose.pgstate.yaml
$PG up -d
C=$PG
wait_pg
rc=0
mel state list || rc=$?
echo "exit code: $rc"

echo "== Postgres system database: create it, first run"
dst_db warehouse "create database meltano_state"
state_settings
mel state list
mel run tap-postgres target-postgres
mel state get $ID
dst_db meltano_state "select table_schema, table_name from information_schema.tables where table_schema not in ('pg_catalog','information_schema') order by 1,2"
dst_db meltano_state "select state_id, completed_state from state"

echo "== Postgres system database: new, empty .meltano volume, dst untouched"
cat compose.freshvol.yaml
$PGFRESH up -d meltano
C=$PGFRESH
mounts mlab4-meltano-1
in_meltano "ls -la .meltano"
mel state list
mel run tap-postgres target-postgres

echo "== Postgres system database: full docker compose down -v"
mounts mlab4-dst-1
$PGFRESH down -v
$BASE down -v
$PG up -d
C=$PG
wait_pg
mounts mlab4-dst-1
dst_db warehouse "select datname from pg_database order by 1"
dst_db warehouse "create database meltano_state"
mel state list
mel run tap-postgres target-postgres

echo "== file state backend on a host directory"
$PG down -v
cat compose.filestate.yaml
$FILE up -d
C=$FILE
wait_pg
state_settings
mounts mlab4-meltano-1
mel run tap-postgres target-postgres
mel state get $ID
in_meltano "find /state-store -type f; cat /state-store/*/state.json; echo"
in_meltano "ls -la /state-store"
echo "\$ ls -la state-store   (host, Git Bash)"
ls -la state-store
echo "\$ cmd /c dir /b state-store   (host, cmd.exe)"
cmd //c "dir /b state-store"
echo "\$ powershell Get-ChildItem state-store   (host, name and UTF-16 code units)"
powershell.exe -NoProfile -Command "Get-ChildItem -LiteralPath '$(cygpath -w "$PWD/state-store")' | ForEach-Object { \$_.Name + '   ' + ((\$_.Name.ToCharArray() | ForEach-Object { 'U+{0:X4}' -f [int]\$_ }) -join ' ') }"
echo "== docker compose down -v, then up"
$FILE down -v
$FILE up -d
wait_pg
in_meltano "ls -la .meltano; find /state-store -type f"
mel run tap-postgres target-postgres
$FILE down -v
echo "== state backends done"

keys-stream-variants.sh, the replication-key and emit_activate_version_messages variants of the keys extractor:

#!/usr/bin/env bash
# Usage: bash keys-stream-variants.sh 2>&1 | tee keys-stream-variants.log
set -eu
C="docker compose -p mlab4"
src() { echo "shop> $1"; $C exec -T src psql -U postgres -d shop -At -c "$1"; }
dst() { echo "warehouse> $1"; $C exec -T dst psql -U postgres -d warehouse -At -c "$1"; }
mel() { echo "\$ meltano $*"; $C exec -T meltano meltano "$@"; }
in_meltano() { echo "meltano-container\$ $1"; $C exec -T meltano sh -c "$1"; }

set_keys_variant() {
  echo "== tap-postgres-keys variant: replication-key $1, emit_activate_version_messages $2"
  $C exec -T meltano python - "$1" "$2" <<'PY'
import sys
import yaml

key_mode, activate_version = sys.argv[1], sys.argv[2]
path = "meltano.yml"
project = yaml.safe_load(open(path))
for plugin in project["plugins"]["extractors"]:
    if plugin["name"] == "tap-postgres-keys":
        stream = plugin["metadata"]["public-customers"]
        if key_mode == "absent":
            stream.pop("replication-key", None)
        elif key_mode == "null":
            stream["replication-key"] = None
        else:
            stream["replication-key"] = ""
        if activate_version == "on":
            plugin["config"]["emit_activate_version_messages"] = True
        else:
            plugin["config"].pop("emit_activate_version_messages", None)
        print(yaml.safe_dump(plugin, sort_keys=False))
yaml.safe_dump(project, open(path, "w"), sort_keys=False)
PY
}
invoke_keys() {
  rc=0
  in_meltano "meltano invoke tap-postgres-keys > /tmp/keys.jsonl" || rc=$?
  echo "exit code: $rc"
  in_meltano "grep -c '\"RECORD\"' /tmp/keys.jsonl; grep -c 'ACTIVATE_VERSION' /tmp/keys.jsonl; true"
}

echo "== clean start"
$C down -v
$C up -d --build
until $C exec -T src pg_isready -h 127.0.0.1 -U postgres -q \
   && $C exec -T dst pg_isready -h 127.0.0.1 -U postgres -q; do sleep 2; done

set_keys_variant absent on
invoke_keys
set_keys_variant null on
invoke_keys
set_keys_variant empty on
invoke_keys

set_keys_variant empty off
mel run tap-postgres target-postgres
mel run tap-postgres-keys target-postgres-keys
src "delete from customers where id in (900,901,902)"
mel run tap-postgres-keys target-postgres-keys
dst "select count(*), count(_sdc_table_version) from raw.customer_keys"
dst "select w.id from raw.customers w left join raw.customer_keys k using (id) where k.id is null order by w.id"

set_keys_variant empty on
mel run tap-postgres-keys target-postgres-keys
dst "select count(*), count(_sdc_table_version) from raw.customer_keys"
dst "select w.id from raw.customers w left join raw.customer_keys k using (id) where k.id is null order by w.id"

$C down -v

source-reads.sh, the SDK and plugin source lines cited above and the wal2json check, run with the drill.sh stack up:

#!/usr/bin/env bash
# Usage (stack from drill.sh running): bash source-reads.sh 2>&1 | tee source-reads.log
set -eu
C="docker compose -p mlab4"
TAP=.meltano/extractors/tap-postgres/venv/lib/python3.12/site-packages
TGT=.meltano/loaders/target-postgres/venv/lib/python3.12/site-packages
in_meltano() { echo "meltano-container\$ $1"; $C exec -T meltano sh -c "$1"; }
lines() { in_meltano "awk 'NR>=$2 && NR<=$3 {printf \"%d: %s\n\", NR, \$0}' $1"; }

echo "== package versions in the plugin venvs"
in_meltano "$TAP/../../../bin/python -c 'import importlib.metadata as m; print(m.version(\"singer-sdk\"), m.version(\"meltanolabs-tap-postgres\"))'"
in_meltano "$TGT/../../../bin/python -c 'import importlib.metadata as m; print(m.version(\"singer-sdk\"), m.version(\"meltanolabs-target-postgres\"))'"

echo "== replication filter: inclusive bookmark"
lines $TAP/singer_sdk/sql/stream.py 195 208
lines $TAP/tap_postgres/client.py 218 231

echo "== A: where a SQL stream gets its key properties"
lines $TAP/singer_sdk/streams/core.py 1394 1408
lines $TAP/singer_sdk/sql/stream.py 60 67
lines $TAP/singer_sdk/sql/stream.py 114 130
lines $TAP/singer_sdk/mapper.py 793 802
lines $TAP/singer_sdk/mapper.py 535 543
lines $TAP/singer_sdk/streams/core.py 879 890

echo "== F: ACTIVATE_VERSION on FULL_TABLE streams"
lines $TAP/singer_sdk/streams/core.py 719 722
lines $TAP/singer_sdk/streams/core.py 1339 1345
lines $TAP/singer_sdk/streams/core.py 709 717

echo "== I: load_method in singer_sdk (target venv)"
in_meltano "grep -rn --include=*.py 'load_method\|TargetLoadMethods' $TGT/singer_sdk/"
lines $TGT/singer_sdk/helpers/capabilities.py 303 334
in_meltano "grep -rn --include=*.py 'append_only' $TGT/singer_sdk/; echo \"grep exit: \$?\""
echo "== I: target_postgres overrides"
in_meltano "grep -rn --include=*.py 'load_method' $TGT/target_postgres/; echo \"grep exit: \$?\""
in_meltano "grep -n 'def prepare_table\|def setup\|def upsert\|def process_batch\|append_only' $TGT/target_postgres/sinks.py $TGT/target_postgres/connector.py"
lines $TGT/target_postgres/sinks.py 49 56
lines $TGT/target_postgres/sinks.py 285 316
lines $TGT/target_postgres/connector.py 369 404
lines $TGT/singer_sdk/sql/connector.py 1410 1427

echo "== E: LOG_BASED in tap-postgres 0.10.0"
in_meltano "grep -n -i 'log.based\|wal2json\|wal_level\|max_replication_slots\|pg_create_logical_replication_slot\|REPLICATION\b\|_sdc_lsn\|LOGIN' $TAP/meltanolabs_tap_postgres-0.10.0.dist-info/METADATA"
lines $TAP/meltanolabs_tap_postgres-0.10.0.dist-info/METADATA 309 380
lines $TAP/tap_postgres/client.py 360 380
lines $TAP/tap_postgres/client.py 649 660
lines $TAP/tap_postgres/tap.py 756 780
echo "\$ docker compose exec src: wal2json in the stock postgres:17.11 image?"
$C exec -T src sh -c "ls /usr/lib/postgresql/17/lib/ | grep -i -E 'wal2json|pgoutput'; echo \"ls|grep exit: \$?\"; apt list --installed 2>/dev/null | grep -i wal2json; echo \"apt grep exit: \$?\""
echo "\$ git ls-remote --tags https://github.com/MeltanoLabs/tap-postgres v0.10.0"
git ls-remote --tags https://github.com/MeltanoLabs/tap-postgres v0.10.0

echo "== G: Meltano extras"
in_meltano "python -c 'import importlib.metadata as m; [print(r) for r in m.requires(\"meltano\") if \"postgres\" in r]'"
in_meltano "pip list 2>/dev/null | grep -i -E '^(meltano|psycopg|psycopg-binary|sqlalchemy) '"
in_meltano "$TAP/../../../bin/python -m pip list 2>/dev/null | grep -i -E 'psycopg|sqlalchemy' || $TAP/../../../bin/python -c 'import importlib.metadata as m; [print(d.metadata[\"Name\"], d.version) for d in m.distributions() if \"psycopg\" in d.metadata[\"Name\"].lower()]'"
in_meltano "find $TAP $TGT -maxdepth 1 -iname '*psycopg*'"
in_meltano "which git gcc pg_config; echo \"which exit: \$?\""

extras.sh, the alert wrappers, --test=schema, load_method=upsert, state IDs and key-properties:

#!/usr/bin/env bash
# Usage: bash extras.sh 2>&1 | tee extras.log
set -eu
C="docker compose -p mlab5"
ID=dev:tap-postgres-to-target-postgres

dst() { echo "warehouse> $1"; $C exec -T dst psql -U postgres -d warehouse -At -c "$1"; }
mel() { echo "\$ meltano $*"; $C exec -T meltano meltano "$@"; }
in_meltano() { echo "meltano-container\$ $1"; $C exec -T meltano sh -c "$1"; }
wait_pg() { until $C exec -T src pg_isready -h 127.0.0.1 -U postgres -q \
              && $C exec -T dst pg_isready -h 127.0.0.1 -U postgres -q; do sleep 2; done; }
events_counts() { dst "select count(*), count(distinct id), count(*) - count(distinct id) as duplicate_rows from raw.events"; }
wrapper() {
  local name=$1 script=$2
  echo "meltano-container\$ bash -c '<$name wrapper>'"
  local rc=0
  $C exec -T meltano bash -c "$script" || rc=$?
  echo "$name wrapper exit code: $rc"
}

OLD=$(cat <<'EOF'
set -o pipefail; meltano run tap-postgres target-postgres 2>&1 | tee run.log; if grep -q 'No state was found, complete import.' run.log; then echo "FULL RELOAD" >&2; exit 2; fi
EOF
)
FIXED=$(cat <<'EOF'
set -o pipefail; meltano run tap-postgres target-postgres 2>&1 | tee run.log; rc=$?; if grep -q 'No state was found, complete import.' run.log; then echo "FULL RELOAD" >&2; exit 2; fi; exit "$rc"
EOF
)

echo "== clean start"
$C down -v
$C up -d --build
wait_pg
mel --version

echo "== 1. alert wrappers"
echo "OLD:   $OLD"
echo "FIXED: $FIXED"
echo "== 1a. brand-new pipeline, first run, FIXED wrapper"
mel state list
wrapper FIXED "$FIXED"
echo "== 1b. state present, run succeeds"
wrapper OLD "$OLD"
wrapper FIXED "$FIXED"
echo "== 1c. state absent, run succeeds"
mel state clear $ID --force
wrapper OLD "$OLD"
mel state clear $ID --force
wrapper FIXED "$FIXED"
echo "== 1d. destination Postgres stopped, state present"
mel state list
$C stop dst
wrapper OLD "$OLD"
wrapper FIXED "$FIXED"
$C start dst
wait_pg
mel state get $ID

echo "== 2. --test=schema"
in_meltano "meltano invoke tap-postgres-events-mapkey --test=schema; echo \"exit code: \$?\""
in_meltano "meltano invoke tap-postgres-events --test=schema; echo \"exit code: \$?\""
echo "== 2b. SCHEMA message from a normal invoke (records discarded)"
in_meltano "meltano invoke tap-postgres-events-mapkey 2>/dev/null | grep '\"type\":\"SCHEMA\"'"
in_meltano "meltano invoke tap-postgres-events 2>/dev/null | grep '\"type\":\"SCHEMA\"'"

echo "== 3. keyless events stream, TARGET_POSTGRES_LOAD_METHOD=upsert"
in_meltano "TARGET_POSTGRES_LOAD_METHOD=upsert meltano config list target-postgres 2>/dev/null | grep -A1 load_method"
in_meltano "TARGET_POSTGRES_LOAD_METHOD=upsert meltano config print target-postgres 2>/dev/null | grep load_method"
rc=0
mel state clear dev:tap-postgres-events-to-target-postgres --force || rc=$?
echo "exit code: $rc"
echo "\$ TARGET_POSTGRES_LOAD_METHOD=upsert meltano run tap-postgres-events target-postgres"
$C exec -T -e TARGET_POSTGRES_LOAD_METHOD=upsert meltano meltano run tap-postgres-events target-postgres
events_counts
echo "\$ TARGET_POSTGRES_LOAD_METHOD=upsert meltano run tap-postgres-events target-postgres"
$C exec -T -e TARGET_POSTGRES_LOAD_METHOD=upsert meltano meltano run tap-postgres-events target-postgres
events_counts
mel state clear dev:tap-postgres-events-to-target-postgres --force
echo "\$ TARGET_POSTGRES_LOAD_METHOD=upsert meltano run tap-postgres-events target-postgres"
$C exec -T -e TARGET_POSTGRES_LOAD_METHOD=upsert meltano meltano run tap-postgres-events target-postgres
events_counts
$C exec -T dst psql -U postgres -d warehouse -c '\d raw.events'

echo "== 4. state ID after a rename and in another environment"
mel state list
mel add --plugin-type extractor tap-shop --inherit-from tap-postgres
in_meltano "grep -n -A1 'name: tap-shop' meltano.yml"
mel run tap-shop target-postgres
mel state list
mel --environment=prod run tap-postgres target-postgres
mel state list
mel --environment=prod state list

echo "== 5. metadata key-properties on the keyless events stream"
in_meltano "grep -n -B6 'key-properties: \[id\]' meltano.yml"
dst "drop table raw.events"
mel run tap-postgres-events-keyprops target-postgres
events_counts
mel run tap-postgres-events-keyprops target-postgres
events_counts
mel state clear dev:tap-postgres-events-keyprops-to-target-postgres --force
mel run tap-postgres-events-keyprops target-postgres
events_counts
in_meltano "meltano invoke tap-postgres-events-keyprops --test=schema; echo \"exit code: \$?\""
in_meltano "meltano invoke tap-postgres-events-keyprops 2>/dev/null | grep '\"type\":\"SCHEMA\"'"
in_meltano "python -c 'import json; c=json.load(open(\".meltano/run/tap-postgres-events-keyprops/tap.properties.json\")); s=[x for x in c[\"streams\"] if x[\"tap_stream_id\"]==\"public-events\"][0]; print(\"key_properties\", s.get(\"key_properties\")); print([m for m in s[\"metadata\"] if m[\"breadcrumb\"]==[]])'"
in_meltano "grep -rn --include=*.py 'key-properties\|key_properties' /usr/local/lib/python3.12/site-packages/meltano/core/; echo \"grep exit: \$?\""
$C exec -T dst psql -U postgres -d warehouse -c '\d raw.events'

$C down -v
echo "== extras done"

Get in touch

Meltano documentation, “Replicate Data” (retrieved October 13, 2026). https://docs.meltano.com/guide/integration ↩︎¹ ↩︎² ↩︎³ ↩︎⁴
Meltano documentation, “Command Line”: meltano run --full-refresh, meltano state (retrieved October 13, 2026). https://docs.meltano.com/reference/command-line-interface ↩︎¹ ↩︎²
Meltano documentation, “Settings,” database_uri (retrieved October 13, 2026). https://docs.meltano.com/reference/settings ↩︎
PostgreSQL Global Development Group, “psql,” PostgreSQL 17 Documentation, option -c. https://www.postgresql.org/docs/17/app-psql.html ↩︎
PostgreSQL Global Development Group, “Date/Time Functions and Operators,” section “Current Date/Time,” PostgreSQL 17 Documentation. https://www.postgresql.org/docs/17/functions-datetime.html ↩︎
Meltano SDK documentation, “At Least Once Delivery Promise” (retrieved October 13, 2026). https://sdk.meltano.com/en/latest/implementation/at_least_once.html ↩︎
MeltanoLabs, tap-postgres README at tag v0.10.0, log-based replication setup. https://github.com/MeltanoLabs/tap-postgres/blob/v0.10.0/README.md ↩︎

Mastering Meltano: Meltano Incremental Replication Is One State Record Away From a Full Reload

How the bookmark drives each run

Where state lives, and how to keep it

Why a no-change run re-sends rows

Streams without key properties duplicate rows

What incremental replication misses

Troubleshooting Meltano state

Reproduce it

Related insights

Delete the Database, Then Time the Restore: A pgBackRest Recovery Drill

Complete CI/CD Pipeline with Docker, GitHub Actions, and Hetzner Deployment

Building a Production Data Pipeline on PPDM with Airflow and DuckDB