How Iceberg Query Engines Work Internally

What this guide covers: When you run SELECT * FROM customers WHERE country = 'US' against an Iceberg table, what actually happens? This guide traces every step — from the query engine contacting the catalog, to reading metadata from S3, to scanning only the relevant Parquet files. We cover Trino, Spark, and the REST Catalog protocol in detail.

The Big Picture: What Happens When You Query an Iceberg Table?

Every Iceberg query follows the same fundamental pattern, regardless of which engine you use. Understanding this pattern is the key to understanding Iceberg internals.

The 5-Step Query Flow (Every Engine Follows This)

Key insight: Steps 1-4 are metadata operations. The actual data reading only happens in Step 5. This is why Iceberg can skip 90%+ of data files — it knows exactly which files contain relevant data before reading any of them.

The Iceberg REST Catalog Protocol

The REST Catalog is the standard way for query engines to interact with Iceberg catalogs. It’s a language-agnostic HTTP API that replaced the older pluggable Java-only catalog approach.

Why REST Catalog Exists

Before the REST protocol, every catalog (Hive Metastore, AWS Glue, Nessie) needed a separate client implementation in every language (Java, Python, Go, Rust). This created a combinatorial explosion:

Old approach (pluggable catalogs):
  Trino (Java) → needs: HiveCatalog.java, GlueCatalog.java, NessieCatalog.java
  Spark (Java) → needs: HiveCatalog.java, GlueCatalog.java, NessieCatalog.java
  PyIceberg (Python) → needs: hive_catalog.py, glue_catalog.py, nessie_catalog.py
  
  = 3 engines × 3 catalogs × custom code each = 9 implementations

REST approach:
  Trino → RESTCatalog (HTTP client)
  Spark → RESTCatalog (HTTP client)
  PyIceberg → RESTCatalog (HTTP client)
  
  = 3 engines × 1 standard HTTP client = 3 implementations
  Catalog server handles all backend logic

REST Catalog API — The Key Endpoints

Here are the critical REST API endpoints that every query touches:

Configuration & Authentication

# Step 0: Get catalog configuration
GET /v1/config
→ Returns: warehouse location, default settings, overrides

# Step 0b: Get OAuth token (if using OAuth2)
POST /v1/oauth/tokens
Body: { "grant_type": "client_credentials", "client_id": "...", "client_secret": "..." }
→ Returns: { "access_token": "...", "token_type": "bearer", "expires_in": 3600 }

Namespace & Table Operations

# List namespaces (databases)
GET /v1/{prefix}/namespaces
→ Returns: { "namespaces": [["analytics"], ["raw_data"]] }

# Load a table (THE KEY OPERATION)
GET /v1/{prefix}/namespaces/{namespace}/tables/{table}
→ Returns: {
    "metadata-location": "s3://bucket/db/table/metadata/v42.metadata.json",
    "metadata": { ... full table metadata ... },
    "config": {
      "s3.access-key-id": "...",        ← credential vending!
      "s3.secret-access-key": "...",
      "s3.session-token": "..."
    }
  }

Credential Vending — The Security Model

One of the REST Catalog’s most powerful features is credential vending. Instead of the query engine having permanent S3 credentials, the catalog issues short-lived, scoped credentials:

Why this matters:
  ✅ Engine never has broad S3 access
  ✅ Credentials expire automatically (15-60 min)
  ✅ Access is scoped to specific table paths
  ✅ Catalog controls WHO can access WHAT
  ✅ No static credentials stored in engine config

Table Commits (Writes)

# Commit a table update (atomic metadata swap)
POST /v1/{prefix}/namespaces/{namespace}/tables/{table}
Body: {
  "requirements": [
    { "type": "assert-current-schema-id", "current-schema-id": 1 },
    { "type": "assert-table-uuid", "uuid": "abc-123" }
  ],
  "updates": [
    { "action": "add-snapshot", "snapshot": { ... } },
    { "action": "set-current-schema", "schema-id": 2 }
  ]
}
→ Returns: updated table metadata (or 409 Conflict if requirements fail)

