Why Columnar Storage¶

Code review, 11:14. Maya is reviewing a query — two columns, one time filter, one GROUP BY — and argues it should be near-instant: "it only touches two columns out of eighty." In prod it takes 8 seconds and reads 17 GB.

Predict before you read on: is the two-column claim wrong, or is the disk layout the actual problem — even though the query itself is fine?

The claim about which columns matter is right and the assumption about what the engine reads off disk is wrong: a dashboard over hundreds of millions of events a day almost never wants a row, it wants two or three columns, aggregated, and the storage layout that matches that question is columns on disk, not tuples — the physics ClickHouse and Pinot operationalize, and the reason Trino is only fast when the files underneath it are columnar too.

Start with the situation¶

Observability / SaaS analytics table:

timestamp | customer_id | service | endpoint | status | latency_ms | bytes | trace_id | user_agent | ...

Twenty to eighty columns. The tile that pays the bills:

SELECT endpoint, count(*), avg(latency_ms)
FROM requests
WHERE timestamp >= now() - INTERVAL 1 HOUR
GROUP BY endpoint;

Two grouping/measure columns, a time filter. trace_id and user_agent are dead weight for this query.

Why the obvious approach breaks¶

Disk and memory buses move bytes, not “logical rows.” If a row is 800 bytes and you need 16, you still move 800 unless the engine can avoid touching the other 784.

At 5×10⁸ events/day × 800 B ≈ 400 GB/day raw. A 7-day scan of a row store is terabytes of IO for a chart with 40 endpoints on it. SSDs do not save you; they just fail later than spinning rust.

Compression does not rescue a row layout. Adjacent bytes in a row are a timestamp, a UUID, a URL, an integer, a string — high entropy. Adjacent bytes in the status column are the same three values for meters of disk.

Build the mental picture¶

Imagine a spreadsheet. A row-store page is several full rows stacked. A column-store file is one column, many rows deep.

To GROUP BY endpoint and average latency_ms you walk two columns. You never load trace_id.

flowchart TB
    subgraph rows["Row pages — query still walks every field"]
        P1["row0: ts, cust, ep, lat, ua, trace, ..."]
        P2["row1: ts, cust, ep, lat, ua, trace, ..."]
    end
    subgraph cols["Column files — query opens two"]
        EP["endpoint.bin  ep ep ep ep ..."]
        LAT["latency_ms.bin  12 40 9 800 ..."]
        UA["user_agent.bin  (not opened)"]
    end

Vectorised engines then feed those columns to the CPU as arrays. One SIMD instruction adds 8–16 latencies. A row interpreter cannot do that without gathering from tuples first.

Under the hood¶

Row vs column IO for GROUP BY two columns¶

Take 1 billion rows, 20 columns, ~50 bytes average per column → ~1 KB/row.

Row store (Postgres heap / InnoDB):

A B-tree on timestamp can limit the row range for “last hour,” which is why OLTP is not uniformly terrible. Once the range is “last 7 days” and you still want two columns, you are back to reading fat tuples.

Column store (Parquet / MergeTree / Pinot segment):

If ORDER BY (endpoint, timestamp) (ClickHouse) or a sorted index (Pinot), equal endpoints sit together: run-length encoding turns 50 million /health rows into a handful of runs. The same GROUP BY becomes closer to a merge of already-sorted keys.

Worked ratio for one hour at 400 million events/day ≈ 16.7 million rows/hour:

The dashboard latency gap (50 ms vs 8 s) is this table, not a magic join algorithm.

Compression is a columnar feature¶

Codecs that matter in production:

ClickHouse lets you set codecs per column. Parquet uses dictionary + run-length + general purpose per page. Pinot similar inside segments.

Rule: sort first, compress second. ORDER BY (service, endpoint, timestamp) makes service and endpoint locally constant. ORDER BY (timestamp) makes time compressible and service a random walk — still columnar, much worse compression on dimensions.

SIMD and vectorised execution¶

CPUs offer SIMD (AVX2 / AVX-512 / NEON): one instruction, many lanes.

Row-at-a-time:

for row in table:
    if row.endpoint == "/checkout":
        acc.sum += row.latency_ms
        acc.n += 1

Branchy, tuple-oriented, wrecks prefetch.

Vectorised (granule / batch of 8,192 rows in ClickHouse):

load 256-bit register of endpoint dictionary ids
compare-eq → bitmask
load latencies
masked add into accumulator

This is why granule size 8,192 is not an arbitrary constant: it is large enough to amortize decoding and fill SIMD loops, small enough that skipping a granule is worth a mark lookup.

Trino pages and Spark whole-stage codegen are the same idea on data they read from Parquet. They do not get ClickHouse’s sparse primary index unless the file format + layout provides skip info.

File layout pieces you will see in EXPLAIN / parts¶

Columnar without skip metadata is still better than rows (you read fewer columns) but you cannot jump. Columnar with a sort key is the OLAP jackpot.

For the specific byte-level anatomy of a Parquet file — row groups, column chunks, pages, the footer, and how predicate pushdown and small-file pathology follow directly from that structure — see Parquet Internals.

Put it to work¶

You choose layout when you write, not when you query.

Parquet (lake, Trino/Spark):

-- Spark: write partitioned, reasonably large files, snappy/zstd
df.repartition("ds")
  .write.mode("append")
  .partitionBy("ds")
  .option("compression", "zstd")
  .parquet("s3://lake/events")

ClickHouse (serving):

CREATE TABLE requests
(
    timestamp   DateTime,
    customer_id LowCardinality(String),
    service     LowCardinality(String),
    endpoint    LowCardinality(String),
    status      UInt16,
    latency_ms  UInt32,
    bytes       UInt32,
    trace_id    String CODEC(ZSTD(3))
)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (service, endpoint, timestamp)
SETTINGS index_granularity = 8192;

