“

Series: S8 Technology Deep Dives · Article 19 | Level: Advanced | Reading Time: 20 min

Trino Query Federation Deep Dive: Cross-Engine Unified Queries

#TL;DR

Trino is coomia-dip Data Layer's federation query engine, enabling unified SQL queries across Iceberg (Lakehouse), Doris (OLAP), and PostgreSQL (metadata)
This article analyzes Trino's Coordinator/Worker architecture, Connector plugin system, query plan optimization (CBO/predicate pushdown/projection pushdown), and dynamic filtering with runtime adaptivity
Covers coomia-dip's 3 Catalog configurations, cross-engine Join optimization, memory management with spilling, fault tolerance, and performance tuning

#1. Trino in coomia-dip

#1.1 Federation Query Architecture

Code

┌────────────────────────────────────────────────┐
│                 Trino Cluster                   │
│  ┌──────────────────────────────────────────┐  │
│  │            Coordinator                    │  │
│  │  SQL Parser → Planner → Optimizer        │  │
│  └──────────────┬───────────────────────────┘  │
│      ┌──────────┼──────────┐                   │
│  ┌───┴───┐  ┌───┴───┐  ┌───┴───┐             │
│  │Worker1│  │Worker2│  │Worker3│             │
│  └───┬───┘  └───┬───┘  └───┬───┘             │
└──────┼──────────┼──────────┼──────────────────┘
  ┌────┴───┐ ┌───┴────┐ ┌───┴──────┐
  │Iceberg │ │ Doris  │ │PostgreSQL│
  └────────┘ └────────┘ └──────────┘

#1.2 coomia-dip Catalog Configuration

PROPERTIES

# catalog/iceberg.properties
connector.name=iceberg
iceberg.catalog.type=nessie
iceberg.catalog.uri=http://nessie-server:19120/api/v2

# catalog/doris.properties
connector.name=jdbc
connection-url=jdbc:mysql://doris-fe:9030/ontology

# catalog/metadata.properties
connector.name=postgresql
connection-url=jdbc:postgresql://pg-metadata:5432/coomia-dip

#2. Coordinator-Worker Architecture

#2.1 Query Execution Flow

Code

Client → SQL → Coordinator
  1. SQL Parser (ANTLR) → AST
  2. Analyzer → Resolve tables/columns/types via Catalogs
  3. Planner → Logical plan + RBO
  4. Optimizer → CBO + Predicate/Projection pushdown + Join reorder
  5. Fragmenter → Split plan into Fragments for Workers
  6. Scheduler → Schedule Fragment execution, collect results

Worker: Execute Fragment → Read from Connector → Process in memory → Send results

#2.2 Resource Configuration

PROPERTIES

# Coordinator
coordinator=true
query.max-memory=50GB
query.max-memory-per-node=10GB

# Worker
coordinator=false
query.max-memory-per-node=10GB

#3. Connector Plugin System

Java

public interface Connector {
    ConnectorMetadata getMetadata();          // Schema/Table metadata
    ConnectorSplitManager getSplitManager();  // Data splits
    ConnectorPageSourceProvider getPageSourceProvider();  // Read data
    ConnectorPageSinkProvider getPageSinkProvider();      // Write data
}

#4. Query Planning and Optimization

#4.1 Predicate Pushdown

SQL

SELECT e.name, d.department_name
FROM iceberg.ontology.employee e
JOIN doris.ontology.department d ON e.department_id = d.id
WHERE e.world_id = 'world-001' AND e.salary > 100000 AND d.location = 'Shanghai'

-- Optimized: predicates pushed to each data source
-- Iceberg: WHERE world_id = 'world-001' AND salary > 100000
-- Doris: WHERE location = 'Shanghai'

#4.2 Cost-Based Optimization (CBO)

SQL

EXPLAIN (TYPE DISTRIBUTED)
SELECT e.name, d.department_name
FROM iceberg.ontology.employee e
JOIN doris.ontology.department d ON e.department_id = d.id;

-- CBO auto-selects: small table → Broadcast Join, large tables → Distributed Hash Join
ANALYZE iceberg.ontology.employee;  -- Collect statistics

