Distributed Query | Documentation Spice.ai Enterprise Documentation | Spice.ai Cloud Documentation

Spice.ai Enterprise supports distributed query execution, built on Apache Ballista and Apache DataFusion, for horizontally scaling SQL workloads across multiple nodes. A cluster runs in multi-active mode with shared state in an S3-compatible object store, executes queries with partition-aware routing across executors, and supports distributed accelerations with per-partition data ownership and write-through semantics.

Architecture

A distributed query cluster is composed of three tiers:

Schedulers — Coordinate query planning, partition discovery and assignment, async job submission, and federated task_history aggregation.
Executors — Hold accelerated data for their assigned partitions, execute query stages, perform shuffles, and stream results.
Object store — S3-compatible storage that holds shared cluster state (cluster.json, scheduler heartbeats, async job results, and Cayenne acceleration snapshots).

     ┌──────────────────────────────────────┐
     │           Load Balancer              │
     └──────────────────────────────────────┘
                        │
     ┌──────────────────┼──────────────────┐
     ▼                  ▼                  ▼
┌──────────┐     ┌──────────┐     ┌──────────┐
│Scheduler │     │Scheduler │     │Scheduler │◄──► Object Store
└──────────┘     └──────────┘     └──────────┘
     ▲                  ▲                  ▲
     │                  │                  │
┌──────────┐     ┌──────────┐     ┌──────────┐
│ Executor │     │ Executor │     │ Executor │
└──────────┘     └──────────┘     └──────────┘

Schedulers register themselves and discover peers through the object store. Executors are shared across all schedulers — they are not bound to or owned by any single scheduler. On startup, an executor connects to its bootstrap scheduler, fetches the full scheduler membership list, and then opens a Ballista task poll_loop plus a persistent bidirectional ControlStream to every scheduler over the internal gRPC port (50052). It refreshes the membership list every 10 s and adds/drops connections as schedulers join or leave. Any scheduler can dispatch tasks to any executor, and an executor's partition assignment is owned by the cluster (persisted in the object store), not by the scheduler that allocated it.

Multi-Active High Availability

Multiple schedulers run simultaneously without an external coordinator:

Shared state — Cluster membership and per-table partition metadata are persisted in cluster.json in the configured object store. All mutations use optimistic concurrency control (ETag / If-Match) with up to 8 retry attempts; the schema is versioned (CLUSTER_STATE_SCHEMA_VERSION = 1) and incompatible versions are rejected.
Heartbeats — Each scheduler writes a small heartbeats/{scheduler_id}.json file independent of cluster.json to avoid contention. Heartbeats refresh at one third of the configured TTL (default ~10 s for a 30 s TTL).
Discovery — Every scheduler refreshes its peer map every 5 s by reading cluster.json plus live heartbeat files.
Reaper — Schedulers whose last heartbeat exceeds ttl_ms + 5 s clock-skew tolerance are evicted on the next reaper tick (jittered ±20 % around ttl_ms).
Takeover safety — Each scheduler process generates a fresh instance_id (UUID) at startup, so a restarted process can safely reclaim its previous scheduler_id without colliding with itself.
Executor failover — Executors maintain a poll loop (Fibonacci backoff, max 5 s) and a ControlStream to every scheduler concurrently. If a scheduler dies, the executor's other connections remain live and surviving schedulers continue to drive partition assignment and refresh commands without re-binding the executor.

The object store backing state_location must support conditional writes. Native AWS S3, S3-compatible stores with PutIfNotExists/PutIfMatch semantics, and the file:// backend (for local development) are supported.

Configuration

Spicepod (`runtime.scheduler`)

The presence of runtime.scheduler activates scheduler role on a node. Setting --scheduler-address at launch time activates executor role.

runtime:
  scheduler:
    state_location: "s3://my-bucket/spice-cluster/"
    params:
      region: us-east-1
      auth: iam_role                         # iam_role (default) | key
      key: ${secrets:AWS_ACCESS_KEY_ID}
      secret: ${secrets:AWS_SECRET_ACCESS_KEY}
    partition_assignment_interval: "30s"
    max_partition_assignments_per_interval: 100
    max_partitions_per_executor: 1000
    partition_discovery_timeout: "60s"

