perf: Logical replication perf experiments#667
Conversation
|
Hi @bdewilde, thanks for taking the time to post this!
I'm not sure what this refers to. Is there a specific location in the code where you see this happening?
I'm reading through the performance docs1 and I don't see anything obvious we're missing other than using 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 Footnotes |
Sorry about the confusion! This was in reference to an earlier comment in the PR about apparent perf bottlenecks:
I'll try to take a peek at the sdk's usage of msgspec, thanks for the link to perf tips 👍 |
…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>
changes
msgspecas package dependency and setsMsgSpecWriteras tap's message writer classcontext
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-spyflame graph based on this repo's main branch:And here's the (very similar) equivalent based on the latest version of my branch:
My changes only touch a small part of the tap's call stack:
Unfortunately, I wasn't able to do anything about the slowest parts of the tap:
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
msgspec?