SQE — Sovereign Query Engine

SQE is a Rust-based distributed SQL query engine for Apache Iceberg tables. It replaces a patched Trino fork with a purpose-built engine based on Apache DataFusion and iceberg-rust.

graph LR
    Client["JDBC / Flight SQL Client"] --> Coordinator
    Coordinator --> Worker1["Worker 1"]
    Coordinator --> Worker2["Worker 2"]
    Coordinator --> WorkerN["Worker N"]
    Worker1 --> S3["S3 / MinIO"]
    Worker2 --> S3
    WorkerN --> S3
    Coordinator --> Polaris["Polaris Catalog"]
    Coordinator --> Keycloak["Keycloak OIDC"]

Key Properties

• No service account — every query runs as the authenticated user. Bearer tokens pass through from client to Polaris catalog and S3 storage.
• Arrow-native — columnar data flows from Parquet files through the entire query pipeline to the client. No row-based serialization anywhere.
• Iceberg-native — built on iceberg-rust, not a connector bolted onto a generic engine. Partition pruning, metadata caching, and Iceberg v3 support are first-class.
• Fine-grained security — row filters and column masks enforced at the logical plan level, before the optimizer runs. Invisible columns, transparent row filtering, no information leakage.
• Rust performance — single binary, no JVM, no GC pauses, predictable memory usage, fast startup.

Quick Start (embedded, no server)

cargo install --path crates/sqe-cli
sqe-cli --embedded                    # ~/.sqe/warehouse persistent Iceberg catalog

sqe> SELECT * FROM '/data/sales.parquet' LIMIT 5;
sqe> SELECT * FROM read_csv('s3://bucket/orders.tsv.gz');
sqe> SELECT * FROM 'hf://datasets/squad/plain_text/train-00000-of-00001.parquet';
sqe> SELECT * FROM read_delta('/data/delta/sales', version => '5');

Full embedded reference: cli-embedded.md. DuckDB comparison: duckdb-comparision.md.

Quick Start (cluster mode)

# Build
cargo build --release --bin sqe-coordinator --bin sqe-cli

# Start coordinator
SQE_CONFIG=sqe.toml ./target/release/sqe-coordinator

# Connect
./target/release/sqe-cli --host localhost --port 50051

Project Status

SQE is production-ready against Apache Iceberg. The cluster mode runs distributed (coordinator + stateless workers) with OIDC bearer-token passthrough, Polaris / Nessie / Glue / HMS / S3 Tables / JDBC / Hadoop catalogs, and 167/189 (88.4%) on the public Iceberg matrix scoreboard. The embedded mode (V8 through V12.1) adds DuckDB-style file-format TVFs (read_csv, read_json, read_delta), HuggingFace hf:// URLs, and a single-binary CLI for laptop analytics.

Why SQE

The Problem

Our data platform runs on Apache Iceberg tables stored in S3, cataloged by Apache Polaris (Iceberg REST Catalog), with authentication through Keycloak OIDC. We need a SQL query engine that can:

1. Authenticate users through Keycloak
2. Pass the user’s bearer token through to Polaris and S3 (no service account)
3. Enforce fine-grained security (row filters, column masks) per user
4. Support full SQL for analytics, dbt transformations, and ad-hoc queries
5. Run on Kubernetes with minimal operational overhead

Why Not Trino?

We started with Trino — the industry-standard SQL engine for data lakehouses. It works, but:

Why DataFusion + Rust?

graph TB
    subgraph "Old: Trino Fork"
        T[Trino JVM] -->|service account| P1[Polaris]
        T -->|service account| S1[S3]
        T -.- Fork[DCAF Fork<br/>maintenance burden]
    end

    subgraph "New: SQE"
        C[SQE Coordinator<br/>Rust binary] -->|user bearer token| P2[Polaris]
        W[SQE Workers] -->|user credentials| S2[S3]
        C --> W
    end

    style Fork fill:#f96,stroke:#333
    style C fill:#6f9,stroke:#333
    style W fill:#6f9,stroke:#333

Apache DataFusion is a Rust-native query engine that gives us:

• Extensible query planning — we can inject security filters into the LogicalPlan before optimization, which is exactly where row filters and column masks need to go
• iceberg-rust integration — native Rust Iceberg library, no JNI bridge, no serialization overhead
• Per-query context — each query gets its own SessionContext with the user’s bearer token. No shared service account.
• Single binary — the coordinator and worker ship as one ~50MB binary. Starts in milliseconds.
• No GC — predictable latency, no stop-the-world pauses during large scans

The Name

Sovereign — because every query runs with the identity and permissions of the user who submitted it. No service account intermediary. No privilege escalation. The user’s token is sovereign.

From Trino to DataFusion

Architecture Comparison

graph LR
    subgraph Trino
        TC[Coordinator<br/>JVM ~2GB heap] --> TW1[Worker<br/>JVM ~8GB heap]
        TC --> TW2[Worker<br/>JVM ~8GB heap]
        TC -->|Hive Metastore<br/>protocol| HMS[Hive Metastore<br/>or Polaris]
        TW1 -->|service account| TS3[S3]
        TW2 -->|service account| TS3
    end

    subgraph SQE
        SC[sqe-server<br/>coordinator<br/>~50MB binary] --> SW1[sqe-server<br/>worker]
        SC --> SW2[sqe-server<br/>worker]
        SC -->|user bearer token<br/>Iceberg REST| POL[Polaris]
        SW1 -->|user credentials| SS3[S3]
        SW2 -->|user credentials| SS3
    end

What Changes

What Stays the Same

• Apache Iceberg as the table format
• Apache Polaris as the REST catalog
• Keycloak as the identity provider
• S3 as the storage layer
• dbt as the transformation framework (new native adapter instead of Trino adapter)
• JDBC connectivity (via Arrow Flight SQL JDBC driver instead of Trino JDBC)

Migration Path

SQE includes an optional Trino-compatible HTTP endpoint (/v1/statement) that speaks enough of the Trino wire protocol to support existing dashboards and tools during the migration period. This is not a full Trino emulation — it covers SELECT, SHOW, and basic DDL, enough to keep things running while teams migrate to Flight SQL.

timeline
    title Migration Timeline
    Phase 1 : SQE single-node : Flight SQL : CLI
    Phase 2 : Write path : dbt-sqe adapter : Views
    Phase 3 : Distributed execution : Workers
    Phase 4 : Trino compat layer : Dashboard migration
    Phase 5 : Security policies : Row filters : Column masks
    Phase 6 : Decommission Trino fork

The Auth Challenge

The core design constraint that drove us to build SQE: no service account.

The Problem with Service Accounts

In a typical data platform, the query engine authenticates to the catalog and storage with a service account — a single identity with broad permissions. The engine then enforces per-user access control internally.

sequenceDiagram
    participant User
    participant Trino
    participant Polaris
    participant S3

    User->>Trino: Query (user token)
    Note over Trino: Validates user token
    Trino->>Polaris: List tables (SERVICE ACCOUNT)
    Polaris-->>Trino: Table metadata
    Trino->>S3: Read Parquet (SERVICE ACCOUNT IAM role)
    S3-->>Trino: Data
    Trino-->>User: Results

This means:

• Polaris sees one identity for all queries — audit logs show the service account, not the actual user
• S3 access is all-or-nothing — the service account can read everything, security depends entirely on the engine enforcing it correctly
• Credential rotation is a blast-radius event — rotating the service account key affects all users simultaneously
• Compliance gap — auditors want to see that Alice read table X, not that sqe-service-account did

SQE’s Approach: Bearer Token Passthrough

SQE never stores or uses a service account for data access. Instead, the user’s Keycloak bearer token flows through the entire stack:

sequenceDiagram
    participant User
    participant SQE as SQE Coordinator
    participant KC as Keycloak
    participant Polaris
    participant S3

    User->>SQE: Handshake (username, password)
    SQE->>KC: OIDC Password Grant
    KC-->>SQE: Access token + refresh token
    SQE-->>User: Session (bearer token)

    User->>SQE: Query (bearer token)
    SQE->>Polaris: List tables (USER's bearer token)
    Polaris-->>SQE: Table metadata + S3 credentials
    Note over Polaris: Polaris vends scoped S3<br/>credentials for THIS user
    SQE->>S3: Read Parquet (user-scoped credentials)
    S3-->>SQE: Data
    SQE-->>User: Arrow Flight results

Key Implications

Per-Session Catalog

Each user session gets its own SessionCatalog instance, initialized with the user’s bearer token:

graph TB
    subgraph "Session: alice"
        SC1[SessionCatalog<br/>token: alice_jwt] --> P[Polaris REST]
    end

    subgraph "Session: bob"
        SC2[SessionCatalog<br/>token: bob_jwt] --> P
    end

    P -->|alice's token| S3A[S3: alice sees<br/>tables A, B, C]
    P -->|bob's token| S3B[S3: bob sees<br/>tables A, B only]

Polaris enforces catalog-level access control based on the token. If Alice has access to tables A, B, C but Bob only has access to A and B, this is enforced at the catalog level — SQE doesn’t need to duplicate this logic.

Token Lifecycle

SQE manages token refresh transparently. A background task checks all active sessions every 10 seconds and refreshes tokens that are about to expire:

stateDiagram-v2
    [*] --> Active: Handshake (ROPC grant)
    Active --> Refreshing: Token expiry - 60s buffer
    Refreshing --> Active: New token from Keycloak
    Refreshing --> Expired: Refresh fails
    Active --> Expired: Session timeout
    Expired --> [*]: Session removed

    note right of Active: Queries use current token
    note right of Refreshing: Background task (10s interval)

The token fingerprint (last 8 characters of the access token) is used to invalidate iceberg-rust’s internal catalog session cache when a token is refreshed, ensuring the catalog client always uses the current token.

System Overview

Components

graph TB
    subgraph Clients
        JDBC["JDBC / ODBC<br/>(Flight SQL driver)"]
        CLI["sqe-cli"]
        DBT["dbt-sqe adapter"]
        DASH["Dashboards<br/>(Trino compat)"]
    end

    subgraph "SQE Cluster"
        subgraph "Coordinator (sqe-server --mode coordinator)"
            FLS["Flight SQL Server<br/>:50051"]
            TH["Trino HTTP<br/>:8080"]
            SM["Session Manager"]
            QH["Query Handler"]
            PE["Policy Enforcer"]
            SCHED["Scheduler"]
        end

        subgraph "Workers (sqe-server --mode worker)"
            W1["Worker 1<br/>DataFusion executor"]
            W2["Worker 2<br/>DataFusion executor"]
            WN["Worker N<br/>DataFusion executor"]
        end
    end

    subgraph "External Services"
        KC["OIDC provider<br/>(Keycloak / Auth0 / Entra)"]
        CAT["Catalog backend<br/>(Polaris / Nessie / Glue REST /<br/>S3 Tables / Unity / HMS / JDBC)"]
        S3["S3-compatible storage<br/>(AWS / Ceph / R2 / rustfs)"]
    end

    JDBC --> FLS
    CLI --> FLS
    DBT --> FLS
    DASH --> TH

    SM --> KC
    QH --> PE
    QH --> SCHED
    SCHED --> W1
    SCHED --> W2
    SCHED --> WN

    QH --> CAT
    W1 --> S3
    W2 --> S3
    WN --> S3

The catalog backend is selectable at runtime. Polaris is the primary target and the only one verified end-to-end for production write paths today. Nessie, AWS Glue, AWS S3 Tables, Unity Catalog OSS, Hive Metastore, JDBC (Postgres), and Hadoop storage-only are all reachable through the same iceberg::Catalog trait, with live integration tests in sqe-catalog/tests/backends_integration.rs. AWS endpoints share the OSS Iceberg REST code path through the aws-sigv4 cargo feature on the vendored iceberg-catalog-rest crate. See features/iceberg.md for the catalog-by-catalog state.

Request Flow

A query flows through SQE in these stages:

sequenceDiagram
    participant C as Client
    participant F as Flight SQL Server
    participant SM as Session Manager
    participant QH as Query Handler
    participant PE as Policy Enforcer
    participant DF as DataFusion
    participant CAT as Polaris Catalog
    participant S3 as S3 Storage

    C->>F: do_handshake(user, pass)
    F->>SM: authenticate(user, pass)
    SM->>SM: Keycloak OIDC → session
    F-->>C: bearer token

    C->>F: execute(SQL, token)
    F->>SM: get_session(token)
    F->>QH: execute(session, SQL)

    QH->>QH: parse & classify SQL
    QH->>CAT: create SessionCatalog(user_token)
    QH->>DF: plan SQL → LogicalPlan
    QH->>PE: enforce(user, plan)
    PE-->>QH: secured plan (row filters, column masks)
    QH->>DF: optimize & execute
    DF->>S3: read Parquet files
    S3-->>DF: Arrow RecordBatches
    DF-->>QH: results
    QH-->>F: RecordBatches
    F-->>C: Arrow Flight stream

Single-Node vs Distributed

SQE starts in single-node mode by default — the coordinator executes queries locally using DataFusion. No workers needed.

For larger deployments, enable workers:

graph LR
    subgraph "Single-Node (default)"
        C1[sqe-server] -->|local DataFusion| S1[S3]
    end

    subgraph "Distributed"
        C2[Coordinator] -->|plan fragments| W1[Worker 1]
        C2 --> W2[Worker 2]
        W1 --> S2[S3]
        W2 --> S2
    end

Ports

Coordinator

The coordinator is the brain of SQE. It handles SQL parsing, query planning, security enforcement, and result delivery. In single-node mode, it also executes queries directly.

Responsibilities

graph TB
    subgraph Coordinator
        FLS["Flight SQL Server"] --> SM["Session Manager"]
        FLS --> QH["Query Handler"]

        QH --> PARSE["SQL Parser<br/>(sqlparser-rs)"]
        QH --> CLASS["Statement Classifier"]
        QH --> PLAN["Query Planner<br/>(DataFusion)"]
        QH --> PE["Policy Enforcer"]
        QH --> CAT["Catalog Ops<br/>(DDL)"]
        QH --> WH["Write Handler<br/>(CTAS, INSERT,<br/>DELETE, UPDATE,<br/>MERGE — CoW)"]

        SM --> AUTH["Authenticator<br/>(Keycloak OIDC)"]
        SM --> TC["Token Cache"]

        PLAN --> DF["DataFusion<br/>SessionContext"]
        DF --> SC["SessionCatalog<br/>(per-user token)"]
    end

Statement Routing

The coordinator classifies every SQL statement and routes it to the appropriate handler:

Session Context

Each query gets a fresh DataFusion SessionContext with the user’s catalog:

sequenceDiagram
    participant QH as Query Handler
    participant SC as SessionCatalog
    participant POL as Polaris
    participant DF as DataFusion

    QH->>SC: new(catalog_url, warehouse, user_token)
    SC->>POL: list_namespaces() [user_token]
    POL-->>SC: [ns1, ns2, ns3]
    QH->>DF: register_catalog(warehouse, CatalogProvider)
    QH->>DF: sql("SELECT * FROM ns1.table1")
    DF->>SC: get_table("ns1", "table1")
    SC->>POL: load_table("ns1.table1") [user_token]
    POL-->>SC: table metadata + S3 location
    SC-->>DF: TableProvider (Iceberg scan)

This means two users running the same query may see different tables, schemas, or data — depending on what Polaris grants them.

Worker

Workers are stateless DataFusion executors. They receive plan fragments (scan tasks) from the coordinator, read Parquet files from S3, and stream Arrow results back.

Architecture

graph TB
    subgraph Worker["sqe-server --mode worker"]
        FS["Flight Service<br/>:50052"]
        EX["Executor"]
        PR["Parquet Reader"]
    end

    COORD["Coordinator"] -->|ScanTask ticket| FS
    FS --> EX
    EX --> PR
    PR -->|read| S3["S3 / MinIO"]
    PR -->|Arrow RecordBatches| FS
    FS -->|Flight stream| COORD

Scan Task

The coordinator sends workers a ScanTask — a lightweight JSON message containing everything the worker needs:

{
  "fragment_id": "frag-001",
  "data_file_paths": [
    "s3://warehouse/ns/table/data/00001.parquet",
    "s3://warehouse/ns/table/data/00002.parquet"
  ],
  "projected_columns": ["id", "name", "amount"],
  "s3_endpoint": "http://s3:9000",
  "s3_region": "us-east-1",
  "s3_access_key": "...",
  "s3_secret_key": "...",
  "s3_path_style": true
}

Workers don’t need access to Polaris or Keycloak. The coordinator resolves table metadata, applies security, and provides the worker with direct S3 credentials and file paths.

Health Checking

The coordinator monitors workers with a background health check task:

stateDiagram-v2
    [*] --> Unhealthy: Registered
    Unhealthy --> Healthy: Health check OK
    Healthy --> Healthy: Health check OK
    Healthy --> Degraded: 1-2 consecutive failures
    Degraded --> Unhealthy: 3 consecutive failures
    Degraded --> Healthy: Health check OK
    Unhealthy --> Healthy: Health check OK

• Health checks run every 5 seconds via Flight Action("health_check")
• A worker is marked unhealthy after 3 consecutive failures
• Unhealthy workers are excluded from query scheduling
• Recovery is automatic — a healthy response resets the failure counter

Scaling

Workers are stateless, so scaling is just changing the replica count:

# Helm
helm upgrade sqe deploy/helm/sqe/ --set worker.replicas=5

# kubectl
kubectl scale deployment sqe-worker --replicas=5

The coordinator discovers workers from the config or service discovery. No re-registration needed.

Authentication Flow

SQE supports two OAuth2 flows for initial authentication, then manages token lifecycle transparently:

• OIDC Password Grant (ROPC) – for user-interactive authentication where Flight SQL sends username and password. Works with Keycloak or any OIDC provider that supports the Resource Owner Password Credentials grant.
• OAuth2 Client Credentials – for service-to-service auth, test environments, or OIDC providers that do not support ROPC. Configured by setting token_endpoint directly instead of keycloak_url.

The mode is selected automatically based on configuration: if keycloak_url is set, SQE uses ROPC; if token_endpoint is set (and keycloak_url is empty), SQE uses client credentials.

Why ROPC?

Flight SQL’s handshake sends username and password directly. There’s no browser redirect flow possible over gRPC. ROPC is the standard mechanism for non-interactive clients (JDBC drivers, CLI tools, dbt adapters).

Client Credentials Mode

When token_endpoint is set (and keycloak_url is empty), SQE uses the client_credentials grant instead of ROPC. In this mode:

• The coordinator obtains a service token using client_id + client_secret posted directly to the configured token endpoint.
• The username from the Flight SQL handshake is informational only – it is used for session labeling and audit logs, but is not sent to the token endpoint.
• There is no refresh_token in client credentials responses. When a token nears expiry, SQE re-fetches a new token via another client_credentials request.
• This is the mode used by the lightweight test stack (Polaris built-in OAuth), where Polaris itself acts as the token issuer.

Example Configuration

[auth]
token_endpoint = "http://polaris:8181/api/catalog/v1/oauth/tokens"
client_id = "root"
client_secret = "s3cr3t"

Client Credentials Sequence

sequenceDiagram
    participant Client
    participant SQE as SQE Coordinator
    participant TE as Token Endpoint
    participant POL as Polaris
    participant S3

    Note over Client,TE: Authentication
    Client->>SQE: Flight Handshake<br/>Basic auth (user:pass)
    SQE->>TE: POST /token<br/>grant_type=client_credentials<br/>client_id, client_secret
    TE-->>SQE: access_token, expires_in
    SQE->>SQE: Create Session<br/>(username from handshake,<br/>token from endpoint)
    SQE-->>Client: Bearer token (session_id)

    Note over Client,S3: Query Execution
    Client->>SQE: execute(SQL)<br/>Authorization: Bearer session_id
    SQE->>SQE: Lookup session → get access_token
    SQE->>POL: GET /namespaces<br/>Authorization: Bearer access_token
    POL-->>SQE: [namespace list]
    SQE->>POL: POST /tables/load<br/>Authorization: Bearer access_token
    POL-->>SQE: table metadata + S3 credentials
    SQE->>S3: GetObject (vended credentials)
    S3-->>SQE: Parquet data
    SQE-->>Client: Arrow Flight stream

    Note over SQE,TE: Background Token Re-fetch
    SQE->>TE: POST /token<br/>grant_type=client_credentials<br/>client_id, client_secret
    TE-->>SQE: new access_token, expires_in
    SQE->>SQE: Update session token

ROPC Flow

The following sections describe the ROPC (password grant) flow in detail.

Complete Flow

sequenceDiagram
    participant Client
    participant SQE as SQE Coordinator
    participant KC as Keycloak
    participant POL as Polaris
    participant S3

    Note over Client,KC: Authentication
    Client->>SQE: Flight Handshake<br/>Basic auth (user:pass)
    SQE->>KC: POST /token<br/>grant_type=password<br/>username, password, client_id
    KC-->>SQE: access_token, refresh_token, expires_in
    SQE->>SQE: Create Session<br/>(id, user, roles, tokens)
    SQE-->>Client: Bearer token (session_id)

    Note over Client,S3: Query Execution
    Client->>SQE: execute(SQL)<br/>Authorization: Bearer session_id
    SQE->>SQE: Lookup session → get access_token
    SQE->>POL: GET /namespaces<br/>Authorization: Bearer access_token
    POL-->>SQE: [namespace list]
    SQE->>POL: POST /tables/load<br/>Authorization: Bearer access_token
    POL-->>SQE: table metadata + S3 credentials
    Note over POL: Polaris vends S3<br/>credentials scoped to<br/>this user + table
    SQE->>S3: GetObject (vended credentials)
    S3-->>SQE: Parquet data
    SQE-->>Client: Arrow Flight stream

    Note over SQE,KC: Background Token Refresh
    SQE->>KC: POST /token<br/>grant_type=refresh_token
    KC-->>SQE: new access_token, new refresh_token
    SQE->>SQE: Update session tokens

Token Refresh

A background task runs every 10 seconds, scanning all active sessions:

#![allow(unused)]
fn main() {
// Pseudocode
loop {
    sleep(10 seconds);
    for session in sessions_expiring_within(60 seconds) {
        match keycloak.refresh_token(session.refresh_token) {
            Ok(new_tokens) => session.update(new_tokens),
            Err(_) => session.mark_expired(),
        }
    }
}
}

The 60-second buffer ensures tokens are refreshed well before expiry, avoiding mid-query auth failures.

Client Credentials mode: There is no refresh_token in client credentials responses. The background task detects this and re-fetches a fresh token via a new client_credentials request to the token endpoint when the current token is near expiry. The same 60-second buffer applies.

Token Fingerprinting

When a token is refreshed, the iceberg-rust catalog client’s internal HTTP session cache still holds the old token. SQE uses a token fingerprint (last 8 characters of the access token) as part of the catalog session key. When the fingerprint changes, a new catalog session is created with the fresh token.

graph LR
    T1["Token: ...abc12345<br/>fingerprint: abc12345"] -->|refresh| T2["Token: ...xyz98765<br/>fingerprint: xyz98765"]
    T1 --> CS1["CatalogSession 1"]
    T2 --> CS2["CatalogSession 2<br/>(new, fresh token)"]

Role Extraction

SQE extracts user roles from the JWT realm_access.roles claim. These roles are stored in the session and used for policy evaluation:

{
  "realm_access": {
    "roles": ["data-analyst", "finance-reader", "admin"]
  }
}

Roles flow through to the Policy Enforcer, which uses them to determine row filters and column masks for each query.

Client Credentials mode: Role extraction only applies in OIDC (ROPC) mode, where the JWT contains user-specific claims. In client credentials mode, the token represents the service itself and typically does not carry realm_access.roles. The session’s role list is empty, and all authorization decisions are delegated to Polaris (which enforces access based on the service principal’s catalog grants).

Security & Policy

SQE enforces fine-grained security through LogicalPlan rewriting — injecting row filters and column masks into the query plan before DataFusion’s optimizer runs.

Status: The policy enforcement framework is designed and stubbed (Phase 5). Currently, a PassthroughEnforcer is active, which returns plans unmodified.

Design Principle

Security enforcement happens at the logical plan level, not at the data level:

graph TB
    SQL["SQL: SELECT * FROM sales"] --> PARSE["Parse"]
    PARSE --> PLAN["LogicalPlan<br/>Projection → TableScan(sales)"]
    PLAN --> POLICY["Policy Enforcer<br/>inject row filter + column mask"]
    POLICY --> SECURED["Secured LogicalPlan<br/>Projection → Filter(region='EU') → TableScan(sales)<br/>+ mask(ssn)"]
    SECURED --> OPT["DataFusion Optimizer"]
    OPT --> EXEC["Execute"]

    style POLICY fill:#f96,stroke:#333

This approach means:

• Row filters are transparent — the user doesn’t know they exist
• Column masks block predicate pushdown on raw values — you can’t WHERE ssn = '123-45-6789' to probe masked data
• Denied columns are invisible — they don’t appear in SELECT *, not as errors
• The optimizer can push user predicates through row filters but not through column masks

Policy Enforcer Trait

#![allow(unused)]
fn main() {
#[async_trait]
pub trait PolicyEnforcer: Send + Sync {
    async fn evaluate(
        &self,
        user: &SessionUser,
        plan: LogicalPlan,
    ) -> Result<LogicalPlan>;
}
}

Implementations:

• PassthroughEnforcer — returns plan unchanged (current default)
• OPA Enforcer — queries Open Policy Agent for policies (planned)
• Cedar Enforcer — evaluates AWS Cedar policies locally (planned)

Planned SQL Extensions

-- Grant row filter
GRANT SELECT ON sales TO ROLE analyst
  ROWS WHERE region = 'EU';

-- Grant column mask
GRANT SELECT ON customers TO ROLE support
  MASKED WITH (ssn AS '***-**-' || RIGHT(ssn, 4));

-- View effective policies
SHOW EFFECTIVE POLICY ON sales FOR ROLE analyst;

-- View grants
SHOW GRANTS ON sales;

No Information Leakage

Following the PostgreSQL RLS model:

Runtime Security Controls

SQE includes several runtime security mechanisms that are active by default or can be enabled via configuration.

Rate Limiting

Throttles query submission to prevent abuse or runaway clients. Uses a token-bucket algorithm (via the governor crate).

[rate_limit]
enabled = true
per_user_queries_per_minute = 60
global_queries_per_minute = 1000

When a limit is exceeded, the client receives a RESOURCE_EXHAUSTED Flight error. Rate limiting is disabled by default.

Query Timeouts

Every query is subject to an execution timeout. If the query exceeds the limit, it is cancelled and the client receives an error.

[query]
timeout_secs = 300              # Default: 5 minutes

[query.role_overrides]
admin = 3600                    # Admins get 1 hour
analyst = 600                   # Analysts get 10 minutes

Role overrides allow different timeout limits per role. The user’s longest-matching role timeout wins.

Session Lifecycle

Sessions have both idle and absolute timeouts. A background sweeper runs every 60 seconds to clean up expired sessions.

[session]
idle_timeout_secs = 900         # 15 min idle timeout
absolute_timeout_secs = 28800   # 8 hour hard cap

• Idle timeout: sessions with no query activity for this long are expired
• Absolute timeout: sessions older than this are expired regardless of activity

Query Cancellation

SQE supports Arrow Flight’s native cancellation mechanism. When a client cancels a query (or disconnects), the CancellationToken is triggered and propagated to workers, stopping execution promptly.

Error Sanitization

In production mode (debug = false, the default), error messages returned to clients are sanitized:

• Internal details (stack traces, file paths, internal error types) are stripped
• Clients receive a short error message and a request ID for correlation
• Full details are logged server-side for debugging

Enable debug = true during development to see full error details:

[coordinator]
debug = true

TLS Encryption

Flight SQL connections can be encrypted with TLS. Optional mTLS adds client certificate verification.

[coordinator.tls]
cert_file = "/etc/sqe/server.crt"
key_file  = "/etc/sqe/server.key"
ca_file   = "/etc/sqe/ca.crt"    # Optional: mTLS

See Configuration for details.

Streaming Execution

SQE’s streaming execution engine enables 1TB-scale queries on memory-constrained servers. The implementation is split into two phases: Phase A (safe) handles single-node memory management and scan optimization, while Phase B (fast) distributes computation across workers via Arrow Flight DoExchange.

The Problem

A coordinator-centric query engine hits a hard wall: every intermediate result flows through one process. An ORDER BY on 1TB of data requires 1TB of memory (or spill space) on the coordinator, regardless of how many workers scanned the data. A four-way hash join between large tables can exhaust coordinator memory long before the result set is assembled.

The fundamental tension is between sovereignty (run on your own hardware, which may be small) and scale (query datasets that don’t fit in memory). SQE solves this in two stages: first, make the coordinator survive large queries through spill-to-disk and scan optimization (Phase A); then, push computation to workers so the coordinator handles only final aggregation (Phase B).

Phase A: Safe (Single-Node)

Phase A ensures that a single coordinator with limited memory (e.g., 512MB) can execute large analytical queries without OOM kills.

Coordinator Spill-to-Disk

DataFusion’s FairSpillPool divides available memory across all active operators. When an operator (sort, hash aggregate, hash join) exceeds its share, it spills intermediate results to disk as sorted runs.

Key components:

• FairSpillPool – configured via memory_limit in sqe.toml. Divides memory equally among registered MemoryConsumer instances. Triggers spill when any consumer exceeds its fair share.
• Watermark system – four levels (green/yellow/orange/red) based on pool utilization percentage. Green (<60%) allows normal execution. Yellow (60-75%) triggers advisory warnings. Orange (75-90%) forces spillable operators to spill. Red (>90%) activates admission control, queueing new queries until memory drops below the orange threshold.
• Admission control – when the pool is in the red zone, new queries wait in a bounded queue rather than competing for memory. This prevents cascade failures where N concurrent queries each grab 1/N of memory and all spill simultaneously.
• External merge sort – when a SortExec spills, it writes sorted runs to spill_dir (default: /tmp/sqe-spill). On final output, a k-way merge reads all runs simultaneously, producing a globally sorted stream with constant memory overhead.

Configuration in sqe.toml:

[coordinator]
memory_limit = "512MB"
spill_dir = "/tmp/sqe-spill"
spill_compression = "zstd"  # lz4, zstd, or none

Late Materialization

Standard Parquet scans read all projected columns from every row group. Late materialization splits this into two phases:

1. Predicate phase – read only the columns referenced in WHERE clauses. Apply filters. Produce a set of surviving row indices.
2. Projection phase – for surviving rows only, read the remaining projected columns.

This is implemented as a two-phase RowFilter scan in the Iceberg scan planning layer. For queries with selective predicates (e.g., WHERE status = 'CLOSED' on a table where 5% of rows match), late materialization reduces I/O by up to 95% on the non-predicate columns.

The optimization is transparent to the rest of the plan – the TableScan still produces the same Arrow schema. The difference is entirely in how many bytes are read from Parquet.

Iceberg Scan Planning

Three optimizations happen before any Parquet data is read:

• File-level min/max pruning – Iceberg manifest files contain per-column min/max statistics for each data file. SQE reads these statistics and skips files where the predicate cannot match. For example, WHERE order_date > '2025-01-01' skips any file whose order_date max is before 2025.
• Sort-order detection – Iceberg metadata records the sort order of each data file. When a query includes ORDER BY on the sort column, SQE can skip the sort operator entirely and produce output directly from the pre-sorted scan. When multiple sorted files need merging, a merge-sort is cheaper than a full re-sort.
• PageIndex pruning – for Parquet files with page-level statistics (column index), SQE prunes individual pages within a row group, further reducing I/O for selective predicates.
• TopK optimization – ORDER BY ... LIMIT N queries use a heap-based TopK operator that maintains only N rows in memory, avoiding a full sort and spill.

S3 I/O Pipeline

Reading Parquet files from S3 involves many small HTTP GET requests (one per column chunk per row group). SQE optimizes this with:

• Request coalescing – adjacent byte ranges within coalesce_threshold (default: 1MB) are merged into a single GET request. This reduces the number of HTTP round-trips, which dominate latency on high-latency S3 endpoints.
• Footer cache – Parquet file footers (schema, row group metadata, column chunk offsets) are cached in a footer_cache_size-bounded LRU cache. Repeated queries against the same table skip the footer read entirely.
• Prefetch – while the executor processes the current row group, the next row group’s column chunks are fetched in the background, hiding S3 latency behind compute.

Configuration:

[storage]
coalesce_threshold = "1MB"
footer_cache_size = 256  # number of footers

SortMergeJoin Fallback

DataFusion’s default join strategy is hash join, which builds a hash table from the build side in memory. For large joins, this hash table can exceed the memory limit. DataFusion does not yet support hash join spill-to-disk upstream.

SQE registers a SortMergeJoin fallback: when the estimated build-side size exceeds hash_join_memory_threshold, the optimizer rewrites the join as a sort-merge join. Both sides are sorted (spilling to disk if needed via the external merge sort) and then merged with constant memory. This is slower than an in-memory hash join but avoids OOM on large joins.

[optimizer]
hash_join_memory_threshold = "256MB"

Phase B: Fast (Distributed)

Phase B pushes computation past the scan boundary. Instead of workers sending raw Arrow batches to the coordinator for all processing, workers perform filters, partial aggregations, partial sorts, and join probes locally.

DoExchange Shuffle

Arrow Flight’s DoExchange RPC enables bidirectional streaming between workers. SQE uses this to implement a hash-partitioned shuffle:

1. The coordinator decomposes the physical plan into stages separated by shuffle boundaries (e.g., a hash join requires both sides to be hash-partitioned on the join key).
2. Each stage runs on a set of workers. When a stage completes, its output is hash-partitioned by the shuffle key and streamed to the next stage’s workers via DoExchange.
3. The partitioning function uses the same hash(key) % num_partitions scheme as DataFusion’s RepartitionExec, ensuring compatibility with the existing hash join and hash aggregate operators.

Distributed Sort (Range-Partition)

A distributed ORDER BY proceeds in three steps:

1. Sample – each worker samples its local partition and sends the sample to the coordinator.
2. Range boundaries – the coordinator computes quantile boundaries from the samples, producing N-1 split points for N workers.
3. Range-partition and merge – each worker range-partitions its data and sends each range to the designated worker. Each receiving worker sorts its range locally (spilling if needed). The coordinator merges the sorted ranges via a k-way merge.

This distributes both the memory cost and the CPU cost of sorting. A 1TB ORDER BY with 8 workers requires roughly 125GB of spill per worker instead of 1TB on the coordinator.

Two-Phase Aggregation

Aggregation queries (GROUP BY) use a two-phase approach:

1. Partial aggregation – each worker computes partial aggregates on its local data. For SUM(amount) GROUP BY region, each worker produces a partial sum per region from its partition.
2. Final aggregation – partial results are shuffled by the grouping key to a set of finalizer workers (or the coordinator for small result sets). Each finalizer merges the partial aggregates into the final result.

This solves the q18 problem: TPC-H query 18 has a high-cardinality GROUP BY that produces millions of groups. On a single coordinator with 512MB, the GroupedHashAggregate exceeds memory. With two-phase aggregation, each worker handles a fraction of the groups, and memory pressure is distributed.

Distributed Joins

SQE supports four join strategies in distributed mode:

• Broadcast join – when one side of the join is small (below broadcast_threshold, default 10MB), it is broadcast to all workers. Each worker probes its local partition of the large side against the broadcast table. No shuffle required.
• Shuffle hash join – both sides are hash-partitioned on the join key and shuffled to matching workers. Each worker performs a local hash join on its partition.
• Pre-sorted merge join – when both sides are already sorted on the join key (detected via Iceberg sort-order metadata), workers perform a merge join without re-sorting. This avoids the sort cost entirely.
• Predicate transfer – before executing a join, the build side’s distinct join keys are collected and pushed as an IN-list filter to the probe side’s scan. This skips probe-side files that contain no matching keys, reducing I/O by 90%+ for selective joins. Based on the predicate transfer technique from Yang et al. (SIGMOD 2025).

[optimizer]
broadcast_threshold = "10MB"

Multi-Endpoint Flight SQL

In Phase A, all results flow through the coordinator’s single Flight SQL endpoint. Phase B adds multi-endpoint support: get_flight_info can return multiple FlightEndpoint entries, each pointing to a different worker. The client fetches results directly from workers, bypassing the coordinator for data transfer.

This eliminates the coordinator NIC bottleneck for large result sets. A query returning 4GB across 4 workers streams 1GB directly from each worker to the client, achieving 4x the effective bandwidth.

Stage Decomposition

The coordinator decomposes the physical plan into stages:

1. Scan stage – workers read Parquet files, apply predicates and projections.
2. Shuffle stage – workers hash-partition or range-partition output for the next stage.
3. Join/Aggregate stage – workers perform local joins or aggregations on shuffled data.
4. Final stage – coordinator (or a designated worker) performs final aggregation, sort, or limit.

Each stage boundary is a shuffle point. The coordinator tracks stage completion and triggers the next stage when all workers in the current stage have finished.

Memory Model

SQE uses a four-level watermark system to manage memory pressure:

The FairSpillPool divides the total memory_limit equally among all registered MemoryConsumer instances. When a consumer’s try_grow call would push the pool past the orange threshold, the pool asks other spillable consumers to spill first. If spilling frees enough memory, the allocation succeeds. If not, the allocation fails with ResourceExhausted.

Per-operator behavior:

• SortExec – spills sorted runs to disk, later merged via k-way merge.
• HashAggregateExec – spills partition groups to disk (when supported by DataFusion).
• HashJoinExec – not spillable upstream; SQE rewrites to SortMergeJoin when estimated size exceeds threshold.
• SortMergeJoinExec – both sides sort-and-spill independently, then merge with constant memory.

Configuration Reference

Benchmark Results

TPC-H at scale factor 1 (approximately 1GB of data) on a coordinator with 512MB memory and spill-to-disk enabled:

• 21 of 22 queries pass. All queries produce correct results within the memory budget.
• 1 failure: q18. TPC-H query 18 uses a high-cardinality GROUP BY with HAVING that produces millions of intermediate groups. DataFusion’s GroupedHashAggregate does not yet support spill-to-disk for hash aggregation, so the operator exceeds the 512MB limit. This is a known upstream limitation. With Phase B’s two-phase aggregation (distributing the groups across workers), q18 passes.

These results demonstrate that SQE can run analytical workloads on hardware that would be considered undersized for traditional query engines. The combination of spill-to-disk, late materialization, and scan planning keeps memory usage bounded regardless of data size.

Research Papers

The streaming execution engine draws on decades of database systems research. This chapter lists the papers that most directly influenced SQE’s design, with notes on how each idea is implemented.

Papers and Their Influence

Detailed Notes

Graefe, “Volcano” (1994)

Volcano introduced the exchange operator as the universal mechanism for parallelism in query evaluation. An exchange operator sits between two plan tree segments and handles data redistribution – hash partitioning, round-robin, or broadcast – without the operators above or below knowing about parallelism. SQE’s DoExchange shuffle is a direct implementation of this model over Arrow Flight’s bidirectional streaming RPC. Each shuffle boundary in the stage decomposition corresponds to an exchange operator. The key insight from Volcano that SQE preserves: the operators themselves are single-threaded and unaware of distribution; all parallelism is encapsulated in the exchange.

Shapiro, “Hybrid Hash Join” (1986)

Shapiro’s hybrid hash join partitions the build side into buckets, keeping one bucket in memory and spilling the rest to disk. During the probe phase, the in-memory bucket is probed immediately; the spilled buckets are read back and probed sequentially. This is the standard approach for hash join spill in systems like PostgreSQL and Trino. DataFusion does not yet implement hash join spill upstream, so SQE cannot use this technique directly. Instead, SQE falls back to SortMergeJoin for large joins – both sides are sorted (spilling via external merge sort) and then merged with constant memory. The SortMergeJoin fallback is the safe path; when DataFusion adds hash join spill, SQE can adopt the hybrid approach from Shapiro for better performance on unsorted inputs.

Sethi et al., “Presto: SQL on Everything” (2019)

Presto’s architecture – a stateless coordinator that plans and schedules, stateless workers that execute, no shared storage between them – is the direct model for SQE’s coordinator/worker split. The paper describes Presto’s evolution from a no-spill engine (all intermediate data in memory) to one that supports spill-to-disk under memory pressure. SQE followed the same evolution: Phase A added coordinator spill (the “safe” path), and Phase B added distributed computation (pushing work to workers so the coordinator handles less data). The paper’s observation that “most queries fit in memory; spill is for the tail” matches SQE’s experience – 20 of 22 TPC-H queries run without spill on 512MB; only the largest sorts and aggregations trigger it.

Leis et al., “Morsel-Driven Parallelism” (2014)

The morsel-driven model assigns small, fixed-size chunks of work (morsels) to worker threads, enabling NUMA-aware scheduling without explicit thread pinning. DataFusion’s pull-based execution model, where each operator produces batches on demand, is conceptually similar. SQE’s batch sizing (default 8192 rows per RecordBatch) is informed by this paper’s finding that small, uniform work units lead to better load balancing and cache utilization. The paper also highlights the importance of avoiding global synchronization in the hot path – a principle SQE follows by using per-operator memory consumers and tokio::sync::watch channels for credential refresh rather than shared mutexes.

Pedreira et al., “Velox” (2022)

Velox introduces cooperative memory arbitration: operators register with a central arbitrator and respond to memory pressure by spilling or shrinking their buffers. This is more sophisticated than DataFusion’s FairSpillPool, which divides memory equally and triggers spill when any consumer exceeds its share. Velox’s arbitrator can make global decisions – asking a low-priority operator to spill so a high-priority one can proceed. SQE does not implement priority-based arbitration today, but the FairSpillPool’s watermark system (green/yellow/orange/red) provides a simpler version of the same concept. The Velox paper’s cooperative model is the planned future direction for SQE’s memory management, particularly for mixed workloads where interactive queries should preempt batch jobs.

Raasveldt & Muhleisen, “DuckDB” (2019)

DuckDB proves that a single-node engine with proper buffer management can process datasets far larger than available memory. Its out-of-core hash join and sort implementations use disk-backed buffers that transparently page data in and out. SQE’s Phase A is built on the same principle: spill-to-disk is not an error path, it is the normal execution path for large queries on small machines. The difference is that SQE operates over remote storage (S3) rather than local files, so the I/O pipeline (coalescing, footer cache, prefetch) is more critical. DuckDB’s benchmark results – 1TB queries on 16GB machines – provided the confidence that SQE’s 512MB target was achievable with the right memory management.

Pang et al., “Memory-Adaptive External Sorting” (1993)

This paper addresses the problem of external sorting when available memory fluctuates during execution (due to other concurrent operators). The key technique is dynamic run splitting: instead of committing to a fixed run size at the start of the sort, the algorithm adapts the run size based on currently available memory. SQE’s FairSpillPool implements a version of this: as other operators allocate and release memory, the pool available to a SortExec changes. When memory shrinks (another query starts), the sort produces smaller runs and spills more frequently. When memory grows (another query finishes), the sort can produce larger runs. The k-way merge at the end adapts to whatever set of runs was produced.

Abadi et al., “Materialization Strategies” (2007)

Abadi’s paper compares early materialization (read all columns, then filter) with late materialization (read predicate columns, filter, then read remaining columns for survivors) in column-oriented databases. Late materialization wins when predicates are selective because it avoids reading non-predicate columns for filtered-out rows. SQE implements late materialization in the Iceberg scan layer: the scan planner splits the column set into predicate columns and projection-only columns, reads the predicate columns first, applies the RowFilter, and then reads the projection columns only for rows that survived the filter. For a query like SELECT * FROM orders WHERE status = 'CLOSED' on a table where 5% of rows match, this reduces column-chunk reads by up to 19x (for a 20-column table where status is one column).

Yang et al., “Predicate Transfer” (2025)

Predicate transfer pushes join key values from the build side of a join to the probe side’s scan, filtering probe-side data before it enters the join operator. In the simplest form, the distinct join keys from the build side are collected into an IN-list and injected as a predicate on the probe side’s table scan. SQE implements this for distributed joins: after the build side is scanned and its distinct keys are known, the coordinator pushes the key set to probe-side workers as a scan predicate. Combined with Iceberg’s file-level min/max statistics, this can skip entire data files on the probe side. For selective joins (e.g., a dimension table join where only 100 of 10,000 distinct key values appear), predicate transfer skips 90%+ of probe-side files, dramatically reducing I/O. This is particularly effective for star-schema queries common in analytical workloads.

SQL Support

SQE inherits DataFusion’s comprehensive SQL support and adds Iceberg-specific operations.

Query Language

SELECT & Expressions

-- Full ANSI SQL
SELECT customer_id, SUM(amount) AS total
FROM orders
WHERE order_date >= '2024-01-01'
GROUP BY customer_id
HAVING SUM(amount) > 1000
ORDER BY total DESC
LIMIT 10;

-- CTEs
WITH monthly AS (
    SELECT DATE_TRUNC('month', order_date) AS month, SUM(amount) AS total
    FROM orders GROUP BY 1
)
SELECT month, total, LAG(total) OVER (ORDER BY month) AS prev_month
FROM monthly;

-- Subqueries, EXISTS, IN
SELECT * FROM customers
WHERE customer_id IN (SELECT customer_id FROM orders WHERE amount > 500);

Window Functions

SELECT
    employee_id,
    department,
    salary,
    ROW_NUMBER() OVER (PARTITION BY department ORDER BY salary DESC) AS rank,
    AVG(salary) OVER (PARTITION BY department) AS dept_avg,
    salary - LAG(salary) OVER (ORDER BY hire_date) AS salary_diff
FROM employees;

Supported: ROW_NUMBER, RANK, DENSE_RANK, NTILE, LAG, LEAD, FIRST_VALUE, LAST_VALUE, NTH_VALUE, CUME_DIST, PERCENT_RANK, with PARTITION BY, ORDER BY, and frame clauses (ROWS BETWEEN, RANGE BETWEEN).

Joins

-- All join types
SELECT * FROM a INNER JOIN b ON a.id = b.id;
SELECT * FROM a LEFT JOIN b ON a.id = b.id;
SELECT * FROM a RIGHT JOIN b ON a.id = b.id;
SELECT * FROM a FULL OUTER JOIN b ON a.id = b.id;
SELECT * FROM a CROSS JOIN b;

-- Anti and semi joins (via EXISTS/NOT EXISTS)
SELECT * FROM a WHERE NOT EXISTS (SELECT 1 FROM b WHERE b.id = a.id);

Set Operations

SELECT id FROM a UNION ALL SELECT id FROM b;
SELECT id FROM a INTERSECT SELECT id FROM b;
SELECT id FROM a EXCEPT SELECT id FROM b;

DDL

-- Schemas
CREATE SCHEMA analytics;
DROP SCHEMA staging;

-- Tables (via CTAS)
CREATE TABLE analytics.summary AS
SELECT region, SUM(revenue) AS total FROM sales GROUP BY region;

-- CREATE OR REPLACE
CREATE OR REPLACE TABLE analytics.summary AS
SELECT region, SUM(revenue) AS total FROM sales GROUP BY region;

-- Views
CREATE VIEW active_customers AS
SELECT * FROM customers WHERE status = 'active';

DROP VIEW active_customers;

-- Drop
DROP TABLE analytics.summary;
DROP TABLE IF EXISTS analytics.summary;

DML

-- Insert from query
INSERT INTO target_table
SELECT * FROM source_table WHERE condition;

-- CTAS
CREATE TABLE new_table AS SELECT * FROM existing_table;

-- DELETE (Copy-on-Write by default)
DELETE FROM orders WHERE status = 'cancelled';
DELETE FROM orders WHERE customer_id IN (SELECT id FROM blacklist);

-- DELETE (Merge-on-Read; opt in via table property)
ALTER TABLE orders SET TBLPROPERTIES ('write.delete.mode' = 'merge-on-read');
DELETE FROM orders WHERE status = 'cancelled';  -- writes a position delete file

-- UPDATE (Copy-on-Write)
UPDATE orders SET status = 'shipped' WHERE tracking_id IS NOT NULL;
UPDATE orders SET amount = CASE WHEN amount > 1000 THEN amount * 0.9 ELSE amount END;

-- MERGE INTO (Copy-on-Write)
MERGE INTO target USING source ON target.id = source.id
WHEN MATCHED THEN UPDATE SET value = source.value
WHEN NOT MATCHED THEN INSERT (id, value) VALUES (source.id, source.value);

All row-level write operations (DELETE, UPDATE, MERGE INTO) default to Copy-on-Write via the RisingWave iceberg-rust fork’s rewrite_files() transaction API. Affected data files are read, filtered/transformed, and rewritten as new files in a single atomic commit.

DELETE also supports Merge-on-Read when write.delete.mode = 'merge-on-read' is set on the table. SQE writes a position-delete file (or an equality-delete file when the table declares an identifier-field-id) and commits via FastAppendAction / RowDeltaAction. MoR avoids rewriting whole data files for small deletes against large tables.

Data Types

SQE accepts the standard ANSI SQL type set plus a few Iceberg-specific extensions:

CREATE TABLE events (
    id              BIGINT,
    payload         JSON,                       -- Aliases to Utf8 underneath
    occurred_at     TIMESTAMP(6),
    occurred_time   TIME(6),                    -- Time-of-day, microseconds
    occurred_at_tz  TIMESTAMP(6) WITH TIME ZONE,
    occurred_ns     TIMESTAMP_NS,               -- V3-only: nanosecond precision
    region_id       INTEGER,
    amount          DECIMAL(18, 2)
);

• JSON columns store as Utf8. CAST(json_col AS BIGINT|VARCHAR|DOUBLE) rides DataFusion’s built-in coercion. JSON-shaped extraction works through json_extract, json_extract_scalar, json_array_length, json_parse, json_get_str, json_get_int, json_get_float, json_get_bool.
• TIME / TIME(p) maps to Arrow Time64(Microsecond) since Iceberg’s time primitive is microsecond-only across V2 and V3. Precisions 0..=6 collapse to microsecond. TIME(p > 6) rejects with a clear NotImplemented; use TIMESTAMP(9) for sub-microsecond resolution.
• TIME WITH TIME ZONE rejects at CREATE TABLE: Arrow has no equivalent. Use TIMESTAMP WITH TIME ZONE instead.
• TIMESTAMP_NS (and TIMESTAMP_NS WITH TIME ZONE) is a V3-only nanosecond timestamp. SQE auto-upgrades the table to format-version 3 when one of these types appears in a CREATE.
• localtime() returns Time64. EXTRACT(HOUR|MINUTE|SECOND FROM time_col) works through the Trino-aliased hour() / minute() / second() UDFs. year() / month() / day() on a TIME column raise a clear plan error per Trino spec.

Metadata Queries

SHOW CATALOGS;
SHOW SCHEMAS;
SHOW TABLES;
SHOW TABLES IN schema_name;

-- information_schema
SELECT * FROM information_schema.tables;
SELECT * FROM information_schema.schemata;
SELECT * FROM information_schema.columns WHERE table_name = 'orders';

-- Query plan (logical + physical)
EXPLAIN SELECT * FROM orders WHERE amount > 100;

-- With actual execution metrics
EXPLAIN ANALYZE SELECT * FROM orders WHERE amount > 100;

-- With Iceberg file/row estimates (no execution)
EXPLAIN FULL SELECT * FROM orders WHERE amount > 100;

Feature Comparison

Query Plan Inspection (EXPLAIN)

SQE provides three variants of EXPLAIN for inspecting how queries are planned and executed.

EXPLAIN

Returns the logical and physical query plan without executing the query.

EXPLAIN SELECT * FROM orders WHERE amount > 100;

Output: Two rows — logical_plan and physical_plan — each containing a text representation of the plan tree. The plan shown is the policy-enforced plan: any row filters or column masks applied by the security layer are visible.

EXPLAIN ANALYZE

Executes the query and returns per-operator timing and row counts.

EXPLAIN ANALYZE
SELECT dept_id, COUNT(*), AVG(salary)
FROM employees
GROUP BY dept_id;

Output columns: step, operation, output_rows, elapsed_ms

Rows are ordered leaf-to-root (execution order). output_rows and elapsed_ms are NULL for operators that do not expose DataFusion metrics.

EXPLAIN FULL

Returns the plan enriched with Iceberg table statistics — without executing the query.

EXPLAIN FULL SELECT * FROM large_table WHERE region = 'EU';

Output columns: step, operation, estimated_rows, estimated_bytes, files_scanned, files_total

For IcebergScanExec nodes, statistics come from the Iceberg snapshot summary (fast, no data file reads). estimated_rows reflects the total rows in the snapshot at plan time. files_scanned equals files_total because predicate-pushdown to file level is not yet implemented.

For other operators (Filter, Aggregate, Sort) estimated_rows comes from DataFusion’s cardinality analysis where available; file columns are NULL.

Notes

• All three variants apply policy enforcement — the plan reflects what will actually execute for the authenticated user.
• EXPLAIN FULL on non-Iceberg tables (e.g., information_schema) returns NULL for all statistics columns without error.

Custom SQL Extensions

SQE extends standard SQL with statements for security policy management. These are parsed by wrapping sqlparser-rs — we don’t fork the parser.

Status: Parser extensions are designed (Phase 5). The parser currently recognizes GRANT and REVOKE as standard SQL but does not yet handle the custom ROWS WHERE and MASKED WITH clauses.

Policy Statements

GRANT with Row Filter

GRANT SELECT ON schema.table TO ROLE role_name
  ROWS WHERE condition;

Example:

-- Analysts can only see European data
GRANT SELECT ON sales.orders TO ROLE eu_analyst
  ROWS WHERE region = 'EU';

-- Finance team sees only their cost center
GRANT SELECT ON hr.expenses TO ROLE finance
  ROWS WHERE cost_center = current_user_attr('cost_center');

GRANT with Column Mask

GRANT SELECT ON schema.table TO ROLE role_name
  MASKED WITH (column AS expression);

Example:

-- Support sees masked SSN (last 4 digits only)
GRANT SELECT ON customers TO ROLE support
  MASKED WITH (ssn AS '***-**-' || RIGHT(ssn, 4));

-- Partial email masking
GRANT SELECT ON users TO ROLE viewer
  MASKED WITH (email AS CONCAT(LEFT(email, 2), '***@', SPLIT_PART(email, '@', 2)));

REVOKE

REVOKE SELECT ON schema.table FROM ROLE role_name;

SHOW Statements

-- All grants on a table
SHOW GRANTS ON schema.table;

-- Effective policy for a role (combines all grants, resolves conflicts)
SHOW EFFECTIVE POLICY ON schema.table FOR ROLE analyst;

Parser Strategy

SQE wraps sqlparser-rs rather than forking it:

graph LR
    SQL["SQL Input"] --> SP["sqlparser-rs<br/>(standard parse)"]
    SP --> AST["Standard AST"]
    AST --> PP["Post-Parse Transform"]
    PP -->|Standard SQL| STD["Standard Statement"]
    PP -->|GRANT with ROWS WHERE| PS["PolicyStatement::GrantRowFilter"]
    PP -->|GRANT with MASKED WITH| PS2["PolicyStatement::GrantColumnMask"]
    PP -->|SHOW GRANTS| PS3["PolicyStatement::ShowGrants"]

The post-parse transform detects GRANT/REVOKE statements with the custom extensions and converts them to PolicyStatement AST nodes. Standard GRANT/REVOKE (without extensions) passes through unchanged.

Statement Classification

Every SQL statement is classified for routing, metrics, and audit:

#![allow(unused)]
fn main() {
pub enum StatementKind {
    Query,          // SELECT
    Ctas,           // CREATE TABLE AS SELECT
    Insert,         // INSERT INTO
    Merge,          // MERGE INTO (planned)
    Delete,         // DELETE FROM (planned)
    Drop,           // DROP TABLE/VIEW
    Rename,         // ALTER TABLE RENAME
    CreateView,     // CREATE VIEW
    DropView,       // DROP VIEW
    CreateSchema,   // CREATE SCHEMA
    DropSchema,     // DROP SCHEMA
    ShowCatalogs,   // SHOW CATALOGS
    ShowSchemas,    // SHOW SCHEMAS
    ShowTables,     // SHOW TABLES
    Policy,         // GRANT, REVOKE
    Utility,        // EXPLAIN, SET, etc.
}
}

Each kind maps to a stable lowercase label ("query", "ctas", "insert") used in Prometheus metrics and audit logs.

Iceberg Integration

SQE is built on the iceberg-rust library (vendored fork) and speaks the Iceberg REST Catalog protocol natively. Iceberg is the only table format SQE supports.

Iceberg version

• iceberg-rust: SQE-rebased fork of risingwavelabs/iceberg-rust dev_rebase_main_20260303 at commit 645f02a4b533, vendored at vendor/iceberg-rust/. Provides RewriteFilesAction / OverwriteFilesAction (Copy-on-Write DELETE/UPDATE), PositionDeleteFileWriter (Merge-on-Read position deletes), and DeletionVectorWriter (Iceberg V3) on top of upstream v0.9.0.
• DataFusion: 53.0
• Arrow: 58
• Parquet: 58
• Iceberg table format: V2 and V3. V3 features verified end-to-end (TIMESTAMP_NS, column defaults, equality-delete UPDATE on identifier-fields, partition evolution).

The matrix score against icebergmatrix.org is 167/189 (88.4%), placing SQE fifth on the public scoreboard behind only Spark distributions (EMR, AWS Glue, OSS Spark, Dataproc). See iceberg-matrix.md for the per-cell breakdown and iceberg-matrix-compare.md for the side-by-side V2/V3 comparison against every other engine on the public scoreboard.

Architecture

graph TB
    subgraph SQE
        SC["SessionCatalog<br/>per-user token"]
        TP["IcebergTableProvider<br/>DataFusion TableProvider"]
        WR["Writer<br/>Parquet output"]
    end

    SC -->|REST API + bearer or SigV4| CAT["Catalog backend<br/>Polaris, Nessie, Glue REST,<br/>S3 Tables, Unity, HMS, JDBC, Hadoop"]
    CAT -->|table metadata| SC
    CAT -->|S3 credentials<br/>credential vending| SC

    TP -->|read Parquet| OS["Object storage<br/>AWS S3, GCS, ADLS Gen2, R2,<br/>Ceph, MinIO, rustfs"]
    WR -->|write Parquet| OS

    SC -->|commit| CAT

Supported catalogs

SQE keeps the catalog choice as a runtime configuration concern. Every catalog below ships compiled in by default; pick one in [catalog.backend]. See Catalog backends for the full per-backend recipe with TOML examples and verification queries. For where the data files live (S3, R2, GCS, ADLS Gen2, HTTPS, hf://), see Storage backends.

There are two ways into AWS-managed Iceberg. The REST path (type = "rest" with the Glue REST or S3 Tables REST endpoint as catalog_url) speaks the Iceberg REST protocol that AWS exposes on top of these services. The native SDK path (type = "glue" or type = "s3tables") goes directly through aws-sdk-glue or aws-sdk-s3tables without REST. Both work; the SDK path is currently more maintained upstream and avoids the SigV4 signing patches.

The five non-Hadoop backends share one dispatch path through the upstream iceberg-catalog-loader crate. SQE’s for_session_other_backend translates the typed CatalogBackend enum into a uniform (catalog_type, name, props) tuple and hands it to iceberg_catalog_loader::load(type). The loader picks the right CatalogBuilder (REST / Glue / S3 Tables / HMS / SQL) and returns an Arc<dyn iceberg::Catalog>. Hadoop is the lone outlier because it is filesystem-only; its dispatch stays in sqe-catalog/src/backends/hadoop.rs.

Two SQE-only patches sit on top of the vendored loader. The first feature-gates each backend so a slim build does not transitively pull every backend’s AWS SDK / Thrift / sqlx weight. The second adds Send + Sync to BoxedCatalogBuilder so the boxed builder can cross await points in async contexts. Both are documented in vendor/iceberg-rust/README.md and forward-compatible with upstream.

Live integration tests for HMS, Nessie, JDBC Postgres, AWS Glue, AWS S3 Tables, and Unity OSS live in sqe-catalog/tests/backends_integration.rs. Each is #[ignore] and runs against a docker-compose overlay or a real cloud account configured via .env. Glue and S3 Tables are also live-verified (2026-05-05) against AWS account ACCOUNT_ID in eu-example-1 and eu-example-2.

The AWS REST endpoints share the OSS Iceberg REST code path. Phase P added an aws-sigv4 cargo feature to the vendored iceberg-catalog-rest crate that swaps the OAuth/Bearer authenticator for an AWS SigV4 signer when rest.sigv4-enabled=true lands in the catalog properties (or in the server’s /v1/config defaults). The signer reads credentials from the standard AWS provider chain.

Catalog REST surface

For Polaris, Nessie, Unity OSS, AWS Glue, and AWS S3 Tables, SQE talks to the catalog via the Iceberg REST API. Key interactions:

Every request includes the user’s bearer token (Polaris, Unity, Nessie, anything OIDC) or is signed with AWS SigV4 (Glue, S3 Tables). The catalog enforces access control.

Credential vending

When SQE loads a table, the catalog returns the table metadata and scoped storage credentials for accessing the data files:

sequenceDiagram
    participant SQE
    participant Catalog
    participant Storage as S3 / Object store

    SQE->>Catalog: Load table (user token / SigV4)
    Catalog-->>SQE: Table metadata<br/>+ scoped storage credentials<br/>(STS / table-scoped key)
    Note over SQE: Credentials are per-user,<br/>per-table, time-limited
    SQE->>Storage: Read Parquet (scoped credentials)
    Storage-->>SQE: Data (allowed prefix only)

This means:

• No service account with broad storage access.
• Each user’s storage access is scoped to exactly the tables they are querying.
• Credentials are short-lived (STS or equivalent).

Read path

graph LR
    PLAN["LogicalPlan<br/>TableScan"] --> META["Load table metadata<br/>from catalog"]
    META --> PRUNE["Partition pruning<br/>(manifest filtering)"]
    PRUNE --> FILES["Data file list"]
    FILES --> READ["Read Parquet<br/>(columnar, predicate pushdown,<br/>row-group skipping, RowFilter)"]
    READ --> BATCH["Arrow RecordBatches"]

Read-side optimizations:

• Partition pruning: Iceberg manifest stats skip whole partitions that cannot match the query predicate.
• Column projection: only requested columns leave Parquet.
• Predicate pushdown: filters land at the row group level, the page-index level, and the Parquet RowFilter.
• Runtime filter pushdown: Phase P shipped a DynamicPredicate API that absorbs DataFusion 53 hash-join build-side runtime filters into the same pruning surface. SF10 TPC-H lineitem-heavy queries saw q06 -51%, q07 -31%, q14 -33%. Engineering log at docs/features/runtime-filter-pushdown.md.
• Bloom filter consultation: write.parquet.bloom-filter-columns lands bloom offsets in the file footer; DataFusion consults them automatically for literal equality predicates at scan time.
• 5-layer caching: REST catalog cache, table metadata cache, manifest cache, SessionContext cache, OAuth token cache. Warm queries hit sub-millisecond planning.

Write path

graph LR
    SQL["CTAS / INSERT / DELETE / UPDATE / MERGE"] --> EXEC["Execute SELECT"]
    EXEC --> BATCH["RecordBatches"]
    BATCH --> WRITE["Write Parquet<br/>(WriterProperties from table props)"]
    WRITE --> COMMIT["Commit to Iceberg<br/>(append / row-delta / rewrite)"]

Supported DML, both V2 and V3 verified:

• CREATE TABLE AS SELECT: Apache Iceberg V2 and V3, including TIMESTAMP_NS columns and DEFAULT literals.
• INSERT INTO: streaming, with proper schema validation against the catalog.
• DELETE FROM: Copy-on-Write via RewriteFilesAction, or Merge-on-Read via PositionDeleteFileWriter when write.delete.mode=merge-on-read.
• UPDATE: CoW or MoR (equality deletes when the table declares an identifier-field-id).
• MERGE INTO: full WHEN MATCHED / WHEN NOT MATCHED semantics, dispatching to CoW or MoR based on the table’s write.update.mode.
• ALTER TABLE: ADD/DROP/RENAME COLUMN, SET/DROP NOT NULL, type promotion, ADD/DROP/REPLACE PARTITION FIELD (partition evolution), CREATE/DROP BRANCH/TAG, SET WRITE BRANCH.

The writer respects write.parquet.bloom-filter-columns and write.parquet.bloom-filter-fpp for any column the schema knows about. The footer-inspection test in sqe-catalog/src/parquet_writer_config.rs proves bloom offsets land in the resulting Parquet file.

V3 features verified

• TIMESTAMP_NS / TIMESTAMPTZ_NS: V3 nanosecond timestamps round-trip end-to-end.
• Column defaults: CREATE TABLE ... DEFAULT <literal> applies write_default; ALTER TABLE ADD COLUMN ... DEFAULT applies initial_default.
• Position deletes (V3): MoR DELETE on a V3 table writes position-delete files.
• Equality deletes (V3): UPDATE with a declared identifier-field-id commits a single RowDelta with new data file plus equality-delete row.
• Partition evolution (V3): ALTER TABLE ADD/DROP/REPLACE PARTITION FIELD evolves the spec on V3 tables, including with day(ts) on TIMESTAMP_NS columns.
• Time travel (V3): FOR SYSTEM_TIME AS OF and FOR VERSION AS OF work against V3 tables through the same snapshot walk as V2.
• Schema evolution (V3): ADD COLUMN, DROP COLUMN, RENAME COLUMN, SET DATA TYPE all work on V3.

V3 features still blocked upstream:

• Variant: pending iceberg-rust #2188.
• Geometry: pending DataFusion UDT #12644.
• Vector / Embedding: V3 spec not finalised.

The deferred list is tracked in docs/iceberg-matrix-state.json under caveats for each cell.

Write Path

SQE supports writing data to Iceberg tables through SQL. Writes go through the coordinator, which executes the SELECT portion, writes Parquet files to S3, and commits to the Iceberg catalog.

Supported Operations

CREATE TABLE AS SELECT (CTAS)

CREATE TABLE analytics.monthly_sales AS
SELECT
    DATE_TRUNC('month', order_date) AS month,
    region,
    SUM(amount) AS total
FROM raw.orders
GROUP BY 1, 2;

Flow:

1. Parse SQL, extract target table name and SELECT query
2. Execute SELECT → get Arrow RecordBatches
3. Convert Arrow schema to Iceberg schema
4. Create table in Polaris catalog
5. Write RecordBatches as Parquet files to S3
6. Commit data files to Iceberg via AppendAction

CREATE OR REPLACE TABLE

CREATE OR REPLACE TABLE analytics.monthly_sales AS
SELECT ... ;

Drops the existing table (if it exists) and creates a new one. Useful for dbt table materializations.

INSERT INTO

INSERT INTO analytics.monthly_sales
SELECT
    DATE_TRUNC('month', order_date) AS month,
    region,
    SUM(amount) AS total
FROM raw.orders
WHERE order_date >= '2024-06-01'
GROUP BY 1, 2;

Flow:

1. Parse SQL, extract target table and SELECT query
2. Execute SELECT → get Arrow RecordBatches
3. Write RecordBatches as Parquet files to S3
4. Commit data files to Iceberg via AppendAction (new snapshot)

Write Architecture

sequenceDiagram
    participant QH as Query Handler
    participant DF as DataFusion
    participant WH as Write Handler
    participant S3
    participant POL as Polaris

    QH->>DF: Execute SELECT query
    DF-->>QH: RecordBatches

    QH->>WH: handle_ctas(table_name, schema, batches)
    WH->>POL: Create table (schema)
    POL-->>WH: Table created

    WH->>S3: Write Parquet data files
    S3-->>WH: Written (paths + sizes)

    WH->>POL: Commit AppendAction<br/>(data file list → new snapshot)
    POL-->>WH: Committed
    WH-->>QH: Success

Row-Level Operations (Copy-on-Write)

Row-level write operations are implemented via Copy-on-Write using the RisingWave iceberg-rust fork’s rewrite_files() transaction API. Affected data files are read, filtered/transformed, and rewritten as new files in a single atomic Iceberg commit.

DELETE FROM

DELETE FROM sales.orders WHERE status = 'cancelled';

-- Cross-table subqueries in WHERE
DELETE FROM sales.orders
WHERE customer_id IN (SELECT id FROM blacklist);

-- DELETE without WHERE = truncate
DELETE FROM sales.orders;

Flow:

1. Scan table metadata to identify affected data files
2. Read each affected file, apply the WHERE filter
3. If all rows match: mark file for removal
4. If partial match: rewrite file without matching rows
5. Commit via rewrite_files() (remove old files, add rewritten files)

UPDATE

UPDATE sales.orders SET status = 'shipped' WHERE tracking_id IS NOT NULL;

-- CASE WHEN transformations
UPDATE sales.orders SET amount = CASE
    WHEN amount > 1000 THEN amount * 0.9
    ELSE amount
END;

Flow:

1. Scan table metadata to identify affected data files
2. Read each affected file, apply the WHERE filter
3. For matching rows: apply SET expressions
4. Rewrite file with modified rows
5. Commit via rewrite_files()

MERGE INTO

MERGE INTO target USING source ON target.id = source.id
WHEN MATCHED THEN UPDATE SET value = source.value
WHEN NOT MATCHED THEN INSERT (id, value) VALUES (source.id, source.value);

Flow:

1. Execute a full outer join of source and target via DataFusion
2. Classify each result row: matched (UPDATE/DELETE) or not matched (INSERT)
3. Rewrite affected target data files with modifications applied
4. Add new data files for INSERT rows
5. Commit via rewrite_files() (remove old files, add new + rewritten files)

Iceberg Dependency

Row-level writes depend on the risingwavelabs/iceberg-rust fork (rev 1978911ec4) which provides the rewrite_files() transaction support not yet available in upstream iceberg-rust. When upstream ships OverwriteAction, the dependency can be migrated back to the official crate.

dbt Compatibility

The write path is designed to support dbt Core via a native dbt-sqe adapter:

read_parquet TVF

read_parquet() is a table-valued function registered on every SQE SessionContext. It reads Parquet files from local disk or S3-compatible storage and returns them as a DataFusion table scan, making Parquet files directly queryable without first loading data into Iceberg.

Syntax

SELECT * FROM read_parquet(
  '<path>',
  [access_key => '<key>',]
  [secret_key => '<secret>',]
  [endpoint => '<url>',]
  [region => '<region>']
)

The first argument is the file path or glob pattern. All other arguments are named (keyword) parameters for S3 credentials. Named parameters are optional and fall back to the engine’s configured storage defaults when omitted.

Local files

Absolute paths and glob patterns both work:

-- Single file
SELECT * FROM read_parquet('/data/tpch/sf1/lineitem/part-0000.parquet');

-- All files in a directory
SELECT * FROM read_parquet('/data/tpch/sf1/lineitem/*.parquet');

-- Recursive glob
SELECT * FROM read_parquet('/data/tpch/sf1/lineitem/**/*.parquet');

The schema is inferred from the Parquet metadata of the first matched file. All matched files must share the same schema.

S3 with inline credentials

Pass credentials directly in the SQL statement. This is the primary mechanism used by sqe-bench load to inject credentials at load time without relying on environment variables or configuration files.

SELECT * FROM read_parquet(
  's3://bench-data/tpch/sf1/lineitem/*.parquet',
  access_key => 'AKIAIOSFODNN7EXAMPLE',
  secret_key => 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY',
  endpoint   => 'http://localhost:9000',
  region     => 'us-east-1'
);

All four named parameters are optional independently. Omit endpoint for AWS S3 (uses the default AWS endpoint for the given region). Omit region to default to us-east-1.

S3 with default credentials

When no inline credentials are provided, read_parquet() falls back to the storage configuration in sqe.toml:

-- Uses [storage] section from sqe.toml
SELECT * FROM read_parquet('s3://bench-data/tpch/sf1/lineitem/*.parquet');

This is convenient for internal workloads where the engine already has ambient S3 credentials configured.

Glob patterns

read_parquet() supports the same glob syntax as object_store:

For S3 paths, globbing is applied to the key prefix after the bucket name.

Using with CTAS for data loading

The primary use case for read_parquet() is ingesting external Parquet data into Iceberg tables via CTAS. This avoids an intermediate format conversion step — the Parquet files are read directly and written as Iceberg data files in one operation.

-- Load TPC-H lineitem from local disk
CREATE TABLE tpch_sf1.lineitem AS
SELECT * FROM read_parquet('/data/tpch/sf1/lineitem/*.parquet');

-- Load from S3 with inline credentials
CREATE TABLE tpch_sf1.lineitem AS
SELECT * FROM read_parquet(
  's3://bench-data/tpch/sf1/lineitem/*.parquet',
  access_key => 'AKIA...',
  secret_key => '...',
  endpoint   => 'http://localhost:9000',
  region     => 'us-east-1'
);

-- Transform during load
CREATE TABLE analytics.orders_summary AS
SELECT
  o_orderdate,
  o_orderstatus,
  COUNT(*) AS order_count,
  SUM(o_totalprice) AS total_revenue
FROM read_parquet('/data/tpch/sf1/orders/*.parquet')
GROUP BY o_orderdate, o_orderstatus;

Because read_parquet() returns a standard DataFusion table scan, it participates in the full optimizer pipeline: predicate pushdown, projection pruning, and partition pruning all apply.

Querying without loading

read_parquet() can also be used as a one-off query target, without creating an Iceberg table:

-- Inspect schema
DESCRIBE SELECT * FROM read_parquet('/data/tpch/sf1/orders/*.parquet') LIMIT 0;

-- Quick aggregation over raw Parquet
SELECT o_orderstatus, COUNT(*) AS cnt
FROM read_parquet('/data/tpch/sf1/orders/*.parquet')
GROUP BY o_orderstatus;

-- Join Parquet with an Iceberg table
SELECT p.p_name, l.l_quantity
FROM read_parquet('/data/tpch/sf1/lineitem/*.parquet') AS l
JOIN warehouse.tpch_sf1.part AS p ON l.l_partkey = p.p_partkey
LIMIT 20;

Implementation

read_parquet() is registered in sqe-catalog (or sqe-functions) as a DataFusion TableFunction. On each invocation:

1. The path argument is parsed to detect s3:// vs local (/ or file://) paths.
2. For S3: an AmazonS3Builder is constructed from the inline named parameters, with fallback to the StorageConfig from sqe-core for any omitted fields.
3. For local paths: the built-in DataFusion local filesystem ObjectStore is used.
4. Glob patterns are expanded against the chosen ObjectStore.
5. A ListingTable is constructed over the matched files and returned as the table scan node.

The function is registered on every SessionContext at startup, so it is always available without any special configuration.

Limitations

• All matched Parquet files must share an identical Arrow schema. Schema evolution across files in the same glob is not supported.
• read_parquet() is read-only. It cannot be used as the target of an INSERT INTO.
• Credential parameters are passed as SQL literals. Avoid logging or displaying these queries in audit logs without redaction. SQE’s audit logger redacts named parameter values that match access_key, secret_key, and session_token patterns.
• Very large numbers of matched files (>10,000) may cause slow planning due to the object listing step.

File-format TVFs

The four TVFs read_parquet, read_csv, read_json, and read_delta query files directly without registering an external table. They share a uniform calling convention and a uniform path-resolution layer (local filesystem, S3, HTTPS, HuggingFace hf://).

This chapter covers read_csv, read_json, and read_delta. The dedicated read_parquet chapter covers Parquet specifically.

Common path forms

All four TVFs accept the same path shapes:

-- Local
SELECT * FROM read_csv('/data/sales.csv');

-- S3 (anywhere object_store understands)
SELECT * FROM read_csv('s3://bucket/key.csv',
    access_key => 'AKIA...', secret_key => '...',
    endpoint => 'http://localhost:9000', region => 'us-east-1');

-- HTTP / HTTPS (V10)
SELECT * FROM read_csv('https://raw.githubusercontent.com/.../data.csv');

-- HuggingFace (V10)
SELECT * FROM read_csv('hf://datasets/squad/plain_text/train.csv');

-- HuggingFace with revision (V12.1)
SELECT * FROM read_parquet('hf://datasets/foo/[email protected]/data.parquet');

-- HuggingFace auto-generated parquet view
SELECT * FROM read_parquet('hf://datasets/foo/bar@~parquet/default/train/0.parquet');

S3 credentials default to the engine’s [storage] block when not supplied inline. HTTPS and hf:// paths flow through V10’s LazyHttpObjectStoreRegistry, which constructs an HttpStore for the host on first request.

Quoted-string auto-detect

V8 introduced a shortcut. With the embedded CLI, the engine recognises a quoted string in a FROM clause as a file URL and dispatches to the right TVF based on extension:

SELECT * FROM '/data/sales.parquet';
SELECT * FROM 's3://bucket/orders.csv';
SELECT * FROM 'hf://datasets/foo/bar/data.csv';

Format dispatch happens by extension. .parquet -> read_parquet, .csv / .tsv / .psv / .ssv -> read_csv, .json / .jsonl / .ndjson -> read_json, .avro -> the Avro reader. Compressed extensions are recognised: .csv.gz, .tsv.zst, .json.bz2 all dispatch to the right reader with the right codec.

read_csv

SELECT * FROM read_csv(
    '<path>',
    [delimiter | delim | sep => '<byte>',]
    [has_header | header => '<bool>',]
    [quote => '<byte>',]
    [escape => '<byte>',]
    [comment => '<byte>',]
    [null_regex | nullstr => '<regex>',]
    [compression | compress => 'auto|none|gzip|bz2|xz|zstd',]
    [file_extension => '<.ext>']
);

Smart defaults:

• Delimiter detected from the path extension. .csv is ,, .tsv is tab, .psv is |, .ssv is ;. Compression suffixes are stripped first, so .tsv.gz still picks tab.
• Compression detected from the path extension. .gz, .bz2, .xz, .zst are recognised.
• has_header defaults to true (DataFusion default).

DuckDB-style aliases: sep, delim for delimiter; header for has_header; nullstr for null_regex; compress for compression.

-- All three are equivalent
SELECT * FROM read_csv('events.tsv');
SELECT * FROM read_csv('events.tsv', sep => '\t');
SELECT * FROM read_csv('events.tsv', delimiter => '\t', has_header => 'true');

-- Compressed, with explicit override
SELECT * FROM read_csv('events.tsv.zst', compression => 'auto');

-- Semicolon-separated file
SELECT * FROM read_csv('financial.ssv', sep => ';');

read_json

SELECT * FROM read_json(
    '<path>',
    [access_key | secret_key | endpoint | region | file_extension]
);

Reads NDJSON (one JSON document per line). Schema inference samples the first batch.

SELECT * FROM read_json('/var/log/events.jsonl');
SELECT * FROM read_json('s3://logs/2026-05-07/events.json.gz');
SELECT * FROM read_json('hf://datasets/nyu-mll/glue/cola/train.jsonl');

read_delta

SELECT * FROM read_delta(
    '<path>',
    [access_key | secret_key | endpoint | region,]
    [version => '<u64>',]
    [timestamp => '<RFC3339>']
);

Read-only Delta Lake reader, V11. Wraps deltalake-core 0.32.1. Time travel via version (snapshot id) or timestamp (RFC3339); the two are mutually exclusive.

SELECT * FROM read_delta('/data/delta/sales');

SELECT * FROM read_delta('s3://bucket/delta/orders',
    access_key => 'AKIA...');

-- Time travel
SELECT * FROM read_delta('/data/delta/sales', version => '5');
SELECT * FROM read_delta('/data/delta/sales',
    timestamp => '2026-04-01T00:00:00Z');

Writes are not exposed. The Delta transaction pipeline is substantial; the read path covers the most common ad-hoc query case.

HuggingFace specifics

The hf:// path uses a slightly different shape than S3 or HTTPS because HuggingFace expects a revision in the URL.

Two revision spellings work:

1. Inline @<rev> (DuckDB-style):

   SELECT * FROM read_parquet('hf://datasets/foo/[email protected]/train.parquet');
   

2. Query parameter ?revision=<rev>:

   SELECT * FROM read_parquet('hf://datasets/foo/bar/train.parquet?revision=v1.0');
   

Default is main when neither is specified. Specifying both rejects with a clear error.

@~parquet is special. HuggingFace auto-generates a Parquet conversion of every dataset on the refs/convert/parquet branch. The TVF translates this:

-- Equivalent to https://huggingface.co/datasets/foo/bar/resolve/refs%2Fconvert%2Fparquet/data.parquet
SELECT * FROM read_parquet('hf://datasets/foo/bar@~parquet/default/train/0.parquet');

Glob expansion (**/*.parquet) is on the V12.2 roadmap; today the path must point to a specific file.

When to use which

• read_parquet: ad-hoc queries against Parquet on disk, S3, HTTPS, or hf://. Anything Iceberg-aware that does not need the catalog.
• read_csv: ETL ingestion, log analysis, dataset preview before deciding to load into Iceberg.
• read_json: NDJSON logs, HuggingFace train.jsonl style splits.
• read_delta: query a Delta Lake table without converting to Iceberg.

For tables with metadata that you want to write back to, register them in a catalog. The TVFs are reads only.

Implementation references

• sqe-catalog/src/read_parquet.rs
• sqe-catalog/src/read_csv.rs
• sqe-catalog/src/read_json.rs
• sqe-catalog/src/read_delta.rs
• sqe-catalog/src/file_tvf_common.rs: shared parsing + S3 / HTTPS / hf:// resolver
• sqe-catalog/src/lazy_object_store.rs: V10’s lazy HTTPS object-store registry

Observability

SQE provides comprehensive observability through Prometheus metrics, OpenTelemetry traces/logs, and structured audit logging.

Metrics (Prometheus)

Available at http://coordinator:9090/metrics in Prometheus text format.

Histogram buckets: 10ms, 25ms, 50ms, 100ms, 250ms, 500ms, 1s, 2.5s, 5s, 10s, 30s, 60s.

Statement types: query, ctas, insert, merge, delete, drop, create_view, create_schema, show_catalogs, show_schemas, show_tables, policy, utility.

Example Queries (PromQL)

# Query rate (queries per second)
rate(sqe_query_count_total[5m])

# Error rate
rate(sqe_query_count_total{status="error"}[5m])

# P99 query duration
histogram_quantile(0.99, rate(sqe_query_duration_seconds_bucket[5m]))

# Active sessions
sqe_active_sessions

Health Endpoints

Available on port 9091 (metrics port + 1) for both coordinator and workers.

Kubernetes Probes

Cluster Status (Ballista/DataFusion-style)

GET /api/v1/status returns a JSON snapshot of the node and cluster:

{
  "status": "ACTIVE",
  "node": {
    "role": "coordinator",
    "version": "0.1.0",
    "datafusionVersion": "51",
    "uptimeSeconds": 3600
  },
  "workers": {
    "total": 2,
    "healthy": 2,
    "healthyUrls": ["http://worker-0:50052", "http://worker-1:50052"]
  }
}

For worker nodes, the workers field is null.

Trino-Compatible Info (port 8080)

When the Trino compat layer is enabled, standard Trino info endpoints are available on the Trino HTTP port:

These endpoints are compatible with Trino JDBC drivers, DBeaver, and other Trino-aware tools for auto-detecting node state.

OpenTelemetry

When otlp_endpoint is configured, SQE exports traces, metrics, and logs via OTLP/gRPC:

graph LR
    SQE["sqe-server"] -->|OTLP gRPC| COLL["OTel Collector"]
    COLL --> JAEGER["Jaeger<br/>(traces)"]
    COLL --> PROM["Prometheus<br/>(metrics)"]
    COLL --> LOKI["Loki<br/>(logs)"]

Configuration:

[metrics]
otlp_endpoint = "http://otel-collector:4317"

When the endpoint is empty (default), SQE falls back to structured JSON logs on stdout — no external dependency required.

Trace Spans

Key spans emitted:

• sqe.query.execute — full query lifecycle
• sqe.query.plan — SQL parsing and planning
• sqe.policy.evaluate — policy enforcement
• sqe.flight.do_get — result streaming
• sqe.auth.handshake — authentication
• sqe.worker.scan — worker scan execution

Audit Log

SQE writes a JSONL audit log capturing every query:

{
  "timestamp": "2025-03-15T10:30:00Z",
  "username": "alice",
  "session_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "query_text": "SELECT * FROM sales.orders WHERE region = 'EU'",
  "query_hash": "sha256:e3b0c44298fc1c149afb...",
  "statement_type": "query",
  "client_ip": "10.0.1.42",
  "duration_ms": 142,
  "rows_returned": 1583,
  "status": "ok"
}

The query_hash field is a SHA-256 hash of the SQL text, useful for correlating repeated queries without storing the full text. When audit logging is enabled, all fields are always present.

Configuration:

[metrics]
audit_log_path = "/var/log/sqe/audit.jsonl"

When the path is empty (default), audit logging is disabled (no-op).

Structured Logging

All SQE components use tracing with JSON output:

{
  "timestamp": "2025-03-15T10:30:00.142Z",
  "level": "INFO",
  "target": "sqe_coordinator::query_handler",
  "message": "Query executed",
  "user": "alice",
  "statement_type": "query",
  "duration_ms": 142,
  "rows": 1583
}

Log level controlled via RUST_LOG environment variable:

RUST_LOG=info             # Default
RUST_LOG=sqe=debug        # Debug SQE crates only
RUST_LOG=sqe=trace        # Everything

Kubernetes Integration

The Helm chart includes optional ServiceMonitor for Prometheus Operator:

serviceMonitor:
  enabled: true
  interval: 30s
  labels:
    release: prometheus

Trino Compatibility

SQE includes a Trino-compatible HTTP endpoint that allows existing Trino clients (JDBC drivers, DBeaver, CLI tools) to connect without modification.

Enabling

The Trino HTTP endpoint is enabled by default on port 8080. Set port to 0 to disable:

[coordinator]
trino_http_port = 8080    # 0 to disable

Endpoints

Authentication

The Trino endpoint supports two authentication methods:

Bearer Token

Pass an existing access token directly:

curl -H "Authorization: Bearer eyJhbG..." \
     -H "X-Trino-User: alice" \
     -d "SELECT 1" \
     http://localhost:8080/v1/statement

Basic Auth

Username and password are exchanged for a token via the configured OIDC/OAuth2 backend:

curl -u alice:password \
     -d "SELECT 1" \
     http://localhost:8080/v1/statement

Client Headers

SQE respects standard Trino client headers:

Result Pagination

Query results are paginated. The initial response includes a nextUri field. Follow nextUri links to retrieve subsequent pages:

{
  "id": "query-uuid",
  "stats": { "state": "FINISHED" },
  "columns": [{"name": "result", "type": "integer"}],
  "data": [[1]],
  "nextUri": "http://localhost:8080/v1/statement/query-uuid/1"
}

When nextUri is absent, all results have been consumed.

Using with the CLI

sqe-cli --protocol http --host localhost --port 8080 --user alice

Connecting External Tools

DBeaver

1. Create a new Trino connection
2. Host: localhost, Port: 8080
3. Authentication: Username/Password
4. Driver properties: no special settings needed

JDBC (Java)

String url = "jdbc:trino://localhost:8080";
Properties props = new Properties();
props.setProperty("user", "alice");
props.setProperty("password", "secret");
Connection conn = DriverManager.getConnection(url, props);

Limitations

• The Trino endpoint returns results as JSON (Trino wire format), not Arrow. For maximum performance, use Flight SQL.
• Prepared statements are not supported via the Trino protocol.
• Transaction control (START TRANSACTION, COMMIT) is not supported — queries execute in auto-commit mode.
• Type mapping covers common types; complex nested types may differ from native Trino behavior.

Flight SQL vs Trino HTTP

Use Flight SQL for performance-sensitive workloads. Use Trino HTTP for compatibility with existing tools.

Benchmark Suite

SQE ships with sqe-bench, a Rust CLI tool that generates benchmark data, loads it into SQE via the read_parquet() TVF, and runs query suites to validate SQL correctness and measure query performance.

For the longitudinal view (every benchmark JSON in benchmarks/results/ plotted across time, per-suite, per-scale, per-query heatmaps), see docs/benchmark/. Charts auto-regenerate from the committed JSONs via make benchmark-charts.

Available Benchmarks

Why these benchmarks? Each covers a different slice of SQL correctness:

• TPC-H and SSB validate the analytical core: joins, aggregates, GROUP BY, ORDER BY, date arithmetic. TPC-H is the standard first check for any SQL engine.
• TPC-DS is the hardest. Its 99 queries exercise correlated subqueries, CTEs, window functions, GROUPING SETS, and complex multi-table joins. Passing TPC-DS well means the engine handles real analytical workloads.
• TPC-C and TPC-E cover OLTP patterns: point lookups, small aggregates, indexed access by key ranges, plus write operations (DELETE, UPDATE) exercised via Copy-on-Write.
• TPC-BB exercises semi-structured data alongside the TPC-DS schema — useful for validating string functions and JSON handling.

Generating Data

The generate command produces Parquet files on local disk or S3. Data is deterministic (seeded) so results are reproducible.

# Generate TPC-H at scale factor 1 (~1 GB, 8 tables)
cargo run -p sqe-bench -- generate tpch --scale 1 --output ./data

# Scale factor 10 (~10 GB)
cargo run -p sqe-bench -- generate tpch --scale 10 --output ./data

# Write directly to S3
cargo run -p sqe-bench -- generate tpch --scale 1 \
  --output s3://bench-data/ \
  --s3-access-key AKIA... \
  --s3-secret-key ... \
  --s3-endpoint http://localhost:9000 \
  --s3-region us-east-1

# Generate all benchmarks at once
./scripts/benchmark-generate-all.sh

Scale factors explained

The scale factor controls dataset size. Scale factor 1 produces roughly 1 GB for TPC-H and TPC-DS; SSB is ~600 MB at SF1.

Files are split at 128 MB for parallelism. Output is structured as:

./data/
└── tpch/
    └── sf1/
        ├── lineitem/
        │   ├── part-0000.parquet
        │   └── part-0001.parquet
        ├── orders/
        │   └── part-0000.parquet
        └── ... (8 tables total)

Loading Data

The load command connects to SQE and creates Iceberg tables using read_parquet() + CTAS. No intermediate format conversion is needed — Parquet files are read directly and written as Iceberg.

# Load TPC-H from local disk
cargo run -p sqe-bench -- load tpch \
  --scale 1 \
  --data ./data \
  --host localhost \
  --port 60051 \
  --username root \
  --password ""

# Load from S3
cargo run -p sqe-bench -- load tpch \
  --scale 1 \
  --data s3://bench-data/ \
  --s3-access-key AKIA... \
  --s3-secret-key ... \
  --s3-endpoint http://localhost:9000 \
  --s3-region us-east-1 \
  --host localhost \
  --port 60051 \
  --username root \
  --password ""

# Drop and recreate tables before loading
cargo run -p sqe-bench -- load tpch --scale 1 --data ./data --clean \
  --host localhost --port 60051 --username root --password ""

# Use the Trino HTTP protocol instead of Flight SQL
cargo run -p sqe-bench -- load tpch --scale 1 --data ./data \
  --protocol trino \
  --host localhost \
  --port 8080 \
  --username root \
  --password ""

The loader creates a namespace named <benchmark>_sf<N> (e.g., tpch_sf1) and sends one CTAS statement per table:

CREATE TABLE tpch_sf1.lineitem AS
SELECT * FROM read_parquet('/data/tpch/sf1/lineitem/*.parquet');

For S3 sources, inline credentials are injected:

CREATE TABLE tpch_sf1.lineitem AS
SELECT * FROM read_parquet(
  's3://bench-data/tpch/sf1/lineitem/*.parquet',
  access_key => 'AKIA...',
  secret_key => '...',
  endpoint => 'http://localhost:9000',
  region => 'us-east-1'
);

See read_parquet TVF for full syntax documentation.

Running Tests

The test command executes all queries in the benchmark suite against the loaded data and reports correctness and timing.

# Run all TPC-H queries (Flight SQL, default)
cargo run -p sqe-bench -- test tpch \
  --scale 1 \
  --host localhost \
  --port 60051 \
  --username root \
  --password ""

# Run a single query
cargo run -p sqe-bench -- test tpch --scale 1 --query q03 \
  --host localhost --port 60051 --username root --password ""

# Use Trino HTTP protocol
cargo run -p sqe-bench -- test tpch --scale 1 \
  --protocol trino \
  --host localhost \
  --port 8080 \
  --username root \
  --password ""

# Run all benchmarks end-to-end
./scripts/benchmark-test.sh tpch
./scripts/benchmark-test.sh tpcds
./scripts/benchmark-test.sh ssb

Query result statuses

DIFF is not treated as a failure in CI — it is a signal for investigation. Decimal precision differences are expected when comparing float-heavy aggregates across different engines.

Queries can declare their requirements in a header comment:

-- name: Revenue by nation
-- requires: delete, merge
-- timeout: 30s
SELECT ...

Any query with -- requires: will be SKIPped if SQE does not support that feature, rather than FAILing the suite.

Understanding Results

Terminal output

TPC-H SF1 — Flight SQL (localhost:50051)
─────────────────────────────────────────
q01  PASS   1.23s   6001215 rows
q02  PASS   0.45s       460 rows
q03  PASS   0.89s     11620 rows
...
q17  DIFF   2.10s         1 rows  (decimal precision)
q22  PASS   0.33s         7 rows

Results: 20/22 PASS, 1 DIFF, 1 SKIP
Total time: 28.4s
Report: benchmarks/results/tpch-sf1-flight-2026-03-24T14:30:00.json

JSON report format

Reports are written to benchmarks/results/<benchmark>-sf<N>-<protocol>-<timestamp>.json:

{
  "benchmark": "tpch",
  "scale_factor": 1,
  "protocol": "flight",
  "host": "localhost:50051",
  "timestamp": "2026-03-24T14:30:00Z",
  "summary": {
    "total": 22,
    "pass": 20,
    "fail": 0,
    "diff": 1,
    "skip": 1,
    "error": 0,
    "total_duration_ms": 28400
  },
  "queries": [
    {
      "id": "q01",
      "status": "pass",
      "duration_ms": 1230,
      "rows": 6001215
    },
    {
      "id": "q17",
      "status": "diff",
      "duration_ms": 2100,
      "rows": 1,
      "diff_detail": "decimal precision mismatch: expected 1.0000, got 0.9999"
    }
  ]
}

JSON reports are machine-readable and suitable for tracking regressions over time in CI.

Historical Performance Tracking

Benchmark JSON results are committed to benchmarks/results/ for historical comparison. This enables tracking performance regressions and improvements across releases.

TPC-H SF1 — Historical Comparison (Apr 2 baseline vs. Apr 6 streaming execution)

After implementing the streaming execution engine (coordinator spill-to-disk, late materialization, file-level pruning, S3 I/O pipeline, distributed execution), TPC-H SF1 improved 3.1x on a distributed cluster (coordinator + 2 workers) compared to the single-node baseline:

Query   Apr 2 (single-node)   Apr 6 (distributed)   Speedup
────────────────────────────────────────────────────────────
q01               3.21s                1.29s     2.5x
q02               0.89s                0.27s     3.3x
q03               2.23s                0.94s     2.4x
q04               1.14s                0.32s     3.6x
q05               1.89s                0.55s     3.4x
q06               1.13s                0.30s     3.7x
q07               2.07s                0.85s     2.4x
q08               1.81s                0.54s     3.4x
q09               1.78s                0.60s     3.0x
q10               2.47s                0.63s     3.9x
q11               0.74s                0.11s     6.8x
q12               1.71s                0.57s     3.0x
q13               1.10s                0.18s     6.1x
q14               1.46s                0.55s     2.7x
q15               2.24s                0.72s     3.1x
q16               0.75s                0.10s     7.4x
q17               1.89s                0.63s     3.0x
q18               3.19s                0.74s     4.3x
q19               1.68s                0.79s     2.1x
q20               1.39s                0.53s     2.6x
q21               2.11s                0.68s     3.1x
q22               0.67s                0.09s     7.7x
────────────────────────────────────────────────────────────
TOTAL             37.5s                12.0s     3.1x

Key observations:

• Metadata-light queries (q11, q13, q16, q22) see 6-8x speedup — footer cache, file pruning, and scan distribution eliminate I/O overhead
• Scan-heavy queries (q01, q03, q07) see 2-2.5x speedup — proportional to worker count (2 workers)
• q18 (the hardest TPC-H query) improved from 3.19s to 0.74s (4.3x) — benefits from distributed aggregation across workers
• Single-node with 512MB spill: 21/22 pass — only q18 fails due to DataFusion hash aggregate memory limitation (DF#17334). With 1GB+ memory or with workers, all 22 pass.

Full Benchmark Matrix (Apr 7, 2026 — SF1)

Spill behavior across configs

The counterintuitive finding: 8GB spills more than 512MB. This is because 8GB successfully runs TPC-E queries that 512MB cannot — those TPC-E queries involve massive multi-table joins that produce 27GB of intermediate sorted data. With 512MB, the same queries OOM before reaching the spill point.

With distribution (2 workers), spill drops to 49MB. Workers handle scan and partial aggregation; the coordinator only merges small result sets.

Scheduler observations

At SF1, all distributed queries ran locally on the coordinator (scheduler_decisions{local}=120+). This is correct — SF1 tables have 1-2 data files each, below the distribution threshold (default: 4 files). The 2.5x speedup comes from streaming execution improvements (spill, scan planning), not from worker distribution. To observe actual worker distribution, run at SF10+ where tables have 10+ files.

TPC-E: the outlier

TPC-E has the lowest pass rate (56-72%) across all configs:

• 5 queries blocked by DataFusion’s IN(subquery) limitation (cannot decorrelate)
• Deep join chains across 33 tables produce massive intermediate results
• Some queries timeout at 600s after spilling 27GB

Metrics gaps

Several Phase A/B metrics show 0 because the increment calls are not yet wired into the execution path (the infrastructure exists but metric.inc() calls are missing):

• Footer cache hits/misses — FooterCache not wired into IcebergScanExec
• File pruning counts — PruningPredicate built but counter not incremented
• Late materialization bytes — RowFilter wired but byte tracking not connected
• Time to first row — histogram registered but not observed

These are wiring tasks for the next iteration.

Comparing against Trino

The benchmark harness can run the same suite against a real Trino on the same data, so you can compare SQE and Trino directly. There are two modes:

• Correctness parity — --compare-trino diffs SQE’s results against Trino’s row-for-row. This is how SQL correctness is validated at scale, not just timing. Small decimal differences on float-heavy aggregates are expected and flagged for investigation rather than treated as failures.
• Timing — the same run records per-query wall-clock for both engines, so a head-to-head speed comparison falls out of the parity run.

SQE’s own distributed execution path (coordinator + workers, spill-to-disk, late materialization, file-level pruning) gives a measured ~3.1× speedup over single-node on TPC-H SF1, with metadata-light queries seeing more.

Run a comparison yourself and see the captured numbers in the benchmark quickstart: Benchmarks: TPC-H / TPC-DS / SSB, or in the repo under benchmarks/.

CI/CD Integration

All three scripts support automated use without a TTY:

# Generate data once (idempotent — skip if files exist)
./scripts/benchmark-generate-all.sh

# Load all benchmarks
./scripts/benchmark-load.sh

# Run and report
./scripts/benchmark-test.sh tpch
./scripts/benchmark-test.sh tpcds
./scripts/benchmark-test.sh ssb

# Exit code is 0 if all queries are PASS or SKIP
# Exit code is 1 if any query is FAIL or ERROR

A typical CI pipeline runs TPC-H at SF1 as a smoke test on every PR, and the full suite (TPC-H + TPC-DS + SSB) nightly.

Query Files

Query SQL files are stored under benchmarks/queries/<benchmark>/:

benchmarks/
├── queries/
│   ├── tpch/     q01.sql ... q22.sql
│   ├── tpcds/    q01.sql ... q99.sql
│   ├── ssb/      q1.1.sql ... q4.3.sql
│   ├── tpcc/     order_status.sql, stock_level.sql, ...
│   ├── tpce/     trade_lookup.sql, customer_position.sql, ...
│   └── tpcbb/    q01.sql ... q10.sql
├── expected/
│   ├── tpch/sf1/    q01.csv ... q22.csv
│   └── ...
└── schemas/
    ├── tpch.sql
    ├── tpcds.sql
    └── ...

Expected results under benchmarks/expected/ are CSV files containing the correct output at the given scale factor. They are committed to the repository and used for regression checking.

Adding New Benchmarks

Implement the BenchmarkGenerator trait in sqe-bench/src/generate/:

#![allow(unused)]
fn main() {
pub trait BenchmarkGenerator {
    fn name(&self) -> &str;
    fn tables(&self) -> Vec<TableDef>;
    fn generate_table(
        &self,
        table: &str,
        scale: f64,
        output: &dyn ObjectStore,
        prefix: &str,
    ) -> Result<GenerateStats>;
}

pub struct TableDef {
    pub name: String,
    pub schema: Arc<Schema>,             // Arrow schema
    pub row_count_fn: fn(f64) -> usize,  // scale factor → row count
}
}

Steps to add a benchmark:

1. Create sqe-bench/src/generate/<name>.rs implementing BenchmarkGenerator.
2. Add SQL query files under benchmarks/queries/<name>/.
3. Add expected result CSVs under benchmarks/expected/<name>/sf1/.
4. Register the generator in sqe-bench/src/generate/mod.rs.
5. Add the benchmark name to the CLI subcommand list in sqe-bench/src/cli.rs.
6. Add a schema DDL file under benchmarks/schemas/<name>.sql for reference.

Supported catalog backends

SQE supports multiple Iceberg catalog backends and wire-protocol client adapters. The same binary works with every option below; choose by setting the [catalog.backend] block in your TOML. Catalog weight is opt-in through cargo features: a default REST-only build pulls no AWS SDK, no Thrift, no sqlx. Add glue, s3tables, hms, or sql as needed; rest and hadoop are always available.

For per-backend TOML configuration, credential setup, and troubleshooting, see Catalog backends.

Polaris (and any Iceberg REST catalog)

Apache Polaris exposes the Iceberg REST catalog specification. SQE uses this as its default backend. Any Iceberg REST-compatible service (Polaris, Lakeformation REST, custom) works with the same config block.

[catalog]
polaris_url = "http://localhost:18181/api/catalog"
warehouse   = "test_warehouse"

See the quickstart: Polaris + Keycloak (client credentials).

AWS Glue Data Catalog

Native AWS SDK integration against the regional Glue Data Catalog. Credentials come from the standard provider chain (AWS_PROFILE, instance profile, SSO).

[catalog.backend]
type      = "glue"
region    = "eu-example-1"
warehouse = "s3://my-bucket/warehouse"

See the quickstart: AWS Glue Data Catalog.

AWS S3 Tables

Managed Iceberg via the federated Glue Iceberg REST endpoint with AWS SigV4 authentication on every request. Backed by the vendored iceberg-catalog-rest crate with the aws-sigv4 feature.

[catalog.backend]
type             = "s3tables"
table_bucket_arn = "arn:aws:s3tables:eu-example-1:ACCOUNT:bucket/NAME"

See the quickstart: AWS S3 Tables (managed Iceberg).

Unity Catalog OSS

Unity Catalog OSS exposes an Iceberg REST adapter at /api/2.1/unity-catalog/iceberg/. The OSS image is read-only on create/drop; use for query workloads.

docker compose -f docker-compose.unity.yml up -d
set -a; source .env; set +a
cargo test -p sqe-catalog --test backends_integration -- --ignored unity_catalog::

See the quickstart: Unity Catalog OSS (Iceberg REST, read-only).

Hive Metastore

Thrift metastore protocol. Requires the hms cargo feature.

docker compose -f docker-compose.hms.yml up -d
set -a; source .env; set +a
cargo test -p sqe-catalog --features hms --test backends_integration -- --ignored hms::

Covered by the suite (hms::live_hms_namespace_round_trip, a create / list / drop round-trip against the Thrift metastore).

Project Nessie

Git-like Iceberg REST catalog with branch/tag semantics.

docker compose -f docker-compose.nessie.yml up -d
set -a; source .env; set +a
cargo test -p sqe-catalog --test backends_integration -- --ignored nessie::

See the quickstart: Project Nessie (Iceberg REST catalog).

Hadoop (filesystem warehouse, no catalog service)

No metadata service. SQE walks the warehouse prefix for metadata.json files. See Embedded mode for the catalog-free embedded flow.

[catalog.backend]
type      = "hadoop"
warehouse = "s3://my-bucket/warehouse"

Quack (DuckDB wire protocol)

Quack is not an Iceberg catalog; it is a client/server wire protocol that lets DuckDB clients (and other Quack-compatible tools) issue SQL to SQE and receive Arrow-serialised results. It sits alongside the Iceberg catalog backends, not in competition with them.

See Quack.

How catalog backends are tested

• sqe-catalog/tests/backends_integration.rs: live round-trips per backend (create / list / drop namespace, or read smoke), gated on #[ignore] plus the .env warehouse variables.
• sqe-catalog/tests/mount_*_test.rs: mount-time validation per backend (rejects bad secrets, requires a warehouse, and so on).

Embedded mode

SQE can run the full query engine in-process, with no server and no network. sqe-cli in embedded mode starts DataFusion, the Iceberg reader, and the same SQL planner locally — inside the CLI process. This is the fastest path for querying a warehouse from a laptop, a CI job, or a script.

Four warehouse modes:

• In-memory (--memory): a transient DataFusion catalog. Nothing is persisted. Good for ad-hoc SQL and testing SQL functions.
• Filesystem warehouse (--warehouse PATH): an Iceberg warehouse on local disk or object storage with no catalog service. SQE walks the path for metadata.json files and treats the prefix as the catalog. The “Iceberg without a catalog” case.
• Persistent SQLite catalog (--catalog-backend sqlite): a durable single-node catalog backed by a local SQLite file. Survives restarts.
• Cloud catalogs embedded: Glue and S3 Tables can be attached directly, with no coordinator, using the standard AWS credential chain.

See the quickstarts:

• Embedded: query local and remote files
• Embedded: persistent local catalog (SQLite)
• Embedded: attach multiple catalogs

In-memory

sqe-cli --embedded --memory -e "SELECT 1 AS one"

+-----+
| one |
+-----+
| 1   |
+-----+

Filesystem warehouse (no catalog service)

Point at a directory; SQE reads the Iceberg metadata directly. No Polaris, no Glue, no metastore.

sqe-cli --embedded --warehouse /data/warehouse \
    -e "SELECT COUNT(*) FROM sales.orders"

This is the catalog-free Hadoop mode. Writes need atomic rename, which object stores do not all provide, so this mode is read-oriented; for writes use a real catalog. The same backend powers the [catalog.backend] type = "hadoop" server config.

Cloud catalogs embedded

The embedded engine can attach a Glue or S3 Tables catalog directly, with no coordinator. Pass --catalog-backend plus the cloud warehouse; credentials come from the standard AWS provider chain (AWS_PROFILE, instance profile, SSO). These catalogs attach read-only (query, not write); use the server for writes. Requires the aws cargo feature (default-on).

# AWS Glue Data Catalog (warehouse is an s3:// prefix)
AWS_PROFILE=analytics sqe-cli --embedded \
    --catalog-backend glue \
    --catalog-warehouse s3://my-bucket/warehouse --region eu-example-1 \
    -e "SELECT * FROM glue.analytics.events LIMIT 10"

# AWS S3 Tables (warehouse is the table-bucket ARN)
AWS_PROFILE=analytics sqe-cli --embedded \
    --catalog-backend s3tables \
    --catalog-warehouse arn:aws:s3tables:eu-example-1:ACCOUNT:bucket/NAME \
    --region eu-example-1 \
    -e "SHOW SCHEMAS"

The catalog mounts under the backend name by default (glue. / s3tables.); override with --catalog-name.

How it is tested

• crates/sqe-cli/tests/cli_smoke.rs: binary-level flag parsing, exit codes, mutually-exclusive flag rejection, and the --embedded --memory happy path.
• The catalog spec parser (NAME=PATH) is validated for empty names, missing separators, and dotted names.

Notes

• --memory and --warehouse are mutually exclusive.
• Local-path TVFs (read_csv and friends) work in embedded mode; the embedded engine enables allow_local_paths so a laptop user can read local files.
• Embedded mode authenticates the OS user against the configured catalog’s credential source, not OIDC; there is no server to pass tokens through.

Flight SQL

Arrow Flight SQL is SQE’s primary wire protocol. It is columnar end to end: results come back as Arrow record batches over gRPC, with no row-by-row serialization tax. Every official client (the sqe-cli, ADBC drivers, the dbt-sqe adapter) speaks it. This page covers both topologies against Apache Polaris.

Single server

One coordinator parses, plans, and executes. Good for development, small deployments, and the default Docker image.

Prerequisites

docker compose -f docker-compose.test.yml up -d
./scripts/bootstrap-test.sh

This brings up Polaris (http://localhost:18181) and an S3-compatible store, then creates the test_warehouse warehouse and the default / test_ns namespaces.

Configuration

The coordinator reads one TOML file. The Flight SQL listener is on 50051:

[auth]
token_endpoint = "http://localhost:18181/api/catalog/v1/oauth/tokens"
client_id      = "root"

[catalog]
polaris_url = "http://localhost:18181/api/catalog"
warehouse   = "test_warehouse"

[storage]
s3_endpoint   = "http://localhost:19000"
s3_region     = "us-east-1"
s3_path_style = true

Run

# Start the coordinator (Flight SQL on 50051).
./target/release/sqe-server --config sqe.toml

# Connect with the CLI over Flight.
./target/release/sqe-cli --protocol flight --host localhost --port 50051 \
    -u root -e "SELECT 1 AS one"

Expected output

 one
-----
 1
(1 row)

The in-process equivalent is exercised by integration_test.rs: test_authentication (OIDC client-credentials against Polaris), test_simple_select, and the file-format TVF tests all run the full Flight SQL query path against this stack.

How it is tested

• crates/sqe-coordinator/tests/integration_test.rs (run with ./scripts/integration-test.sh): authentication, SELECT, CTAS round-trip, information_schema, and read_parquet/read_csv/read_json.

Distributed (coordinator + workers)

The coordinator parses and plans, then ships secured plan fragments to stateless workers over Arrow Flight. Workers hold no catalog state; they receive the plan and the user’s bearer token and execute.

Prerequisites

docker compose -f docker-compose.test.yml -f docker-compose.distributed.yml up --build -d
./scripts/bootstrap-test.sh

This adds a coordinator (Flight SQL on 60051) and two workers (internal 50052, exposed as 60061 / 60062) on the shared Polaris and storage.

Run

./scripts/distributed-test.sh

The script builds sqe-cli, runs SQL over Flight on 60051, and verifies worker dispatch through system.runtime.tasks (proving fragments actually reach the workers rather than silently falling back to local execution).

Expected output

The script runs a sequence of SQL statements over Flight on 60051 and prints a pass line per check, ending with the worker-dispatch verification. The single-node Flight path (auth, CTAS, SELECT against live Polaris) was re-run this round through the TVF integration tests (3/3, 23.6s); the distributed harness is covered by the suite and the committed benchmark baselines (TPC-H SF1 distributed 22/22 in 12.0s, 3.1x over single node).

How it is tested

• crates/sqe-coordinator/tests/integration_test.rs::test_distributed_select (ignored by default; needs workers listening on :50052).
• scripts/distributed-test.sh: full coordinator + worker harness.
• scripts/concurrent-test.sh: N parallel Flight clients, cache behaviour.

Notes

• Auth is bearer-token passthrough: the CLI authenticates the user via OIDC, and the token rides to Polaris and S3. There is no service account.
• The internal Flight port is 50051; compose maps it to 60051 to avoid colliding with a local coordinator.
• Workers are stateless. Scale by adding worker replicas; the coordinator load-balances fragments across registered workers.

Trino HTTP compatibility

SQE speaks enough of the Trino HTTP protocol that Trino clients, JDBC drivers, and BI tools can point at it unchanged. The coordinator exposes the /v1/statement endpoint with nextUri pagination, /v1/info, and /v1/info/state. This is a compatibility surface, not a re-implementation of Trino; it covers the query-submission path that clients actually use.

Single server

Prerequisites

docker compose -f docker-compose.test.yml up -d
./scripts/bootstrap-test.sh

Configuration

The Trino HTTP listener runs alongside Flight SQL. The default port is 8080 (the test/distributed compose files map it to 28080):

[trino_compat]
enabled = true
port    = 8080

Run

# Submit a query. Basic auth carries the user; the password is the OIDC secret
# (empty for the local root client).
curl -s -u root: \
  -H "X-Trino-User: root" \
  -d "SELECT 1 AS one" \
  http://localhost:28080/v1/statement

Trino clients follow the nextUri field until results are exhausted. A JDBC client connects with the Trino driver against http://localhost:28080.

Expected output

The first response carries a nextUri; following it returns the data:

{
  "columns": [{"name": "one", "type": "bigint"}],
  "data": [[1]],
  "stats": {"state": "FINISHED"}
}

How it is tested

• crates/sqe-coordinator/tests/integration_test.rs::test_trino_http_query: server startup, Basic auth, /v1/statement POST, pagination.
• test_trino_type_mapping, test_trino_batches_to_json: Arrow to Trino JSON.
• scripts/trino-parity-test.sh and scripts/trino-compat-test.sh: run the same SQL against SQE and a real Trino and diff the results.

Distributed

The Trino HTTP endpoint lives on the coordinator. Distribution across workers is identical to the Flight path: the coordinator plans, workers execute. The client sees a single Trino-compatible endpoint regardless of worker count.

Prerequisites

docker compose -f docker-compose.test.yml -f docker-compose.distributed.yml up --build -d
./scripts/bootstrap-test.sh

Run

./scripts/distributed-test.sh

The script exercises the Trino HTTP endpoint on 28080 (test 11) alongside the Flight path and confirms worker dispatch.

Expected output

distributed-test.sh test 11 submits a query to the Trino HTTP endpoint on the coordinator (28080) and follows nextUri to completion, alongside the Flight path on the same cluster. This is covered by the suite; the docker-dependent re-run was not repeated this round (see the validation matrix note on local Docker capacity).

Trino SQL parity

SQE adds a Trino-compatibility function layer (date/time helpers like year(), month(), day_of_week(), JSON casts, and more) so dbt models and Trino SQL run with fewer rewrites. The current parity surface is tracked in Trino Compatibility. The parity scripts above are the regression guard.

Notes

• Authentication needs Basic auth (-u user:password) to populate the session, not just the X-Trino-User header. For the local root client the password is empty (-u root:).
• The Trino layer is optional; enable it with [trino_compat] enabled = true. Flight SQL is the recommended protocol for SQE-native clients.

SQL Reference

A function-by-function and statement-by-statement reference for everything SQE can parse and execute. Every entry lists where the implementation lives so you can jump from “what does this do” to “where do I read the code”.

The reference focuses on what ships in the running engine, not the SQL standard in the abstract. If a function is not listed here, it is not registered in our SessionContext.

How to read these tables

Every page uses the same column shape. The first three columns describe the function in SQE; the four right columns describe how the same idea looks in other engines.

Origins

Every function in SQE has exactly one origin. Eight values appear:

The two registration entry points are crates/sqe-coordinator/src/session_context.rs (cluster mode) and crates/sqe-cli/src/embedded.rs (single-binary mode). Both register the same UDFs / UDTFs in the same order, so a function works the same way in both personas.

Pages

Scalar functions

• Conditional and null-handling: if, iff, case, coalesce, nullif, greatest, least, nvl, nvl2, typeof, try.
• String: concat, substring, trim, lower, upper, regex, normalisation, split, format, hash digests.
• Math: trig, rounding, logs, exponents, sign, modular, base conversion.
• Date and time: timestamp construction, extraction, formatting, parsing, arithmetic, time-zone handling.
• Array, map, struct: the 40+ functions from datafusion-functions-nested plus Trino aggregate constructors (map_agg, histogram).
• JSON: two layered surfaces: Trino-named (json_extract, json_parse) and the datafusion-functions-json json_get_* family.
• Encoding, hashing, URL: base64, hex, md5, sha224..512, url_extract_*, url_encode, url_decode.

Aggregate and window

• Aggregate functions: count, sum, avg, statistical, regression, array_agg, string_agg / listagg, histogram, map_agg, approximation.
• Window functions: row_number, rank, lag, lead, first_value, frame syntax (ROWS BETWEEN, RANGE BETWEEN, GROUPS BETWEEN).

Table-valued functions

• Table-valued functions: file format (read_parquet, read_csv, read_json, read_delta), Iceberg metadata (table_snapshots, table_history, table_files, table_partitions, table_manifests, table_refs), generators (generate_series, unnest).

Statements

• DDL: CREATE, ALTER, DROP for tables, schemas, views; partition evolution; branches and tags; column defaults.
• DML: SELECT, INSERT, UPDATE, DELETE, MERGE, COPY TO, TRUNCATE, time travel (FOR VERSION AS OF, FOR SYSTEM_TIME AS OF, FOR INCREMENTAL BETWEEN), SET WRITE_BRANCH.
• CALL procedures: system.rewrite_data_files, expire_snapshots, remove_orphan_files, rewrite_manifests, suggest_bloom_filter_columns.
• GRANT and REVOKE: SQE-specific security extensions. GRANT MASKED WITH, GRANT ROWS WHERE, SHOW GRANTS, SHOW EFFECTIVE GRANTS, CHECK ACCESS.
• SHOW and EXPLAIN: metadata queries and plan inspection. SHOW CATALOGS, SHOW STATS, EXPLAIN FULL.
• Operators: arithmetic, string, comparison, null tests, casting (::), set membership.

Embedded CLI

• Dot-commands: .help, .tables, .schema, .describe, .summarize, .timer, .read, .format. Embedded CLI only.

What is intentionally not in SQE

Some functions appear in the dialect comparison columns as missing. The reasoning:

• PIVOT, UNPIVOT, QUALIFY, ASOF JOIN, FROM-first syntax: DataFusion’s parser does not accept them. Tracked upstream.
• Lambda expressions, list comprehensions: DataFusion has no AST node for closures.
• Oracle / Snowflake DECODE: name collides with DataFusion’s binary decode(input, encoding) helper. CASE WHEN covers the use case.
• IIF (T-SQL): covered by if (Trino) and iff (Snowflake), both registered.
• postgres_table_scanner, mysql_table_scanner, sqlite_scanner: out of scope. SQE is Iceberg-first; if you need a non-Iceberg engine, query it where it lives.
• spatial, vss, fts, excel: niche. Use a tool built for the job (PostGIS, a vector DB, an FTS engine).

The full DuckDB-comparison audit lives at duckdb-comparision.md. The Trino-comparison audit lives at trino-compatibility.md.

Conditional and null-handling

Functions for choosing between values, replacing nulls, and inspecting types. Most are scalar UDFs; CASE WHEN is a SQL expression handled by the planner.

Function table

Patterns

Replace NULL with a default

SELECT coalesce(comment, 'no comment') FROM orders;
SELECT nvl(comment, 'no comment') FROM orders;          -- two-arg shorthand

Treat empty strings as NULL

SELECT coalesce(nullif(name, ''), 'unknown') FROM users;

nullif(name, '') returns NULL when name is the empty string, then coalesce substitutes the default.

Branch on a boolean

SELECT
    iff(amount > 1000, 'large', 'small') AS bucket,    -- Snowflake
    if(amount > 1000, 'large', 'small') AS bucket_t    -- Trino
FROM orders;

Both calls produce the same result. Use whichever matches your team’s existing dbt models. dbt-snowflake projects ported to SQE keep iff() working unmodified.

Complex branching: prefer CASE

SELECT
    CASE
        WHEN amount < 100 THEN 'small'
        WHEN amount < 1000 THEN 'medium'
        ELSE 'large'
    END AS bucket
FROM orders;

Reach for CASE when there are more than two branches or the condition is not a single boolean expression.

Take the safer cast

SELECT try(CAST(payload AS BIGINT)) AS amount FROM events;

try() swallows the conversion error and returns NULL for rows that fail. Without it, one bad row aborts the query.

Type promotion

coalesce, greatest, least, if, iff all coerce arguments to a common supertype. The rules follow SQL standard widening: integer + decimal -> decimal; integer + double -> double; date + timestamp -> timestamp. If the arguments have no common supertype the planner returns an error before execution.

case is stricter: every branch must produce the same type, or the planner adds explicit casts when it can. Mixed types without an obvious supertype fail at plan time.

NULL handling cheat sheet

Why no IIF (T-SQL)

T-SQL’s IIF(cond, then, else) is the same shape as iff. SQE registers iff (Snowflake) and if (Trino), both pointing at the same implementation, so a T-SQL IIF rename is the only change needed. We deliberately did not register a third name to keep the function table tight.

Why no Oracle / Snowflake DECODE

Snowflake’s DECODE(expr, search1, result1, ..., default) is a multi-way conditional with NULL = NULL match semantics. Two reasons it is not in SQE:

1. The name collides with DataFusion’s built-in decode(input, encoding), which decodes base64 / hex strings to binary. Registering a Snowflake-style DECODE under the same name would shadow the encoding helper and break any existing callsite.
2. CASE WHEN expr IS NOT DISTINCT FROM s1 THEN r1 ... END covers the same ground in standard SQL. (IS NOT DISTINCT FROM treats NULL = NULL as true.)

The audit row lives in features.md so the conflict is visible.

String functions

DataFusion contributes ~35 string functions plus a unicode submodule and a regex submodule. SQE adds Trino-name aliases and a few extras.

Concatenation

Length and offsets

Slicing

Trimming

Case and padding

Predicates

Regular expressions

Hashing

Distance / phonetic

Encoding

See Encoding, hashing, URL for from_base64, to_base64, from_hex, to_hex, encode, decode.

Unicode normalization

Examples

Cleanse user input

SELECT
    initcap(trim(lower(name))) AS name_clean,
    regexp_replace(email, '\\s+', '') AS email_clean,
    coalesce(nullif(trim(comment), ''), 'no comment') AS comment_clean
FROM users;

Extract domain from URL

SELECT
    url,
    regexp_extract(url, 'https?://([^/]+)', 1) AS host_via_regex,
    url_extract_host(url) AS host_via_helper
FROM access_logs;

url_extract_host is dramatically faster than the regex version. Prefer it whenever the input is well-formed URLs.

Tokenise a sentence

SELECT id, token
FROM articles, UNNEST(split(body, ' ')) AS t(token)
WHERE length(token) > 3;

Mask sensitive data

SELECT
    user_id,
    sha256(email) AS email_hash,
    concat('***', right(phone, 4)) AS phone_masked
FROM users;

(For declarative masking via grants, see GRANT and REVOKE.)

What is NOT supported

• SOUNDEX_DIFFERENCE (T-SQL). use levenshtein(soundex(a), soundex(b)).
• PARSE_URL family with named parts. Use the url_extract_* family in Encoding, hashing, URL.
• STRING_SPLIT_TO_ARRAY. use split(s, delim).

Math functions

DataFusion contributes ~30 math functions. SQE adds a small set of Trino-named extras (e(), mod(), truncate(), sign()) plus base conversion and IEEE specials (infinity, nan).

Sign and rounding

Powers, logs, roots

Trigonometry

Modular and bit / base

SELECT to_base(255, 16);     -- 'ff'
SELECT from_base('ff', 16);  -- 255
SELECT to_base(8, 2);        -- '1000'

Random

IEEE specials

Statistical helpers

For aggregates (stddev, variance, corr, covar_*, regr_*), see Aggregate functions. The math page covers scalars only.

Examples

Bucketing and binning

SELECT
    floor(amount / 100) * 100 AS bucket,
    count(*)
FROM orders
GROUP BY 1
ORDER BY 1;

Geometric mean via logs

SELECT exp(avg(ln(price))) AS geo_mean FROM products WHERE price > 0;

DataFusion has no built-in geo_mean; the log identity is the standard workaround.

Distance from a reference point (Pythagorean)

SELECT
    name,
    sqrt(pow(x - 100, 2) + pow(y - 200, 2)) AS distance
FROM points
ORDER BY distance
LIMIT 10;

Hex and binary representations

SELECT
    n,
    to_base(n, 16) AS hex,
    to_base(n, 2)  AS bin,
    to_base(n, 8)  AS oct
FROM generate_series(0, 255) AS t(n);

Numeric type promotion

pow, log, exp always return Double regardless of input. abs, floor, ceil, round preserve the input type. +, -, * follow SQL standard widening: integer + decimal -> decimal; integer + double -> double; decimal + decimal -> decimal with combined precision.

/ between two integers in DataFusion returns Double, not integer. For integer division use floor(a / b) or div(a, b).

Decimal precision

DECIMAL(p, s) arithmetic widens precision per SQL standard. Two DECIMAL(18, 2) values multiplied produce DECIMAL(36, 4). Going beyond DECIMAL(38, ...) overflows; CAST or use Double.

Date and time

The largest single category in SQE. Two layers stack:

1. DataFusion native functions: date_part, date_trunc, date_bin, extract, now, the to_timestamp_* family, make_date. Powerful but uses DataFusion-specific names.
2. Trino aliases registered by sqe-trino-functions: year(), month(), day(), date_add(), date_diff(), format_datetime(). These cover every Trino date function so dbt-trino models work unmodified.

Snowflake and Spark also map well via the Trino layer.

Construction and current value

Extraction (year, month, day, …)

These return integer parts. SQE registers Trino names; DataFusion’s extract and date_part are also available for SQL standard syntax.

Truncation and binning

Arithmetic

date_add example:

SELECT date_add('day', 7, DATE '2026-05-08');     -- 2026-05-15
SELECT date_add('month', -1, DATE '2026-05-08');  -- 2026-04-08
SELECT date_add('hour', 36, TIMESTAMP '2026-05-08 09:00:00');  -- 2026-05-09 21:00:00

date_diff example:

SELECT date_diff('day', DATE '2026-01-01', DATE '2026-01-06');   -- 5
SELECT date_diff('month', DATE '2026-01-01', DATE '2026-04-01'); -- 3
SELECT date_diff('year', DATE '2020-01-01', DATE '2026-01-01');  -- 6

Formatting and parsing

Two parallel formatting families:

• Trino / MySQL %Y-%m-%d style via date_format / date_parse. Familiar to anyone who’s used strftime.
• Java / Joda yyyy-MM-dd style via format_datetime / parse_datetime. The Trino default for new code.

Trino-style example:

SELECT date_format(TIMESTAMP '2026-05-08 14:30:00', '%Y-%m-%d %H:%i:%s');
-- 2026-05-08 14:30:00

SELECT date_parse('2026-05-08 14:30:00', '%Y-%m-%d %H:%i:%s');
-- 2026-05-08 14:30:00 (TIMESTAMP)

Java-style example (preferred for new code):

SELECT format_datetime(TIMESTAMP '2026-05-08 14:30:00', 'yyyy-MM-dd HH:mm:ss');
-- 2026-05-08 14:30:00

Unix epoch

Time zones

Misc

Worked example

A daily revenue rollup, time-zone aware, formatted for human reports:

SELECT
    date_format(
        with_timezone(date_trunc('day', order_ts), 'Europe/Amsterdam'),
        '%Y-%m-%d'
    ) AS day_local,
    quarter(order_ts) AS qtr,
    day_of_week(order_ts) AS dow,
    SUM(amount) AS revenue
FROM orders
WHERE order_ts >= TIMESTAMP '2026-01-01 00:00:00' AT TIME ZONE 'UTC'
GROUP BY 1, 2, 3
ORDER BY 1;

Iceberg V3 nanosecond timestamps

TIMESTAMP_NS and TIMESTAMP_NS WITH TIME ZONE only exist in Iceberg format-version 3. Adding such a column to a CREATE TABLE auto-bumps the table format version. All datetime functions operate on these the same way they operate on TIMESTAMP(6); the underlying Arrow type is Timestamp(Nanosecond, ...) instead of Timestamp(Microsecond, ...).

When you query a V3 ns column from a Trino client, Trino downscales to microseconds. SQE keeps the full precision when serving Arrow Flight SQL clients.

Array, map, struct

DataFusion’s datafusion-functions-nested crate ships ~40 array and map helpers. SQE adds Trino-named aggregate constructors (map_agg, histogram, multimap_agg).

Array construction

Array inspection

Array transformation

Array set operations

Array reductions

Array unnesting (lateral)

-- One row per (order, item) pair
SELECT order_id, item
FROM orders, UNNEST(items) AS t(item);

-- Numbered
SELECT order_id, item, idx
FROM orders, UNNEST(items) WITH ORDINALITY AS t(item, idx);

Map functions

Aggregates that build maps / arrays

See Aggregate functions for array_agg, map_agg, histogram, multimap_agg, map_union. The names differ slightly across engines:

Struct / row

SELECT named_struct('host', host, 'port', port) AS endpoint
FROM servers;

SELECT endpoint.host, endpoint.port FROM ...;

Examples

Tag-set membership

-- Find products with both 'sale' and 'new' tags
SELECT * FROM products
WHERE array_has_all(tags, ARRAY['sale', 'new']);

-- Find products with any of the listed tags
SELECT * FROM products
WHERE array_has_any(tags, ARRAY['sale', 'clearance']);

Top-K frequencies via histogram

SELECT k, v
FROM events, UNNEST(map_keys(histogram(event_type)), map_values(histogram(event_type))) AS t(k, v)
ORDER BY v DESC
LIMIT 10;

Build a map from joined tables

SELECT
    user_id,
    map_agg(setting_key, setting_value) AS preferences
FROM user_settings
GROUP BY user_id;

map_agg errors on duplicate keys. For multimap-style behaviour use multimap_agg.

Lateral pattern: filter then unnest

SELECT order_id, tag
FROM orders, UNNEST(tags) AS t(tag)
WHERE order_id > 100 AND tag LIKE 'priority_%';

Lambda functions

DataFusion’s parser does not support lambda syntax (x -> x + 1). Trino, Spark, DuckDB do. The audit rows in features.md note this. Workarounds:

• Pre-compute via a CTE plus unnest.
• Use map_filter / transform from datafusion-functions-nested once parser support lands upstream.

What is NOT registered

• zip(a, b) (parallel-iterate two arrays). Use unnest against an indexed pair instead.
• reduce(a, init, lambda, finish). Aggregate within a CTE instead.
• Snowflake flatten table function (with PATH and OUTER options). Use UNNEST directly.

These are tracked but blocked on lambda support in DataFusion.

JSON

Two layered JSON surfaces. Both are registered in every session.

1. Trino-named layer in sqe-trino-functions: json_extract, json_extract_scalar, json_array_length, json_parse, json_format, json_object, is_json_scalar, json_array_contains, json_size, json_array_get, to_json. Maps directly to dbt-trino model expectations.
2. DataFusion JSON layer via datafusion-functions-json (registered in session_context.rs:361 and embedded.rs:172): json_get, json_get_str, json_get_int, json_get_float, json_get_bool, json_get_json, json_get_array, json_contains, json_as_text, json_length. Type-specific extractors that avoid an outer CAST.

JSON columns are stored as VARCHAR (Iceberg has no native JSON type yet). Both layers parse on every call, so for hot paths consider extracting once into a typed column.

Trino-named layer

DataFusion JSON layer

These return typed scalars directly, which means no outer CAST and DataFusion can push predicates through.

Path syntax

The Trino layer uses JSONPath ($.foo.bar[0]).

The DataFusion JSON layer uses single-step keys: a string for object access or an integer for array access.

-- Trino-style: full JSONPath
SELECT json_extract_scalar(payload, '$.user.id')
FROM events;

-- DataFusion: chained single-step
SELECT json_get_str(json_get_json(payload, 'user'), 'id')
FROM events;

Both compose. Both work on the same VARCHAR column. Choose by ergonomics:

• Trino-style is one call per path; cleaner SQL.
• DataFusion JSON layer returns native types; no outer CAST.

For a deeply nested path used in a hot WHERE clause, the DataFusion layer wins on speed (no JSON re-encoding between steps).

Comparison

Examples

Extract a typed field for filtering

-- Slowpath: parse JSON, cast to int, compare
SELECT * FROM events
WHERE CAST(json_extract_scalar(payload, '$.user_id') AS BIGINT) = 12345;

-- Faster: typed extractor, no CAST
SELECT * FROM events
WHERE json_get_int(json_get_json(payload, 'user'), 'id') = 12345;

Project several fields at once

SELECT
    json_extract_scalar(payload, '$.user.id')        AS user_id,
    json_extract_scalar(payload, '$.user.email')     AS email,
    json_extract_scalar(payload, '$.session.locale') AS locale,
    CAST(json_extract_scalar(payload, '$.amount') AS DECIMAL(18, 2)) AS amount
FROM events;

Path-existence filter

SELECT * FROM events
WHERE json_contains(payload, '$.metadata.experimental_flag') = true;

Build a structured field for downstream consumers

SELECT json_object(
    'id', id,
    'host', url_extract_host(url),
    'path', url_extract_path(url),
    'received_at', cast(occurred_at AS VARCHAR)
) AS event_doc
FROM events;

Performance tips

• Decode once at the boundary: when a JSON column is queried in many downstream views, consider materialising the relevant subfields into typed columns at ingest time. Repeated parsing is the biggest JSON cost.
• Path index pruning: the DataFusion JSON layer can push json_get_str(payload, 'user_id') = 'X' through to a manifest min/max statistic when the underlying JSON has been pre-extracted. The Trino layer is opaque.
• Avoid round-trip: json_format(json_parse(j)) is a no-op semantically but burns CPU. Skip the round-trip unless you need re-canonicalisation.

Iceberg JSON column type

Iceberg V3 has no native JSON primitive yet. SQE accepts JSON in CREATE TABLE and stores it as VARCHAR underneath. Predicates and projections work as text. When Iceberg adds a JSON primitive, the storage layout will change but the SQL surface stays the same.

What is not registered

• json_array(...) (Trino name for build-an-array). Use cast(make_array(...) AS VARCHAR) followed by json_format, or to_json(make_array(...)).
• json_size at the root: covered. With a path: covered.
• JSONPath wildcards ($..foo, $[*]): not supported by DataFusion’s path parser. Use unnest(json_get_array(...)) for array-level wildcarding.

Encoding, hashing, URL

Three small, related families: binary encoding (base64 / hex), cryptographic hashes, URL parsing. SQE inherits the encoding and crypto helpers from DataFusion and adds Trino-named aliases plus eight URL extractors.

Binary encoding

SELECT to_base64(CAST('hello' AS bytea));         -- 'aGVsbG8='
SELECT from_base64('aGVsbG8=');                   -- bytea -> 'hello'
SELECT encode(CAST('hi' AS bytea), 'hex');        -- '6869'
SELECT decode('6869', 'hex');                     -- bytea -> 'hi'

Cryptographic hashes

SELECT sha256(email) AS email_hash FROM users;
SELECT digest('hello', 'blake3');       -- 64-char blake3 hex
SELECT to_hex(checksum(payload)) FROM events;

URL parsing

All eight URL functions live in sqe-trino-functions. They wrap the url crate from crates.io for correct RFC 3986 handling.

SELECT
    url,
    url_extract_protocol(url) AS proto,
    url_extract_host(url)     AS host,
    url_extract_port(url)     AS port,
    url_extract_path(url)     AS path,
    url_extract_query(url)    AS query,
    url_extract_parameter(url, 'utm_source') AS utm
FROM events;

For example, on https://example.com:8443/api?utm_source=newsletter&page=2:

When to use which hash

When to use which URL parser

url_extract_host, url_extract_query, url_extract_parameter use a real RFC 3986 parser, so they handle internationalised domain names (IDN), unusual ports, percent-encoded query values, and trailing fragments correctly. The regex equivalent (regexp_extract(url, 'https?://([^/]+)', 1)) breaks on these edge cases.

Use the regex form only when the input is known to be a fixed shape that the regex covers.

Why no Snowflake-style PARSE_URL

Snowflake’s parse_url(url [, permissive]) returns an OBJECT with named keys. The eight separate url_extract_* functions are equivalent in expressive power and easier to compile statically. dbt-snowflake to dbt-sqe migrations need the rewrite, but it is mechanical: parse_url(u, 'HOST') becomes url_extract_host(u), etc.

Why DataFusion’s decode is not shadowed

A Snowflake-style DECODE(expr, search1, result1, ...) would conflict with DataFusion’s binary decode(input, encoding). SQE keeps the binary one and exposes the conditional logic via CASE WHEN. See Conditional for the trade-off.

Aggregate functions

Functions used in GROUP BY queries and OVER clauses. SQE inherits ~40 aggregates from DataFusion plus 12 Trino UDAFs from sqe-trino-functions.

Standard aggregates

Statistical / regression

Distinct and approximation