CLI flags

The same spiced binary runs as scheduler or executor; the role is selected by flag combination.

Flag

Default

Description

--role

inferred

scheduler or executor. Inferred as executor when --scheduler-address is set.

--node-bind-address

0.0.0.0:50052

Internal gRPC bind address for both roles.

--node-advertise-address

required

Hostname/IP this node advertises to peers. Forms scheduler_id as {advertise}:{port}.

--scheduler-address

required (executor)

URL of the scheduler's internal gRPC service. Scheme inferred from TLS configuration.

--node-mtls-ca-certificate-file

—

PEM CA used to verify peer mTLS certificates.

--node-mtls-certificate-file

—

PEM server + client certificate for this node.

--node-mtls-key-file

—

Private key for the mTLS certificate.

--allow-insecure-connections

false

Disable mTLS. Development / test only.

Manual launch (non-Kubernetes)

# Scheduler
spiced --role scheduler \
  --node-mtls-ca-certificate-file ca.pem \
  --node-mtls-certificate-file scheduler.pem \
  --node-mtls-key-file scheduler-key.pem \
  --node-advertise-address scheduler1.cluster.local

# Executor
spiced --role executor \
  --node-mtls-ca-certificate-file ca.pem \
  --node-mtls-certificate-file executor.pem \
  --node-mtls-key-file executor-key.pem \
  --scheduler-address https://scheduler1.cluster.local:50052 \
  --node-advertise-address executor1.cluster.local

Kubernetes (recommended)

Use the SpicepodCluster CRD. The Spice Operator provisions mTLS certificates, child SpicepodSet resources for the scheduler and executor pools, services, and PodMonitors automatically. See High Availability for AZ spread, anti-affinity, and PodDisruptionBudget guidance.

Internal gRPC (port 50052)

The internal ClusterService gRPC surface is mTLS-protected and never exposed externally. All cluster coordination flows through it:

RPC

Caller

Purpose

GetAppDefinition

Executor at startup

Fetches the full Spicepod definition (datasets, catalogs, views, UDFs) so executors do not need a local manifest.

ExpandSecret

Executor at startup

Resolves a secret key through the scheduler's secret store.

GetSchedulers

Executor at startup

Returns the list of live scheduler advertise addresses; executor opens a poll loop to each.

AllocateInitialPartitions

Executor at startup

Returns the executor's assigned per-table partition filter expressions (serialized DataFusion Expr).

ControlStream (bidirectional)

Executor → Scheduler

Carries heartbeats and metric responses; receives UpdatePartitions, PollNow, RefreshDataset, and CancelTasks commands.

GetTaskHistory

Scheduler → peer schedulers

Federated runtime.task_history fan-out across the cluster.

GetMetrics

Scheduler → peers / executors

On-demand OTLP metrics collection.

Partitioning, Sharding, and Partition-Aware Queries

Distributed Spice clusters shard accelerated tables horizontally across executors using user-declared partition keys. Query planning is partition-aware: each executor only scans the partitions it owns, and the scheduler unions and merges results.

Declaring partitions

Every accelerated dataset and view in cluster mode must declare at least one partition key via acceleration.partition_by. Startup fails otherwise with:

Accelerated {component_type} '{name}' has no partition keys configured. Add 'partition_by' to its acceleration config to participate in cluster partition assignment.

Each entry is a SQL expression over the source schema. Entries can be anonymous, named, or column references:

datasets:
  - from: s3://lake/sales/
    name: sales
    acceleration:
      enabled: true
      engine: cayenne                       # required for write-through; any engine works for read-only
      partition_by:
        - "YEAR(order_date)"                # anonymous expression -> name `expr0`
        - year: "YEAR(order_date)"          # named expression
        - region: "region"                  # column value as partition key
        - "bucket(100, customer_id)"        # static hash bucketing; no source query required

The bucket(N, col) function is treated as a static partition key: values 0..N-1 are enumerated without querying the source.

Partition discovery and assignment

