Querying Sharded PostgreSQL with DuckDB

This guide walks through using DuckDB’s postgres extension to run ad-hoc analytics across multiple PostgreSQL shards from a single laptop or jump host — no ETL pipeline, no coordinator service — as part of a broader federated query execution strategy within Cross-Partition Querying & Aggregation Strategies.

Where postgres_fdw (covered in executing federated queries across multiple PostgreSQL instances) requires DDL on a coordinator database, DuckDB inverts the model: the analyst’s own process becomes a throwaway coordinator. It attaches each shard over the wire, scans them in parallel with a vectorized engine, and disappears when the session ends. That makes it ideal for incident investigation and exploratory analysis — and fundamentally unsuited to serving federated queries to applications, a boundary this walkthrough makes explicit at the end.

DuckDB as a throwaway coordinator over PostgreSQL shards An analyst runs one DuckDB process with the postgres extension loaded. The process attaches shard one, shard two, and shard three read-only over the PostgreSQL wire protocol and unions them into a local view. Hot aggregates are materialized downward into local DuckDB tables and Parquet snapshot files for repeated analysis. Analyst SQL / Python DuckDB process postgres extension attached: shard1..3 all_orders view vectorized aggregation PostgreSQL shard 1 10.0.2.11:5432 PostgreSQL shard 2 10.0.2.12:5432 PostgreSQL shard 3 10.0.2.13:5432 Local snapshots DuckDB tables + Parquet wire scan wire scan wire scan materialize

Prerequisites

Step 1 — Install DuckDB and load the postgres extension

DuckDB ships as a single binary, and the postgres extension installs from the built-in extension repository with one statement. Install once per host; load in each session (or set it to autoload).

# Install the DuckDB CLI (Linux x86_64) via your package manager or artifact mirror
sudo apt-get install -y duckdb

# Start a persistent local database file for this investigation
duckdb ~/analysis/shards.duckdb
-- Inside the DuckDB shell: fetch and load the postgres extension
INSTALL postgres;
LOAD postgres;

Operational note: Using a persistent database file (shards.duckdb) rather than the default in-memory mode means views, secrets, and materialized tables survive between sessions — an incident notebook you can reopen tomorrow. In-memory mode loses everything on exit.

SRE tip: On locked-down hosts with no outbound internet, INSTALL postgres fails at the extension download. Mirror the DuckDB extension repository internally and point SET custom_extension_repository at it, the same way you already mirror OS packages.

Step 2 — Attach each shard read-only

ATTACH connects a remote PostgreSQL database into the DuckDB namespace. Attach every shard with READ_ONLY so no ad-hoc statement can ever write to production, and keep credentials out of the connection string by using DuckDB secrets.

-- One secret per shard keeps passwords out of ATTACH strings and shell history
CREATE PERSISTENT SECRET shard1_pg (
  TYPE postgres, HOST '10.0.2.11', PORT 5432,
  DATABASE 'orders_db', USER 'analytics_ro', PASSWORD 'REDACTED'
);
-- Repeat for shard2_pg (10.0.2.12) and shard3_pg (10.0.2.13)

ATTACH '' AS shard1 (TYPE postgres, SECRET shard1_pg, READ_ONLY);
ATTACH '' AS shard2 (TYPE postgres, SECRET shard2_pg, READ_ONLY);
ATTACH '' AS shard3 (TYPE postgres, SECRET shard3_pg, READ_ONLY);

SHOW DATABASES;

Operational note: Attach read replicas instead of primaries where they exist. DuckDB parallelizes scans by opening several connections per attached shard, and a wide scan against a busy primary competes with OLTP traffic for I/O and buffer cache.

DBA tip: Each attached database consumes shard connections at query time (roughly one per scan thread). If several analysts adopt this workflow, put PgBouncer in front of the shards or set a low connection_limit on the analytics_ro role so investigations can never starve the application pool.

Step 3 — Union the shards into one local view

A single view over a UNION ALL of the attached shards gives every subsequent query the shape of one logical table. Tag each branch with its shard name — indispensable when a number looks wrong and you need to know which shard produced it.

CREATE OR REPLACE VIEW all_orders AS
SELECT 'shard1' AS shard, order_id, tenant_id, total, currency, created_at
FROM shard1.public.orders
UNION ALL
SELECT 'shard2' AS shard, order_id, tenant_id, total, currency, created_at
FROM shard2.public.orders
UNION ALL
SELECT 'shard3' AS shard, order_id, tenant_id, total, currency, created_at
FROM shard3.public.orders;

