diff --git a/catalogs/postgres-cdc/.env.example b/catalogs/postgres-cdc/.env.example new file mode 100644 index 00000000..62d08b6e --- /dev/null +++ b/catalogs/postgres-cdc/.env.example @@ -0,0 +1 @@ +PG_USER=postgres diff --git a/catalogs/postgres-cdc/.gitignore b/catalogs/postgres-cdc/.gitignore new file mode 100644 index 00000000..7dcdeecb --- /dev/null +++ b/catalogs/postgres-cdc/.gitignore @@ -0,0 +1,3 @@ +.env +.spice/ +*.log diff --git a/catalogs/postgres-cdc/README.md b/catalogs/postgres-cdc/README.md new file mode 100644 index 00000000..94832f47 --- /dev/null +++ b/catalogs/postgres-cdc/README.md @@ -0,0 +1,365 @@ +# PostgreSQL Catalog CDC Acceleration + +Works with `v2.2+` + +The PostgreSQL Catalog Connector can automatically discover every table in a +database _and_ keep a local, always-fresh copy of each one using Change Data +Capture (CDC). Adding `acceleration: { refresh_mode: changes }` to the catalog +tells Spice to, with zero per-table configuration: + +1. **Bootstrap** every discovered table by snapshotting it into the + [Cayenne](https://docs.spiceai.org/components/data-accelerators/cayenne) accelerator, and +2. **Keep it live** by streaming inserts, updates, and deletes from the + PostgreSQL write-ahead log (WAL) through a single shared replication slot. + +Queries are then served from the local accelerated copy, and source mutations +show up automatically — no polling, no per-table `refresh_sql`, no manual +dataset definitions. + +This recipe uses the standard TPC-H benchmark dataset (Scale Factor 1). Every +TPC-H table has a primary key, so all eight are eligible for CDC acceleration. +You can point a catalog at any PostgreSQL database, though: each table is +accelerated according to its `REPLICA IDENTITY` — a primary key, or a unique +index via `REPLICA IDENTITY USING INDEX` — and any table with no usable replica +identity is skipped with a warning rather than failing the whole catalog. + +## How it differs from the [PostgreSQL Catalog Connector](../postgres) recipe + +| | `catalogs/postgres` | This recipe | +|---|---|---| +| Discovers all tables | ✅ | ✅ | +| Query path | Federated (each query hits PostgreSQL) | Local accelerated copy (Cayenne) | +| Freshness | Live (source is queried directly) | Live via CDC from the WAL | +| Requires `wal_level=logical` | ❌ | ✅ | +| Each accelerated table needs a usable `REPLICA IDENTITY` (primary key or unique index); others are skipped | ❌ | ✅ | + +## Prerequisites + +- [Docker](https://docs.docker.com/get-docker/) installed +- Spice installed (see the [Getting Started](https://docs.spiceai.org/getting-started) documentation) +- **PostgreSQL 13 or newer.** Catalog CDC creates its publication + `WITH (publish_via_partition_root = true)` so that a partitioned table's + changes are published under the parent relation; that publication option was + introduced in PostgreSQL 13. + +## Step 1. Start the PostgreSQL database + +The database is started with `wal_level=logical` and replication slots enabled, +which is what lets Spice stream changes from the WAL. The `postgres-init` +container downloads the TPC-H SF1 parquet files and loads them into PostgreSQL, +creating all tables with primary keys and foreign keys. + +```bash +git clone https://github.com/spiceai/cookbook.git +cd cookbook/catalogs/postgres-cdc +docker compose up -d +``` + +Wait for the init container to finish loading data (it downloads ~700 MB of +parquet files on first build): + +```bash +docker compose logs -f postgres-init +``` + +You should see: + +``` +tpch-cdc-postgres-init | PostgreSQL is ready. +tpch-cdc-postgres-init | Creating schema ... +tpch-cdc-postgres-init | Schema created. +... +tpch-cdc-postgres-init | All TPC-H tables loaded successfully! +``` + +## Step 2. Create a new directory and initialize a Spicepod + +```bash +mkdir postgres-cdc-catalog-recipe +cd postgres-cdc-catalog-recipe +spice init +``` + +## Step 3. Configure credentials + +Create a `.env` file with the database credentials: + +```bash +echo "PG_USER=postgres" > .env +``` + +## Step 4. Add the accelerated PostgreSQL catalog to `spicepod.yaml` + +The only difference from the plain catalog connector is the `acceleration` +block: + +```yaml +version: v1 +kind: Spicepod +name: postgres-cdc-catalog-recipe + +catalogs: + - from: pg + name: pg + include: + - 'public.*' + params: + pg_host: localhost + pg_port: 5432 + pg_db: tpch + pg_user: ${secrets:PG_USER} + pg_sslmode: disable + acceleration: + refresh_mode: changes +``` + +> `refresh_mode: changes` is the only supported catalog-level acceleration mode, +> and the engine defaults to `cayenne`. Each discovered table is accelerated +> according to its PostgreSQL `REPLICA IDENTITY`: a primary key (`DEFAULT`) or a +> unique index (`USING INDEX`) becomes the CDC key, and `REPLICA IDENTITY FULL` +> works too **as long as the table also has a primary key** (it is heavier — the +> full old-row image is written to the WAL on every change). Note that `FULL` +> alone is **not** enough: `FULL` still needs a primary key (or a `USING INDEX` +> key) to route upserts, so a `FULL` table with no key is **still skipped**, just +> like `NOTHING` and a keyless `DEFAULT`. A skipped table gets a warning and is +> left out of the catalog rather than failing catalog setup. Use `include`/`exclude` +> to scope out tables (or whole schemas) you don't want accelerated, which also +> silences the skip warning for +> known-ineligible tables. + +## Step 5. Start the Spice runtime + +```bash +spice run +``` + +Spice discovers all tables, snapshots each into Cayenne, and opens a single +shared replication slot to keep them live: + +``` +INFO runtime::catalogconnector::postgres_accelerated: Catalog 'pg': accelerating 8 table(s) via CDC (8 via primary key, 0 via REPLICA IDENTITY USING INDEX, 0 via REPLICA IDENTITY FULL; shared replication slot 'spice_pg_f0da15_3f484bfe'); 0 table(s) excluded by include/exclude filters; 0 table(s) skipped (no usable replica identity -- see warnings); tables added to these schema(s) later are picked up on the periodic catalog refresh; schema changes to existing tables, and renamed or dropped tables, are not tracked. +INFO runtime::init::catalog: Registered catalog 'pg' with 1 schema and 8 tables +INFO data_components::postgres_replication::slot: Created new replication slot slot=spice_pg_f0da15_3f484bfe publication=spice_pg_f0da15_3f484bfe_pub +INFO data_components::postgres_replication::shared: dataset joined shared replication slot table=public.customer slot=spice_pg_f0da15_3f484bfe members=2 +INFO data_components::postgres_replication::bootstrap: initial snapshot bootstrap complete dataset=...customer rows=150000 expected=Some(150000) +INFO data_components::postgres_replication::bootstrap: initial snapshot bootstrap complete dataset=...lineitem rows=6001215 expected=Some(6001101) +INFO runtime: All components are loaded. Spice runtime is ready! +``` + +All 8 tables share **one** replication slot and **one** publication — a +multi-table catalog opens a single replication connection, not one per table: + +```bash +docker exec tpch-cdc-postgres psql -U postgres -d tpch \ + -c "SELECT slot_name, plugin, slot_type, active FROM pg_replication_slots;" +``` + +``` + slot_name | plugin | slot_type | active +--------------------------+----------+-----------+-------- + spice_pg_f0da15_3f484bfe | pgoutput | logical | t +(1 row) +``` + +## Step 6. Query the accelerated catalog + +In a new terminal, start the Spice SQL REPL: + +```bash +spice sql +``` + +Query a table using the three-part `catalog.schema.table` name — this reads +from the local accelerated copy, not from PostgreSQL: + +```sql +SELECT c_custkey, c_name, c_mktsegment, c_acctbal +FROM pg.public.customer +WHERE c_custkey <= 5 +ORDER BY c_custkey; +``` + +``` ++-----------+--------------------+--------------+-----------+ +| c_custkey | c_name | c_mktsegment | c_acctbal | ++-----------+--------------------+--------------+-----------+ +| 1 | Customer#000000001 | BUILDING | 711.56 | +| 2 | Customer#000000002 | AUTOMOBILE | 121.65 | +| 3 | Customer#000000003 | AUTOMOBILE | 7498.12 | +| 4 | Customer#000000004 | MACHINERY | 2866.83 | +| 5 | Customer#000000005 | HOUSEHOLD | 794.47 | ++-----------+--------------------+--------------+-----------+ + +Time: 0.029 seconds. 5 rows. +``` + +## Step 7. Mutate the source and watch the change propagate + +This is the point of CDC acceleration: change the data in PostgreSQL, and the +accelerated copy converges automatically. + +The `mutate.sh` helper runs `INSERT`, `UPDATE`, and `DELETE` statements +**directly against PostgreSQL** (never against Spice). Run each step, then +re-run the query in the Spice SQL REPL to watch the change appear. + +### Insert + +```bash +./mutate.sh insert +``` + +```sql +SELECT c_custkey, c_name, c_mktsegment, c_acctbal +FROM pg.public.customer +WHERE c_custkey = 9999999; +``` + +Within about a second, the new row appears in the accelerated catalog: + +``` ++-----------+-------------------+--------------+-----------+ +| c_custkey | c_name | c_mktsegment | c_acctbal | ++-----------+-------------------+--------------+-----------+ +| 9999999 | Customer#CDC-DEMO | BUILDING | 100.00 | ++-----------+-------------------+--------------+-----------+ + +Time: 0.011 seconds. 1 rows. +``` + +### Update + +```bash +./mutate.sh update +``` + +```sql +SELECT c_custkey, c_name, c_acctbal +FROM pg.public.customer +WHERE c_custkey = 9999999; +``` + +The updated balance propagates: + +``` ++-----------+-------------------+-----------+ +| c_custkey | c_name | c_acctbal | ++-----------+-------------------+-----------+ +| 9999999 | Customer#CDC-DEMO | 999999.99 | ++-----------+-------------------+-----------+ + +Time: 0.011 seconds. 1 rows. +``` + +### Delete + +```bash +./mutate.sh delete +``` + +```sql +SELECT count(*) AS n +FROM pg.public.customer +WHERE c_custkey = 9999999; +``` + +The row is gone: + +``` ++---+ +| n | ++---+ +| 0 | ++---+ + +Time: 0.011 seconds. 1 rows. +``` + +No refresh, no restart — the accelerated catalog stays in lock-step with the +source through the WAL. + +## Step 8. Clean up + +```bash +docker compose down --volumes --rmi local +``` + +## How it works: slots, publications, and restarts + +**One slot and one publication per catalog.** All eligible tables in a catalog +share a single replication slot and a single publication, so a multi-table +catalog decodes the WAL once and opens one replication connection — not one per +table. The names are derived deterministically from the catalog (e.g. +`spice_pg_f0da15_3f484bfe` above). + +**The publication lists only the eligible tables.** Spice builds the publication +explicitly with `FOR TABLE ... ` / `ADD TABLE ...` over the tables it +accelerates — never `FOR ALL TABLES`. This means: + +- Tables with no usable `REPLICA IDENTITY` (keyless, `NOTHING`), views, + materialized views, and foreign tables are **never** publication members, so + the source never has to log changes Spice would only discard. +- Views and materialized views are not CDC-accelerable (they have no replica + identity). Each one is reported with a "not replicated" warning and left out + of the accelerated catalog — it is not an error, and it does not stop the + eligible tables from replicating. +- Discovery re-runs on the catalog's periodic refresh, so a table added to a + selected schema **after** startup is picked up and accelerated on the next + refresh. Schema changes to existing tables, and renamed or dropped tables, are + not tracked. + +**Restart vs. re-snapshot.** The replication slot persists on the PostgreSQL +server across a Spice restart. When the same Spice instance restarts, it resumes +from the slot's `restart_lsn` and replays only the WAL accumulated while it was +down — it does **not** re-snapshot tables from scratch. The slot name is +deterministic for a given instance, which is what lets it find and reuse its own +slot on restart. + +**Multiple Spice instances.** Two instances pointed at the same catalog get +distinct slot names, so they do not fight over one physical slot (PostgreSQL +permits a single consumer per slot). Rescheduling the same logical service onto a +different node is a distinct concern tracked by the slot-lifecycle enhancement +([#12018](https://github.com/spiceai/spiceai/issues/12018)), which covers making +the slot identity independent of the host and cleaning up slots that are no +longer used. + +## Troubleshooting + +**`Failed to setup the catalog pg (pg). PostgreSQL connection failed.`** + +Another PostgreSQL is already listening on `localhost:5432` — commonly a +host-local install (Homebrew `postgresql@16`, Postgres.app, or another +container). Because it binds the loopback address directly, it shadows this +recipe's Docker container for connections from the host, so Spice connects to +the wrong server (which has no `tpch` database) and fails. + +Confirm what's on the port: + +```bash +lsof -nP -iTCP:5432 -sTCP:LISTEN +``` + +Then either free the port — e.g. `brew services stop postgresql@16`, or stop +the other container — **or** run this recipe on a different port by changing +the published port in `compose.yaml` (e.g. `"5433:5432"`) and `pg_port` in +`spicepod.yaml` to match. + +**`... 0 of N discovered table(s) are eligible for CDC acceleration ...`** + +The catalog matched no CDC-eligible tables, so it fails to load rather than +registering an empty catalog. The error reports how many of the discovered +tables were skipped for lacking a usable `REPLICA IDENTITY` and how many were +excluded by `include`/`exclude`. Common causes: an `include`/`exclude` pattern +that matches nothing (it +is matched against `schema.table`, e.g. `public.*`), or a database whose tables +have no primary key and no `REPLICA IDENTITY USING INDEX`/`FULL`. Fix the +patterns, or give the tables a usable replica identity (a primary key, or a +`UNIQUE NOT NULL` index set via `ALTER TABLE ... REPLICA IDENTITY USING INDEX +...`), then restart — a catalog that discovers zero eligible tables fails to +load, so it won't pick them up until Spice is restarted. + +## References + +- [Spice.ai PostgreSQL Catalog Connector documentation](https://docs.spiceai.org/components/catalogs/postgres) +- [Cayenne Data Accelerator documentation](https://docs.spiceai.org/components/data-accelerators/cayenne) +- [TPC-H Benchmark Specification](https://www.tpc.org/tpc_documents_current_versions/pdf/tpc-h_v2.17.1.pdf) +- [Spice SQL CLI reference](https://docs.spiceai.org/cli/reference/sql) diff --git a/catalogs/postgres-cdc/compose.yaml b/catalogs/postgres-cdc/compose.yaml new file mode 100644 index 00000000..0a38407b --- /dev/null +++ b/catalogs/postgres-cdc/compose.yaml @@ -0,0 +1,38 @@ +services: + postgres: + image: postgres:16 + container_name: tpch-cdc-postgres + environment: + POSTGRES_HOST_AUTH_METHOD: trust + POSTGRES_DB: tpch + # Catalog-level CDC acceleration streams changes from the PostgreSQL + # write-ahead log. Logical decoding must be enabled, and the server must + # allow at least one replication slot and WAL sender for Spice's shared + # replication slot. + command: + - "postgres" + - "-c" + - "wal_level=logical" + - "-c" + - "max_replication_slots=10" + - "-c" + - "max_wal_senders=10" + ports: + - "5432:5432" + healthcheck: + test: ["CMD-SHELL", "pg_isready -U postgres -d tpch"] + interval: 5s + timeout: 5s + retries: 10 + + postgres-init: + container_name: tpch-cdc-postgres-init + build: postgres-init/ + depends_on: + postgres: + condition: service_healthy + environment: + PG_HOST: postgres + PG_PORT: "5432" + PG_DB: tpch + PG_USER: postgres diff --git a/catalogs/postgres-cdc/mutate.sh b/catalogs/postgres-cdc/mutate.sh new file mode 100755 index 00000000..194143d5 --- /dev/null +++ b/catalogs/postgres-cdc/mutate.sh @@ -0,0 +1,62 @@ +#!/usr/bin/env bash +# +# Mutate the *source* PostgreSQL database while Spice is running, so you can +# watch catalog-level CDC acceleration reflect each change. Every statement +# below runs directly against PostgreSQL (never against Spice) -- Spice picks +# the changes up from the write-ahead log through its shared replication slot. +# +# Each step prints the exact command it runs against PostgreSQL before running +# it, so it's clear what the source mutation is. +# +# After each step, re-run the matching query from the Spice SQL REPL: +# +# SELECT c_custkey, c_name, c_mktsegment, c_acctbal +# FROM pg.public.customer +# WHERE c_custkey = 9999999; +# +# Usage: ./mutate.sh [insert|update|delete|all] + +set -euo pipefail + +PSQL=(docker exec -i tpch-cdc-postgres psql -U postgres -d tpch -v ON_ERROR_STOP=1) + +# Print the exact PostgreSQL command, then run it against the source database. +run_sql() { + local label=$1 sql=$2 + echo "==> ${label}" + echo " \$ docker exec -i tpch-cdc-postgres psql -U postgres -d tpch -c \"${sql}\"" + "${PSQL[@]}" -c "${sql}" +} + +insert() { + run_sql "INSERT customer 9999999 into PostgreSQL" \ + "INSERT INTO customer (c_custkey, c_name, c_address, c_nationkey, c_phone, c_acctbal, c_mktsegment, c_comment) VALUES (9999999, 'Customer#CDC-DEMO', '1 Change Data Capture Way', 0, '00-000-000-0000', 100.00, 'BUILDING', 'inserted via CDC demo');" +} + +update() { + run_sql "UPDATE customer 9999999 acctbal in PostgreSQL" \ + "UPDATE customer SET c_acctbal = 999999.99 WHERE c_custkey = 9999999;" +} + +delete() { + run_sql "DELETE customer 9999999 from PostgreSQL" \ + "DELETE FROM customer WHERE c_custkey = 9999999;" +} + +case "${1:-all}" in + insert) insert ;; + update) update ;; + delete) delete ;; + all) + insert + echo " -> query Spice; you should see 1 row with acctbal 100.00"; echo + update + echo " -> query Spice; acctbal should become 999999.99"; echo + delete + echo " -> query Spice; the row should be gone (0 rows)"; echo + ;; + *) + echo "Usage: $0 [insert|update|delete|all]" >&2 + exit 1 + ;; +esac diff --git a/catalogs/postgres-cdc/postgres-init/Dockerfile b/catalogs/postgres-cdc/postgres-init/Dockerfile new file mode 100644 index 00000000..3e6afc55 --- /dev/null +++ b/catalogs/postgres-cdc/postgres-init/Dockerfile @@ -0,0 +1,24 @@ +FROM python:3.12-slim + +RUN apt-get update && apt-get install -y --no-install-recommends curl \ + && rm -rf /var/lib/apt/lists/* + +RUN pip install --no-cache-dir \ + pyarrow==20.0.0 \ + psycopg2-binary==2.9.10 + +RUN mkdir -p /data/tpch_sf1 + +RUN curl -fL https://public-data.spiceai.org/tpch_sf1/region/region.parquet -o /data/tpch_sf1/region.parquet \ + && curl -fL https://public-data.spiceai.org/tpch_sf1/nation/nation.parquet -o /data/tpch_sf1/nation.parquet \ + && curl -fL https://public-data.spiceai.org/tpch_sf1/part/part.parquet -o /data/tpch_sf1/part.parquet \ + && curl -fL https://public-data.spiceai.org/tpch_sf1/supplier/supplier.parquet -o /data/tpch_sf1/supplier.parquet \ + && curl -fL https://public-data.spiceai.org/tpch_sf1/customer/customer.parquet -o /data/tpch_sf1/customer.parquet \ + && curl -fL https://public-data.spiceai.org/tpch_sf1/partsupp/partsupp.parquet -o /data/tpch_sf1/partsupp.parquet \ + && curl -fL https://public-data.spiceai.org/tpch_sf1/orders/orders.parquet -o /data/tpch_sf1/orders.parquet \ + && curl -fL https://public-data.spiceai.org/tpch_sf1/lineitem/lineitem.parquet -o /data/tpch_sf1/lineitem.parquet + +WORKDIR /app +COPY load-tpch.py . + +CMD ["python", "load-tpch.py"] diff --git a/catalogs/postgres-cdc/postgres-init/load-tpch.py b/catalogs/postgres-cdc/postgres-init/load-tpch.py new file mode 100644 index 00000000..26861fd8 --- /dev/null +++ b/catalogs/postgres-cdc/postgres-init/load-tpch.py @@ -0,0 +1,269 @@ +#!/usr/bin/env python3 +""" +Download TPC-H SF1 parquet files and load them into PostgreSQL with +primary keys, proper types, foreign key constraints, and the standard +TPC-H secondary indexes. + +Creation order respects FK dependencies: + region → nation → part, supplier, customer + supplier + part → partsupp + customer → orders → lineitem + +Secondary indexes are created after the bulk load (faster than maintaining +them during COPY). Without the composite lineitem(l_partkey, l_suppkey) index, +correlated-subquery queries such as TPC-H Q20 fall back to full 6M-row scans +and are effectively unrunnable. +""" + +import time +import pyarrow.parquet as pq +import pyarrow as pa +import psycopg2 +from psycopg2 import sql +import io +import os + +PG_HOST = os.getenv("PG_HOST", "postgres") +PG_PORT = int(os.getenv("PG_PORT", "5432")) +PG_DB = os.getenv("PG_DB", "tpch") +PG_USER = os.getenv("PG_USER", "postgres") + +DATA_DIR = "/data/tpch_sf1" + +# --------------------------------------------------------------------------- +# DDL – tables in FK-safe creation order +# --------------------------------------------------------------------------- + +DDL = """ +CREATE TABLE IF NOT EXISTS region ( + r_regionkey INTEGER NOT NULL, + r_name CHAR(25) NOT NULL, + r_comment VARCHAR(152), + PRIMARY KEY (r_regionkey) +); + +CREATE TABLE IF NOT EXISTS nation ( + n_nationkey INTEGER NOT NULL, + n_name CHAR(25) NOT NULL, + n_regionkey INTEGER NOT NULL, + n_comment VARCHAR(152), + PRIMARY KEY (n_nationkey), + CONSTRAINT fk_nation_region + FOREIGN KEY (n_regionkey) REFERENCES region (r_regionkey) +); + +CREATE TABLE IF NOT EXISTS part ( + p_partkey INTEGER NOT NULL, + p_name VARCHAR(55) NOT NULL, + p_mfgr CHAR(25) NOT NULL, + p_brand CHAR(10) NOT NULL, + p_type VARCHAR(25) NOT NULL, + p_size INTEGER NOT NULL, + p_container CHAR(10) NOT NULL, + p_retailprice DECIMAL(15,2) NOT NULL, + p_comment VARCHAR(23) NOT NULL, + PRIMARY KEY (p_partkey) +); + +CREATE TABLE IF NOT EXISTS supplier ( + s_suppkey INTEGER NOT NULL, + s_name CHAR(25) NOT NULL, + s_address VARCHAR(40) NOT NULL, + s_nationkey INTEGER NOT NULL, + s_phone CHAR(15) NOT NULL, + s_acctbal DECIMAL(15,2) NOT NULL, + s_comment VARCHAR(101) NOT NULL, + PRIMARY KEY (s_suppkey), + CONSTRAINT fk_supplier_nation + FOREIGN KEY (s_nationkey) REFERENCES nation (n_nationkey) +); + +CREATE TABLE IF NOT EXISTS customer ( + c_custkey INTEGER NOT NULL, + c_name VARCHAR(25) NOT NULL, + c_address VARCHAR(40) NOT NULL, + c_nationkey INTEGER NOT NULL, + c_phone CHAR(15) NOT NULL, + c_acctbal DECIMAL(15,2) NOT NULL, + c_mktsegment CHAR(10) NOT NULL, + c_comment VARCHAR(117) NOT NULL, + PRIMARY KEY (c_custkey), + CONSTRAINT fk_customer_nation + FOREIGN KEY (c_nationkey) REFERENCES nation (n_nationkey) +); + +CREATE TABLE IF NOT EXISTS partsupp ( + ps_partkey INTEGER NOT NULL, + ps_suppkey INTEGER NOT NULL, + ps_availqty INTEGER NOT NULL, + ps_supplycost DECIMAL(15,2) NOT NULL, + ps_comment VARCHAR(199) NOT NULL, + PRIMARY KEY (ps_partkey, ps_suppkey), + CONSTRAINT fk_partsupp_part + FOREIGN KEY (ps_partkey) REFERENCES part (p_partkey), + CONSTRAINT fk_partsupp_supplier + FOREIGN KEY (ps_suppkey) REFERENCES supplier (s_suppkey) +); + +CREATE TABLE IF NOT EXISTS orders ( + o_orderkey INTEGER NOT NULL, + o_custkey INTEGER NOT NULL, + o_orderstatus CHAR(1) NOT NULL, + o_totalprice DECIMAL(15,2) NOT NULL, + o_orderdate DATE NOT NULL, + o_orderpriority CHAR(15) NOT NULL, + o_clerk CHAR(15) NOT NULL, + o_shippriority INTEGER NOT NULL, + o_comment VARCHAR(79) NOT NULL, + PRIMARY KEY (o_orderkey), + CONSTRAINT fk_orders_customer + FOREIGN KEY (o_custkey) REFERENCES customer (c_custkey) +); + +CREATE TABLE IF NOT EXISTS lineitem ( + l_orderkey INTEGER NOT NULL, + l_partkey INTEGER NOT NULL, + l_suppkey INTEGER NOT NULL, + l_linenumber INTEGER NOT NULL, + l_quantity DECIMAL(15,2) NOT NULL, + l_extendedprice DECIMAL(15,2) NOT NULL, + l_discount DECIMAL(15,2) NOT NULL, + l_tax DECIMAL(15,2) NOT NULL, + l_returnflag CHAR(1) NOT NULL, + l_linestatus CHAR(1) NOT NULL, + l_shipdate DATE NOT NULL, + l_commitdate DATE NOT NULL, + l_receiptdate DATE NOT NULL, + l_shipinstruct CHAR(25) NOT NULL, + l_shipmode CHAR(10) NOT NULL, + l_comment VARCHAR(44) NOT NULL, + PRIMARY KEY (l_orderkey, l_linenumber), + CONSTRAINT fk_lineitem_orders + FOREIGN KEY (l_orderkey) REFERENCES orders (o_orderkey), + CONSTRAINT fk_lineitem_partsupp + FOREIGN KEY (l_partkey, l_suppkey) REFERENCES partsupp (ps_partkey, ps_suppkey) +); +""" + +# --------------------------------------------------------------------------- +# Secondary indexes – created after the bulk load. These mirror the standard +# TPC-H index set; the lineitem(l_partkey)/l_suppkey indexes in particular are +# required for correlated-subquery queries like Q20 to run in reasonable time. +# --------------------------------------------------------------------------- + +INDEXES = """ +CREATE INDEX IF NOT EXISTS idx_region_name ON region(r_name); +CREATE INDEX IF NOT EXISTS idx_nation_region ON nation(n_regionkey); +CREATE INDEX IF NOT EXISTS idx_supplier_nation ON supplier(s_nationkey); +CREATE INDEX IF NOT EXISTS idx_customer_nation ON customer(c_nationkey); +CREATE INDEX IF NOT EXISTS idx_orders_custkey_orderdate ON orders(o_custkey, o_orderdate); +-- Composite (l_partkey, l_suppkey): required for the Q20 correlated subquery to +-- resolve via an index lookup instead of a full lineitem scan. Also serves +-- l_partkey-prefix lookups, so a standalone l_partkey index is unnecessary. A +-- single-column l_partkey index alone leaves Q20 effectively unrunnable +-- (observed ~1h49m vs ~0.1s with the composite) on a modestly-resourced host. +CREATE INDEX IF NOT EXISTS idx_lineitem_partkey_suppkey ON lineitem(l_partkey, l_suppkey); +CREATE INDEX IF NOT EXISTS idx_lineitem_suppkey ON lineitem(l_suppkey); +CREATE INDEX IF NOT EXISTS idx_lineitem_orderkey ON lineitem(l_orderkey); +CREATE INDEX IF NOT EXISTS idx_partsupp_part ON partsupp(ps_partkey); +CREATE INDEX IF NOT EXISTS idx_partsupp_supplier ON partsupp(ps_suppkey); +CREATE INDEX IF NOT EXISTS idx_part_brand_container ON part(p_brand, p_container); +""" + +# Tables in load order (parent tables before child tables) +TABLES = ["region", "nation", "part", "supplier", "customer", "partsupp", "orders", "lineitem"] + + +def wait_for_postgres(max_retries=30, delay=2): + """Retry until postgres is accepting connections.""" + for attempt in range(max_retries): + try: + conn = psycopg2.connect(host=PG_HOST, port=PG_PORT, dbname=PG_DB, + user=PG_USER) + conn.close() + print("PostgreSQL is ready.") + return + except psycopg2.OperationalError: + print(f"Waiting for PostgreSQL... ({attempt + 1}/{max_retries})") + time.sleep(delay) + raise RuntimeError("PostgreSQL did not become ready in time.") + + +def arrow_to_csv_bytes(table: pa.Table) -> bytes: + """Serialise an Arrow table to CSV bytes for COPY FROM.""" + buf = io.BytesIO() + import pyarrow.csv as pa_csv + pa_csv.write_csv(table, buf) + buf.seek(0) + return buf.read() + + +def load_table(conn, table_name: str): + path = f"{DATA_DIR}/{table_name}.parquet" + print(f" Reading {path} ...") + arrow_table = pq.read_table(path) + print(f" {arrow_table.num_rows:,} rows, schema: {arrow_table.schema}") + + # Cast any decimal columns to float64 for parquet compatibility, + # then postgres COPY will coerce back to DECIMAL via the table DDL. + casts = [] + for field in arrow_table.schema: + if pa.types.is_decimal(field.type): + casts.append((field.name, pa.float64())) + elif pa.types.is_date(field.type): + # keep as-is; CSV export renders as YYYY-MM-DD which postgres accepts + casts.append((field.name, field.type)) + else: + casts.append((field.name, field.type)) + + new_schema = pa.schema([pa.field(name, typ) for name, typ in casts]) + arrays = [] + for field in arrow_table.schema: + col = arrow_table.column(field.name) + if pa.types.is_decimal(field.type): + col = col.cast(pa.float64()) + arrays.append(col) + arrow_table = pa.table(dict(zip(arrow_table.schema.names, arrays))) + + buf = io.BytesIO() + import pyarrow.csv as pa_csv + pa_csv.write_csv(arrow_table, buf) + buf.seek(0) + + with conn.cursor() as cur: + cur.copy_expert( + f"COPY {table_name} FROM STDIN WITH (FORMAT CSV, HEADER TRUE)", + buf, + ) + conn.commit() + print(f" Loaded {arrow_table.num_rows:,} rows into {table_name}.") + + +def main(): + wait_for_postgres() + + conn = psycopg2.connect(host=PG_HOST, port=PG_PORT, dbname=PG_DB, user=PG_USER) + conn.autocommit = False + + print("Creating schema ...") + with conn.cursor() as cur: + cur.execute(DDL) + conn.commit() + print("Schema created.") + + for table in TABLES: + print(f"\nLoading {table} ...") + load_table(conn, table) + + print("\nCreating secondary indexes ...") + with conn.cursor() as cur: + cur.execute(INDEXES) + conn.commit() + print("Secondary indexes created.") + + conn.close() + print("\nAll TPC-H tables loaded successfully!") + + +if __name__ == "__main__": + main() diff --git a/catalogs/postgres-cdc/spicepod.yaml b/catalogs/postgres-cdc/spicepod.yaml new file mode 100644 index 00000000..306cc014 --- /dev/null +++ b/catalogs/postgres-cdc/spicepod.yaml @@ -0,0 +1,20 @@ +version: v1 +kind: Spicepod +name: postgres-cdc-catalog-recipe + +catalogs: + - from: pg + name: pg + include: + - 'public.*' + params: + pg_host: localhost + pg_port: 5432 + pg_db: tpch + pg_user: ${secrets:PG_USER} + pg_sslmode: disable + # Bootstrap and CDC-accelerate every discovered table with zero per-table + # configuration. Each table is snapshotted once, then kept live from the + # PostgreSQL WAL through a single shared replication slot. + acceleration: + refresh_mode: changes