docs-mintlify/docs/pre-aggregations/cube-store-architecture.mdx
Cube Store is the purpose-built storage and query engine that serves pre-aggregations. It ingests pre-aggregation data built from your data sources, stores it in a columnar format optimized for analytical queries, and answers queries with sub-second latency at high concurrency.
Cube Store also plays a second role: it hosts Cube's cache and queue store — the in-memory cache used for query result caching and refresh keys, and the query queue that coordinates work across API instances and refresh workers. Both roles were historically served by Redis; Cube Store replaced it, so a Cube deployment needs no external cache or queue infrastructure.
This page explains how Cube Store works under the hood: the design decisions behind it, how data is stored and partitioned, how queries are executed across a cluster, and how the cache and queue store works.
Cloud data warehouses are excellent at storing and transforming large volumes of data at low maintenance cost, but they are not designed to serve user-facing analytics directly. They typically fall short on:
Cube Store closes this gap. It sits between your data sources and your applications, turning warehouse output into a query-optimized store that delivers the latency and concurrency of a dedicated OLAP database while preserving the warehouse's ease of use.
It was designed against the following requirements:
Technology stack: Rust, Apache Arrow, DataFusion, Parquet, RocksDB.
┌──────────────┐
│ Cube API │
└──────┬───────┘
│ queries
▼
┌──────────────┐
│ Router │
│ (metastore) │
└──┬───┬───┬──┘
│ │ │
┌──────┘ │ └──────┐
▼ ▼ ▼
┌──────────┐┌──────────┐┌──────────┐
│ Worker 1 ││ Worker 2 ││ Worker N │
└────┬─────┘└────┬─────┘└────┬─────┘
│ │ │
└─────┬─────┴─────┬─────┘
▼ ▼
┌────────────────────────┐
│ Cloud object store │
│ (S3 / GCS / MinIO) │
└────────────────────────┘
A Cube Store cluster consists of a router and one or more workers:
In a single-process deployment (the default in development), one process plays all roles. See Running in production for cluster deployment.
Cube Store's architecture follows from seven key design decisions, each serving one or more of the requirements above:
The following sections describe each decision.
The metastore holds all metadata: schemas, tables, indexes, partitions, chunks, and background jobs. It is the most frequently accessed component in Cube Store — small relative to the data it describes, but still reaching tens of millions of rows, with strong read-after-write consistency requirements and heavy read traffic alongside significant write concurrency.
RocksDB backs the metastore. Metadata entities form a hierarchy:
Schema
└─ Table
└─ Index (one or more per table)
└─ Partition (a sorted range of the index)
└─ Chunk (a fragment of a partition:
a Parquet file or an in-memory buffer)
All metadata mutations go through the router to maintain consistency. Workers reach the metastore over the network and cache metadata locally.
Every index in Cube Store is a sorted copy of the table data. Sorting is the most efficient way to compress columnar data, and it is optimal for filtering — range predicates can skip entire blocks of data using min-max statistics.
In a columnar format, values of one column are stored next to each other, and encoders exploit local redundancy: repeated or slowly changing neighboring values encode into far fewer bits. Sorting is what creates that redundancy. Consider a column of product categories in arrival order versus sorted order:
Unsorted: books, toys, books, garden, toys, books, garden, toys, ...
Sorted: books, books, books, garden, garden, toys, toys, toys, ...
The unsorted column has to be stored more or less value by value. The
sorted column collapses into a handful of runs — effectively
books×3, garden×2, toys×3 — regardless of how many million rows it spans.
Concretely, sorting feeds each of the standard columnar encodings:
(value, count) pairs. A column with 20 distinct values
across a billion rows compresses to almost nothing.The effect cascades down the sort key: the first key column is perfectly
sorted, and within each of its runs the second column is sorted too, and so
on. Columns not in the sort key still benefit whenever they correlate
with key columns (for example, city correlating with country).
Smaller files are not just a storage win — they are a latency win. Scan speed is bounded by bytes read, and general-purpose compression applied on top works better on already-regular data. The same property also tightens min-max statistics: in sorted data, each block covers a narrow, mostly disjoint slice of the key space, so pruning can rule blocks out with precision instead of finding every block's range overlapping every query.
Because all data is sorted, every operation in the engine can be merge-based: merge sort, merge join, and merge aggregation all work naturally on pre-sorted inputs without expensive hash tables or shuffles.
Cube Store deliberately pays write amplification (maintaining several
sorted copies of the same data) to buy read speed. Indexes are selected
based on query filters: the planner examines the WHERE clause and picks
the index whose sort key best matches the filtered columns. This is why
defining indexes that match
your query patterns is the single most effective pre-aggregation
optimization.
Cube Store supports two index types: regular and aggregating.
A regular index is a sorted copy of every row in the table. An aggregating index goes further: it stores only the dimensions listed in its definition plus pre-aggregated measures, rolling the data up over every dimension that is not in the index. It is effectively a rollup of the rollup, maintained incrementally by the storage engine itself.
Each measure column in an aggregating index carries an aggregate function that defines how rows with the same sort key combine:
| Function | Used for |
|---|---|
SUM | Additive measures (sum, count) |
MIN / MAX | Minimum / maximum measures |
MERGE | HyperLogLog sketches (count_distinct_approx) |
The rollup does not happen at query time. Whenever chunks are merged during compaction, Cube Store runs the merged, sorted stream through a group-by on the index's sort key and applies the aggregate functions — rows with equal dimension values collapse into one. Because the data is already sorted by the grouping columns, this aggregation is a cheap, streaming merge rather than a hash aggregation.
The result: a query that matches an aggregating index reads a table that is often orders of magnitude smaller than the parent pre-aggregation, and the aggregation work has already been paid for at ingestion time.
Note that this is also why aggregating indexes only support additive aggregate functions — a non-additive measure (such as an exact distinct count) cannot be combined incrementally from partial rollups.
An index's sort key defines a totally ordered key space, and each partition owns a contiguous range of it. A partition's range is described by two bounds, where an absent bound means "unbounded":
Key space of an index sorted by (created_at, user_id):
├────────────────┼────────────────┼────────────────┤
Partition 1 Partition 2 Partition 3
(−∞ .. K₁) [K₁ .. K₂) [K₂ .. +∞)
Two properties of ranges matter in practice:
A freshly created table starts with a single partition covering the entire key space (both bounds unbounded). Partitions never split on ingestion — splitting is deferred to compaction, when the data is being rewritten anyway.
During compaction, Cube Store checks whether the partition has outgrown its split threshold and, if so, writes the merged output into several new partitions instead of one:
Before compaction: After split at compaction:
Partition [K₁ .. K₂) Partition [K₁ .. S₁)
┌───────────────────────┐ Partition [S₁ .. S₂)
│ main file + N chunks │ ──► Partition [S₂ .. K₂)
└───────────────────────┘
The number of child partitions is computed from both row count and file size:
CUBESTORE_PARTITION_SPLIT_THRESHOLD, default 2,097,152 rows; can also
be set per pre-aggregation).CUBESTORE_PARTITION_SIZE_SPLIT_THRESHOLD, default 100 MiB).The larger of the two wins, capped at 16 children per compaction pass to bound write amplification. Because the merged output is sorted, the split points fall out naturally: the writer streams rows into the first file until it reaches its row budget, closes it at the next split-key boundary, and moves on. Each child's range bounds are derived from the actual first and last keys written to it.
Uniform partition sizes keep query work evenly distributed across workers and make partition pruning predictable.
All persistent data is stored as Parquet files. Parquet implements the PAX (Partition Attributes Across) layout: rows are grouped into row groups, and values within a row group are stored column by column. This gives columnar compression and scan speed while preserving efficient row reconstruction.
Cube Store leans on Parquet's min-max statistics: every row group records the minimum and maximum value of each column. Combined with sorted data, this lets the query planner skip partitions — and row groups within partitions — whose ranges cannot match the query's filters, often eliminating most of the data before reading a single byte.
Cube Store separates storage from compute. All Parquet files live in a cloud object store (S3, GCS, MinIO, or a local directory in development), and workers keep local copies only as a cache.
This has several consequences:
To avoid download latency on the query path, Cube Store performs a partition warmup: before a table comes online, every worker downloads the partitions assigned to it. Queries never wait for a cold download.
Worker nodes never communicate with each other. Each worker owns a set of partitions, assigned deterministically by consistent hashing over the partition's range and index. Every node computes the same assignment independently — no coordination service is needed.
A query executes as follows:
Router
│ distribute plan
┌───────────┼───────────┐
▼ ▼ ▼
Worker 1 Worker 2 Worker N
┌───────┐ ┌───────┐ ┌───────┐
│Parquet│ │Parquet│ │Parquet│ ← persisted data
├───────┤ ├───────┤ ├───────┤
│Memory │ │Memory │ │Memory │ ← real-time data
│chunks │ │chunks │ │chunks │
└───┬───┘ └───┬───┘ └───┬───┘
│ partial │ partial │
│ results │ results │
└──────┬─────┴─────┬────┘
▼ ▼
┌──────────────────────┐
│ Coordinator: merge │
│ sorted streams, │
│ finalize aggregation,│
│ LIMIT │
└──────────────────────┘
Workers do as much as possible locally: they prune partitions with min-max statistics, scan Parquet and in-memory chunks, and compute partial aggregations. The coordinator receives already-aggregated partial results and only needs to merge them — merging sorted streams is cheap, so coordination never becomes a data-processing bottleneck.
The coordinator here is itself one of the workers, chosen per query — not the router node that hosts the metastore, which only plans queries and relays results (see One plan, two physical plans).
This is also how lambda pre-aggregations work: a single query transparently combines persisted Parquet data with fresh streaming data held in memory on the same workers.
Streaming sources (such as Kafka) deliver rows continuously, but writing single rows to Parquet would be prohibitively expensive. Instead, incoming rows are routed to the worker that owns the target partition and buffered there as in-memory chunks — columnar Arrow buffers that are immediately queryable.
Streaming source ──► Router ──► partition owner worker
│ in-memory chunks
│ (queryable immediately)
▼ compacted at ~1-minute intervals
Parquet partition in object store
In-memory chunks are compacted into persisted Parquet partitions at roughly one-minute intervals, or sooner when count or size thresholds are exceeded. During compaction, Cube Store sorts the data, deduplicates rows by unique key, and evaluates aggregate columns — so unique-key semantics hold across batch and streaming data alike. See Compaction in detail for the full process.
One of Cube Store's founding requirements is that input data always has a unique primary key. Streaming sources exploit this: an update to a row arrives simply as a new version of the same key. Cube Store never updates data in place — chunks and partitions are immutable — so upsert semantics are implemented as last row by unique key: among all stored versions of a key, the newest one wins. Deduplication happens twice: logically on every read, and physically at compaction.
Tables with a unique key carry a hidden, monotonically increasing sequence
column, __seq. For streaming tables it is derived from the source's own
ordering (such as Kafka offsets), so "newest" reflects the order in which
the source emitted the versions.
Two structural guarantees make deduplication cheap:
__seq. When an
index is created on a unique-key table, any unique key columns missing
from its definition are appended to the sort key, followed by __seq as
the final sort column. All versions of a key are therefore physically
adjacent in every sorted file, with the newest version last.__seq is excluded from the partition split key. Partitions may
split on the unique key, but never between versions of the same key —
all versions of a key always live in the same partition, and therefore
on the same worker. Deduplication never requires cross-worker
coordination, preserving the shared-nothing model.A query against a unique-key table dedupes on the fly, seeing streamed updates the moment they arrive:
__seq — they
are needed to decide which version of each row survives.__seq last. Versions of the same key from different sources —
an old version in Parquet, a fresh update in memory — end up adjacent,
newest last.The plan then re-projects to the columns the query actually asked for. The extra cost over a plain scan is modest: the merge was needed anyway, and the last-row filter is a linear scan over already-sorted data.
Compaction runs the very same merge-plus-last-row pipeline (see Compaction in detail): chunks and the main file are merged in sort order, the last-row filter keeps only the newest version of each key, and the result is written out. The difference is what happens to the output — it replaces the inputs, so superseded versions are physically deleted.
This division of labor keeps read-time deduplication bounded. At any moment, a key has at most one version in the compacted main file plus however many updates arrived since the last compaction — and compaction continuously folds those in. Read-time dedup only ever pays for the recent, not-yet-compacted tail, while storage converges toward exactly one row per key.
Note that unique-key deduplication and aggregating indexes are mutually exclusive ways of merging rows: a regular index on a unique-key table keeps the last version per key, while an aggregating index combines rows with aggregate functions.
Cube Store builds on DataFusion, the Arrow-native query engine, and extends it with distribution-aware planning. All planning happens on the router; the outcome is a plan split into a router part and a worker part at an explicit cluster boundary.
The SQL text is parsed into an abstract syntax tree and converted into a DataFusion logical plan — a tree of relational operators (scan, filter, project, aggregate, sort, limit). At this stage, table scans are placeholders: no index has been chosen and no partitions are attached yet. DataFusion's standard logical optimizations run here, most importantly pushing filters and projections down toward the scans, which sets up the constraint collection that follows.
The planner walks the optimized plan and collects, for every scanned table, the constraints relevant to choosing an index:
GROUP BY.It then fetches all candidate indexes for all referenced tables from the metastore in a single round trip and picks the winner per table:
CREATE INDEX statement to fix it —
distributed joins without a matching index would require a data
shuffle or a query-time sort, which Cube Store deliberately does not
do. See Distributed joins for the full
mechanics.With indexes chosen, the planner attaches the concrete partition list to each scan, applying partition pruning as it goes, and inserts a cluster send node into the plan above the scans. That node marks the boundary: everything below it will run on workers, everything above it on the router.
Partition pruning happens on the router at planning time, before any work is distributed. The goal is to touch only the partitions whose key ranges can possibly satisfy the query.
The planner converts the query's WHERE clause into a set of min-max
conditions over the index's sort key columns. Each condition is a pair of
per-column lower and upper bounds; the whole filter is a union of such
conditions (an OR across them). The extraction understands:
=, <, <=, >, >=IN lists (each value becomes its own condition)AND / OR combinations of the aboveAnything the extractor cannot represent as a range over sort key columns — functions on columns, predicates on non-key columns, and so on — is simply left out. That is always safe: pruning may only ever over-include partitions, never skip one that holds matching rows. The remaining predicates are still applied during the scan.
Each partition's [min, max] key interval is then checked against the
conditions with a lexicographic comparison, mirroring how the data is
sorted. If no condition can intersect the partition's interval, the
partition is dropped from the plan — its worker never sees it, and its
files are never opened.
Pruning applies at three successively finer levels:
Combined with sorted data — which makes ranges tight and non-overlapping — this typically eliminates the vast majority of data for selective queries before a single byte is read.
Before the plan is finalized, Cube Store rewrites it so that as much work as possible lands below the cluster boundary:
GROUP BY … ORDER BY … LIMIT n pattern is
rewritten into a distributed top-k: each worker computes its local
candidates and the coordinator merges them, instead of materializing the
full grouped result.The rewritten logical plan — along with the chosen index snapshots and partition lists — is serialized once and shipped to every participating worker. Each side then builds its own physical plan from it:
Workers are assigned their partition subsets by the same consistent hashing that governs data placement, so the plan always executes where the data already lives. Execution streams Arrow record batches from the workers to the node executing the router plan, which merges and returns the result.
Importantly, "router plan" names a role in the plan, not the router node. In a multi-node cluster, the actual router node — the one hosting the metastore — does not execute the router plan itself. For each query, it picks one of the workers at random and delegates the router plan to it: that worker becomes the query's coordinator, fanning the worker plans out to its peers (including possibly itself), merging their streams, and running the final aggregation and limit. The true router only plans the query and passes the finished result through.
This has two effects. The metastore node — a shared dependency of the whole cluster — is kept free of heavy data processing, so a large merge never degrades metadata operations for everyone else. And because the coordinator is re-picked per query, the merge workload itself is load balanced across the cluster rather than concentrating on a single node.
Putting it all together for a typical query:
SELECT product_category, SUM(order_total)
FROM orders_rollup
WHERE created_at >= '2026-01-01'
GROUP BY 1
ORDER BY 2 DESC
LIMIT 10
created_at filter down to
the scan.(created_at, product_category).created_at >= '2026-01-01'.LIMIT 10, and returns the result.The resulting distributed physical plan looks like this (with the top-k rewrite left out for clarity — shown as a plain distributed aggregation over six surviving partitions and three workers):
Router node (metastore)
plans the query, picks Worker 2 as coordinator
│
▼
Worker 2 — executes the router plan
┌─────────────────────────────────────┐
│ Projection │
│ └─ InlineAggregate (Final) │
│ └─ SortPreservingMerge │
│ └─ ClusterSend │
└────────┬──────────┬──────────┬──────┘
│ │ │ worker plans
▼ ▼ ▼
┌────────────┐ ┌────────────┐ ┌────────────┐
│ Worker 1 │ │ Worker 2 │ │ Worker 3 │
│ │ │ │ │ │
│ InlineAgg │ │ InlineAgg │ │ InlineAgg │
│ (Partial) │ │ (Partial) │ │ (Partial) │
│ └─ Filter │ │ └─ Filter │ │ └─ Filter │
│ └─ Merge │ │ └─ Merge │ │ └─ Merge │
│ ├─ P1 │ │ ├─ P2 │ │ ├─ P3 │
│ ├─ P4 │ │ ├─ P5 │ │ ├─ P6 │
│ └─ mem │ │ └─ mem │ │ └─ mem │
│ chunks │ │ chunks │ │ chunks │
└────────────┘ └────────────┘ └────────────┘
P1..P6 — pruned-in partitions, assigned to
workers by consistent hashing
Reading it bottom-up: each worker merges its partitions' sorted streams — Parquet files and in-memory chunks together (the lambda architecture query) — filters them, and computes a partial streaming aggregation. The coordinating worker (Worker 2 here, which also processes its own share of partitions) merges the three sorted partial streams, finalizes the aggregation, and returns the result through the router.
Cube Store supports joins between tables, with two properties that follow directly from its merge-based, shared-nothing design: every joined table must have an index on the join key, and the join is distributed so that workers execute it locally.
Cube Store executes joins exclusively as merge joins — hash joins are disabled. A merge join consumes two streams that are both sorted by the join key and zips them together in a single linear pass, with no hash table to build and no memory proportional to the input size. This is the same principle that runs through the whole engine: every operation is merge-based over sorted data.
The sorted inputs have to come from somewhere, and re-sorting billions of rows at query time would defeat the purpose. So the planner requires each side of a join to be scanned through an index whose sort key starts with the join columns — the join columns must form a prefix of the index sort key, in order. Such a scan produces join-key-ordered output for free, and the merge join can start streaming immediately.
If no such index exists on one of the joined tables, planning fails
immediately with an error that spells out the fix, including the exact
CREATE INDEX statement to run. This is deliberate: the alternative —
silently falling back to a sort or shuffle — would turn a sub-second query
into a multi-second one, and Cube Store treats that as an error rather
than a degradation.
The join itself runs entirely on the workers, below the cluster boundary. When both sides of a join resolve to worker-bound scans, the planner places the join under a single cluster-send node — so each worker joins its own data locally and the coordinator merely merges already-joined results.
Since the joined tables are partitioned independently, their partition
boundaries don't line up, so work is batched by the left-most (root)
table: the root table's partitions are split across batches, and every
right-side table is included in full in every batch. Including the right
side whole is required for correctness — batching a right table would make
a LEFT JOIN produce spurious NULLs and duplicates — but it means the
right side is effectively replicated into each batch. This works well when
the right side is small (dimension tables joined to a large fact table)
and poorly when both sides are large. A configurable guard
(CUBESTORE_MAX_JOINED_PARTITIONS, default 5 partitions per batch) fails
the query fast when the right side is too fragmented, rather than letting
it fan out into an expensive execution.
Each worker then executes the same plan: scan its batch's partitions through the join-key indexes (each stream already sorted), merge-join them, apply whatever partial aggregation was pushed down, and stream results to the coordinator.
Ingestion optimizes for write latency: every batch or streaming buffer becomes a new chunk — a small, independently sorted file (or in-memory buffer) attached to a partition. Queries, however, want few large sorted files. Compaction is the background process that continuously converts the former into the latter. It runs at two levels.
Streaming ingestion produces many tiny in-memory chunks. A scheduled job checks each partition and compacts its in-memory chunks when either:
CUBESTORE_IN_MEMORY_CHUNKS_MAX_LIFETIME_THRESHOLD, default 60
seconds).Chunks that are still small and young are merged into a single larger in-memory chunk (staying queryable in RAM); chunks that have grown past the size limit or aged past the lifetime threshold are merged and written out as persistent Parquet chunks. This is the ~1-minute cadence at which streaming data flows from memory into durable storage.
A per-partition compaction job merges accumulated Parquet chunks into the partition's main file:
CUBESTORE_CHUNKS_TOTAL_SIZE_THRESHOLD, default
2,097,152 rows) and an on-disk byte budget. Taking the smallest chunks
first maximizes the reduction in file count per unit of work.SUM, MIN, MAX, MERGE) combine matching
rows — this is where the incremental rollup happens.If data arrives faster than one compaction pass can absorb, the cycle simply repeats: each pass halves the file count or splits the partition further, converging toward a small number of large, sorted, uniformly-sized files per partition.
Beyond storing pre-aggregations, Cube Store hosts a second, independent subsystem: the cache and queue store. It is the engine behind Cube's in-memory cache (query result caching, refresh keys) and the query queue that coordinates query execution across Cube nodes. Historically these roles were played by Redis; Cube Store replaced it, removing an entire moving part from the deployment.
The cache and queue store lives on the router node and is backed by its own
RocksDB instance, fully separate from the metastore. Hot entries are served
from memory (RocksDB memtables and block cache), while the disk backing
means the cache and any in-flight queue state survive a router restart.
Cube talks to it over the same SQL interface as everything else, using
dedicated CACHE and QUEUE command families.
The cache is a Redis-style key-value store. Each entry has:
prefix:key, so related entries (for example, all
refresh keys of one deployment) can be listed and truncated by prefix,Because unbounded caches eventually eat their host, a background eviction
manager enforces two budgets: a total size limit
(CUBESTORE_CACHE_MAX_SIZE, default 4 GiB) and a key count limit
(CUBESTORE_CACHE_MAX_KEYS, default 100,000). When a budget is exceeded,
it evicts entries according to a configurable policy
(CUBESTORE_CACHE_POLICY):
| Policy | Keeps |
|---|---|
allkeys-lru (default) | Most recently used entries |
allkeys-lfu | Most frequently used entries |
allkeys-ttl | Entries with the most remaining time-to-live |
sampled-lru / sampled-lfu / sampled-ttl | Same criteria, evaluated on a sample for speed |
To support this, every entry tracks a last-access timestamp and an access frequency counter. Expired entries are also collected proactively in the background rather than only on access.
The query queue is how Cube serializes and prioritizes expensive work — data source queries and pre-aggregation builds — across any number of API instances and refresh workers. Centralizing the queue in Cube Store is what makes those nodes stateless and horizontally scalable: any node can add work, process work, or pick up results, because the queue state lives in one place.
There is not one global queue but one queue per workload, separated by key prefix: each data source gets its own query queue, and pre-aggregation builds run through their own queue. Each queue has its own concurrency budget and its own ordering.
Every queue item carries a key, a payload (the serialized query), a priority, and a status that moves through a simple lifecycle:
Pending ──► Active ──► Finished
(added) (claimed by a (acknowledged with
processing node) a result)
Pending items are picked by priority (from -10000 to 10000, higher
first), with older items winning ties — a priority queue with FIFO
fairness.
A queue item's key is a deterministic hash of the query itself — the SQL, its parameters, and the security context that shapes it. Identical queries always hash to the same key, and adding to the queue is an atomic insert-if-absent on that key: the first request creates the item, and every subsequent identical request — from the same API instance or any other — matches the existing item instead. Cube Store reports back whether the item was actually created, so the orchestrator knows whether it is the originator or just another waiter.
The result path is deduplicated the same way. Before adding to the queue at all, an API instance checks whether a result for that key is already available from a just-finished execution. After adding, all waiters — however many instances they are spread across — block on the same key and wake on the same result.
The net effect: a dashboard fanning the same expensive query out through a dozen concurrent requests costs exactly one data source execution.
There is no dedicated scheduler node. Every API instance and refresh worker runs the same reconciliation loop against the shared queue — after adding an item, when a blocked wait times out, and on a background interval. Reconciliation does two things:
Because many instances reconcile the same queue concurrently, several of
them may attempt to claim the same pending item at once. The claim itself —
QUEUE RETRIEVE — is an atomic compare-and-act inside Cube Store: it
re-checks the concurrency budget and flips the item from Pending to
Active in a single serialized write. Exactly one instance wins the race
and receives the payload; the others get a "not enough concurrency" or
"already taken" response and simply move on. Coordination requires no
locks, no leader election, and no scheduler — the queue itself is the
synchronization point.
When a queue is already saturated (active count at the concurrency limit), reconciling instances limit themselves to a single claim attempt rather than trying to grab a full budget's worth — this keeps a large cluster from stampeding the queue with doomed claim attempts.
The concurrency you configure for a data source is a cluster-wide cap
on simultaneously executing queries for that queue — not a per-node
setting. If concurrency is 4 and you scale from one API instance to ten,
at most 4 queries run against the data source at any moment; the ten
instances merely compete to be among the claimants. The cap is enforced
atomically at claim time inside Cube Store — an item is only flipped to
Active if the queue's active count is below the cap — so it cannot be
overshot by racing nodes.
This is exactly the semantics you want for protecting a data source with a connection or workload limit: scaling out the Cube tier never multiplies the pressure on the warehouse behind it.
The instance that wins a claim executes the query — dispatching it to the data source or running the pre-aggregation build. While the query runs, the processor heartbeats the queue item on a fixed interval (30 seconds by default), continuously proving that the work is still alive.
On completion, the processor acknowledges the item with its serialized
result. The acknowledgment atomically stores the result, finishes the
item, and fires an event. Waiters use a blocking read (QUEUE RESULT_BLOCKING) that subscribes to that event: they wake the moment the
result lands, with no polling interval to pay for.
A blocking wait does not hold an API request open forever. If the result
does not arrive within the continue-wait window (10 seconds by default),
the API responds with Continue wait and the client re-requests; the
re-request finds the queue item already present (idempotency again) and
resumes waiting. Long-running queries thus survive any number of HTTP
request cycles without ever being re-executed. Delivered results are
garbage-collected after a configurable expiry.
Two failure modes are handled by reconciliation, both detected from timestamps Cube Store keeps on each item:
Active items whose heartbeat has gone quiet
for longer than the heartbeat timeout (4× the heartbeat interval — two
minutes by default). This means the node that claimed the query died
mid-execution: a crashed process, a killed pod, a network partition.Pending items that outlived their orphaned
deadline (set when the item is added; 120 seconds by default). This
typically means everyone who cared about the result has gone away —
clients disconnected and stopped re-requesting — so executing the query
would be wasted work.Any reconciling instance can collect both kinds via a single queue call, remove them, and invoke the registered cancel handlers — which, for a stalled data source query, also issues the kill/cancel against the warehouse so a dead node's query does not keep burning warehouse compute.
Because detection is timestamp-based and cancellation is idempotent, it does not matter which instance notices first: recovery works even if the failed node never comes back.
Folding the cache and queue into Cube Store is itself a design decision in the spirit of the rest of the system: one storage engine, one deployment unit, one consistency model. The queue benefits directly from RocksDB's write durability (in-flight queue state survives restarts), and the event-driven blocking reads give lower-latency result delivery than the polling patterns typical of external queue stores.