Operational note: List columns explicitly instead of SELECT *. If one shard’s schema drifts — an extra column from a half-rolled-out migration — UNION ALL by position silently misaligns columns, and explicit lists turn that into an immediate binder error instead of wrong answers.

SRE tip: Store the view definition (and the ATTACH statements) in a version-controlled setup.sql and start sessions with duckdb ~/analysis/shards.duckdb -init setup.sql. Everyone on the on-call rotation then gets an identical, reproducible federation environment in one command.

Step 4 — Run cross-shard aggregations and check filter pushdown

Queries against the view now fan out to all three shards; DuckDB streams rows back and aggregates them locally with its vectorized engine. The extension pushes column projections and simple WHERE predicates into the remote scan, so filtered aggregations move only matching rows over the network.

-- Revenue per tenant over the last 7 days, across all shards
SELECT tenant_id,
       count(*)   AS orders,
       sum(total) AS revenue
FROM all_orders
WHERE created_at >= now() - INTERVAL 7 DAY
GROUP BY tenant_id
ORDER BY revenue DESC
LIMIT 20;

-- Confirm the filter reaches the shards rather than running locally
EXPLAIN SELECT count(*) FROM all_orders
WHERE created_at >= now() - INTERVAL 7 DAY;

In the EXPLAIN output, each POSTGRES_SCAN operator should show the created_at filter attached to the remote scan. If the filter sits only in a DuckDB FILTER node above the scan, the predicate was not pushable (an expression on the column, or an unsupported type) and each shard is streaming its entire table.

Operational note: Aggregations themselves are not pushed down — sum and count always run in DuckDB over the streamed rows. That is acceptable when filters cut the row volume; it is painful when they cannot, which is what Step 5’s materialization exists for.

DBA tip: Long-running wide scans hold their transaction snapshot open on the shard for the duration, delaying vacuum the same way any long transaction does. Keep exploratory scans time-bounded with WHERE clauses on indexed timestamp columns, and check pg_stat_activity on shards for stragglers after a session ends.

Step 5 — Materialize hot aggregates into local tables and Parquet

An investigation typically hammers the same slice repeatedly. Instead of re-scanning three shards for every variation, pull the aggregate once into a local DuckDB table, and export a Parquet snapshot when the result should outlive the session or be shared.

-- Materialize once: daily revenue per shard and tenant for the last 90 days
CREATE OR REPLACE TABLE daily_revenue AS
SELECT shard,
       tenant_id,
       date_trunc('day', created_at) AS day,
       count(*)   AS orders,
       sum(total) AS revenue
FROM all_orders
WHERE created_at >= now() - INTERVAL 90 DAY
GROUP BY ALL;

-- Every follow-up query is now local and sub-second
SELECT day, sum(revenue) FROM daily_revenue GROUP BY day ORDER BY day;

-- Snapshot for sharing or archival
COPY daily_revenue TO 'daily_revenue_2026-07-06.parquet' (FORMAT parquet);

Operational note: A materialized table is a snapshot, not a subscription — it ages from the moment CREATE TABLE AS finishes. Embed the snapshot time in the filename (as above) or add a snapshot_at column, so a week-old Parquet file can never masquerade as live data in someone else’s analysis.

SRE tip: Set SET memory_limit = '12GB' and SET temp_directory = '/data/duckdb_spill' before big materializations. DuckDB spills oversized aggregations and sorts to disk instead of dying — but only if the temp directory sits on a volume with real capacity, not the root filesystem.

Step 6 — Script the refresh with Python

Once an aggregate proves useful beyond one incident, wrap the attach-union-materialize cycle in a short Python script so it can run from cron or a notebook without hand-typed setup.

#!/usr/bin/env python3
"""Refresh cross-shard daily_revenue snapshot into DuckDB + Parquet."""
import os
import duckdb

SHARDS = {
    "shard1": "10.0.2.11",
    "shard2": "10.0.2.12",
    "shard3": "10.0.2.13",
}

con = duckdb.connect("/data/analysis/shards.duckdb")
con.execute("INSTALL postgres; LOAD postgres;")

for name, host in SHARDS.items():
    con.execute(f"""
        ATTACH 'host={host} port=5432 dbname=orders_db
                user=analytics_ro password={os.environ["PG_RO_PASSWORD"]}'
        AS {name} (TYPE postgres, READ_ONLY)
    """)

