Running Federated Queries Across Shards with Trino

This guide walks through deploying a single-node Trino server that federates SQL across independent PostgreSQL shards — one catalog per shard — as part of a broader federated query execution strategy within Cross-Partition Querying & Aggregation Strategies.

Trino sits in the middle ground between coordinator extensions like Citus (covered in setting up Citus for distributed PostgreSQL queries) and in-database foreign tables: it leaves the shards completely untouched. No extension is installed on any shard, no schema changes are required, and the shards keep serving OLTP traffic while Trino runs read-mostly analytics against them over JDBC.

Single-node Trino federating three PostgreSQL shards A client sends SQL to a single-node Trino server acting as both coordinator and worker. The server holds catalogs shard1, shard2, and shard3, plus a memory catalog for views. It scans each PostgreSQL shard over JDBC with pushed-down filters and merges results locally. Client trino CLI / BI Trino (single node) coordinator + worker catalogs: shard1..3 memory catalog (views) merge + final GROUP BY PostgreSQL shard 1 10.0.2.11:5432 PostgreSQL shard 2 10.0.2.12:5432 PostgreSQL shard 3 10.0.2.13:5432 JDBC scan JDBC scan JDBC scan

Prerequisites

Step 1 — Deploy a single-node Trino server

A single Trino process can act as coordinator and worker simultaneously, which is the right shape for federating a handful of shards. Run the official container with the configuration directory mounted from the host so catalog files survive restarts.

# docker-compose.yml on the Trino host
services:
  trino:
    image: trinodb/trino:476
    container_name: trino
    ports:
      - "8080:8080"
    volumes:
      - ./etc:/etc/trino          # config.properties, jvm.config, catalog/
    mem_limit: 20g
    restart: unless-stopped

The mounted etc/config.properties enables single-node mode by letting the coordinator schedule work on itself:

# etc/config.properties
coordinator=true
node-scheduler.include-coordinator=true
http-server.http.port=8080
discovery.uri=http://localhost:8080

Operational note: Pin the image tag (trinodb/trino:476) rather than latest. Trino releases frequently and connector property names occasionally change between versions; an unpinned image can break catalogs on a routine host reboot.

SRE tip: Expose port 8080 only to your analytics network or behind an authenticating proxy. A default Trino install has no authentication, and every configured catalog — meaning every shard — is queryable by anyone who can reach the port.

Step 2 — Size the JVM and query memory budgets

Federated aggregation happens in Trino’s heap: rows stream in from every shard, and the final GROUP BY and merge run locally. Set the JVM heap to most of the container’s memory, then bound individual queries so one runaway aggregation cannot take down the server.

# etc/jvm.config (one flag per line)
-server
-Xmx16G
-XX:+UseG1GC
-XX:InitialRAMPercentage=80
-XX:+ExitOnOutOfMemoryError
-XX:+HeapDumpOnOutOfMemoryError

# etc/config.properties — append these lines
query.max-memory=12GB
query.max-memory-per-node=8GB
memory.heap-headroom-per-node=2GB

Operational note: query.max-memory-per-node plus memory.heap-headroom-per-node must stay below -Xmx, or Trino refuses to start with a configuration validation error. On a single node, query.max-memory effectively equals the per-node limit — keep them consistent to avoid confusing failure messages.

SRE tip: Keep -XX:+ExitOnOutOfMemoryError enabled and let your restart policy (restart: unless-stopped) bring Trino back. A JVM that limps on after an OOM returns wrong or hung results; a clean crash-and-restart is strictly better for a stateless query layer.

Step 3 — Define one PostgreSQL catalog per shard

Each file in etc/catalog/ becomes a catalog — a top-level namespace in Trino SQL. Create one per shard so shard1.public.orders, shard2.public.orders, and shard3.public.orders address the same logical table on different shards. Add a memory catalog to hold federated views.

# etc/catalog/shard1.properties
connector.name=postgresql
connection-url=jdbc:postgresql://10.0.2.11:5432/orders_db
connection-user=trino_reader
connection-password=${ENV:SHARD1_PASSWORD}
join-pushdown.enabled=true
aggregation-pushdown.enabled=true