Partition assignment is coordinated through the object store: cluster.json is the single source of truth for which executor owns which partition. Schedulers never assign partitions purely from in-memory state, and executors never own a partition that is not durably committed to the object store first.

The scheduler runs a PartitionAssignmentTask on partition_assignment_interval (default 30 s):

Discover — For each accelerated table, runs SELECT DISTINCT {expressions} FROM {federated_source} against the source connector (within partition_discovery_timeout). Static bucket(N, col) keys skip the query.
Diff — Compares discovered values against cluster.json (accelerations sub-map). New partitions are recorded as unassigned; stale partitions are removed.
Assign — Picks executors for unassigned partitions using a greedy minimum set-cover algorithm, respecting max_partitions_per_executor and max_partition_assignments_per_interval. Tie-breaks are deterministic by executor ID.
Commit — Writes the assignment (assigned_executors field of each partition entry) to cluster.json via an OCC conditional write (up to 8 retries). Concurrent schedulers serialize on this write: if two schedulers race on the same partition, one commits first and the other re-reads fresh state and retries, so an executor cannot be assigned the same partition twice.
Notify — After the object-store commit succeeds, the scheduler pushes an UpdatePartitions message over the executor's ControlStream so it can immediately materialize the new partitions locally without waiting to re-read cluster.json.

How executors learn their assignments

Phase

Mechanism

Startup / restart

Executor calls AllocateInitialPartitions (RPC pull) on its bootstrap scheduler, which reads from cluster.json and returns the executor's assigned partition filter expressions per table.

Ongoing changes

Scheduler commits the change to cluster.json (OCC), then pushes UpdatePartitions over the ControlStream. The executor updates its in-memory partition_assignments map and registers/removes local accelerations accordingly.

Authoritative state

cluster.json in the object store. The executor's in-memory map and the scheduler's PartitionStore cache are derived views; both are rebuilt from the object store after a process restart.

Partition-aware query planning

The DataFusion analyzer rule PartitionedTableScanRewrite (scheduler-only) rewrites every TableScan on an accelerated table into a UNION ALL over per-executor FlightSQL scans, pushing down the executor's partition filter and any user predicates:

Before:
  TableScan: sales [filters: status = 'Disputed']

After:
  UNION ALL
    TableScan: sales@executor-1 [filters: status='Disputed' AND year=2024 AND region='us-east']
    TableScan: sales@executor-2 [filters: status='Disputed' AND year=2024 AND region='us-west']
    TableScan: sales@executor-3 [filters: status='Disputed' AND year=2025 AND region='us-east']

When the plan contains a Limit → Sort → Union pattern (top-K), the Sort is pushed into each union leg and the executor returns at most Limit rows before the scheduler performs a final merge-sort.

Executor selection uses a greedy minimum set-cover algorithm: pick the executor that covers the most still-required partitions, repeat until coverage is complete, break ties deterministically by executor ID. If any required partition is unassigned, the query fails with:

Cannot execute query: N partition(s) not assigned to any executor

On executors, the AcceleratedPartitionProvider resolves partitions to local TableProviders. Row-level partition predicates are not re-evaluated on executors — each executor only holds its own partitions, so filtering happens by ownership.

Partition-aware writes

Write-through INSERT, UPDATE, DELETE, and MERGE INTO flow through the same partition-aware Arrow Flight DoPut path on the scheduler:

The scheduler decodes the inbound DoPut schema header.
For each RecordBatch, it evaluates the table's partition expressions against every row to classify rows into partitions.
Rows for partitions already assigned to an executor are streamed via that executor's FlightSQL DoPut.
Rows whose partition values are new are assigned on-the-fly to the least-loaded executor (recorded in cluster.json via OCC), then forwarded.
All per-executor sub-streams run concurrently. Idle streams emit a keepalive sentinel to avoid the DoPut idle timeout (120 s default; override with SPICE_DO_PUT_IDLE_TIMEOUT_SECS).

Write-through is currently constrained to the Cayenne accelerator (see Distributed Accelerations).

Constraints