union_sql = " UNION ALL ".join(
    f"SELECT '{name}' AS shard, tenant_id, total, created_at "
    f"FROM {name}.public.orders "
    f"WHERE created_at >= now() - INTERVAL 90 DAY"
    for name in SHARDS
)
con.execute(f"""
    CREATE OR REPLACE TABLE daily_revenue AS
    SELECT shard, tenant_id, date_trunc('day', created_at) AS day,
           count(*) AS orders, sum(total) AS revenue
    FROM ({union_sql}) GROUP BY ALL
""")
con.execute("COPY daily_revenue TO '/data/exports/daily_revenue.parquet' (FORMAT parquet)")
print(con.execute("SELECT count(*), max(day) FROM daily_revenue").fetchone())

Operational note: The password comes from the environment (PG_RO_PASSWORD), keeping it out of the script and of process listings that log argv. In shared automation, prefer a secrets manager injecting the variable at runtime over exporting it in a shell profile.

SRE tip: Print (and alert on) max(day) from the refreshed table, exactly as you would for any scheduled rollup. A refresh that silently attaches two of three shards produces plausible-looking but wrong totals; a per-shard row count check catches it immediately.

Verification

Confirm attachments, view coverage, and pushdown from a fresh DuckDB session:

SHOW DATABASES;

SELECT shard, count(*) AS rows, max(created_at) AS newest
FROM all_orders
GROUP BY shard
ORDER BY shard;

Expected output for SHOW DATABASES: the local database plus shard1, shard2, and shard3. Expected output for the coverage query: one row per shard with a newest timestamp within the last few minutes on an active system, for example:

┌────────┬─────────┬─────────────────────────┐
│ shard  │  rows   │         newest          │
├────────┼─────────┼─────────────────────────┤
│ shard1 │ 4128390 │ 2026-07-06 09:41:02+00  │
│ shard2 │ 4090217 │ 2026-07-06 09:41:05+00  │
│ shard3 │ 4211045 │ 2026-07-06 09:40:58+00  │
└────────┴─────────┴─────────────────────────┘

A missing shard row means an ATTACH failed or the view was created while a shard was absent; a stale newest on one shard usually means you attached a lagging replica rather than the intended node.

Failure mode table

Failure mode Root cause SRE mitigation
DuckDB process killed or Out of Memory during a cross-shard join Both join sides streamed unfiltered into a single process; intermediate state exceeded memory and spill capacity Set memory_limit and a temp_directory on a large volume; pre-filter or pre-aggregate each side before joining; move recurring heavy joins to a distributed engine such as Trino
Shards show full-table scan load whenever the analyst queries the view Filter predicate not pushed into POSTGRES_SCAN — expression-wrapped columns or unsupported types keep filtering local Check EXPLAIN for filters on the scan operator; rewrite predicates as bare indexed-column comparisons; materialize a filtered slice once and iterate locally
Decisions made on stale numbers from a materialized snapshot CREATE TABLE AS and Parquet exports are point-in-time copies that silently age Embed snapshot timestamps in table columns and filenames; schedule the Python refresh via cron; alert when max(day) in the snapshot falls behind expected freshness

FAQ

Is DuckDB querying live shard data or a copy?

Live data. Each query against an attached database opens connections to that shard and streams rows over the PostgreSQL wire protocol at execution time, so results reflect committed data as of the moment the scan runs. Nothing is cached between queries unless you explicitly materialize results into local DuckDB tables or Parquet files — which is exactly what Step 5 does for slices you query repeatedly.

How large can the shards be before the DuckDB approach breaks down?

The constraint is the working set per query, not total shard size. Filtered aggregations that push predicates to the shards handle multi-hundred-gigabyte shards comfortably, because only matching rows cross the network. Cross-shard joins over unfiltered tables are the breaking point: both sides stream into one process, and once intermediate state exceeds memory_limit plus disk spill, the query dies. There is no distributed execution to fall back on — that is the moment to move the workload to Trino.

Can DuckDB write results back to the PostgreSQL shards?

Yes — attaching without the READ_ONLY flag enables INSERT and CREATE TABLE against the shard. For analytics workflows, keep every shard attachment read-only and write results to local DuckDB tables or Parquet instead, so an ad-hoc session can never mutate production. If rollups must land back in PostgreSQL for applications to read, that is a scheduled-pipeline job better served by the Trino pattern than by an analyst’s session.

Back to Federated Query Execution