# etc/catalog/shard2.properties — identical except the host
# connection-url=jdbc:postgresql://10.0.2.12:5432/orders_db

# etc/catalog/shard3.properties
# connection-url=jdbc:postgresql://10.0.2.13:5432/orders_db

# etc/catalog/federated.properties — in-memory catalog for views
connector.name=memory

Operational note: The ${ENV:...} substitution reads credentials from the container environment, keeping passwords out of the properties files. Restart Trino after adding or editing catalog files — catalogs are loaded at startup, not hot-reloaded, on a default install.

DBA tip: Point connection-url at a read replica of each shard where one exists. Analytical scans from Trino compete with OLTP for buffer cache and I/O on the primary; a replica isolates that pressure and costs nothing in correctness for reporting workloads.

Step 4 — Query across catalogs with a UNION ALL view

Federation across shards is expressed as a UNION ALL over the per-catalog tables. Save it once as a view in the memory catalog so analysts query a single object, with a shard column preserved for skew debugging.

CREATE SCHEMA IF NOT EXISTS federated.views;

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

-- Cross-shard aggregation, exactly as if it were one table
SELECT date_trunc('day', created_at) AS day,
       count(*)   AS orders,
       sum(total) AS revenue
FROM federated.views.all_orders
WHERE created_at >= date_add('day', -30, current_date)
GROUP BY 1
ORDER BY 1;

Operational note: The memory connector stores view definitions in the Trino process, so they vanish on restart. Keep the CREATE VIEW statements in a version-controlled SQL file and replay them from a container entrypoint or your scheduler after each deploy.

SRE tip: Always include the literal shard tag column. When one shard’s numbers look wrong, GROUP BY shard immediately isolates which branch of the union is misbehaving — without it you are re-running per-catalog queries by hand during an incident.

Step 5 — Verify pushdown behaviour and know its limits

Trino pushes filters, projections, LIMIT, and many single-table aggregations into the remote PostgreSQL query for each catalog. What it cannot do is push work across catalogs: the final GROUP BY over the UNION ALL, and any join between two different catalogs, always execute inside Trino. Inspect the plan before trusting a query’s cost.

EXPLAIN
SELECT tenant_id, sum(total)
FROM federated.views.all_orders
WHERE created_at >= DATE '2026-06-01'
GROUP BY tenant_id;

In the plan, each TableScan fragment shows the remote query sent to that shard — confirm the WHERE filter (and ideally a partial aggregate) appears inside it:

Fragment 2 [SOURCE]
  ScanFilterProject[table = shard1:public.orders ...]
      constraint on created_at pushed into connector
  ...
Fragment 1 [HASH]
  Aggregate[type = FINAL, keys = [tenant_id]]

If a filter appears only in a Trino FilterProject node above the scan, pushdown failed for that predicate — the usual culprits are functions applied to the column (date_trunc on the filter side), type mismatches, or varchar comparisons with differing collations.

Operational note: Joins across catalogs (e.g. shard1.public.orders to shard2.public.customers) pull both sides over JDBC and join in Trino memory. Keep cross-catalog joins to pre-filtered or pre-aggregated inputs, or materialise one side first — this is the same coordinator-merge discipline described in cross-shard aggregation patterns.

DBA tip: Each Trino query opens its own JDBC connections per shard, bypassing any application-side pooling. If shards run close to max_connections, front them with PgBouncer and point the catalogs at the pooler port, as with any other federated client.

Step 6 — Wire a scheduled aggregation query

Recurring rollups should not depend on an analyst remembering to run them. Drive the federated query from the Trino CLI on a schedule and materialise the result into a reporting table on a designated shard — the PostgreSQL connector supports INSERT and CREATE TABLE AS.

#!/usr/bin/env bash
# /opt/trino-jobs/daily_revenue_rollup.sh — run from cron at 02:15
set -euo pipefail