The commit protocol is change-based, not state-based. The engine sends a list of requirements (preconditions) and updates (changes). The catalog server validates requirements atomically and applies updates. If another writer modified the table, the commit fails with a 409 Conflict and the engine retries.

How Trino Queries Iceberg Tables

Trino is a distributed SQL query engine with a coordinator-worker architecture. Here’s exactly what happens when you run an Iceberg query in Trino.

Trino Architecture Overview

Step-by-Step: Trino Query Execution on Iceberg

Let’s trace what happens for this query:

SELECT customer_id, name, total_spent
FROM iceberg.analytics.customers
WHERE country = 'US' AND signup_date > DATE '2024-01-01'
ORDER BY total_spent DESC
LIMIT 100;

Phase 1: Query Parsing & Planning (Coordinator)

What the coordinator does:

1. Parses the SQL into an Abstract Syntax Tree (AST)
2. Resolves the table reference iceberg.analytics.customers through the Iceberg connector
3. Fetches table metadata from the catalog (schema, partition spec, current snapshot)
4. Creates a logical plan and optimizes it
5. Pushes down predicates to the Iceberg connector — this is critical for performance

Phase 2: Split Generation (Coordinator)

This is where Iceberg’s metadata really shines. The coordinator uses Iceberg’s metadata to determine exactly which files to read.

Pruning example:
  Table has 500 data files across all partitions
  
  Manifest list pruning:
    → 100 manifests → 15 manifests (country='US' partition range)
    
  Manifest file pruning (partition values):
    → 150 data files → 40 data files (exact country='US' match)
    
  Column statistics pruning (min/max on signup_date):
    → 40 data files → 3 data files (signup_date > 2024-01-01)
    
  Result: Read 3 files instead of 500 = 99.4% data skipped

Each surviving data file becomes a “split” — a unit of work that Trino assigns to a worker.

Phase 3: Distributed Execution (Workers)

Coordinator assigns splits to workers:
  Worker 1 → data-007.parquet (rows 0-125,000)
  Worker 2 → data-012.parquet (rows 0-125,000)
  Worker 3 → data-019.parquet (rows 0-125,000)

Each worker independently:
  1. Opens S3 connection to its assigned Parquet file
  2. Reads Parquet footer to find column chunks
  3. Reads only needed columns: customer_id, name, total_spent
     (Parquet columnar format = skip unneeded columns)
  4. Applies remaining filters in-memory
  5. Streams results back to coordinator

Coordinator:
  1. Receives partial results from all workers
  2. Merges, sorts by total_spent DESC
  3. Takes top 100 rows
  4. Returns to client

Trino Iceberg Connector Configuration

# /etc/trino/catalog/iceberg.properties

# Using REST Catalog
connector.name=iceberg
iceberg.catalog.type=rest
iceberg.rest-catalog.uri=https://polaris.example.com/api/catalog
iceberg.rest-catalog.warehouse=analytics_warehouse

# Using Hive Metastore
connector.name=iceberg
iceberg.catalog.type=hive_metastore
hive.metastore.uri=thrift://hive-metastore:9083

# Using AWS Glue
connector.name=iceberg
iceberg.catalog.type=glue

Trino-Specific Iceberg Behaviors

How Spark Queries Iceberg Tables

Spark has a fundamentally different architecture from Trino. It’s a distributed compute engine with a driver-executor model, and uses the DataSource V2 API to integrate with Iceberg.

Spark Architecture Overview

Step-by-Step: Spark Query Execution on Iceberg

# The query
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .config("spark.sql.catalog.my_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.my_catalog.type", "rest") \
    .config("spark.sql.catalog.my_catalog.uri", "https://polaris.example.com/api/catalog") \
    .config("spark.sql.catalog.my_catalog.warehouse", "analytics_warehouse") \
    .getOrCreate()