Pinot (high QPS): schema + table config declaring sorted column, inverted indexes, star-tree — see Pinot.

Sanity check after load:

-- ClickHouse: compressed vs uncompressed
SELECT
    table,
    formatReadableSize(sum(data_compressed_bytes))   AS compressed,
    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed,
    round(sum(data_uncompressed_bytes) / sum(data_compressed_bytes), 2) AS ratio
FROM system.columns
WHERE table = 'requests'
GROUP BY table;

A ratio of ~2 on an analytics table usually means bad sort key or high-entropy columns dominating. A ratio of 15–30 on dimensions + timestamps is normal.

When columnar is worse¶

Columnar is not a personality. It loses when the access pattern is row-shaped.

Point lookups¶

SELECT * FROM requests WHERE trace_id = '…' needs every column of one row. Row store: one heap fetch (if indexed). Column store: open 20 files, seek to the granule that might contain the id, reconstruct the row.

If trace_id is not in the sort key, you scan (or hit a bloom skip index and still read a granule per hit). Observability “get this trace” belongs in a document/search store or a row-oriented side table, not in the dashboard OLAP table.

Wide updates¶

UPDATE customers SET plan = 'enterprise' WHERE id = 55 is a tuple rewrite (plus WAL) in Postgres. In a column store it is rewrite the part / segment (or a mutation queue that rewrites later). One column change dirties the layout because parts are immutable.

Streaming appends are fine. CDC that mutates 2% of a 10 TB table every hour is how you recreate Hadoop.

Tiny rows, transactional writes¶

Inserting one event per HTTP request: row store appends to a page. ClickHouse creates a part. Ten thousand parts later you are in too many parts. Columnar engines want batches.

Highly selective OLTP¶

WHERE customer_id = ? AND order_id = ? returning 1 row, 200 QPS, 5 ms: indexes on a row store. A sparse 8,192-row granule index will read thousands of rows of each column you project to reconstruct one tuple.

Wide SELECT * over a few rows is still OK-ish¶

The disaster is SELECT * over billions of rows. Reconstructing 10 rows from columns is fine. Reconstructing 10⁹ rows from 80 columns is how a “data export” melts the cluster.

Where teams get caught¶

How it fails¶

Practice the idea¶

Use the Bloom-filter playground. Predict whether a missing value can be reported as “possibly present,” then increase the inserted-item count without increasing the bit array. Connect the rising false-positive rate to extra reads, not incorrect query results.

What happened next¶

The teammate was right about the columns and wrong about the bytes. Two columns out of eighty is the query; 17 GB is what came off disk, because the rows were stored as tuples and reading any column meant reading the whole row.

Columnar layout is what makes the claim true: store each column contiguously and a two-column query reads two columns, compresses them far better because neighbouring values are similar, and skips whole blocks whose min/max cannot match the filter. The query never changed.

Which is the physics under the next three pages — ClickHouse and Pinot operationalize it, and Trino is only fast when the files beneath it are columnar too. Eight seconds was never a query-planning problem.

Check your understanding¶

Table events 80 columns, 2 billion rows. Query A: SELECT customer_id, sum(bytes) FROM events WHERE ds = '2024-06-12' GROUP BY customer_id. Query B: SELECT * FROM events WHERE request_id = 'abc'. Query C: UPDATE events SET bytes = 0 WHERE customer_id = 'free-tier'.

For each, say row store vs column store vs “neither / copy,” and what IO you expect on a cold cache.

Reference¶

Behaviour at the next orders of magnitude, the trade-offs, the alternatives, and what to check when inheriting someone else's version of this — kept here rather than in the walkthrough above.

How to investigate¶

Postgres (prove the row-store tax):

EXPLAIN (ANALYZE, BUFFERS)
SELECT endpoint, avg(latency_ms)
FROM requests
WHERE timestamp > now() - interval '1 hour'
GROUP BY endpoint;

Look at shared read / read bytes. If buffers dwarf the two columns, you found it.

ClickHouse:

EXPLAIN PIPELINE
SELECT endpoint, avg(latency_ms)
FROM requests
WHERE timestamp > now() - 3600
GROUP BY endpoint;

SELECT
    query,
    read_rows,
    formatReadableSize(read_bytes) AS read,
    formatReadableSize(memory_usage) AS mem
FROM system.query_log
WHERE type = 'QueryFinish'
ORDER BY event_time DESC
LIMIT 20;

read_bytes should be in the same neighborhood as compressed size of the columns you named, times granules you did not skip. If you read 80% of the table for a 1-hour filter, timestamp is not helping the primary index — see ClickHouse ORDER BY.

Parquet / Trino: EXPLAIN should list projected columns and partition filter. Footer-heavy plans: too many files.

Marks/granules: system.parts (rows, bytes_on_disk), system.parts_columns. A part with 200 rows is a write-pattern bug, not a columnar-theory bug.

Scale 10× / 100× / 1000׶

Columnar does not scale a point-lookup API 1000×. It scales scans and aggregations whose working set is a few columns.

Trade-offs¶

Alternatives¶

Apply¶

Before you pick an engine, write the query and count:

1. How many columns are named?
2. How many rows survive the time filter?
3. Is this a lookup of one entity or an aggregate of a population?
4. Do you update those rows?

Aggregates over a few columns, append-only, dashboard latency → columnar. Lookups, updates, tiny writes → row store (or a hybrid: OLAP copy + OLTP source).

Then implement the copy. Do not “turn Postgres into a column store” with a dozen covering indexes; you will still update tuples.