Trino Federated Query Architecture

Trino Federated Query Architecture

When to Use

• Joining data from multiple systems (data lake + operational DB + Kafka) in a single query
• Deciding whether to use live federation or pre-materialization
• Designing a cross-domain reporting layer without moving data
• Understanding performance implications before writing a cross-catalog query

Federation Architecture

┌──────────────────────────────────────────────────────────────────┐
│                     Trino Federated Query Layer                  │
│                                                                  │
│  SELECT f.amount, c.region, p.name                               │
│  FROM iceberg.gold.fact_orders f                                 │
│  JOIN postgresql.public.customers c ON f.customer_id = c.id     │
│  JOIN mysql.inventory.products p ON f.product_id = p.id         │
│                                                                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐           │
│  │  Iceberg     │  │ PostgreSQL   │  │   MySQL      │           │
│  │  (S3/Parquet)│  │ (JDBC)       │  │  (JDBC)      │           │
│  └──────────────┘  └──────────────┘  └──────────────┘           │
└──────────────────────────────────────────────────────────────────┘

Connector Performance Characteristics

Understanding how each connector fetches data is essential for correct federation design:

Key insight: Iceberg/Hive reads scale with worker count. JDBC connectors (PostgreSQL, MySQL) make a single serial connection — Trino fetches all rows then filters/joins in memory. For large JDBC tables, always push predicates down.

Cross-Catalog JOIN Patterns

Pattern 1: Small Dimension from PostgreSQL (Broadcast)

-- GOOD: PostgreSQL table is small (< 100MB), broadcasts to all workers
-- Trino fetches entire table from PG into worker memory, then broadcasts
SELECT
    f.order_id,
    f.amount,
    c.region,
    c.tier
FROM iceberg.silver.orders f
JOIN postgresql.public.customers c
  ON f.customer_id = c.id
WHERE f.order_date = DATE '2024-06-01'   -- predicate pushed into Iceberg (partition prune)
  AND c.tier = 'premium';               -- predicate pushed into PostgreSQL query

-- EXPLAIN shows: BROADCAST exchange for the PG join

Pattern 2: Large JDBC Table → Pre-Materialize First

-- BAD: joining 50M-row PG table directly — Trino fetches 50M rows serially over JDBC
SELECT f.amount, p.product_name
FROM iceberg.silver.orders f
JOIN postgresql.catalog.products p ON f.product_id = p.id;   -- 50M rows from PG!

-- GOOD: materialize the large PG table into Iceberg daily
CREATE TABLE iceberg.silver.products_snapshot
WITH (format = 'PARQUET', compression_codec = 'ZSTD')
AS SELECT id, product_name, category, price, updated_at
   FROM postgresql.catalog.products;

-- Now join two Iceberg tables (parallel, pushdown, fast)
SELECT f.amount, p.product_name
FROM iceberg.silver.orders f
JOIN iceberg.silver.products_snapshot p ON f.product_id = p.id;

Pattern 3: Kafka Event Preview (Read-Once)

-- Query Kafka topic as a table (reads current offset window)
-- Note: results are non-repeatable — each query reads different offsets
SELECT
    _timestamp,
    json_extract_scalar(cast(_message as VARCHAR), '$.event_type') AS event_type,
    json_extract_scalar(cast(_message as VARCHAR), '$.user_id')    AS user_id
FROM kafka.default."prod.user.events.v1"
WHERE _timestamp > NOW() - INTERVAL '1' HOUR
LIMIT 1000;

Pattern 4: Cross-Schema Enrichment (Same Catalog)

-- Cross-schema within same Iceberg catalog — same connector, full pushdown
SELECT
    o.order_id,
    o.amount,
    c.name     AS customer_name,
    c.region,
    p.product_name
FROM iceberg.silver.orders o
JOIN iceberg.silver.customers c ON o.customer_id = c.customer_id
JOIN iceberg.gold.dim_product p ON o.product_id  = p.product_id
WHERE o.order_date = DATE '2024-06-01';

Predicate Pushdown Strategy

Always write predicates to enable connector-level filtering:

-- GOOD: predicate on PG side is pushed down (becomes WHERE in PG query)
SELECT * FROM postgresql.public.orders
WHERE status = 'shipped'        -- pushed to PG: SELECT * WHERE status='shipped'
  AND created_at > NOW() - INTERVAL '7' DAY;

-- VERIFY: EXPLAIN (TYPE IO) shows constraint on columns
EXPLAIN (TYPE IO)
SELECT * FROM postgresql.public.orders WHERE status = 'shipped';
-- → "constraint": {"columnConstraints": [{"columnName": "status", "domain": "shipped"}]}

