ClickHouse Analytics: Petabyte-Scale Query Performance

ClickHouse can scan billions of rows per second on a single server. Achieving that performance requires understanding its fundamentally different architecture from OLTP databases.

Why ClickHouse Is Fast

1. Columnar storage: Only reads queried columns
2. Compression: Similar values compress 5-10x better than row stores
3. Vectorized execution: Processes 128 rows at a time with SIMD instructions
4. MergeTree granules: Sparse index skips irrelevant 8192-row blocks
5. Parallel execution: All CPU cores used automatically

Choosing the Right Engine

-- ReplacingMergeTree: Deduplication by primary key
CREATE TABLE events (
    event_id UUID,
    user_id UInt64,
    event_type LowCardinality(String),
    created_at DateTime64(3),
    version UInt64
)
ENGINE = ReplacingMergeTree(version)
PARTITION BY toYYYYMM(created_at)
ORDER BY (user_id, event_type, created_at);

-- AggregatingMergeTree: Pre-aggregated analytics states
CREATE TABLE hourly_stats (
    hour DateTime,
    user_id UInt64,
    event_count AggregateFunction(count, UInt8),
    total_amount AggregateFunction(sum, Decimal(18, 2))
)
ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, user_id);

-- SummingMergeTree: Auto-sum counters on merge
CREATE TABLE ad_metrics (
    hour DateTime,
    campaign_id UInt32,
    impressions UInt64,
    clicks UInt64,
    spend Float64
)
ENGINE = SummingMergeTree((impressions, clicks, spend))
ORDER BY (hour, campaign_id);

ORDER BY Design: The Most Critical Decision

ORDER BY is a sort key, not a uniqueness constraint—it determines query performance:

-- WRONG: High-cardinality first, cannot skip granules on user_id
ORDER BY (event_id, user_id, created_at)

-- RIGHT: Low to high cardinality
ORDER BY (user_id, event_type, created_at)
-- Queries on user_id skip 99%+ of granules!

Optimal Data Types

-- Use smallest types that fit your data
CREATE TABLE optimized (
    user_id UInt64,                       -- 8 bytes
    country LowCardinality(String),       -- Dictionary encoding
    status UInt8,                         -- 1 byte
    amount Decimal(18, 2),                -- Sufficient precision
    created_at DateTime64(3)              -- Millisecond precision
);
-- Benchmark: String vs LowCardinality for country (100M rows)
-- String: 15GB, LowCardinality: 2GB, GROUP BY: 15x faster

Materialized Views for Pre-Aggregation

-- Feeds aggregate table on each INSERT automatically
CREATE MATERIALIZED VIEW mv_hourly
TO hourly_stats AS
SELECT
    toStartOfHour(created_at) AS hour,
    user_id,
    countState() AS event_count,
    sumState(amount) AS total_amount
FROM events
GROUP BY hour, user_id;

-- Query: 1000x faster than scanning raw events
SELECT
    hour, user_id,
    countMerge(event_count) AS events,
    sumMerge(total_amount) AS revenue
FROM hourly_stats
WHERE hour >= now() - INTERVAL 24 HOUR
GROUP BY hour, user_id
ORDER BY revenue DESC;

Query Optimization Patterns

-- PREWHERE: Pre-filter using one column before reading others
SELECT user_id, amount
FROM events
PREWHERE event_type = 'purchase'  -- Only reads event_type first
WHERE amount > 100;               -- Then applies amount filter

-- Approximate functions: 2% error, 10-50x faster
SELECT uniqHLL12(user_id) FROM events;         -- vs COUNT(DISTINCT)
SELECT quantileTDigest(0.95)(response_ms) FROM requests;  -- vs exact p95

-- Never SELECT *: reads ALL columns in columnar store
SELECT user_id, event_type, amount FROM events  -- Specify only needed
WHERE created_at > '2026-01-01';

Partitioning and TTL

-- Monthly partitions for time-series
CREATE TABLE logs (
    timestamp DateTime,
    level LowCardinality(String),
    message String
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (level, timestamp);

-- Auto-delete old data
ALTER TABLE logs MODIFY TTL timestamp + INTERVAL 90 DAY;

-- Tiered storage: hot on SSD, cold on HDD
ALTER TABLE events MODIFY TTL
    created_at + INTERVAL 30 DAY TO DISK 'ssd',
    created_at + INTERVAL 1 YEAR TO DISK 'hdd';

Distributed Setup

-- Replicated local table on each shard
CREATE TABLE events_local ON CLUSTER my_cluster (...)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}')
...

-- Distributed table reads from all shards
CREATE TABLE events ON CLUSTER my_cluster
AS events_local
ENGINE = Distributed(my_cluster, default, events_local, rand());

Performance Profiling

-- Verify granule skipping is working
EXPLAIN indexes = 1
SELECT count() FROM events WHERE user_id = 123;
-- "Granules: 5/100000" = 99.995% of data skipped!

-- Find slow queries
SELECT query, read_rows, memory_usage, query_duration_ms
FROM system.query_log
WHERE type = 'QueryFinish' AND query_start_time > now() - INTERVAL 1 HOUR
ORDER BY query_duration_ms DESC LIMIT 10;

Key insight: ClickHouse schema design must be driven by query patterns, not normalization. Denormalize aggressively, choose ORDER BY carefully, and use materialized views for common aggregations.