Implement snapshot planning and row streaming for logical replication

[isdaniel]

Implement snapshot planning and row streaming for logical replication

• Introduced plan.rs to handle the planning logic for initial snapshots, including the construction of COPY statements and catalog queries for published tables.
• Added SnapshotTable and CatalogRow structs to represent tables and their metadata.
• Implemented functions to build SQL queries for catalog retrieval and COPY statements based on publication configurations.
• Created rows.rs to manage the streaming of snapshot rows, including handling the replication stream and decoding row data.
• Developed the SnapshotRows struct to encapsulate the snapshot process, providing methods for row retrieval and stream management.
• Added comprehensive unit tests to ensure the correctness of the snapshot planning and row streaming functionalities.

[codspeed]

Merging this PR will not alter performance

✅ 50 untouched benchmarks

---

_{Comparing feature/initial-snapshot (97c12d0) with main (58092f8)}

[codecov]

Codecov Report

❌ Patch coverage is 94.30212% with 129 lines in your changes missing coverage. Please review.
✅ Project coverage is 94.98%. Comparing base (58092f8) to head (97c12d0).

Files with missing lines	Patch %	Lines
src/snapshot/mod.rs	85.42%	58 Missing ⚠️
src/connection/native/connection.rs	91.05%	22 Missing ⚠️
src/snapshot/rows.rs	93.37%	22 Missing ⚠️
src/stream.rs	81.70%	15 Missing ⚠️
src/snapshot/plan.rs	97.84%	7 Missing ⚠️
src/connection/native/copy_out.rs	98.39%	5 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #116      +/-   ##
==========================================
- Coverage   95.03%   94.98%   -0.05%     
==========================================
  Files          27       33       +6     
  Lines       22295    24557    +2262     
==========================================
+ Hits        21187    23325    +2138     
- Misses       1108     1232     +124     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[isdaniel]

@claude[agent] review

[Claude]

@claude[agent] review

Reviewed. Gate is clean locally: cargo test --lib --features derive (1466 passed), workspace+derive clippy and cargo fmt all pass. No code changes pushed — findings below, the first one is the only substantive one.

1. Generated columns (src/snapshot/plan.rs:94-121). The pre-PG15 branch uses true as the attname predicate, so pg_attribute contributes every non-dropped column including GENERATED ALWAYS AS ... STORED. README:549 says PG14 is supported, so that path is live. pgoutput before PG18 never sends generated columns (no publish_generated_columns), so the snapshot's RelationInfo would carry a column the live stream's Relation message does not — the COPY arity still matches the snapshot's own relation, so nothing errors; the two phases just emit different row shapes for the same table. I believe PG15–17 pg_publication_tables.attnames also lists generated columns (they're filtered in pgoutput, not in the view), which would extend the same mismatch to the default path. A blanket AND a.attgenerated = '' would be wrong on PG18 with publish_generated_columns = true, so this needs a version-conditional predicate and an integration test to pin which side each version falls on — I'd rather flag it than guess.

2. unique_row_filter (plan.rs:251) rejects the case where one publication filters a table and another publishes it unfiltered. PostgreSQL's rule there is that the unfiltered publication wins and every row is published, so this is a case that could be resolved (None beats Some) rather than erroring. Erroring is the safe direction, so this is a judgement call, not a bug — but the message points at retain_tables, which drops the table entirely and is a worse outcome than the resolvable answer.

3. split_publication_names (plan.rs:130) splits on , before unquoting, so a publication name containing a comma splits into two names; trim_matches('"') also strips repeated quotes and doesn't undo "" escaping. Both are unreachable for identifiers that survived CREATE PUBLICATION without quoting, so this is cosmetic unless you want to accept exotic names.

4. Minor: SnapshotRows::abandon, Snapshot::abandon and SnapshotOutcome::skip are async fn ... -> Result<_> with no await and no failing branch. run also sets self.poisoned = true at rows.rs:229 on the handler-error path, which is a dead store since self is dropped on the next line.

The parts I checked hardest and found correct: the COPY always naming its columns (the arity guarantee), relkind = 'p' forcing the subselect form, the ORDER BY ... a.attnum / COPY-order agreement, quote_ident/quote_literal on every interpolated name, the poisoning latch matching run's consume-on-Err, and discard_slot firing on every failure path but not on abandon or a successful finish.

Warning

Firewall rules blocked me from connecting to one or more addresses (expand for details)

I tried to connect to the following addresses, but was blocked by firewall rules:

• 192.0.2.1
  • Triggering command: /home/REDACTED/work/pg-walstream/pg-walstream/target/debug/deps/pg_walstream-bc67eca55add9dc6 (packet block)

If you need me to access, download, or install something from one of these locations, you can either:

• Configure Actions setup steps to set up my environment, which run before the firewall is enabled
• Add the appropriate URLs or hosts to the custom allowlist in this repository's Copilot cloud agent settings (admins only)