#5. Cross-Engine Join Optimization

#5.1 Join Strategies

Strategy	Use Case	Mechanism
Broadcast Join	One side small (< 100MB)	Broadcast small table to all Workers
Partitioned Join	Both sides large	Hash partition by Join key

#5.2 Cross-Engine Join Example

SQL

SELECT e.name, e.hire_date, d.department_name, m.total_sales
FROM iceberg.ontology.employee e
JOIN metadata.ontology.department d ON e.department_id = d.id
JOIN doris.analytics.monthly_sales m ON e.id = m.employee_id
WHERE e.world_id = 'world-001' AND m.month = '2026-03'
ORDER BY m.total_sales DESC LIMIT 100;

#5.3 Join Hints

SQL

SELECT /*+ BROADCAST(d) */ e.name, d.department_name
FROM iceberg.ontology.employee e
JOIN doris.ontology.department d ON e.department_id = d.id;

#6. Dynamic Filtering

SQL

-- Trino automatically applies dynamic filtering:
-- 1. Scan small Doris department table first, get Shanghai department_ids
-- 2. Use department_ids as runtime filter
-- 3. When scanning Iceberg employee table, skip non-matching partitions/files
-- → Dramatically reduces Iceberg scan volume

PROPERTIES

enable-dynamic-filtering=true
dynamic-filtering-max-per-driver-row-count=1000000
dynamic-filtering-max-per-driver-size=10MB

#7. Memory Management and Spilling

PROPERTIES

query.max-memory=50GB
query.max-memory-per-node=10GB
spill-enabled=true
spiller-spill-path=/data/trino/spill
spiller-max-used-space-threshold=0.9

# Spill-supported operations: Hash Join, Aggregation, Sort, Window Functions

#8. Fault Tolerance

PROPERTIES

# Task-level retry (Trino 400+)
retry-policy=TASK
task-retry-attempts-per-task=2
retry-initial-delay=10s
fault-tolerant-execution-target-task-input-size=256MB

Python

class TrinoQueryClient:
    async def execute_with_retry(self, sql: str, max_retries: int = 3):
        for attempt in range(max_retries):
            try:
                return await self._execute(sql)
            except TrinoExternalError as e:
                if attempt < max_retries - 1 and e.error_name in RETRIABLE_ERRORS:
                    await asyncio.sleep(2 ** attempt)
                    continue
                raise

#9. Security and Isolation

#9.1 Multi-World Isolation

JSON

{
  "tables": [
    {
      "catalog": "iceberg",
      "schema": "world_(.*)",
      "table": ".*",
      "privileges": ["SELECT"],
      "filter": "world_id = '${USER}'"
    }
  ]
}

#9.2 Resource Groups

JSON

{
  "rootGroups": [
    {"name": "interactive", "hardConcurrencyLimit": 20, "softMemoryLimit": "40%"},
    {"name": "batch", "hardConcurrencyLimit": 10, "softMemoryLimit": "60%"}
  ]
}

#10. Performance Tuning

PROPERTIES

join-distribution-type=AUTOMATIC
join-reordering-strategy=AUTOMATIC
task.concurrency=16
iceberg.split-size=128MB

SQL

EXPLAIN ANALYZE SELECT ... -- View detailed execution statistics
-- Key metrics: Input/Output rows, Physical bytes, Wall time, Peak memory

#11. Key Takeaways

Topic	Key Conclusion
Positioning	Federation across Iceberg/Doris/PostgreSQL
Architecture	Coordinator plans + Workers execute in parallel
Pushdown	Predicates, projections, aggregations pushed to data sources
Joins	CBO auto-selects strategy; small tables use Broadcast
Dynamic filtering	Runtime Join results fed back as scan filters
Memory	Spill to disk prevents OOM
Fault tolerance	Task-level retry with intermediate result recovery
Isolation	Resource groups + row-level filtering for World isolation

“

Next up: S8-20 dives into nsjail sandbox, exploring secure user-defined function execution in coomia-dip.