Learn how to configure and run Spice in distributed mode to handle larger scale queries across multiple nodes.
:::info Preview
Multi-node distributed query execution based on Apache Ballista is available as a preview feature in Spice v1.9.0.
:::
Spice integrates Apache Ballista (opens in a new tab) to schedule and coordinate distributed queries across multiple executor nodes. This integration is useful when querying large, partitioned datasets in data lake formats such as Parquet, Delta Lake, or Iceberg. For smaller workloads or non-partitioned data, a single Spice instance is typically sufficient.
A distributed Spice cluster consists of two components:
The scheduler holds the cluster-wide configuration for a Spicepod, while executors connect to the scheduler to receive work. A cluster can run with a single scheduler for simplicity, or multiple schedulers for high availability (opens in a new tab).
Spice separates public and internal cluster traffic across different ports:
| Port | Service | Description |
|---|---|---|
| 50051 | Flight SQL | Public query endpoint |
| 8090 | HTTP API | Public REST API |
| 9090 | Prometheus | Metrics endpoint |
| 50052 | Cluster Service | Internal scheduler/executor communication (mTLS enforced, by default) |
Internal cluster services are isolated on port 50052 with mTLS enforced by default.
Distributed query cluster mode uses mutual TLS (mTLS) for secure communication between schedulers and executors. Internal cluster communication includes highly privileged RPC calls like fetching Spicepod configuration and expanding secrets. mTLS ensures only authenticated nodes can join the cluster and access sensitive data.
Each node in the cluster requires:
ca.crt) trusted by all nodesProduction deployments should use the organization's PKI infrastructure to generate certificates with proper SANs for each node.
For local development and testing, the Spice CLI provides commands to generate self-signed certificates:
Certificates are stored in ~/.spice/pki/ by default. Each certificate lists localhost and 127.0.0.1 as Subject Alternative Names. When a node advertises any other hostname or IP address with --node-advertise-address, add that address with --host.
:::warning CLI-generated certificates are not intended for production use. Production deployments should use certificates issued by the organization's PKI or a trusted certificate authority. :::
For local development and testing, mTLS can be disabled using the --allow-insecure-connections flag:
:::warning
Do not use --allow-insecure-connections in production environments. This flag disables authentication and encryption for internal cluster communication.
:::
Cluster deployment typically starts with a scheduler instance, followed by one or more executors that register with the scheduler.
The following examples use CLI-generated development certificates. For production, substitute certificates from your organization's PKI.
The scheduler is the only spiced process that needs to be configured (i.e. have a spicepod.yaml in the current dir). Override the Flight bind address when it must be reachable outside of localhost. Every scheduler and executor must set --node-advertise-address, the hostname or IP address other nodes use to reach it; a node in cluster mode without it fails to start:
Executors connect to the scheduler's internal cluster port (50052) to register and pull work. Executors do not require a spicepod.yaml; they fetch the configuration from the scheduler. Each executor automatically selects a free port if the default is unavailable:
Specifying --scheduler-address implies --role executor.
Each scheduler and executor registers under its instance identity (opens in a new tab): the SPICE_INSTANCE_ID environment variable, else HOSTNAME, else COMPUTERNAME, else the host name. Nodes on separate hosts or Kubernetes pods already have distinct identities. Processes on one host share its host name, so two schedulers, or two executors, on one host each need their own SPICE_INSTANCE_ID. This is in addition to what any two spiced processes on one host already need, such as their own instance directory, set with SPICE_CONFIG_DIR, because each process holds an exclusive lock on its instance directory.
The identity is stable across restarts, so the scheduler recognizes an executor that restarts under the same identity as the same executor. It logs Executor '<id>' restarted, so the tasks and shuffle output of its earlier run are rescheduled and reschedules that work instead of waiting for the earlier run to time out.
A scheduler restarted after a crash under the same identity waits until its predecessor's heartbeat has stayed unchanged for 30 seconds, then registers. A scheduler whose identity is held by a running scheduler logs an error and is refused. See Instance identity (opens in a new tab).
In v2.4.0-rc.2 and earlier, a scheduler registered under its host:port address and an executor under its advertise address, so two nodes on one host did not need distinct identities.
Run the same Spice version on every scheduler and executor in a cluster, and upgrade all of them together. A scheduler checks the cluster protocol version that each executor reports when it registers and polls for work, and rejects an executor built for a different Spice version. A rejected executor receives no work. The scheduler logs Rejecting executor <id>: protocol version mismatch, and the executor logs a warning like the following on each retry:
Queries run against the scheduler endpoint. The EXPLAIN output confirms that distributed planning is active—Spice includes a distributed_plan section showing how stages are split across executors:
:::warning[Limitations]
:::
:::tip[Spice.ai Enterprise]
Distributed Accelerations (opens in a new tab) partition-shard accelerated datasets across executors — each executor materializes and serves only the partitions it owns, with partition-aware read routing and write-through semantics. This is available in Spice.ai Enterprise (opens in a new tab) and requires a SpicepodCluster (opens in a new tab).
:::
For long-running queries, the async queries API enables submitting queries for background execution, polling for status, and retrieving paginated results when ready.
:::warning
The async queries API is experimental and requires a scheduler state location: the shared runtime.state.location (opens in a new tab), or the deprecated runtime.scheduler.state_location.
:::
--role schedulerruntime.state.location configured in the Spicepod (see High Availability > Configuration (opens in a new tab))Configure runtime.state.location (opens in a new tab) in your spicepod.yaml to enable the async queries API. The scheduler keeps its state in the location's cluster/ folder:
The state location is a shared object store (S3, GCS, Azure Blob, or local filesystem via file://) used to persist async query job state and result chunks.
For local development:
runtime.scheduler.state_location and runtime.scheduler.params are deprecated. A scheduler that sets them logs a startup warning that shows how to move to runtime.state. See runtime.scheduler (opens in a new tab).
A query is owned by the principal that submitted it, and only that principal can list, poll, fetch results from, or cancel it. This applies to the HTTP, Arrow Flight, and CLI surfaces alike.
Ownership follows the same principal boundary as Per-Principal Cache Isolation (opens in a new tab), and is not configurable:
| Caller | Owner scope |
|---|---|
| Unauthenticated (or authentication (opens in a new tab) not enabled) | A single shared public scope — every caller sees every query, matching pre-isolation behavior. |
| Authenticated principal (e.g., API key) | The principal's own isolated scope. |
| Internal / background runtime tasks | An internal system scope that can never reach a user-submitted query. |
There is no administrator or organization-wide scope. Every authenticated principal sees only its own queries; no principal can list, poll, or cancel another's, and there is no role or setting that grants a wider view.
For a query it does not own, a caller always receives 404 Not Found — never 403, and never 410. Ownership is checked before result expiry, so the response does not reveal that the query exists:
| Caller | Query is running or complete | Query's results have expired |
|---|---|---|
| The owner | 200 OK | 410 Gone, then 404 Not Found once cleanup (opens in a new tab) deletes the job |
| Any other principal | 404 Not Found | 404 Not Found |
Ownership tracking was introduced in v2.2.0. A job written by an earlier runtime carries no owner and is treated as belonging to the public scope.
Base path: /v1/queries
| Method | Path | Description |
|---|---|---|
POST | /v1/queries | Submit a query for async execution |
GET | /v1/queries | List the caller's queries |
GET | /v1/queries/{query_id} | Get query status and first result chunk |
GET | /v1/queries/{query_id}/status | Get query status only |
GET | /v1/queries/{query_id}/results | Get results (with pagination) |
GET | /v1/queries/{query_id}/results/chunks/{chunk_index} | Get a specific result chunk |
POST | /v1/queries/{query_id}/cancel | Cancel a running query |
POST /v1/queries
Submits a SQL query for asynchronous execution and returns immediately with a job ID.
Request Body (application/json):
| Field | Type | Required | Description |
|---|---|---|---|
sql | string | Yes | SQL statement to execute |
parameters | array | No | Bind variables for parameterized queries ($1, $2, ...) |
timeout_seconds | integer | No | Maximum execution time in seconds. The query is cancelled and failed on timeout. |
maximum_size | integer | No | Maximum result size in bytes. The query is failed if results exceed this limit. |
Request Example:
Response (HTTP 202 Accepted):
GET /v1/queries/{query_id}
Returns the full query status, result manifest, and the first result chunk (if completed successfully).
Response (HTTP 200):
GET /v1/queries/{query_id}/status
Returns the current status of a query without result data. Use this for lightweight polling.
Response (HTTP 200):
When the query has failed:
GET /v1/queries/{query_id}/results
Returns result data for a completed query. Use the partition query parameter to paginate through chunks.
Query Parameters:
| Parameter | Type | Default | Description |
|---|---|---|---|
partition | integer | 0 | Chunk index to retrieve (0-based) |
Response (HTTP 200):
When the last chunk is reached, next_chunk_index and next_chunk_url are null.
GET /v1/queries/{query_id}/results/chunks/{chunk_index}
Returns a specific result chunk by index. Same response format as Get Results.
POST /v1/queries/{query_id}/cancel
Cancels a running query. Also cancels the underlying distributed query on the Ballista scheduler. Only the principal that submitted the query can cancel it; any other caller receives 404 Not Found and the query keeps running — see Query Ownership (opens in a new tab).
Response (HTTP 200):
GET /v1/queries
Lists the queries submitted by the calling principal, optionally filtered by status. Queries submitted by other principals are never listed — see Query Ownership (opens in a new tab).
Query Parameters:
| Parameter | Type | Default | Description |
|---|---|---|---|
status | string | all | Filter by status: queued/pending, running, completed/succeeded, failed, cancelled, closed |
limit | integer | 100 | Maximum number of results |
Response (HTTP 200):
The sql_preview field contains the first 100 characters of the SQL statement.
| HTTP Status | Condition |
|---|---|
| 202 Accepted | Query successfully submitted |
| 200 OK | Status/results retrieved successfully |
| 404 Not Found | Query ID, chunk, or result not found, or the query was submitted by a different principal |
| 409 Conflict | Query not yet complete (when fetching results by chunk) |
| 410 Gone | Query results have expired |
| 425 Too Early | Query still running (results endpoint) |
| 500 Internal Server Error | Execution or serialization failure |
| 503 Service Unavailable | Not running in scheduler cluster mode, or executor not yet initialized |
The async query API is also available via Apache Arrow Flight DoAction requests. This is more efficient for programmatic access since results are returned in Arrow IPC binary format instead of JSON.
Every action below is scoped to the principal that submitted the query, exactly as the HTTP endpoints are — see Query Ownership (opens in a new tab).
| Action Type | Request Body (JSON) | Response |
|---|---|---|
SubmitAsyncQuery | {"sql": "...", "parameters": [...]} | JSON: {"query_id": "...", "status": "PENDING"} |
GetAsyncQueryStatus | {"query_id": "..."} | JSON: query status with error/result metadata |
GetAsyncQueryResult | {"query_id": "...", "chunk_index": 0} | Binary: Arrow IPC stream |
CancelAsyncQuery | {"query_id": "..."} | JSON: {"query_id": "...", "cancelled": true, "status": "CANCELLED"} |
Request:
Response (JSON):
Request:
Response (JSON):
Request:
Response: Arrow IPC binary stream containing the RecordBatch data for the requested chunk.
Request:
Response (JSON):
The spice query command provides a CLI and interactive REPL for the async queries API.
The CLI auto-polls with a spinner and displays results when ready. Press Ctrl+C to stop waiting — the query continues running in the background.
| Option | Default | Description |
|---|---|---|
--no-wait | false | Submit the query and return immediately without waiting for results |
--timeout <DURATION> | none | Maximum client-side wait time (e.g., 30s, 5m). The query itself continues running on timeout. |
-o, --output <FORMAT> | table | Output format: table or json |
These subcommands reach only the queries submitted with the same credentials — see Query Ownership (opens in a new tab).
When invoked without arguments, spice query starts an interactive REPL:
query> SELECT COUNT(*)
> FROM large_table
> WHERE status = 'active';
Submitted query: 01ABC-DEF-456-7890AB (PENDING)
Press Ctrl+C to stop waiting (query continues in background)
⠹ RUNNING (2.3s)...
✓ SUCCEEDED (5.1s)
+----------+
| count(*) |
+----------+
| 42000 |
+----------+
Time: 5.10000000 seconds. 1 rows.
REPL Commands:
| Command | Description |
|---|---|
.list | List all queries tracked in this REPL session |
.status <id> | Show detailed status of a query |
.results <id> | Fetch and display results of a completed query |
.wait <id> | Resume waiting for a query to complete |
.cancel <id> | Cancel a running query |
.clear | Clear the local tracked queries list |
.clear history | Clear command history |
.help | Show available commands |
.exit, .quit, .q | Exit the REPL |
Query IDs can be abbreviated if they uniquely identify a query within the tracked session.
PENDING → RUNNING → SUCCEEDED → CLOSED (after 12h TTL)
→ FAILED
→ CANCELLED
| Status | Description |
|---|---|
PENDING | Job is queued but not yet executing |
RUNNING | Job is actively executing on the distributed cluster |
SUCCEEDED | Job completed successfully, results are available |
FAILED | Job execution failed (see error field for details) |
CANCELLED | Job was cancelled by the user |
CLOSED | Job results have expired and been cleaned up |
The first final status a job reaches is kept. A job that is SUCCEEDED, FAILED, or CANCELLED is not restarted or moved to another status. For example, a cancellation that arrives before the job starts running leaves the job CANCELLED.
When a query fails, the error object contains an error_code field:
| Error Code | Description |
|---|---|
SCHEDULER_UNAVAILABLE | The Ballista scheduler is not reachable, or the scheduler running the query stopped and the query cannot be resumed (see Scheduler Failover (opens in a new tab)) |
SUBMISSION_FAILED | Failed to submit the query to the distributed scheduler, or the scheduler running the query stopped before saving it (see Scheduler Failover (opens in a new tab)) |
EXECUTION_FAILED | The query failed during execution |
FETCHING_RESULTS_FAILED | Failed to retrieve results from executor nodes |
CANCELLED | The query was explicitly cancelled |
PARAMETER_BINDING_FAILED | Failed to bind the provided query parameters |
NOT_FOUND | The referenced query or job was not found |
INTERNAL | An unexpected internal error occurred |
TIMEOUT | The query exceeded the configured timeout_seconds |
Job state and result chunks are stored in the shared object store configured via runtime.state.location, or the deprecated runtime.scheduler.state_location. {base_prefix} is the cluster/ folder below the path of runtime.state.location, or the root of state_location (see Runtime state layout (opens in a new tab)):
{base_prefix}/
├── jobs/
│ ├── {job_id}.json # Job state (JSON)
│ └── {job_id}/
│ ├── chunk_0.arrow # Result chunk 0 (Arrow IPC)
│ ├── chunk_1.arrow # Result chunk 1
│ └── ...
Each scheduler deletes, with their result chunks, the jobs whose results have expired and the jobs that failed or were cancelled at least 12 hours ago. It runs this cleanup about once an hour, at a random time between 30 and 90 minutes after the previous run. An expired job answers its owner with 410 Gone until the next cleanup deletes it, and with 404 Not Found after that. A cleanup that deletes jobs logs how many it deleted.
| Setting | Default |
|---|---|
| Chunk size | 10,000 rows |
| Result TTL | 12 hours |
| List limit | 100 queries |
--role schedulerruntime.state.location, or the deprecated runtime.scheduler.state_location, to be configuredformat query parameter on the results endpoint is declared but not yet implemented (results are always JSON over HTTP, Arrow IPC over Flight)For production deployments, Spice supports running multiple active schedulers in an active/active configuration. This eliminates the scheduler as a single point of failure and enables graceful handling of node failures.
In an HA cluster:
┌─────────────────────┐
│ Load Balancer │
└─────────────────────┘
│
┌────────────────┼────────────────┐
▼ ▼ ▼
┌────────────┐ ┌────────────┐ ┌────────────┐
│ Scheduler │ │ Scheduler │ │ Scheduler │◄──► Object Store
│ │ │ │ │ │ (S3)
└────────────┘ └────────────┘ └────────────┘
▲ ▲ ▲
│ (executor-initiated) │
┌────────────┐ ┌────────────┐ ┌────────────┐
│ Executor │ │ Executor │ │ Executor │
└────────────┘ └────────────┘ └────────────┘
Enable HA by configuring the shared runtime.state.location (opens in a new tab) in the Spicepod to point to an S3-compatible object store. The schedulers keep their shared state in its cluster/ folder:
The object store is used for scheduler registration and discovery, and to persist async query (opens in a new tab) job state (the execution graph plus its status) so that schedulers are effectively stateless for async queries.
When a scheduler state location is configured, each async query's execution graph and status are persisted to the shared object store. If the scheduler driving an async query becomes unavailable, another scheduler detects the orphaned job and resumes it to completion from the persisted execution graph — the query is re-driven rather than replanned, and consumers and executors do not need to know which scheduler is running it. Takeover is single-winner: ownership transfers via a compare-and-set on the job's metadata, and a scheduler never reclaims its own in-flight jobs.
A job submitted by an authenticated principal is not resumed. The job records only an opaque owner ID, not an identity that table access and masking can be applied to, so resuming it would run the query without the submitter's permissions. Instead, the scheduler that detects the orphaned job marks it FAILED with the error code SCHEDULER_UNAVAILABLE and a message to resubmit the query. Jobs submitted without a principal are resumed.
A job submitted without a principal can also be RUNNING before its scheduler has saved the execution graph. If that scheduler stops at this point, there is nothing to resume, so the scheduler that detects the orphaned job marks it FAILED with the error code SUBMISSION_FAILED and a message to resubmit the query.
This failover applies to async queries, which require a scheduler state location. Synchronous queries in flight on a scheduler that becomes unavailable are not resumed automatically; the client should retry them against another scheduler. Without a scheduler state location, job state is held in memory and a single-scheduler cluster behaves as before (no failover).
runtime.state.params supports the following S3 parameters for the scheduler's state:
| Parameter | Description | Default |
|---|---|---|
s3_region | AWS region for the S3 bucket | - |
s3_endpoint | Custom S3-compatible endpoint URL | - |
s3_auth | Authentication method: iam_role or key | iam_role |
s3_key | AWS access key ID (required when s3_auth: key) | - |
s3_secret | AWS secret access key (required when s3_auth: key) | - |
s3_session_token | AWS session token for temporary credentials | - |
client_timeout | S3 client timeout | - |
allow_http | Allow HTTP (non-TLS) connections to S3 endpoint | false |
Example with explicit credentials:
With s3_auth: key and either key missing, the scheduler state store fails to initialize with an error instead of using credentials from the environment. To use environment credentials, set s3_auth: iam_role or omit s3_auth.
A gs:// state location takes the GCS data connector (opens in a new tab) parameters (gcs_service_account_path, gcs_service_account_key, gcs_application_default_credentials, or gcs_skip_signature) instead. See State location parameters (opens in a new tab).
Configure shared state in spicepod.yaml:
Start multiple schedulers, each with unique certificates:
Start executors (they discover all schedulers automatically):
Configure a load balancer to distribute queries across scheduler Flight SQL endpoints (port 50051).
:::info Object Store Requirements The object store must support conditional writes (S3 ETags). Most S3-compatible stores support this, including AWS S3, MinIO, and Google Cloud Storage (with S3 compatibility mode). :::