Filtering only applies to tables, which can lead to stuck replication or exiting with failure

[ldanz]

Describe the bug

When using source.postgres.replication.plugin.add_tables to configure pgstream to only replicate some schemas and not others, only tables are affected, not other postgres objects like types, indices, functions, etc. This is arguably expected behavior based on the documentation (although it's not what I actually want, since my goal is to replicate some schemas and completely ignore others).

However, the behavior as-written leads to failures. If the would-be-ignored schema does not exist downstream, then pgstream repeatedly retries the operation, which blocks future changes. If the schema exists, but the object that's to be created depends on a table (which was ignored by the filtering), then we get a failure that (in strict mode) causes pgstream to exit.

To Reproduce

Here is my config file:

---
source:
  postgres:
    url: "postgres://postgres:postgres@localhost:5432?sslmode=disable"
    mode: snapshot_and_replication # options are replication, snapshot or snapshot_and_replication
    snapshot: # when mode is snapshot or snapshot_and_replication
      tables: ["schema1.*", "schema2.*"]
      mode: full # options are data_and, schema or data
      recorder:
        postgres_url: "postgres://postgres:postgres@localhost:5432?sslmode=disable" # URL of the database where the snapshot status is recorded
      schema: # when mode is full or schema
        pgdump_pgrestore:
          clean_target_db: false # whether to clean the target database before restoring
    replication:
      plugin:
        add_tables: "schema1.*,schema2.*"

target:
  postgres:
    url: "postgres://postgres:postgres@localhost:7654?sslmode=disable"
    strict_mode: true

modifiers:
  filter:
    include_tables: ["schema1.*", "schema2.*"]

The source and target databases are from the docker-compose setup provided by build/docker/docker-compose.yml.

First, I created three schemas upstream -- the two that I want to replicate, plus an additional one:

create schema schema1; create schema schema2; create schema schema3;

Then, I start pgstream:

pgstream run -c ./pgstream_config.yml  --init --log-level trace

From here, I can test a few different cases. (I clean and recreate the test environment in between cases by running docker-compose -f build/docker/docker-compose.yml --profile pg2pg down && docker volume rm pgstream_data-pg18 && docker-compose -f build/docker/docker-compose.yml --profile pg2pg up.)

Case 1: Creating an index on a table

Upstream SQL:

CREATE TABLE schema3.mytable (id int PRIMARY KEY, name text);
CREATE INDEX idx_mytable_name ON schema3.mytable (name);

The "CREATE TABLE" statement is ignored, as expected. However, the "CREATE INDEX" statement is not ignored, and since it depends on the schema and the table existing, it can't be applied.

Resulting pgstream logs:

2026-09-03T13:41:31.534302-07:00 TRC logger.go:32 > module=wal_postgres_listener server_time=2026-09-03T13:41:31.532847-07:00 wal_data="{\"action\":\"M\",\"timestamp\":\"2026-09-03 20:41:31.531307+00\",\"lsn\":\"0/1836F50\",\"transactional\":true,\"prefix\":\"pgstream.ddl\",\"content\":\"{\\\"ddl\\\": \\\"CREATE TABLE schema3.mytable (id int PRIMARY KEY, name text);\\\", \\\"objects\\\": [{\\\"oid\\\": \\\"16440\\\", \\\"type\\\": \\\"index\\\", \\\"schema\\\": \\\"schema3\\\", \\\"identity\\\": \\\"schema3.mytable_pkey\\\"}, {\\\"oid\\\": \\\"16434\\\", \\\"type\\\": \\\"table\\\", \\\"schema\\\": \\\"schema3\\\", \\\"columns\\\": [{\\\"name\\\": \\\"id\\\", \\\"type\\\": \\\"integer\\\", \\\"attnum\\\": 1, \\\"unique\\\": true, \\\"default\\\": null, \\\"identity\\\": null, \\\"nullable\\\": false, \\\"generated\\\": false}, {\\\"name\\\": \\\"name\\\", \\\"type\\\": \\\"text\\\", \\\"attnum\\\": 2, \\\"unique\\\": false, \\\"default\\\": null, \\\"identity\\\": null, \\\"nullable\\\": true, \\\"generated\\\": false}], \\\"identity\\\": \\\"schema3.mytable\\\", \\\"primary_key_columns\\\": [\\\"id\\\"]}], \\\"command_tag\\\": \\\"CREATE TABLE\\\", \\\"schema_name\\\": \\\"schema3\\\"}\"}" wal_end=0/18372B2
2026-09-03T13:41:31.535046-07:00 TRC logger.go:32 > skipping DDL event module=wal_filter schema=schema3 table=mytable
...
2026-09-03T13:41:34.863193-07:00 TRC logger.go:32 > module=wal_postgres_listener server_time=2026-09-03T13:41:34.861487-07:00 wal_data="{\"action\":\"M\",\"timestamp\":\"2026-09-03 20:41:34.859823+00\",\"lsn\":\"0/1837F18\",\"transactional\":true,\"prefix\":\"pgstream.ddl\",\"content\":\"{\\\"ddl\\\": \\\"CREATE INDEX idx_mytable_name ON schema3.mytable (name);\\\", \\\"objects\\\": [{\\\"oid\\\": \\\"16442\\\", \\\"type\\\": \\\"index\\\", \\\"schema\\\": \\\"schema3\\\", \\\"identity\\\": \\\"schema3.idx_mytable_name\\\"}], \\\"command_tag\\\": \\\"CREATE INDEX\\\", \\\"schema_name\\\": \\\"schema3\\\"}\"}" wal_end=0/18380A5
2026-09-03T13:41:34.883293-07:00 TRC logger.go:32 > module=wal_postgres_listener server_time=2026-09-03T13:41:34.861589-07:00 wal_data= wal_end=0/1838028
2026-09-03T13:41:37.154866-07:00 DBG logger.go:39 > sending batch batch_size=1 module=postgres_batch_writer
2026-09-03T13:41:37.17732-07:00 WRN logger.go:47 > retrying Postgres operation after error error.message="ERROR: schema \"schema3\" does not exist (SQLSTATE 3F000)" module=postgres_batch_writer retry_delay=387.991721ms
2026-09-03T13:41:37.380162-07:00 TRC logger.go:32 > module=wal_postgres_listener server_time=2026-09-03T13:41:37.378285-07:00 wal_data= wal_end=0/1838028
2026-09-03T13:41:37.380425-07:00 TRC logger.go:32 > module=wal_postgres_listener server_time=2026-09-03T13:41:37.37846-07:00 wal_data= wal_end=0/1838028
2026-09-03T13:41:37.380585-07:00 DBG logger.go:39 > max batch size reached or keep alive received, draining batch batch_size=0 max_batch_size=20000 module=postgres_batch_writer
2026-09-03T13:41:37.586861-07:00 WRN logger.go:47 > retrying Postgres operation after error error.message="ERROR: schema \"schema3\" does not exist (SQLSTATE 3F000)" module=postgres_batch_writer retry_delay=450.453002ms
2026-09-03T13:41:38.059292-07:00 WRN logger.go:47 > retrying Postgres operation after error error.message="ERROR: schema \"schema3\" does not exist (SQLSTATE 3F000)" module=postgres_batch_writer retry_delay=1.189802405s
...

This keeps retrying and blocks future transactions from getting replicated.

If I try to kill pgstream at this point with SIGINT (ctrl-C), it does not die. The only way I found to kill it was with SIGKILL (kill -9).

If instead of killing pgstream, I create schema3 downstream, then the error changes, and in strict mode, pgstream exits.

2026-09-03T13:42:02.087786-07:00 ERR logger.go:51 > running DDL query error.message="relation does not exist: relation \"schema3.mytable\" does not exist: permanent error, do not retry" module=postgres_batch_writer query_args=null query_sql="CREATE INDEX idx_mytable_name ON schema3.mytable (name);"
2026-09-03T13:42:02.087976-07:00 ERR logger.go:51 > failed to send batch error.message="strict mode: stopping on non-internal DDL failure: relation does not exist: relation \"schema3.mytable\" does not exist: permanent error, do not retry" module=postgres_batch_writer

Case 2: Creating a type

Upstream SQL:

CREATE TYPE schema3.mytype AS ENUM ('foo', 'bar', 'baz');

As before, initially there's an error about the schema not existing, which results in retries:

2026-09-03T13:43:02.306188-07:00 TRC logger.go:32 > module=wal_postgres_listener server_time=2026-09-03T13:43:02.304958-07:00 wal_data="{\"action\":\"M\",\"timestamp\":\"2026-09-03 20:43:02.303842+00\",\"lsn\":\"0/1827500\",\"transactional\":true,\"prefix\":\"pgstream.ddl\",\"content\":\"{\\\"ddl\\\": \\\"CREATE TYPE schema3.mytype AS ENUM ('foo', 'bar', 'baz');\\\", \\\"objects\\\": [{\\\"oid\\\": \\\"16435\\\", \\\"type\\\": \\\"type\\\", \\\"schema\\\": \\\"schema3\\\", \\\"identity\\\": \\\"schema3.mytype\\\"}], \\\"command_tag\\\": \\\"CREATE TYPE\\\", \\\"schema_name\\\": \\\"schema3\\\"}\"}" wal_end=0/1827682
2026-09-03T13:43:02.320518-07:00 TRC logger.go:32 > module=wal_postgres_listener server_time=2026-09-03T13:43:02.304999-07:00 wal_data= wal_end=0/18275E8
2026-09-03T13:43:02.474347-07:00 TRC logger.go:32 > module=wal_postgres_listener server_time=2026-09-03T13:43:02.470133-07:00 wal_data= wal_end=0/18275E8
2026-09-03T13:43:02.474528-07:00 TRC logger.go:32 > module=wal_postgres_listener server_time=2026-09-03T13:43:02.472863-07:00 wal_data= wal_end=0/1827620
2026-09-03T13:43:24.699387-07:00 TRC logger.go:32 > module=wal_postgres_listener server_time=2026-09-03T13:43:24.697767-07:00 wal_data= wal_end=0/1827620
2026-09-03T13:43:24.706951-07:00 DBG logger.go:39 > sending batch batch_size=1 module=postgres_batch_writer
2026-09-03T13:43:24.728158-07:00 WRN logger.go:47 > retrying Postgres operation after error error.message="ERROR: schema \"schema3\" does not exist (SQLSTATE 3F000)" module=postgres_batch_writer retry_delay=553.038342ms

If I create schema3 downstream, this error resolves, and the type gets created.
pgstream log:

2026-09-03T13:43:34.540785-07:00 INF logger.go:43 > retried Postgres operation succeeded module=postgres_batch_writer

Downstream postgres:

postgres=# \dT schema3.*
           List of data types
 Schema  |      Name      | Description 
---------+----------------+-------------
 schema3 | schema3.mytype | 
(1 row)

It may be expected behavior according to pgstream's current feature set that schema3.mytype was created downstream, but that's not the behavior I was trying to achieve with my configuration.

Note that this is not a comprehensive list of cases of non-table objects (for example, creating a function has a similar behavior to creating a type), but it's illustrative of the types of problems that can arise.

Expected behavior

What I would like is to be able to filter by schema. That is, arguably, a feature request, because the only documented filtering is table-based. However, I believe it would cleanly solve the issues that I noted above.

If you want to keep the existing configuration file format without adding new fields, I think it would make sense to say that if all the tables in a schema are filtered out (as is the case for schema3 in my configuration), then no objects from that schema are replicated.

Regardless, in the scenarios that I shared, I would expect that pgstream would continue running and replicating new changes, rather than dying or getting stuck.

Setup:

• pgstream version: v1.4.2
• Postgres version: 18.6
• Postgres environment: self-hosted