Secondary Index on a Low-Cardinality Status Column
An order service stores 400 million rows in orders (customer_id uuid, order_ts timestamp, status text, total decimal, PRIMARY KEY (customer_id, order_ts)). status has 5 distinct values and about 4% of orders are PENDING. Support needs 'all PENDING orders from the last 24 hours, newest first', cluster-wide, on a 6-node cluster with RF 3. Attempt 1 used ALLOW FILTERING and timed out at the 10 s driver timeout. Attempt 2 added CREATE INDEX ON orders (status) and ran the same query without ALLOW FILTERING; it still times out at 10 s, and during the attempt nodetool tpstats shows ReadStage backlogs rising on every node while cluster CPU stays near 40%. Attempt 3 is a nightly reconciliation job that sends one prepared statement SELECT * FROM orders WHERE customer_id IN (?, ...) with 200 ids; it completes, but a few coordinator nodes show roughly 3x the network traffic of the others during the run.
1. Explain why the secondary index on status did not make the query viable.
Read the worked answer
A Cassandra secondary index is local to each node: it indexes only the rows that node already holds, and no node has a global view of the index. The planner therefore cannot know which nodes hold matching rows, so it broadcasts the query to every node (and to every replica set). Each node walks its local index for status = 'PENDING', fetches the matching rows, and returns them; the coordinator merges and pages the combined stream.
With 400M rows and 4% pending, that is roughly 16 million matching rows cluster-wide, on the order of 5 million rows scanned per node for a dashboard that renders 50. CPU stays at 40% because the cost is not compute - it is index and data I/O, per-node ReadStage queueing, and coordinator heap for the merge. The 10 s client timeout follows directly.
The general test for secondary indexes: the queried value must be rare (a small fraction of the rows on each node) and the query should also constrain the partition key. With 5 distinct values, status is the opposite of selective, so an index on it degenerates into a full scan with extra bookkeeping. ALLOW FILTERING and this index are the same scan; the index just makes it look cheaper.
2. Design a table that serves the dashboard query efficiently and explain how it stays correct.
Read the worked answer
Model the query as a table. The read is 'PENDING orders in the last 24 hours, newest first', so make the predicate the partition key and the sort the clustering key:
CREATE TABLE orders_by_status (
status text, bucket date, order_ts timestamp,
customer_id uuid, total decimal,
PRIMARY KEY ((status, bucket), order_ts, customer_id)
) WITH CLUSTERING ORDER BY (order_ts DESC, customer_id ASC);
The dashboard issues one slice for today and one for yesterday and merges them - two single-partition reads instead of a cluster-wide scan. Sizing: a day of PENDING at 4% of daily volume is tens of thousands of rows, comfortably inside one partition. If a peak day ever approaches ~100k rows for a status, shard further (hour bucket, or a hash prefix with reads fanning out over all shards).
Keeping it correct:
- Write the row in the same code path as the order insert, and on every status transition delete the old (status, bucket, order_ts, customer_id) row and insert the new one.
- A status change touches two partitions, so it is not atomic; make the transition idempotent and reconcile periodically from
orders. - If the reconciliation finds drift, rebuild the row from the base table rather than patching.
Materialized views express this declaratively but are still not recommended for new designs in 4.x: they depend on read-before-write, behave badly when base partitions are large, and become unavailable if any view replica is down. If the dashboard later needs many ad-hoc predicates, feed a search engine via CDC instead of adding more tables.
3. Why does the 200-key IN query concentrate load, and how should the reconciliation job be rewritten?
Read the worked answer
A prepared statement with 200 partition keys has no single routing key, so token-aware routing cannot apply: the driver sends the request to one coordinator, which then acts as a super-coordinator, issuing 200 separate read commands - locally where it happens to be a replica, and as forwarded requests to other coordinators where it is not - then gathering every result and paging it back. That is one hot node, 200x fan-out, and a large result buffer, which is exactly the 3x network skew observed.
Rewrite it as 200 separate prepared single-key statements issued asynchronously with bounded concurrency:
Semaphore limit = new Semaphore(48);
for (UUID id : ids) {
limit.acquire();
session.executeAsync(ps.bind(id))
.whenComplete((rs, err) -> { limit.release(); ... });
}
Each statement now carries its partition key, so the driver routes it to a replica, load spreads across the cluster, and memory stays bounded. Project only the columns the reconciliation needs instead of SELECT *, and give the statements a timeout below the job's overall budget.
If the job is truly bulk rather than 200 ids, drop the per-key pattern entirely: page through token ranges (or scan by time bucket), aggregate locally, or run the reconciliation on an analytics path rather than through a coordinator.