Skip to content

perf: Logical replication perf experiments#667

Open
bdewilde wants to merge 7 commits into
MeltanoLabs:mainfrom
bdewilde:burton-hack-log-based-perf
Open

perf: Logical replication perf experiments#667
bdewilde wants to merge 7 commits into
MeltanoLabs:mainfrom
bdewilde:burton-hack-log-based-perf

Conversation

@bdewilde

Copy link
Copy Markdown
Contributor

changes

  • adds msgspec as package dependency and sets MsgSpecWriter as tap's message writer class
  • minimizes variable/function access in loop over messages
  • consolidates row update operations when consuming a message
  • reduces replication status interval from 5s to 1s (afacit this matches what the pipelinewise variant uses; I figured I'd give it a shot)

context

Diving deeper into the code wrt issue #587

These changes may or may not be of interest for merging; they're mostly just a demonstration of what I found.

details

All together, these changes reduced runtimes for my example E+L pipeline from ~112s to ~107s. Not very impressive, I know. Here's the py-spy flame graph based on this repo's main branch:

image

And here's the (very similar) equivalent based on the latest version of my branch:

image

My changes only touch a small part of the tap's call stack:

Screenshot 2025-08-18 at 11 43 39 AM

Unfortunately, I wasn't able to do anything about the slowest parts of the tap:

  • checking if the db cursor has has had any messages within the timeout before breaking out of the loop, here
  • writing messages to stdout via the meltano sdk, here

For what it's worth, here is the corresponding functionality in the pipelinewise variant's code. There's a lot of additional logic there, but I had a hard time understanding the key differences with this variant.

questions

  • Is there any way to make the check for whether or not to break out of the messages loop faster?
  • any idea why writing messages is so slow, even using msgspec?

@edgarrmondragon edgarrmondragon changed the title logical replication perf experiments perf: Logical replication perf experiments Aug 18, 2025
@edgarrmondragon

Copy link
Copy Markdown
Member

Hi @bdewilde, thanks for taking the time to post this!

Is there any way to make the check for whether or not to break out of the messages loop faster?

I'm not sure what this refers to. Is there a specific location in the code where you see this happening?

any idea why writing messages is so slow, even using msgspec?

I'm reading through the performance docs1 and I don't see anything obvious we're missing other than using msgspec.Struct for messages.

The serialization itself with msgspec should be considerably faster even without using Structs, so I think we're doing something wrong elsewhere. I know try/except blocks inside loops can be expensive, and get_records also seems to be taking longer that it probably should, so going one lever down from SQLAlchemy to the native postgres client is another thing to explore there.

Footnotes

  1. https://jcristharif.com/msgspec/perf-tips.html#use-structs

@bdewilde

Copy link
Copy Markdown
Contributor Author

Is there any way to make the check for whether or not to break out of the messages loop faster?

I'm not sure what this refers to. Is there a specific location in the code where you see this happening?

Sorry about the confusion! This was in reference to an earlier comment in the PR about apparent perf bottlenecks:

checking if the db cursor has has had any messages within the timeout before breaking out of the loop, here

any idea why writing messages is so slow, even using msgspec?

I'm reading through the performance docs1 and I don't see anything obvious we're missing other than using msgspec.Struct for messages.

The serialization itself with msgspec should be considerably faster even without using Structs, so I think we're doing something wrong elsewhere. I know try/except blocks inside loops can be expensive, and get_records also seems to be taking longer that it probably should, so going one lever down from SQLAlchemy to the native postgres client is another thing to explore there.

I'll try to take a peek at the sdk's usage of msgspec, thanks for the link to perf tips 👍

bdewilde added a commit to bdewilde/tap-postgres that referenced this pull request Jun 26, 2026
…s#772)

### problem

`PostgresLogBasedStream.get_records()` opens its own
`LogicalReplicationConnection` per selected stream. With N LOG_BASED
streams the tap runs N sequential WAL scans -- each rereads the same
segments, with add-tables discarding most records server-side.
End-to-end sync time scales ~linearly in N. For pipelines with multiple
LOG_BASED streams against a large backlog, this dominates run-time.

### changes

A new `SingleConnectionWALReader` opens one logical replication
connection with `add-tables` covering all selected LOG_BASED tables,
scans the WAL once, and dispatches each parsed wal2json message inline
to the owning stream's new `emit_record()` method for immediate Singer
RECORD emission. STATE flushes every 30s, and the slot is advanced to
the WAL tip on idle/max-run exit.

- new modules: `_wal_helpers.py` (FQN/escaping/parsing helpers) and
`wal_reader.py` (the reader and read loop).
- `client.py` gains `emit_record` method and a config-flag branch in
`get_records`
- `tap.py` adds the `log_based_single_connection` config setting
(default False!) and `_sync_log_based_streams_shared` orchestration
- new tests: `tests/test_wal_helpers.py`, `tests/test_wal_reader.py`,
`tests/test_consume.py`

Full disclosure, I had Claude Code implement those three test modules,
and then I iterated a bit. If it's still excessive / not testing
usefully -- something Claude is known to do, sigh -- just let me know,
and I will take a hatchet to it.

This is a very belated follow-up to PR MeltanoLabs#667 and Issue MeltanoLabs#587.

### constraints

- `Tap.sync_all` is `@typing.final`, so dispatch can't be restructured
at the SDK boundary -- this was a bummer. My next best option was
trigger at the first LOG_BASED stream's `get_records()` call, gated by a
`_shared_wal_run_completed` flag on the tap so siblings become no-ops.
- SCHEMA-before-RECORD across streams: `_sync_log_based_streams_shared`
pre-writes every stream's schema before the reader runs. Since the SDK's
`Stream.sync()` later calls `_write_schema_message()` again, the
override on `PostgresLogBasedStream` is idempotent. (Without that flag
every SCHEMA would be emitted twice, which is not great.)
- Per-stream LSN filter: Replication opens at `min(start_lsn)` across
all streams, so each stream's own bookmark is captured at construction
and used to drop messages that it's already past.
- I had to dip into private SDK calls -- `_write_record_message()` and
`_increment_stream_state()` -- for this to work. I consolidated them in
one place -- `emit_record()` -- so SDK renames hit one method. Not sure
how stable the API is here...

### questions

1. `_write_schema_message` idempotency: I made the smallest fix I could
for the duplicate-SCHEMA bug given the `@final` constraint on
`sync_all`. Is there another / cleaner approach?
2. `emit_record()` uses internal SDK calls. Is this going to be an
issue? Is there a safer / "public" equivalent?
3. `get_records()` as trigger: The "first stream's get_records fires the
shared reader" pattern is a not-great workaround for `sync_all` being
final. Is it okay as documented, or is there a cleaner SDK hook?
4. `replication_max_run_seconds` / `replication_idle_exit_seconds` now
bound the whole LOG_BASED batch instead of each stream. To me that feels
like an improvement, but I don't know the whole system / downstream use
caess. For example, does anything downstream assume per-stream bounds?

---------

Signed-off-by: Edgar Ramírez Mondragón <edgarrm358@gmail.com>
Co-authored-by: Edgar Ramírez Mondragón <16805946+edgarrmondragon@users.noreply.github.com>
Co-authored-by: Edgar Ramírez Mondragón <edgarrm358@gmail.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants