> ## Documentation Index > Fetch the complete documentation index at: https://docs.cube.dev/llms.txt > Use this file to discover all available pages before exploring further. # Cube Store architecture > How Cube Store, the distributed OLAP engine behind pre-aggregations, stores data and executes queries with sub-second latency at high concurrency. Cube Store is the purpose-built storage and query engine that serves [pre-aggregations](/docs/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](#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. ## Why Cube Store exists 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: * **Sub-second query latency** — warehouse queries take seconds to minutes. * **High query concurrency** — most warehouses throttle at dozens of concurrent queries, while user-facing applications need 1000+ queries per second. * **Low ingestion latency** — freshly loaded data is not immediately queryable at interactive speed. 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: * Billions of rows as input * Sub-second response time * High query concurrency * Input data always has a unique primary key * Sources are both batch and real-time streaming * Joins between data sources **Technology stack:** Rust, Apache Arrow, DataFusion, Parquet, RocksDB. ## High-level architecture ```text theme={"dark"} ┌──────────────┐ │ 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**: * The **router** owns the metastore and plans queries. It also hosts the [cache and queue store](#cache-and-queue-store) used by Cube for query result caching and query queue coordination. Result merging is not done on the router: for each query, it delegates execution to one of the workers acting as the query's coordinator. * **Workers** execute partial queries on the partitions they own and return results to the per-query coordinator worker. * All persistent data lives in a **cloud object store**. Compute nodes are disposable and interchangeable — any worker can download the partitions it needs and start serving queries. In a single-process deployment (the default in development), one process plays all roles. See [Running in production](/docs/pre-aggregations/running-in-production) for cluster deployment. ## Design decisions Cube Store's architecture follows from seven key design decisions, each serving one or more of the requirements above: 1. **High-throughput metastore** — metadata is the most accessed data in the system and gets its own optimized store. 2. **Indexes are sorted copies** — sorting is the most efficient way to compress and filter columnar data. 3. **Auto-partitioning** — partitions split automatically to stay uniformly sized. 4. **Parquet as the storage format** — the industry-standard columnar format with min-max statistics for filter pushdown. 5. **Distributed file system as the storage layer** — storage-compute separation makes compute nodes disposable. 6. **Shared-nothing architecture** — workers never talk to each other, which removes coordination overhead and scales concurrency. 7. **Real-time in-memory chunks** — streaming rows are buffered in memory instead of being written to Parquet one at a time. The following sections describe each decision. ## High-throughput metastore 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: ```text theme={"dark"} 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. ## Indexes are sorted copies 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. ### How sorting affects columnar compression 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: ```text theme={"dark"} 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: * **Run-length encoding (RLE).** Sorted low-cardinality columns become a short list of `(value, count)` pairs. A column with 20 distinct values across a billion rows compresses to almost nothing. * **Dictionary encoding.** Values are replaced by small integer codes into a dictionary. Sorting makes the code sequence itself highly repetitive, so the RLE applied on top of the codes compresses further. * **Delta encoding.** A sorted numeric or timestamp column is monotonically non-decreasing, so storing differences between neighbors yields tiny, similar numbers that pack into a few bits each. 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](/docs/pre-aggregations/using-pre-aggregations) that match your query patterns is the single most effective pre-aggregation optimization. ### Aggregating indexes 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](/docs/pre-aggregations/using-pre-aggregations#aggregating-indexes) 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. ## Auto-partitioning ### Partition key space and ranges 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": ```text theme={"dark"} 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: * **Range bounds vs. data bounds.** Alongside its range (the keys it *may* contain), each partition tracks the actual minimum and maximum key present in its data. Pruning uses the tighter of the two, so a partition with a wide range but sparse data is still skipped effectively. * **The partition split key.** Ranges are defined over a *prefix* of the sort key, called the split key. Rows that share the same split key value are guaranteed to land in the same partition — they are never spread across two partitions. This keeps range boundaries clean and lets the engine reason about a partition as a self-contained unit of work. 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. ### Partition splits 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: ```text theme={"dark"} 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: * **By rows:** total pending rows divided by the split threshold (`CUBESTORE_PARTITION_SPLIT_THRESHOLD`, default 2,097,152 rows; can also be set per pre-aggregation). * **By size:** the partition's file size divided by the size threshold (`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. ## Parquet as the storage format 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. ## Distributed file system as the storage layer 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: * **Workers are disposable.** A worker that dies is replaced by a new one that downloads the same partitions and continues serving. * **Replication is simple.** The object store is the single source of truth. * **Scaling is elastic.** Adding workers redistributes partitions without moving the underlying data. 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. ## Shared-nothing architecture 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: ```text theme={"dark"} 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](#one-plan-two-physical-plans)). This is also how [lambda pre-aggregations](/docs/pre-aggregations/lambda-pre-aggregations) work: a single query transparently combines persisted Parquet data with fresh streaming data held in memory on the same workers. ## Real-time in-memory chunks 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. ```text theme={"dark"} 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](#compaction-in-detail) for the full process. ## Real-time deduplication by unique key 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. ### Version ordering with the sequence column 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: * **Every index sort key ends with the unique key and `__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. ### How it is read A query against a unique-key table dedupes on the fly, seeing streamed updates the moment they arrive: 1. **The scan projection is widened.** Even if the query only asks for a measure, the scan also reads the unique key columns and `__seq` — they are needed to decide which version of each row survives. 2. **Only dedup-safe filters are pushed below deduplication.** A predicate that references unique key columns exclusively can be applied early to each stream: it eliminates a key with *all* of its versions. A predicate on any other column cannot — filtering on a measure value before deduplication could drop the newest version and resurrect an overwritten one. Non-key predicates are applied after deduplication. 3. **All sources merge into one sorted stream.** The partition's main Parquet file, its persisted chunks, and its in-memory chunks are each already sorted; a k-way merge combines them, ordered by the sort key with `__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. 4. **A streaming last-row filter drops stale versions.** The operator walks the merged stream and keeps a row only if the next row has a different key (comparing across batch boundaries as well). Since versions are adjacent and the newest sorts last, this keeps exactly the latest version of every key, in a single pass with no hash table. 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. ### How it is compacted Compaction runs the very same merge-plus-last-row pipeline (see [Compaction in detail](#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](#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. ## Query planning Cube Store builds on [DataFusion](https://datafusion.apache.org), 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. ### From SQL to a logical plan 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. ### Index selection The planner walks the optimized plan and collects, for every scanned table, the constraints relevant to choosing an index: * filter predicates that reached the scan, * the set of projected columns, * the aggregates being computed, * sort requirements imposed by joins or `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: 1. **Eligibility.** For joins, an index qualifies only if the join columns form a prefix of its sort key — that is what makes a merge join possible. An [aggregating index](#aggregating-indexes) qualifies only if it covers every projected column, every filtered column, and every requested aggregate; otherwise it cannot answer the query at all. 2. **Scoring.** Among eligible indexes, an aggregating index always beats a regular one (it is strictly smaller). Ties are broken by how early the query's filter columns appear in the index sort key, then by how early the projected columns appear. Earlier is better: a filter on the leading sort key column narrows the scan to a contiguous range, while the same filter on a trailing column cannot skip anything. 3. **Joins.** Every joined table must have an index sorted by the join key. If no suitable index exists for a join, planning fails with an error that includes the exact `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](#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 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: * Comparison operators: `=`, `<`, `<=`, `>`, `>=` * `IN` lists (each value becomes its own condition) * `AND` / `OR` combinations of the above * Boolean columns used directly in predicates Anything 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: 1. **Partitions** are pruned on the router at planning time, using metastore-tracked min/max keys. 2. **Chunks** within a surviving partition are pruned the same way, using per-chunk min/max keys. 3. **Row groups** within a Parquet file are pruned at scan time on the worker, using Parquet's per-row-group min-max statistics. 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. ### Distribution-aware rewrites Before the plan is finalized, Cube Store rewrites it so that as much work as possible lands below the cluster boundary: * **Aggregate pushdown.** An aggregation above the boundary is split into a *partial* aggregation on the workers and a *final* aggregation on the router. Workers send compact partial states (for example, per-group sums and counts) instead of raw rows. * **Top-k pushdown.** The `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. * **Limit pushdown.** Plain limits propagate to workers so they stop producing rows early. * **Sorted (streaming) aggregation.** Where the chosen index's sort order matches the grouping columns, hash aggregation is replaced with a streaming merge aggregation that emits groups as it goes and needs no hash table. ### One plan, two physical plans 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: * The **router plan** is the part above the cluster boundary: receiving worker streams, merging them (sort-preserving where order matters), final aggregation, and the final limit. * A **worker plan** is the part below the boundary for one worker's subset of partitions: scanning local Parquet files and in-memory chunks, applying remaining filters, and computing partial aggregates. 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. ### Planning walkthrough Putting it all together for a typical query: ```sql theme={"dark"} SELECT product_category, SUM(order_total) FROM orders_rollup WHERE created_at >= '2026-01-01' GROUP BY 1 ORDER BY 2 DESC LIMIT 10 ``` 1. Parse and build the logical plan; push the `created_at` filter down to the scan. 2. Collect constraints and pick the best index — say, an aggregating index sorted by `(created_at, product_category)`. 3. Prune partitions to those whose key ranges intersect `created_at >= '2026-01-01'`. 4. Insert the cluster boundary; rewrite the aggregation + sort + limit into a distributed top-k. 5. Ship the serialized plan; each worker scans its surviving partitions, aggregates partially, and returns its top candidates. 6. The coordinating worker merges the candidates, finalizes the aggregation, applies `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): ```text theme={"dark"} 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. ## Distributed joins 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. ### Joins require an index on the join key 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. ### How the join is distributed 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 `NULL`s 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. ## Compaction in detail 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. ### In-memory chunk compaction Streaming ingestion produces many tiny in-memory chunks. A scheduled job checks each partition and compacts its in-memory chunks when either: * the number of in-memory chunks exceeds a count threshold, or * the oldest buffered row exceeds a lifetime threshold (`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. ### Persistent chunk compaction A per-partition compaction job merges accumulated Parquet chunks into the partition's main file: 1. **Select chunks.** Pending chunks are taken smallest-first, up to a total row budget (`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. 2. **Merge as sorted streams.** The partition's main file and every selected chunk are each already sorted, so they are merged with a streaming k-way merge — no re-sorting, no materializing the inputs in memory. Compaction cost is proportional to bytes streamed. 3. **Deduplicate or aggregate.** On the merged stream: * For tables with a unique key, only the **last row per unique key** is kept — this is how upserts and streaming updates resolve, with later ingestion winning (see [Real-time deduplication by unique key](#real-time-deduplication-by-unique-key)). * For aggregating indexes, rows are grouped by the sort key and the aggregate functions (`SUM`, `MIN`, `MAX`, `MERGE`) combine matching rows — this is where the incremental rollup happens. 4. **Split if needed.** If the partition has outgrown its thresholds, the output is written into several child partitions (see [Partition splits](#partition-splits)); otherwise into a single rewritten main file. Writers cut files only at split-key boundaries. 5. **Upload and swap atomically.** New files are uploaded to the object store, then a single metastore transaction deactivates the parent partition and consumed chunks and activates the children with their computed key ranges. Any query planned before the swap still sees the old files; any query planned after sees the new ones. There is no window of inconsistency, and old files are garbage-collected later. 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. ## Cache and queue store 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. ### Cache The cache is a Redis-style key-value store. Each entry has: * a path of the form `prefix:key`, so related entries (for example, all refresh keys of one deployment) can be listed and truncated by prefix, * a string value, * an optional TTL, after which the entry expires. 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. ### Query queue 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: ```text theme={"dark"} 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. #### Idempotency 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. #### How API instances coordinate scheduling 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: 1. **Cancel dead work** — query stalled and orphaned items (see below) and cancel them. 2. **Start new work** — read the queue's active and pending lists, compute how many more items may run (the concurrency budget minus currently active items), and attempt to claim that many pending items, in priority order. 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. #### What the concurrency number actually means 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. #### Execution, heartbeats, and results 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. #### Stalled and orphaned queries Two failure modes are handled by reconciliation, both detected from timestamps Cube Store keeps on each item: * **Stalled queries** are `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. * **Orphaned queries** are `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. ### Why it lives in Cube Store 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.