df = spark.sql("""
    SELECT customer_id, name, total_spent
    FROM my_catalog.analytics.customers
    WHERE country = 'US' AND signup_date > '2024-01-01'
    ORDER BY total_spent DESC
    LIMIT 100
""")
df.show()

Phase 1: Catalog Resolution (Driver)

Phase 2: Scan Planning via DataSource V2 (Driver)

Spark uses the DataSource V2 API — Iceberg implements this interface to control how scans are planned:

Spark calls Iceberg's DataSource V2 implementation:

1. SparkScanBuilder.build()
   → IcebergScan created with pushed-down filters

2. IcebergScan.planInputPartitions()
   → Iceberg library reads manifest list from S3
   → Reads manifest files from S3
   → Applies partition pruning (country = 'US')
   → Applies column statistics pruning (signup_date > 2024-01-01)
   → Returns list of IcebergInputPartition objects
   
3. Each IcebergInputPartition maps to:
   → One or more Parquet files on S3
   → A specific set of columns to read
   → Residual filters to apply after read

The key difference from Trino: In Spark, the Iceberg Java library runs directly inside the Driver JVM. There’s no separate connector process — Iceberg’s scan planning code is embedded in the Spark application.

Phase 3: Task Execution (Executors)

Spark-Specific Iceberg Behaviors

Spark Write Flow

When Spark writes to an Iceberg table, the flow reverses:

# Spark write
df.writeTo("my_catalog.analytics.customers").append()

Trino vs Spark: How They Differ with Iceberg

Deep Dive: Reading Parquet Files from S3

Both Trino and Spark ultimately read Parquet files from S3. Here’s exactly what happens at the storage level.

Parquet File Internal Structure

data-007.parquet (125,000 rows, 15 columns)
│
├── Row Group 0 (rows 0-65,535)
│   ├── Column Chunk: customer_id    [offset: 0,    length: 450KB]
│   ├── Column Chunk: name           [offset: 450K, length: 2.1MB]
│   ├── Column Chunk: email          [offset: 2.5M, length: 1.8MB]
│   ├── Column Chunk: country        [offset: 4.3M, length: 180KB]
│   ├── Column Chunk: signup_date    [offset: 4.5M, length: 290KB]
│   ├── Column Chunk: total_spent    [offset: 4.8M, length: 510KB]
│   └── ... (9 more columns)
│
├── Row Group 1 (rows 65,536-125,000)
│   └── ... (same column chunk structure)
│
└── Footer (at end of file)
    ├── Schema definition
    ├── Row group metadata
    │   ├── Column chunk locations (offset + size)
    │   └── Column statistics (min, max, null count)
    └── Key-value metadata

S3 Read Sequence

Key optimization — HTTP Range Requests:

S3 supports reading specific byte ranges from a file. The query engine uses this to:

1. Read just the footer (to discover column locations)
2. Read only the columns needed by the query
3. Skip entire row groups based on column statistics in the footer

Without column pruning:
  Read entire file: 15 MB × 500 files = 7.5 GB from S3

With Iceberg partition pruning + Parquet column pruning:
  Read 3 columns from 3 files: 3 MB × 3 files = 9 MB from S3
  
  Savings: 99.88% less data read from S3

S3 Request Costs

Every S3 GET request has a cost beyond just data transfer:

Query cost breakdown:
  Metadata reads: ~5 GET requests (metadata.json + manifest list + manifests)
  Data reads: 3 files × 5 range requests each = 15 GET requests
  Total: ~20 GET requests = $0.008 (less than 1 cent)
  
  Without Iceberg pruning:
  Data reads: 500 files × 5 range requests each = 2,500 GET requests
  Total: ~2,505 GET requests = $1.00
  
  125x cost reduction with Iceberg metadata pruning