trino --server http://localhost:8080 --execute "
  INSERT INTO shard1.reporting.daily_revenue (day, shard, orders, revenue)
  SELECT date_trunc('day', created_at), shard, count(*), sum(total)
  FROM federated.views.all_orders
  WHERE created_at >= date_add('day', -1, current_date)
    AND created_at <  current_date
  GROUP BY 1, 2
" || { echo 'rollup failed' >&2; exit 1; }

# crontab entry:
# 15 2 * * * /opt/trino-jobs/daily_revenue_rollup.sh >> /var/log/trino-jobs.log 2>&1

Operational note: Make the job idempotent — either DELETE yesterday’s slice before inserting, or write to a staging table and swap. A cron retry after a partial failure otherwise double-counts a day of revenue in the rollup.

SRE tip: Alert on job absence, not just failure: a dead cron daemon or an unreachable Trino server produces no error line to match on. A freshness check on max(day) in the reporting table catches every silent failure mode with one query.

Verification

Confirm catalogs, shard connectivity, and end-to-end federation from the Trino CLI:

trino --server http://localhost:8080 --execute "SHOW CATALOGS"

trino --server http://localhost:8080 --execute "
  SELECT shard, count(*) AS rows
  FROM federated.views.all_orders
  GROUP BY shard ORDER BY shard"

Expected output for the first command: federated, shard1, shard2, shard3, plus the built-in system catalog. Expected output for the second: one row per shard with plausible row counts, for example:

 shard  |  rows
--------+---------
 shard1 | 4128390
 shard2 | 4090217
 shard3 | 4211045

A missing shard row means that branch of the view failed silently at creation time; a catalog listed but erroring on query means credentials or pg_hba.conf on that shard — test it in isolation with SELECT 1 FROM shardN.public.orders LIMIT 1.

Failure mode table

Failure mode Root cause SRE mitigation
Query exceeded per-node memory limit on federated aggregations Final GROUP BY over the UNION ALL (or a cross-catalog join) materialises too many rows in Trino’s heap Raise query.max-memory-per-node within heap headroom; pre-filter branches on date ranges; materialise one side of cross-catalog joins first
Shards show sequential-scan load spikes during Trino queries Predicate pushdown lost — functions wrapping filter columns, type mismatches, or collation differences keep filters in Trino Check EXPLAIN for filters inside each TableScan; rewrite predicates as bare column comparisons; add matching indexes on shards
Shard max_connections exhaustion during dashboard peaks Every concurrent Trino query opens fresh JDBC connections to every catalog, bypassing app-side pooling Front each shard with PgBouncer in transaction mode and point catalogs at the pooler; cap concurrent queries with Trino resource groups

FAQ

Does Trino push aggregations down to the PostgreSQL shards?

Filters, projections, LIMIT, and many single-table aggregations are pushed into each catalog’s remote SQL when aggregation-pushdown.enabled=true. Once branches are combined with UNION ALL, though, the final GROUP BY across branches always runs inside Trino, and joins between two different catalogs are never pushed down — both sides travel over JDBC and join in Trino memory. Run EXPLAIN and read the remote query inside each TableScan node before assuming a query is cheap.

How much memory does a single-node Trino need for shard federation?

Size the JVM heap to roughly 70–80 percent of the container’s memory and set query.max-memory-per-node to about half the heap. For aggregation-heavy federation over four to eight shards, a 16 GB heap is a comfortable starting point. What blows the budget is not shard count but unfiltered cross-catalog joins — those materialise entire tables in heap, and no amount of shard-side indexing helps.

Can Trino write aggregation results back into PostgreSQL?

Yes. The PostgreSQL connector supports CREATE TABLE AS and INSERT, so a scheduled Trino query can materialise a cross-shard rollup into a reporting schema on one designated shard, as in Step 6. Use a dedicated writer role with privileges limited to the reporting schema, and keep result sets modest — Trino writes row batches over JDBC, so it is a rollup channel, not a bulk-load pipeline.

Back to Federated Query Execution