📦 CDC Orders

Author: Jefferson Marchetti

Real-time analytics pipeline over e-commerce orders, powered by native Postgres Change Data Capture: no Debezium, no Kafka, no trigger on the source table. cdc speaks the Postgres logical replication protocol directly (pgoutput, hand-decoded), pipeline keeps an in-memory snapshot and runs two aggregations with Polars and Rayon every few seconds, and generator produces synthetic order traffic so the demo has something to capture.

No HTTP or gRPC server: the output is Parquet on disk, ready to be read by any analytical engine (DuckDB, Polars, Spark). See docs/trade-offs.md.

• Features
• Architecture
• Requirements
• Running it
• Verification
• Tests
• Development
• Docker
• Project structure
• Technical decisions
• License
• Contact

• Own parser for the pgoutput protocol (no crate on crates.io does this), zero-copy, no unwrap/panic! on malformed input.
• Full logical replication session: creates the publication and slot if needed, decodes Insert/Update/Delete/Relation, replies to keepalives only when the server actually asks (avoids turning into a busy loop).
• Materialized in-memory snapshot (HashMap), rebuilt from scratch if the process restarts, with no persistence of its own beyond the output Parquet.
• Two Polars aggregations (revenue per minute, top products) running concurrently via rayon::join, over the same snapshot.
• 23 tests: unit tests over the parser and the state, plus integration tests against real Postgres (real WAL, not synthetic bytes).
• Assumptions and decisions documented in docs/trade-offs.md.

flowchart LR
    subgraph PG["PostgreSQL (wal_level=logical)"]
        T["orders"]
    end

    T -->|"pgoutput / XLogData"| S["cdc::ReplicationSession"]
    S -->|"ChangeEvent"| ST["pipeline::OrderState (HashMap)"]
    ST -->|"every flush"| AG["analytics::compute\nrayon::join"]
    AG --> R1["revenue_by_minute"]
    AG --> R2["top_products"]
    ST --> SN["orders_snapshot"]
    R1 --> OUT[("Parquet")]
    R2 --> OUT
    SN --> OUT

    G["generator"] -->|"INSERT/UPDATE/DELETE"| T

generator is an ordinary Postgres client, not part of the replication path. It exists only to give the demo a continuous flow of orders without relying on someone running psql by hand. Details of each stage in docs/design.md.

• Rust 1.97+
• Docker + Docker Compose (for Postgres with wal_level=logical, and optionally to run the whole stack in containers)

Via Docker Compose (simplest, brings everything up):

make docker-up

This builds pipeline and generator, starts a Postgres already configured for logical replication, and starts writing Parquet to ./output every 5 seconds.

Locally, without Docker for the binaries (only for the test Postgres):

make test-pg-up                                    # disposable Postgres on :55432
cargo run -p generator -- --port 55432 &            # generates traffic in the background
cargo run -p pipeline -- --port 55432 --output-dir output

git clone <this-repository> && cd rust-cdc-orders
make check

make check runs cargo fmt --check, cargo clippy -D warnings, and the full test suite (automatically starting a disposable Postgres via make test-pg-up). Expected output:

test result: ok. 11 passed; 0 failed   # cdc: pgoutput parser, unit
test result: ok. 2 passed; 0 failed    # cdc: replication against real Postgres
test result: ok. 10 passed; 0 failed   # pipeline: state + analytics

fmt --check and clippy produce no output when there are no issues. Requires Docker running (for the test Postgres).

To verify the full stack (Postgres + pipeline + generator + output Parquet):

make docker-up
sleep 10
ls output/                 # orders_snapshot.parquet, revenue_by_minute.parquet, top_products.parquet
docker compose logs pipeline --tail 5
make docker-down

Expected output in docker compose logs pipeline: one flushed snapshot and analytics line every --flush-interval-secs, with insert/update/delete counters growing.

23 tests across two independent layers (make test):

flowchart TB
    subgraph U["21 unit"]
        U1["protocol: 11 cases: Begin/Commit/Relation/Insert/Update/Delete, NULL, invalid UTF-8, truncated buffer"]
        U2["state: 5 cases: insert/update/delete, missing column, delete of an untracked id"]
        U3["analytics: 4 cases: minute bucket, revenue sum, product ranking, empty snapshot"]
        U4["event: 1 case: Postgres/Unix epoch conversion"]
    end
    subgraph I["2 integration (real Postgres)"]
        I1["full cycle: insert -> update -> delete, via real logical replication"]
        I2["NULL column: correct Some/None round trip"]
    end

The integration tests do not use synthetic fixture bytes: they open a real replication session against a real Postgres, run real SQL through a second client, and verify the ChangeEvents that come back through the complete pgoutput protocol.

make help         # lists every command with a description
make build         # cargo build --workspace
make test-unit       # only the unit tests, no Postgres required
make test              # full suite (starts a test Postgres automatically)
make lint                # clippy -D warnings
make fmt                   # cargo fmt
make check                    # fmt-check + lint + test (what CI runs)

make docker-up      # builds and starts postgres + pipeline + generator
make docker-logs      # follows the logs of all three services
make docker-down        # tears down the containers (keeps the Postgres volume)

pipeline/Dockerfile and generator/Dockerfile are multi-stage: rust:slim-bookworm to build, debian:bookworm-slim as the runtime. Unlike the other projects in this portfolio (which use FROM scratch with a static musl build), the final image here is not scratch. See docs/trade-offs.md for why.

cdc/                          CDC engine, reusable, no business logic
  src/
    protocol.rs                 pgoutput parser (hand-written, zero-copy)
    client.rs                   ReplicationSession: connects, decodes, resolves schema
    event.rs                    ChangeEvent, ChangeKind, epoch conversion
    error.rs                    CdcError (thiserror)
  tests/
    replication_test.rs           integration against real Postgres
pipeline/                     consumer: materialized state + analytics + Parquet
  src/
    state.rs                     OrderState (HashMap<id, OrderRow>)
    analytics.rs                  Polars + rayon::join
    output.rs                       ParquetWriter
    main.rs                           CLI (clap) + flush loop
generator/                    synthetic order traffic (not part of replication)
  src/main.rs
db/
  init.sql                     orders table schema
docs/
  design.md                     architecture decisions
  trade-offs.md                   assumptions and trade-offs made
  assets/banner.svg                this README's banner
docker-compose.yml             postgres (wal_level=logical) + pipeline + generator
.github/workflows/ci.yml       fmt + clippy + test + docker build, on every push

Summary; details and alternatives considered in docs/trade-offs.md:

1. No HTTP/gRPC server: the output is Parquet on disk, not an API.
2. tokio-postgres via a git fork, not crates.io: the published version does not expose copy_both_simple, required for START_REPLICATION.
3. Postgres publications are FOR TABLE, not per connection: a real finding made while building the integration tests, fixed with explicit isolation between tests.
4. Parquet fully overwritten on every flush, not incremental: simple and correct at this project's scale.
5. HashMap<id, OrderRow> as state, not a DataFrame: CDC is point mutation per row; Polars only comes in at query time.

MIT

Jefferson Marchetti jeffersonm.ferreira@gmail.com