Complete End-to-End Flow: Trino + REST Catalog + S3

Here’s the complete picture showing every network call for a single Trino query:

Total network calls for this query:

REST Catalog:  1 API call    (load table)
S3 Metadata:   3 GET requests (metadata.json + manifest list + 1 manifest)
S3 Data:      15 GET requests (3 files × 5 range requests each)
─────────────────────────────────
Total:        19 network calls
Time:         ~2-5 seconds (mostly S3 latency)
Data read:    ~9 MB (out of 7.5 GB total table size)

Complete End-to-End Flow: Spark + REST Catalog + S3

# Spark application
df = spark.sql("""
    SELECT customer_id, name, total_spent
    FROM my_catalog.analytics.customers
    WHERE country = 'US' AND signup_date > '2024-01-01'
""")
df.write.mode("overwrite").saveAsTable("my_catalog.analytics.us_customers")

Performance Tips: Reducing S3 Calls

Understanding the internal flow helps you optimize performance:

1. Partition Your Tables Wisely

-- Good: queries frequently filter by country and date
CREATE TABLE customers (
    customer_id BIGINT,
    name STRING,
    country STRING,
    signup_date DATE,
    total_spent DECIMAL(10,2)
)
USING iceberg
PARTITIONED BY (country, months(signup_date));

-- This means:
-- WHERE country = 'US' → skips all non-US partitions in manifests
-- WHERE signup_date > '2024-01-01' → skips old month partitions

2. Keep Metadata Compact

-- Too many small files = too many manifest entries = slow scan planning
-- Run compaction regularly
CALL system.rewrite_data_files('analytics.customers');
CALL system.rewrite_manifests('analytics.customers');

-- Expire old snapshots to reduce manifest list size
CALL system.expire_snapshots('analytics.customers', TIMESTAMP '2024-06-01 00:00:00');

3. Use Column Statistics

-- Iceberg collects min/max stats per column per file
-- This enables file-level pruning even without partitioning
-- Make sure stats collection is enabled (it is by default)
ALTER TABLE customers SET TBLPROPERTIES (
    'write.metadata.metrics.default' = 'truncate(16)'
);

4. Co-locate Compute and Storage

Bad:  Trino in us-east-1, S3 bucket in eu-west-1
      → Every S3 GET has cross-region latency + $0.09/GB transfer
      
Good: Trino in us-east-1, S3 bucket in us-east-1
      → Low latency, free data transfer

Summary

What Happens When You Query Iceberg

1. ENGINE → CATALOG: "Where is the table?"
   Response: metadata file location + credentials

2. ENGINE → S3: Read metadata.json
   Response: current snapshot, schema, partition spec

3. ENGINE → S3: Read manifest list (Avro)
   Response: list of manifests with partition bounds

4. ENGINE → S3: Read relevant manifests (Avro)
   Response: list of data files with column stats
   (engine prunes files based on query predicates)

5. ENGINE → S3: Read data files (Parquet)
   Response: actual column data
   (engine reads only needed columns via range requests)

Key Takeaways

• REST Catalog is a standard HTTP API — any engine that speaks HTTP can query Iceberg tables
• Trino does scan planning on the coordinator, distributes splits to stateless workers
• Spark does scan planning on the driver via DataSource V2, distributes tasks to executors
• Both engines use Iceberg’s metadata tree (metadata.json → manifest list → manifests) to prune files before reading any data
• Parquet range requests mean engines read only the columns they need, not entire files
• Credential vending from REST Catalog means engines get short-lived, scoped S3 credentials
• The real performance win: Iceberg’s metadata lets engines skip 90-99% of data files, turning a full table scan into a targeted read of a few files

Further Reading

• Iceberg REST Catalog Spec
• Trino Iceberg Connector Docs
• Trino on Ice: Deep Dive into Iceberg Internals
• Apache Iceberg Table Spec
• Spark Iceberg Integration
• Apache Polaris Catalog