PostgreSQL connector limitation: range predicates on VARCHAR columns are NOT pushed down (only equality):

-- This pushes down (equality on VARCHAR):
WHERE country = 'US'

-- This does NOT push down (range on VARCHAR):
WHERE country BETWEEN 'A' AND 'M'   -- full table scan in PG, filter in Trino

Materialization Decision Matrix

Snapshot Pattern: JDBC → Iceberg

-- Daily incremental snapshot (append new/changed rows)
CREATE TABLE IF NOT EXISTS iceberg.silver.pg_orders_snapshot (
    order_id    BIGINT,
    customer_id BIGINT,
    status      VARCHAR,
    amount      DECIMAL(18,2),
    updated_at  TIMESTAMP(6)
) WITH (
    format       = 'PARQUET',
    partitioning = ARRAY['day(updated_at)']
);

-- Incremental load (Airflow task, run daily)
INSERT INTO iceberg.silver.pg_orders_snapshot
SELECT order_id, customer_id, status, CAST(amount AS DECIMAL(18,2)), updated_at
FROM postgresql.public.orders
WHERE updated_at >= TIMESTAMP '{{ yesterday_ds }} 00:00:00'
  AND updated_at <  TIMESTAMP '{{ ds }} 00:00:00';

Multi-Source Reporting Query

-- Report combining Iceberg lake + PostgreSQL OLTP + ClickHouse analytics
-- All predicates pushed to respective connectors
SELECT
    o.order_date,
    c.region                           AS customer_region,
    p.category                         AS product_category,
    COUNT(o.order_id)                  AS order_count,
    SUM(o.amount)                      AS revenue,
    MAX(ch.page_views)                 AS peak_page_views
FROM iceberg.silver.orders o                               -- Iceberg: parallel scan
JOIN postgresql.public.customers c                         -- PG: small table, broadcast
  ON o.customer_id = c.id
JOIN iceberg.gold.dim_product p                            -- Iceberg: broadcast
  ON o.product_id = p.product_id
LEFT JOIN clickhouse.analytics.daily_traffic ch            -- ClickHouse: aggregated
  ON o.order_date = ch.event_date AND c.region = ch.region
WHERE o.order_date BETWEEN DATE '2024-01-01' AND DATE '2024-03-31'
  AND c.tier IN ('premium', 'enterprise')
GROUP BY o.order_date, c.region, p.category
ORDER BY revenue DESC
LIMIT 100;

Catalog Isolation Design

Separate catalogs by domain and access scope:

etc/catalog/
├── iceberg.properties          → lake data (Iceberg on S3)
├── iceberg_external.properties → external Iceberg (read-only partner data)
├── postgresql_orders.properties → orders OLTP DB (read-only replica)
├── postgresql_users.properties  → users OLTP DB (read-only replica)
├── clickhouse.properties        → analytics ClickHouse
├── kafka.properties             → event streaming (read-only)
└── tpch.properties              → TPC-H test data

Use resource groups to restrict which users can query which catalogs:

{
  "selectors": [
    {
      "user": ".*",
      "queryType": "SELECT",
      "queryText": ".*postgresql_orders.*",
      "group": "global.readonly.readonly_${USER}"
    }
  ]
}

Anti-Patterns

1. Joining two large JDBC tables cross-connector — both tables are fetched serially and joined in Trino memory; a 10M×5M JDBC join moves 15M rows over JDBC before any join; always materialize at least one side into Iceberg.
2. No predicate on JDBC side in cross-catalog query — Trino fetches the entire JDBC table when predicates can't be pushed; always include at least one equality predicate on a JDBC table.
3. Querying Kafka topics in production reports — Kafka connector reads are non-repeatable and non-deterministic; use for exploration only; for production, sink Kafka to Iceberg via Flink/Debezium.
4. No metadata caching on JDBC catalogs — each query does schema/stats lookups; set metadata.cache-ttl=5m in JDBC catalog config to avoid thundering herd of metadata requests.
5. Cross-catalog JOINs in scheduled dbt models — dbt incremental on top of JDBC sources creates large full-table scans per run; snapshot JDBC sources into Iceberg first, then run dbt transformations on Iceberg only.

References

• Trino connectors: trino.io/docs/current/connector.html
• PostgreSQL connector: trino.io/docs/current/connector/postgresql.html
• Iceberg connector: trino.io/docs/current/connector/iceberg.html
• Kafka connector: trino.io/docs/current/connector/kafka.html
• Related skills: [[trino-lakehouse-platform-architect]], [[trino-query-optimization]], [[trino-iceberg-best-practices]]