Every accelerated table and view in cluster mode must declare partition_by.
Partition expressions must be deterministic SQL over the source schema.
bucket(N, col) is the supported static hash-partition function.
The scheduler holds no accelerated data; partition discovery queries always go to the federated source.
The DataFusion physical optimizer rule EnsureSupportedFileScan rejects plans containing file-scan nodes that cannot be distributed.

Distributed Accelerations

Full reference: Distributed Accelerations.

In cluster mode, accelerated data is sharded across executors: each executor materializes only the partitions it owns. The scheduler's view of the table is a logical UNION ALL across executors; it never holds row data.

Engine support

Engine

Cluster partition assignment

write_mode: write_through

Notes

Cayenne

✓

Required for write-through. Vortex storage with SQLite metadata. Supports acceleration snapshots.

DuckDB

✓

—

Read-only partitioned acceleration.

Arrow (in-memory)

✓

—

Read-only. Data is lost on pod restart unless backed by snapshots.

SQLite

✓

—

Read-only.

Postgres

✓

—

Read-only.

Attempting write_mode: write_through with a non-Cayenne engine fails fast at startup with Write-through acceleration currently requires the Cayenne accelerator.

Per-executor sharding

Each partition has a single owning executor (1:1 assignment in cluster.json). Executor-local accelerations only contain rows that match their assigned partition filter expressions. Read paths fan out across executors; write paths route by partition key.

Refresh in cluster mode

The scheduler is the single source of truth for partition ownership; there is no separate refresh leader election.
After every source refresh cycle, the PartitionAssignmentTask re-runs discovery; new partition values are assigned, and UpdatePartitions is pushed to the receiving executor, which materializes the new partition.
The scheduler can also explicitly issue RefreshDatasetCommand over the ControlStream to trigger a refresh on a specific executor, which is how API-driven refresh requests are routed.

Acceleration snapshots (Cayenne)

Acceleration snapshots are a Spice.ai Enterprise feature. See Acceleration Snapshots for the full reference. They are not available in Spice.ai OSS.

Cayenne supports object-store-backed acceleration snapshots that integrate naturally with cluster mode, allowing a newly started executor to bootstrap from the shared object store rather than re-fetching from the federated source:

acceleration:
  engine: cayenne
  snapshots: enabled                   # enabled | disabled | bootstrap_only | create_only
  snapshots_trigger: refresh_complete  # refresh_complete | time_interval | stream_batches
  snapshots_compaction: enabled
  snapshots_creation_policy: on_change # on_change | always

Setting snapshots: bootstrap_only is recommended on executors when the source is expensive to scan: executors hydrate from the snapshot at startup but do not produce new snapshots themselves. Cayenne's internal partition metadata supports composite keys (Hive-style key1=v1/key2=v2/... paths).

Sizing

Size each executor's local NVMe to hold its assigned partitions plus a shuffle working set (rule of thumb: ~25 % of dataset volume).
The scheduler does not need fast local storage; provision modest CPU / memory and ensure low-latency access to the object store.

Execution Modes

Mode	Endpoint	Notes
Synchronous	`/v1/sql`, FlightSQL	Client waits for the query to complete and receives results directly. Available in any deployment.
Asynchronous	`/v1/jobs`	Client submits a query and polls a `job_id` for status. Results are stored as chunked Arrow IPC under `jobs/` in the shared object store. Cluster (scheduler) mode only.

Async jobs require runtime.scheduler.state_location to be configured.

Observability

All distributed-query metrics are emitted through the OTel cluster meter and exposed on the standard Prometheus port (9090). The Grafana dashboard shipped with Spice.ai Enterprise (see Observability) plots the most important signals.

Architecture

Multi-Active High Availability

Configuration

Spicepod (runtime.scheduler)

CLI flags

Manual launch (non-Kubernetes)

Kubernetes (recommended)

Internal gRPC (port 50052)

Partitioning, Sharding, and Partition-Aware Queries

Declaring partitions

Partition discovery and assignment

How executors learn their assignments

Partition-aware query planning

Partition-aware writes

Constraints

Distributed Accelerations

Engine support

Per-executor sharding

Refresh in cluster mode

Acceleration snapshots (Cayenne)

Sizing

Execution Modes

Observability

Spicepod (`runtime.scheduler`)