Clickhouse Best Practices

ClickHouse Best Practices

Guidance for ClickHouse covering schema design, query optimization, and data ingestion. Contains 28 atomic rules across 3 categories (schema, query, insert), prioritized by impact. Extended with 14 reference files covering cluster management, backups, monitoring, and integrations.

Official docs: ClickHouse Best Practices

Official docs: ClickHouse Best Practices

---

⚠️ Security Considerations

Credential Placeholders

Example credentials in documentation (password123, AKIAIOSFODNN7EXAMPLE) are placeholders only. Never use these in production. Use proper secret management:

• Environment variables
• Secret managers (AWS Secrets Manager, HashiCorp Vault, etc.)
• Kubernetes secrets (for K8s deployments)
• ClickHouse named collections with external configuration

Installation & Operations

For installation and operational procedures:

• Follow official documentation links provided in reference files
• Prefer package managers (apt, yum, helm) over direct downloads
• Use versioned artifacts instead of latest in production
• Test procedures in non-production environments first

---

IMPORTANT: How to Apply This Skill

Before answering ClickHouse questions, follow this priority order:

1. Check for applicable rules in the rules/ directory 2. If rules exist: Apply them and cite them in your response using "Per rule-name..." 3. If no rule exists: Check references/ for deeper topic coverage 4. If neither covers it: Use general ClickHouse knowledge or search documentation 5. Always cite your source: rule name, reference file, or URL

Why rules take priority: ClickHouse has specific behaviors (columnar storage, sparse indexes, merge tree mechanics) where general database intuition can be misleading. The rules encode validated, ClickHouse-specific guidance.

---

Review Procedures

For Schema Reviews (CREATE TABLE, ALTER TABLE)

Read these rule files in order:

1. rules/schema-pk-plan-before-creation.md — ORDER BY is immutable 2. rules/schema-pk-cardinality-order.md — Column ordering in keys 3. rules/schema-pk-prioritize-filters.md — Filter column inclusion 4. rules/schema-pk-filter-on-orderby.md — Query filter alignment 5. rules/schema-types-native-types.md — Proper type selection 6. rules/schema-types-minimize-bitwidth.md — Numeric type sizing 7. rules/schema-types-lowcardinality.md — LowCardinality usage 8. rules/schema-types-avoid-nullable.md — Nullable vs DEFAULT 9. rules/schema-types-enum.md — Enum for finite value sets 10. rules/schema-partition-low-cardinality.md — Partition count limits 11. rules/schema-partition-lifecycle.md — Partitioning purpose 12. rules/schema-partition-query-tradeoffs.md — Partition pruning trade-offs 13. rules/schema-partition-start-without.md — Start without partitioning 14. rules/schema-json-when-to-use.md — JSON type usage

Check for:

• [ ] PRIMARY KEY / ORDER BY column order (low-to-high cardinality)
• [ ] Data types match actual data ranges
• [ ] LowCardinality applied to appropriate string columns
• [ ] Partition key cardinality bounded (100-1,000 values)
• [ ] ReplacingMergeTree has version column if used

For Query Reviews (SELECT, JOIN, aggregations)

Read these rule files:

1. rules/query-join-choose-algorithm.md — Algorithm selection 2. rules/query-join-use-any.md — ANY vs regular JOIN 3. rules/query-join-filter-before.md — Pre-join filtering 4. rules/query-join-consider-alternatives.md — Dictionaries/denormalization 5. rules/query-join-null-handling.md — join_use_nulls setting 6. rules/query-index-skipping-indices.md — Secondary index usage 7. rules/query-mv-incremental.md — Incremental materialized views 8. rules/query-mv-refreshable.md — Refreshable materialized views

Check for:

• [ ] Filters use ORDER BY prefix columns
• [ ] JOINs filter tables before joining (not after)
• [ ] Correct JOIN algorithm for table sizes
• [ ] Skipping indices for non-ORDER BY filter columns

For Insert Strategy Reviews (data ingestion, updates, deletes)

Read these rule files:

1. rules/insert-batch-size.md — Batch sizing requirements 2. rules/insert-async-small-batches.md — Async insert usage 3. rules/insert-format-native.md — Native format for performance 4. rules/insert-mutation-avoid-update.md — UPDATE alternatives 5. rules/insert-mutation-avoid-delete.md — DELETE alternatives 6. rules/insert-optimize-avoid-final.md — OPTIMIZE TABLE risks

Check for:

• [ ] Batch size 10K-100K rows per INSERT
• [ ] No ALTER TABLE UPDATE for frequent changes
• [ ] ReplacingMergeTree or CollapsingMergeTree for update patterns
• [ ] Async inserts enabled for high-frequency small batches

---

Output Format

Structure review responses as follows:

## Rules Checked
- `rule-name-1` — Compliant / Violation found
- `rule-name-2` — Compliant / Violation found
...

## Findings

### Violations
- **`rule-name`**: Description of the issue
  - Current: [what the code does]
  - Required: [what it should do]
  - Fix: [specific correction]

### Compliant
- `rule-name`: Brief note on why it's correct

## Recommendations
[Prioritized list of changes, citing rules]

---

Rule Categories by Priority

---

Quick Reference

Schema Design — Primary Key (CRITICAL)

• schema-pk-plan-before-creation — Plan ORDER BY before table creation (immutable)
• schema-pk-cardinality-order — Order columns low-to-high cardinality
• schema-pk-prioritize-filters — Include frequently filtered columns
• schema-pk-filter-on-orderby — Query filters must use ORDER BY prefix

Schema Design — Data Types (CRITICAL)

• schema-types-native-types — Use native types, not String for everything
• schema-types-minimize-bitwidth — Use smallest numeric type that fits
• schema-types-lowcardinality — LowCardinality for <10K unique strings
• schema-types-enum — Enum for finite value sets with validation
• schema-types-avoid-nullable — Avoid Nullable; use DEFAULT instead

Schema Design — Partitioning (HIGH)

• schema-partition-low-cardinality — Keep partition count 100-1,000
• schema-partition-lifecycle — Use partitioning for data lifecycle, not queries
• schema-partition-query-tradeoffs — Understand partition pruning trade-offs
• schema-partition-start-without — Consider starting without partitioning

Schema Design — JSON (MEDIUM)

• schema-json-when-to-use — JSON for dynamic schemas; typed columns for known

Query Optimization — JOINs (CRITICAL)

• query-join-choose-algorithm — Select algorithm based on table sizes
• query-join-use-any — ANY JOIN when only one match needed
• query-join-filter-before — Filter tables before joining
• query-join-consider-alternatives — Dictionaries/denormalization vs JOIN
• query-join-null-handling — join_use_nulls=0 for default values

Query Optimization — Indices (HIGH)

• query-index-skipping-indices — Skipping indices for non-ORDER BY filters

Query Optimization — Materialized Views (HIGH)

• query-mv-incremental — Incremental MVs for real-time aggregations
• query-mv-refreshable — Refreshable MVs for complex joins

Insert Strategy — Batching (CRITICAL)

• insert-batch-size — Batch 10K-100K rows per INSERT

Insert Strategy — Async (HIGH)

• insert-async-small-batches — Async inserts for high-frequency small batches
• insert-format-native — Native format for best performance

Insert Strategy — Mutations (CRITICAL)

• insert-mutation-avoid-update — ReplacingMergeTree instead of ALTER UPDATE
• insert-mutation-avoid-delete — Lightweight DELETE or DROP PARTITION

Insert Strategy — Optimization (HIGH)

• insert-optimize-avoid-final — Let background merges work

---

Quick Decision Guides

Which Table Engine?

Need to store data?
├── < 1M rows, dimension → Memory
└── ≥ 1M rows → MergeTree family
    ├── Deduplication? → ReplacingMergeTree(version)
    ├── Changelog? → CollapsingMergeTree(sign)
    ├── Pre-aggregation? → AggregatingMergeTree()
    ├── Replication? → ReplicatedMergeTree(...)
    └── Default → MergeTree()

See references/table-engines.md for complete reference.

Common Issues & Quick Fixes

See references/debugging.md for detailed troubleshooting.

---

Deep Reference Files

For topics beyond the 28 rules, see the references/ directory:

Schema & Table Design

• references/core-concepts.md — Architecture, data model, internals
• references/schema-design.md — Database engines, migrations, version control
• references/table-design.md — ORDER BY, partitioning, column selection
• references/table-engines.md — Complete MergeTree family reference

Query & Performance

• references/sql-reference.md — Complete SQL dialect, data types
• references/query-optimization.md — EXPLAIN, JOINs, projections, skip indexes
• references/advanced-features.md — Materialized views, mutations, TTL, dictionaries

Operations & Cluster

• references/debugging.md — Query debugging, merges, mutations, replication
• references/cluster-management.md — Distributed tables, replication, sharding
• references/backup-restore.md — Backup strategies, disaster recovery
• references/monitoring.md — Query monitoring, health checks, system queries

Integration & Best Practices

• references/integrations.md — Kafka, S3, PostgreSQL, MySQL, BI tools
• references/best-practices.md — Complete checklist and anti-patterns
• references/external.md — Altinity KB links, official docs
• references/system-queries.md — Ready-to-use queries for operations

---

Version: 1.3.0 Rules: Synced with ClickHouse/agent-skills (Apache-2.0) References: Altinity Knowledge Base (200+ articles) + ClickHouse Official Docs

ClickHouse Advanced Features

Materialized views, mutations, TTL, dictionaries, and other advanced capabilities.

Materialized Views

Materialized views automatically process data on INSERT:

Basic Materialized View

-- Pre-aggregate on write
CREATE MATERIALIZED VIEW mv_daily_stats
ENGINE = AggregatingMergeTree()
ORDER BY (date, user_id)
AS SELECT
    toDate(timestamp) as date,
    user_id,
    countState() as hits,
    sumState(revenue) as total_revenue
FROM events
GROUP BY date, user_id;

-- Query MV (fast - already aggregated!)
SELECT
    user_id,
    countMerge(hits) as total_hits,
    sumMerge(total_revenue) as revenue
FROM mv_daily_stats
WHERE date = today()
GROUP BY user_id;

Materialized View to Target Table

-- Create target table
CREATE TABLE daily_stats (
    date Date,
    user_id UInt32,
    hits UInt64,
    revenue Decimal(18, 2)
)
ENGINE = AggregatingMergeTree()
ORDER BY (date, user_id);

-- Create MV that populates target table
CREATE MATERIALIZED VIEW mv_populate_daily_stats
TO daily_stats
AS SELECT
    toDate(timestamp) as date,
    user_id,
    count() as hits,
    sum(revenue) as revenue
FROM events
GROUP BY date, user_id;

POPULATE Option

-- Create MV and backfill existing data
CREATE MATERIALIZED VIEW mv_daily_stats
ENGINE = AggregatingMergeTree()
ORDER BY (date, user_id)
POPULATE  -- Backfill existing data
AS SELECT
    toDate(timestamp) as date,
    user_id,
    count() as hits
FROM events
GROUP BY date, user_id;

Drop Materialized View

-- Drop MV (target table data remains)
DROP TABLE mv_daily_stats;

Materialized view use cases:

• Pre-aggregation (sum, count, avg)
• Rollup data (hourly → daily)
• Data routing (INSERT → multiple tables)
• Real-time analytics

Mutations

Mutations are async UPDATE/DELETE operations:

UPDATE

-- UPDATE (async, expensive - rewrites all data)
ALTER TABLE events UPDATE status = 'done'
WHERE event_id = 123;

-- Multiple columns
ALTER TABLE events UPDATE
    status = 'done',
    processed_at = now()
WHERE event_id = 123;

DELETE

-- DELETE (async, expensive - rewrites all data)
ALTER TABLE events DELETE
WHERE timestamp < now() - INTERVAL 90 DAY;

Monitor Mutations

-- Show all mutations
SELECT * FROM system.mutations;

-- Active mutations
SELECT
    database,
    table,
    command,
    is_done,
    parts_to_do,
    parts_to_do_names,
    elapsed
FROM system.mutations
WHERE is_done = 0;

-- Mutation progress
SELECT
    table,
    command,
    is_done,
    parts_to_do - parts_to_do_names as parts_remaining
FROM system.mutations
ORDER BY parts_to_do DESC;

Force Mutation Completion

-- Wait for mutation to complete
SYSTEM STOP MERGES;
ALTER TABLE ... UPDATE ...;
OPTIMIZE TABLE ... FINAL;
SYSTEM START MERGES;

Mutation characteristics:

• Async: Doesn't block table
• Expensive: Rewrites all data
• Can be slow on large tables
• Better: Use TTL or new tables

TTL (Time To Live)

TTL automatically manages data lifecycle:

Delete Old Data

-- Delete old data
CREATE TABLE events (
    timestamp DateTime,
    data String
)
ENGINE = MergeTree()
ORDER BY timestamp
TTL timestamp + INTERVAL 90 DAY;

Recompress Old Data

-- Change compression over time
CREATE TABLE events (
    timestamp DateTime,
    data String
)
ENGINE = MergeTree()
ORDER BY timestamp
TTL
    timestamp + INTERVAL 7 DAY TO VOLUME 'cold',
    timestamp + INTERVAL 30 DAY TO DISK 's3';

Column TTL

-- Column-level TTL
CREATE TABLE events (
    timestamp DateTime,
    data String,
    metadata String TTL timestamp + INTERVAL 30 DAY DELETE
)
ENGINE = MergeTree()
ORDER BY timestamp;

Modify TTL

-- Modify table TTL
ALTER TABLE events
MODIFY TTL timestamp + INTERVAL 180 DAY;

-- Modify column TTL
ALTER TABLE events
MODIFY COLUMN metadata TTL timestamp + INTERVAL 60 DAY DELETE;

Drop Partition (Instant)

-- Drop entire partition (instant, no mutation)
ALTER TABLE events DROP PARTITION '202401';

TTL operations:

• Delete old data
• Move to cold storage
• Change compression codec
• Drop partition (instant)

Dictionaries

Dictionaries provide fast in-memory lookups:

PostgreSQL Dictionary

-- Create dictionary
CREATE DICTIONARY users_dict (
    user_id UInt32,
    email String,
    name String,
    created_at DateTime
)
PRIMARY KEY user_id
SOURCE(POSTGRESQL(
    port 5432
    host 'localhost'
    db 'mydb'
    table 'users'
    user 'user'
    password 'pass'
))
LIFETIME(60)  -- Refresh every 60 seconds
LAYOUT(HASHED());

-- Use in query
SELECT
    e.user_id,
    dictGet('users_dict', 'email', e.user_id) as email,
    dictGet('users_dict', 'name', e.user_id) as name
FROM events e;

Cache Dictionary

CREATE DICTIONARY users_cache (
    user_id UInt32,
    email String
)
PRIMARY KEY user_id
SOURCE(POSTGRESQL(...))
LIFETIME(300)
LAYOUT(CACHE(SIZE_IN_CELLS 10000));  -- LRU cache

Dictionary Functions

-- Get value
dictGet('dict_name', 'attribute_type', key)
dictGetOrDefault('dict_name', 'attribute_type', key, default_value)
dictHas('dict_name', key)

-- Check dictionary status
SELECT * FROM system.dictionaries WHERE name = 'users_dict';

Dictionary Layouts

Projections

Projections are materialized views at part level:

-- Create projection
ALTER TABLE events ADD PROJECTION pr_user_daily (
    SELECT
        user_id,
        toDate(timestamp) as date,
        count() as events,
        sum(revenue) as total_revenue
    GROUP BY user_id, date
);

-- Query automatically uses projection (fast!)
SELECT
    user_id,
    date,
    count() as events,
    sum(revenue) as total_revenue
FROM events
GROUP BY user_id, date;

-- Drop projection
ALTER TABLE events DROP PROJECTION pr_user_daily;

Projection benefits:

• Automatic usage (no query changes)
• Faster than materialized views
• Maintained per-part (smaller overhead)

FINAL Clause

FINAL forces application of mutations:

-- Force deduplication (expensive!)
SELECT * FROM ReplacingMergeTree_table FINAL WHERE user_id = 123;

-- Check what FINAL does
EXPLAIN SELECT * FROM table FINAL;

-- Better: Design queries to handle duplicates
SELECT
    user_id,
    argMax(profile, updated_at) as latest_profile
FROM user_profiles
GROUP BY user_id;

See Also

• ../SKILL.md - Main skill entry point
• table-engines.md - Complete table engine reference
• debugging.md - Mutation monitoring and troubleshooting

ClickHouse Backup & Restore

Backup strategies, disaster recovery, and data protection.

Backup Strategies Comparison

clickhouse-backup (Recommended)

Installation

See the official clickhouse-backup documentation for installation instructions using:

• Docker images
• DEB/RPM packages
• Helm charts (for Kubernetes)
• Building from source

Configuration

# /etc/clickhouse-backup/config.yml
general:
  remote_storage: s3
  max_file_size: 1073741824
  disable_progress_bar: false
  backups_to_keep_local: 5
  backups_to_keep_remote: 10

clickhouse:
  username: default
  password: ""  # ⚠️ Use proper secret management in production
  host: localhost
  port: 9000
  debug: false
  sync_replicated_tables: true
  skip_table_engines: [Dictionary, View, Set, Join]
  skip_tables: [system.*, temporary_*.*, information_schema*]

s3:
  access_key: AKIAIOSFODNN7EXAMPLE  # ⚠️ Placeholder - use AWS IAM roles or proper secrets
  secret_key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY  # ⚠️ Placeholder
  bucket: clickhouse-backups
  endpoint: https://s3.amazonaws.com
  region: us-west-2
  acl: private
  force_path_style: false
  path: /backups/

# GCS configuration
gcs:
  credentials_file: /path/to/service-account.json
  bucket: clickhouse-backups
  path: /backups/

# Azure configuration
azblob:
  account_name: myaccount
  account_key: mykey
  container: clickhouse-backups
  path: /backups/

Commands

# Create local backup
clickhouse-backup create my_backup

# Upload to remote storage
clickhouse-backup upload my_backup

# List backups
clickhouse-backup list

# Download from remote
clickhouse-backup download my_backup

# Restore from backup
clickhouse-backup restore my_backup

# Restore specific tables
clickhouse-backup restore my_backup -t my_db.my_table

# Create and upload in one command
clickhouse-backup create my_backup --upload

# Delete backup
clickhouse-backup delete local my_backup
clickhouse-backup delete remote my_backup

Incremental Backups

# Differential backup (only changes since last)
clickhouse-backup create --diff-from=previous_backup my_backup

# Upload
clickhouse-backup upload my_backup

Backup Automation

# Cron job for daily backups
0 2 * * * clickhouse-backup create "daily-$(date +\%Y\%m\%d)" --upload >/dev/null 2>&1

# Cron job for weekly full backup
0 3 * * 0 clickhouse-backup create "weekly-$(date +\%Y\%m\%d)" --upload >/dev/null 2>&1

# Cron job to clean old backups
0 4 * * * clickhouse-backup delete local --clickhouse-backup /etc/clickhouse-backup/config.yml >/dev/null 2>&1

BACKUP Statement (v23.6+)

Backup to Disk

-- Backup entire database
BACKUP DATABASE my_db TO DISK('backups/my_db');

-- Backup specific tables
BACKUP TABLE my_db.table1, my_db.table2 TO DISK('backups/partial');

-- Backup with pattern
BACKUP DATABASE my_db.* TO DISK('backups/my_db');

-- Async backup
BACKUP DATABASE my_db TO DISK('backups/my_db') SETTINGS async=true;

Backup to S3

-- Backup to S3
BACKUP DATABASE my_db TO S3('https://bucket.s3.amazonaws.com/backups/my_db');

-- With credentials
BACKUP DATABASE my_db TO S3(
    'https://bucket.s3.amazonaws.com/backups/my_db',
    'access_key',
    'secret_key'
);

Restore

-- Restore database
RESTORE DATABASE my_db FROM DISK('backups/my_db');

-- Restore with new name
RESTORE DATABASE my_db AS my_db_new FROM DISK('backups/my_db');

-- Restore specific tables
RESTORE TABLE my_db.table1 FROM DISK('backups/partial');

-- Async restore
RESTORE DATABASE my_db FROM DISK('backups/my_db') SETTINGS async=true;

Backup Status

-- Check backup status
SELECT * FROM system.backups;

-- List backups
SELECT name, status, size
FROM system.backups
ORDER BY creation_time DESC;

Snapshot-Based Backup

EBS Snapshots (AWS)

# 1. Freeze filesystem
clickhouse-client --query="SYSTEM FREEZE TABLES;"

# 2. Take snapshot
aws ec2 create-snapshot \
  --volume-id vol-xxxxxxxx \
  --description "ClickHouse backup $(date +%Y-%m-%d)"

# 3. Unfreeze
clickhouse-client --query="SYSTEM UNFREEZE TABLES;"

# 4. Cleanup old snapshots (keep last 7 days)
aws ec2 describe-snapshots \
  --filters Name=description,Values="ClickHouse backup*" \
  --query 'Snapshots[?StartTime<`$(date -d '7 days ago' +%Y-%m-%d)`].SnapshotId' \
  --output text | xargs -I {} aws ec2 delete-snapshot --snapshot-id {}

GCE Persistent Disks

# 1. Freeze tables
clickhouse-client --query="SYSTEM FREEZE TABLES;"

# 2. Create snapshot
gcloud compute disks snapshot clickhouse-disk \
  --snapshot-names clickhouse-snapshot-$(date +%Y%m%d)

# 3. Unfreeze
clickhouse-client --query="SYSTEM UNFREEZE TABLES;"

Filesystem-Based Backup

Using rsync

# Stop ClickHouse
systemctl stop clickhouse-server

# Backup data directory
rsync -av --delete /var/lib/clickhouse/ /backup/clickhouse/

# Start ClickHouse
systemctl start clickhouse-server

Using cp

# Stop ClickHouse
systemctl stop clickhouse-server

# Copy data directory
cp -r /var/lib/clickhouse/ /backup/clickhouse/

# Start ClickHouse
systemctl start clickhouse-server

Disaster Recovery

Complete Restore Procedure

For disaster recovery procedures, see the clickhouse-backup documentation.

Key considerations:

• Always verify backup integrity before restore
• Test restore procedures in non-production environments
• Use clickhouse-backup's built-in restore commands rather than manual filesystem operations
• Consider restoring to a new database/cluster first to verify data integrity

Point-in-Time Recovery

-- 1. Find backup
SELECT * FROM system.backups
WHERE creation_time >= '2024-01-01 00:00:00'
  AND creation_time <= '2024-01-01 12:00:00'
ORDER BY creation_time DESC
LIMIT 1;

-- 2. Restore to new database
RESTORE DATABASE my_db AS my_db_pit
FROM DISK('backups/my_db_20240101');

-- 3. Export needed data
-- 4. Import to production

Data Migration

# Using clickhouse-backup
clickhouse-backup create migration_backup
clickhouse-backup upload migration_backup

# On new cluster
clickhouse-backup download migration_backup
clickhouse-backup restore migration_backup

# Using clickhouse-copier for large clusters
# Configure and run:
clickhouse-copier --config=config.xml --base-dir=/tmp/

Backup Verification

Check Backup Integrity

-- Verify row counts after restore
SELECT
    database,
    table,
    sum(rows) as row_count
FROM system.parts
WHERE active = 1
GROUP BY database, table
ORDER BY database, table;

-- Compare checksums
SELECT
    table,
    sum(bytes_on_disk) as size,
    checksum(*) as checksum
FROM system.parts
WHERE active = 1
GROUP BY table;

Test Restore Procedure

# 1. Create test database
clickhouse-client --query="CREATE DATABASE test_restore;"

# 2. Restore to test database
clickhouse-backup restore my_backup -d my_db --db-target test_restore

# 3. Verify data
clickhouse-client --query="SELECT count() FROM test_restore.events"

# 4. Drop test database
clickhouse-client --query="DROP DATABASE test_restore;"

Best Practices

1. Automate backups - Use cron or Kubernetes cronjobs 2. Test restores - Regularly test restore procedure 3. Offsite storage - Store backups in S3/GCS/Azure 4. Retention policy - Keep 7-30 daily, 4-12 weekly backups 5. Monitor backups - Alert on backup failures 6. Document RTO/RPO - Define recovery objectives 7. Encrypt backups - Use encryption for sensitive data 8. Version control - Track backup configuration

Backup Checklist

• [ ] Automated daily backups configured
• [ ] Backups stored offsite (S3, GCS, Azure)
• [ ] Retention policy defined
• [ ] Test restore procedure documented
• [ ] Backup monitoring configured
• [ ] Disaster recovery plan documented
• [ ] RTO/RPO defined
• [ ] Encryption enabled for sensitive data
• [ ] Backup size monitored and optimized
• [ ] Incremental backups configured

See Also

• ../SKILL.md - Main skill entry point
• debugging.md - Troubleshooting backup issues
• cluster-management.md - Replication for high availability
• monitoring.md - Backup monitoring and health checks

ClickHouse Best Practices

Comprehensive checklist and anti-patterns for ClickHouse.

Schema Design Checklist

• [ ] ORDER BY matches query WHERE patterns
• Most selective filter first
• Time-series: timestamp as second column
• Limit to 3-4 columns

• [ ] Partitioning aligned with TTL/DROP needs
• Time-based: Monthly or daily partitions
• Aim for 100-1000 parts total
• Use toYYYYMM() for monthly

• [ ] Primary key is subset of ORDER BY
• Reduces primary key size
• Must be prefix of ORDER BY

• [ ] Used smallest sufficient types
• UInt8 vs UInt32 vs UInt64
• Date vs DateTime (2 bytes vs 4 bytes)
• Decimal for currency

• [ ] Used LowCardinality for enum-like strings
• < 10k distinct values
• Significant compression

• [ ] Avoided Nullable when possible
• Use default values instead
• Nullable has overhead

• [ ] Avoided MODIFY/DROP COLUMN
• Use ADD COLUMN only
• Create new tables for schema changes

Query Writing Checklist

• [ ] Smaller table on RIGHT side of JOIN
• ClickHouse sends RIGHT table to all shards
• Minimize network transfer

• [ ] Used GLOBAL JOIN for distributed queries
• Prevents sending right table multiple times

• [ ] Added skip indexes for frequent filters
• Bloom filter for exact match
• Minmax for range queries
• Set for IN queries

• [ ] Leveraged projections for common aggregations
• Pre-computed aggregations
• Automatic usage

• [ ] Avoided SELECT *
• Reads all columns (columnar penalty)
• Select only needed columns

• [ ] Used EXPLAIN to verify index usage
• Look for "Index" in output
• Avoid "Filter" (full scan)

• [ ] Set reasonable max_memory_usage
• Prevent OOM errors
• Tune based on available memory

• [ ] Used SAMPLE for exploratory queries
• Fast approximate results
• SELECT ... SAMPLE 0.1

Performance Checklist

• [ ] Monitored merges (system.merges)
• Active merges impact performance
• Check merge queue size

• [ ] Checked mutation progress
• Mutations are expensive
• Monitor system.mutations

• [ ] Used TTL instead of DELETE
• Automatic data lifecycle
• No mutation overhead

• [ ] Preferred INSERT over UPDATE/DELETE
• Append-first design
• Mutations rewrite all data

• [ ] Set appropriate max_block_size
• Default: 65536
• Larger for bulk inserts

• [ ] Enabled async_insert for frequent small inserts
• Reduces merge overhead
• async_insert = 1

• [ ] Configured appropriate index_granularity
• Default: 8192
• Smaller = larger index

• [ ] Used AggregatingMergeTree for pre-aggregation
• Materialize aggregations on write
• Fast query performance

Operations Checklist

• [ ] Automated backups (clickhouse-backup)
• Daily backups
• Offsite storage (S3)

• [ ] Monitoring dashboards (Grafana)
• Query performance
• Merge queue
• Replication lag

• [ ] Alert on replication lag > 5s
• Check system.replication_queue
• ZooKeeper health

• [ ] Alert on merge queue > 1000
• Too many parts
• Consider OPTIMIZE FINAL

• [ ] Tested disaster recovery
• Backup restore procedure
• Documented RTO/RPO

• [ ] Documented partition retention policy
• TTL configuration
• DROP PARTITION schedule

• [ ] Configured ZooKeeper session timeout
• Prevent expiration
• Default: 30 seconds

• [ ] Set up query logging
• system.query_log
• TTL for log retention

Cluster Management Checklist

• [ ] Used ReplicatedMergeTree for production
• Data replication
• Automatic failover

• [ ] Configured proper sharding keys
• Even data distribution
• Consider query patterns

• [ ] Set up distributed tables
• Query across cluster
• Transparent to application

• [ ] Monitored cluster health
• All replicas online
• Replication lag minimal

• [ ] Configured load balancing
• Round-robin or random
• load_balancing setting

• [ ] Set up cross-DC replication
• Disaster recovery
• Geographic distribution

• [ ] Documented cluster topology
• Shard/replica mapping
• Network configuration

• [ ] Automated failover testing
• Replica failure
• ZooKeeper failure

Anti-Patterns

Schema Anti-Patterns

Query Anti-Patterns

Operations Anti-Patterns

Code Examples

Good Schema

-- ✅ Good: Matches query patterns
CREATE TABLE events (
    timestamp DateTime,
    user_id UInt32,
    event_type LowCardinality(String),
    revenue Decimal(18, 2) DEFAULT 0
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)  -- Monthly partitions
ORDER BY (user_id, timestamp)     -- Matches queries
SAMPLE BY user_id                  -- Enable sampling
TTL timestamp + INTERVAL 90 DAY;  -- Auto cleanup

Bad Schema

-- ❌ Bad: ORDER BY doesn't match queries
CREATE TABLE events (
    timestamp DateTime,
    user_id UInt32,
    event_type String,
    revenue Nullable(Decimal(18, 2))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (timestamp);  -- Queries filter by user_id first!

Good Query

-- ✅ Good: Uses index, selective columns
SELECT
    user_id,
    count() as events,
    sum(revenue) as total_revenue
FROM events
WHERE timestamp >= today() AND timestamp < tomorrow()
  AND user_id = 123
GROUP BY user_id;

Bad Query

-- ❌ Bad: Function on column, SELECT *
SELECT * FROM events
WHERE toDate(timestamp) = today();

Monitoring Queries

-- Health check
SELECT 'uptime' as metric, toString(uptime()) as value
UNION ALL SELECT 'version', version()
UNION ALL SELECT 'replicas_lagging', toString(count())
FROM system.replication_queue WHERE delay > 5
UNION ALL SELECT 'mutations_running', toString(count())
FROM system.mutations WHERE is_done = 0;

-- Merge queue
SELECT database, table, count() as parts
FROM system.parts
WHERE active = 1
GROUP BY database, table
HAVING parts > 1000;

See Also

• ../SKILL.md - Main skill entry point
• schema-design.md - Database engines and schema organization
• query-optimization.md - Query performance tuning
• monitoring.md - Health checks and monitoring queries

ClickHouse Cluster Management

Distributed tables, replication, sharding, and cluster operations.

Cluster Configuration

Define Cluster

<!-- /etc/clickhouse-server/config.d/remote-servers.xml -->
<clickhouse>
    <remote_servers>
        <my_cluster>
            <shard>
                <replica>
                    <host>node1.example.com</host>
                    <port>9000</port>
                </replica>
            </shard>
            <shard>
                <replica>
                    <host>node2.example.com</host>
                    <port>9000</port>
                </replica>
            </shard>
        </my_cluster>
    </remote_servers>
</clickhouse>

Cluster Macros

<!-- /etc/clickhouse-server/config.d/macros.xml -->
<clickhouse>
    <macros>
        <shard>01</shard>
        <replica>replica_1</replica>
    </macros>
</clickhouse>

Distributed Tables

Create Distributed Table

-- Create local table on each shard
CREATE TABLE local_events ON CLUSTER my_cluster (
    timestamp DateTime,
    user_id UInt32,
    event String
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (user_id, timestamp);

-- Create distributed table (sharding)
CREATE TABLE distributed_events ON CLUSTER my_cluster AS local_events
ENGINE = Distributed(
    'my_cluster',     -- Cluster name
    currentDatabase(), -- Database
    'local_events',   -- Local table name
    rand()            -- Sharding key
);

-- Query distributed table
SELECT * FROM distributed_events;

Sharding Strategies

-- Random sharding
ENGINE = Distributed(cluster, db, table, rand())

-- By user ID (consistent)
ENGINE = Distributed(cluster, db, table, intHash32(user_id))

-- By date
ENGINE = Distributed(cluster, db, table, toYYYYMM(date))

-- Composite sharding
ENGINE = Distributed(cluster, db, table, (tenant_id, user_id))

-- No sharding (all data on all shards)
ENGINE = Distributed(cluster, db, table, const)

Distributed DDL

-- Create table on all shards
CREATE TABLE table_name ON CLUSTER my_cluster (
    id UInt32,
    value String
)
ENGINE = MergeTree()
ORDER BY id;

-- Drop table on all shards
DROP TABLE table_name ON CLUSTER my_cluster;

-- Alter table on all shards
ALTER TABLE table_name ON CLUSTER my_cluster ADD COLUMN new_col String;

Replication

ReplicatedMergeTree Setup

-- On replica 1
CREATE TABLE events ON CLUSTER my_cluster (
    timestamp DateTime,
    user_id UInt32
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{shard}/events',  -- ZooKeeper path
    '{replica}'                           -- Replica name
)
PARTITION BY toYYYYMM(timestamp)
ORDER BY (user_id, timestamp);

-- On replica 2 (same DDL, macros handle replica name)
-- {replica} macro resolves to 'replica_2'

Add Replica to Existing Table

-- Stop replication sends
SYSTEM STOP REPLICATED SENDS database.table;

-- Copy data to new replica
-- (rsync or clickhouse-backup)

-- Start replication
SYSTEM START REPLICATED SENDS database.table;

Replica Management

-- Check replica status
SELECT
    database,
    table,
    is_leader,
    is_readonly,
    queue_size,
    absolute_delay
FROM system.replicas;

-- Check replication queue
SELECT * FROM system.replication_queue
WHERE delay > 5
ORDER BY delay DESC;

Cluster Operations

Cluster Information

-- Show all clusters
SELECT * FROM system.clusters;

-- Cluster nodes
SELECT
    cluster,
    shard_num,
    replica_num,
    host_name,
    port,
    user
FROM system.clusters
WHERE cluster = 'my_cluster';

-- Cluster health
SELECT
    cluster,
    sum(error_count) as errors,
    sum(num_requests) as requests
FROM system.clusters
GROUP BY cluster;

Query Across Cluster

-- Query all shards
SELECT * FROM clusterAllReplicas(
    'my_cluster',
    system.functions
) LIMIT 1;

-- Execute on all nodes
SELECT * FROM remote(
    'node1, node2, node3',
    system.dictionaries
);

-- Query specific shard
SELECT * FROM cluster(
    'my_cluster',
    1,  -- shard
    0,  -- replica
    system.query_log
) LIMIT 1;

Distributed INSERT

-- Insert into distributed table (shards data)
INSERT INTO distributed_events VALUES (now(), 123, 'login');

-- Insert into distributed table with specific shard
INSERT INTO distributed_events SHARD 1 VALUES (now(), 456, 'logout');

Cross-Replication

Multi-Datacenter Setup

-- Multiple datacenters
CREATE TABLE events (
    ...
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{shard}/events',
    '{replica}'
)
-- On DC1: shards 1-2, replicas a-b
-- On DC2: shards 1-2, replicas c-d

Configuration:

<remote_servers>
    <dc_cluster>
        <shard>
            <internal_replication>true</internal_replication>
            <replica>
                <host>dc1-node1</host>
            </replica>
            <replica>
                <host>dc2-node1</host>
            </replica>
        </shard>
    </dc_cluster>
</remote_servers>

Load Balancing

Load Balancing Settings

-- Load balancing modes
SET load_balancing = 'random';           -- Random replica
SET load_balancing = 'nearest_hostname';  -- Hostname-based
SET load_balancing = 'in_order';          -- Sequential

-- Failover settings
set failover = 1;                    -- Enable failover
set failover_by_hostname = 1;       -- Hostname-based failover

-- Defaults for distributed table
SETTINGS
    load_balancing = 'random',
    weights_by_node = 'dc1=1,dc2=1';  -- Weight distribution

Cluster Monitoring

Cluster Health

-- Comprehensive cluster health
SELECT
    host_address() as host,
    'uptime' as metric,
    toString(uptime()) as value
UNION ALL
SELECT host_address(), 'version', version()
UNION ALL
SELECT host_address(), 'replicas_lagging', toString(count())
FROM system.replication_queue WHERE delay > 5
UNION ALL
SELECT host_address(), 'mutations_running', toString(count())
FROM system.mutations WHERE is_done = 0;

-- All cluster nodes status
SELECT * FROM cluster('my_cluster', all, 1, system.replicas);

Replication Lag Monitoring

-- Replication lag across cluster
SELECT
    host_address() as host,
    database,
    table,
    max(delay) as max_lag
FROM system.replication_queue
WHERE delay > 0
GROUP BY host, database, table;

Cluster Maintenance

Add Shard

-- 1. Install ClickHouse on new node
-- 2. Add to remote_servers config
-- 3. Create local tables on new shard
-- 4. Restart cluster
-- 5. Redistribute data (manual or using clickhouse-copier)

Remove Shard

-- 1. Drop data from shard
-- 2. Remove from remote_servers config
-- 3. Restart cluster
-- 4. Update distributed tables

Data Migration

-- Using clickhouse-copier for data migration
-- Config file:
<source>
    <host>old-cluster</host>
    <port>9000</port>
</source>
<destination>
    <host>new-cluster</host>
    <port>9000</port>
</destination>

<tables>
    <table>
        <source_database>db</source_database>
        <source_table>table</source_table>
        <destination_database>db</destination_database>
        <destination_table>table</destination_table>
    </table>
</tables>

Best Practices

1. Always use ReplicatedMergeTree for production 2. Configure proper sharding keys for even distribution 3. Monitor replication lag (alert if > 5s) 4. Use `GLOBAL JOIN` for distributed queries with dimension tables 5. Set up ZooKeeper monitoring (ZK outage = read-only) 6. Test failover regularly 7. Document cluster topology (shard/replica mapping)

See Also

• ../SKILL.md - Main skill entry point
• core-concepts.md - Architecture and data model
• debugging.md - Replication troubleshooting
• monitoring.md - Cluster health checks

ClickHouse Core Concepts

Architecture Overview

Columnar Storage

ClickHouse stores data by columns, not rows:

Benefits:

• Read only needed columns (10-100x faster for analytical queries)
• Excellent compression (similar data types stored together)
• Efficient for wide tables (100+ columns) with selective reads

Trade-offs:

• Slower single-row reads (must read all columns)
• Not optimal for point queries or OLTP

Two-Level Index Structure

┌─────────────────────────────────────┐
│  Sparse Index (one mark per 8192 rows) │
├─────────────────────────────────────┤
│  Mark Files (point to data blocks)   │
├─────────────────────────────────────┤
│  Compressed Column Data              │
└─────────────────────────────────────┘

• Sparse index: One mark per 8192 rows (configurable via index_granularity)
• Mark files: Point to compressed data blocks
• Query execution: Uses marks to skip irrelevant data

-- Check index granularity setting
SELECT index_granularity FROM system.tables
WHERE database = currentDatabase() AND table = 'my_table';

-- Adjust index granularity (affects mark count)
CREATE TABLE ... SETTINGS index_granularity = 4096;

Merge Process

Background merges continuously organize data:

Merge Lifecycle: 1. New data inserted as small parts 2. Background merges combine parts (exponential backoff) 3. Parts grow from MB → GB → TB 4. Merges can be CPU/disk intensive

Monitoring Merges:

-- Active merges
SELECT table, elapsed, bytes_read_uncompressed, rows_read
FROM system.merges
ORDER BY elapsed DESC;

-- Merge queue size
SELECT database, table, count() as parts
FROM system.parts
WHERE active = 1 AND rows > 0
GROUP BY database, table
HAVING parts > 1000;

Distributed Query Execution

┌─────────────┐
│ Coordinator │
└──────┬──────┘
       │
       ├── Shard 1 ── Query Local Data
       ├── Shard 2 ── Query Local Data
       └── Shard 3 ── Query Local Data
       │
       └── Merge Results

• Coordinator node shards query across replicas
• Each shard processes local data in parallel
• Coordinator merges and returns results

Data Model

Append-First Design

ClickHouse is optimized for appends, not in-place updates:

Characteristics:

• No in-place updates (mutations are expensive rewrite operations)
• Data organized as immutable parts
• Parts merge over time (background process)

Parts vs Partitions:

• Partition: Logical division (e.g., monthly 202401, 202402)
• Part: Physical data file on disk
• Typical: 100-1000 parts per partition
• Too many parts = slow queries, high merge overhead

MVCC and Versioning

ClickHouse offers several patterns for versioned data:

ReplacingMergeTree: Keeps latest version per ORDER BY key

CREATE TABLE user_profiles (
    user_id UInt32,
    updated_at DateTime,
    profile String
)
ENGINE = ReplacingMergeTree(updated_at)
ORDER BY user_id;

CollapsingMergeTree: Uses sign column for insert/delete

CREATE TABLE changes (
    id UInt32,
    sign Int8,  -- 1 = insert, -1 = delete
    data String
)
ENGINE = CollapsingMergeTree(sign)
ORDER BY id;

ZooKeeper Coordination

ReplicatedMergeTree uses ZooKeeper for coordination:

ZooKeeper stores:

• Schema metadata
• Part metadata
• Replication status
• Merge queue

ZooKeeper considerations:

• ZK outage = read-only cluster (can't insert)
• High ZK load can impact performance
• Monitor ZK connection health

-- Check ZK connection
SELECT * FROM system.zookeeper WHERE path = '/';

-- Check replication queue
SELECT * FROM system.replication_queue;

When ClickHouse Shines

Ideal Use Cases

✅ Wide tables (100+ columns), read few columns ✅ Time-series with time-based filters ✅ Aggregations over billions of rows ✅ Append-only workloads (events, logs, metrics) ✅ Real-time analytics (sub-second responses) ✅ Histograms and quantiles over large datasets

Example Workloads

-- Events/analytics (ideal)
SELECT user_id, count() as events
FROM events
WHERE timestamp >= today() - INTERVAL 7 DAY
GROUP BY user_id;

-- Time-series (ideal)
SELECT toDate(timestamp) as date, sum(revenue)
FROM events
WHERE timestamp >= now() - INTERVAL 30 DAY
GROUP BY date
ORDER BY date;

-- Aggregations (ideal)
SELECT percentile(duration, [50, 90, 99]) as p50_p90_p99
FROM requests
WHERE timestamp >= today();

When to Avoid ClickHouse

Poor Use Cases

❌ Point updates/deletes: Use row store (PostgreSQL, MySQL) ❌ Heavy JOINs on non-sorted keys: Consider data warehouse ❌ Complex transactions: No ACID support ❌ Low-latency OLTP: Use row store ❌ Unstructured data: Use document store (MongoDB) ❌ Single-row lookups: Use key-value store (Redis)

Anti-Patterns

-- ❌ Point updates (expensive mutation)
UPDATE events SET status = 'done' WHERE event_id = 123;

-- ✅ Use TTL or new table instead
ALTER TABLE events MODIFY TTL timestamp + INTERVAL 90 DAY;
-- OR
INSERT INTO events_done SELECT * FROM events WHERE status = 'done';

-- ❌ Single-row lookup
SELECT * FROM events WHERE event_id = 123;

-- ✅ Use row store or Redis for point queries

Performance Characteristics

Query Performance

Data Types Performance

See Also

• ../SKILL.md - Main skill entry point
• schema-design.md - Database engines and schema organization
• table-design.md - ORDER BY and partitioning strategies
• query-optimization.md - Query performance tuning

ClickHouse Debugging

Query debugging, merge issues, mutations, replication problems, and troubleshooting.

Query Debugging

Enable Query Log

-- Check if enabled
SELECT * FROM system.settings
WHERE name = 'log_queries';

-- Enable query logging
SYSTEM STOP DISTRIBUTED SENDS query_log;
ALTER TABLE system.query_log MODIFY TTL query_start_time + INTERVAL 30 DAY;
SYSTEM START DISTRIBUTED SENDS query_log;

Analyze Slow Queries

-- Find slow queries
SELECT
    query,
    query_duration_ms / 1000 as seconds,
    memory_usage,
    read_rows,
    read_bytes
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query_duration_ms > 1000
ORDER BY query_duration_ms DESC
LIMIT 10;

-- Find expensive queries by memory
SELECT
    query,
    formatReadableSize(memory_usage) as memory,
    query_duration_ms
FROM system.query_log
WHERE type = 'QueryFinish'
ORDER BY memory_usage DESC
LIMIT 10;

Check Query Plan

EXPLAIN SELECT * FROM events WHERE user_id = 123;

-- Readable query plan
EXPLAIN PLAN SELECT * FROM events WHERE user_id = 123;

-- Pipeline details
EXPLAIN PIPELINE SELECT * FROM events WHERE user_id = 123;

-- With estimates
EXPLAIN ESTIMATE SELECT * FROM events WHERE user_id = 123;

Key indicators:

• "Filter" → Not using index (full scan)
• "Index" → Using index marks
• "Projection" → Pre-computed data
• "Filter with index" → Using skip index

Debug Hanging Queries

-- Show running queries
SELECT
    query_id,
    user,
    query,
    elapsed,
    memory_usage,
    read_rows
FROM system.processes
WHERE elapsed > 10
ORDER BY elapsed DESC;

-- Kill query
KILL QUERY WHERE query_id = 'query-id';

-- Cancel mutation
KILL MUTATION WHERE mutation_id = 'mutation-id';

Merge Debugging

Active Merges

-- Active merges
SELECT
    database,
    table,
    elapsed,
    progress,
    bytes_read_uncompressed,
    rows_read,
    is_mutation
FROM system.merges
ORDER BY elapsed DESC;

-- Merge performance by day
SELECT
    table,
    count() as merge_count,
    avg(bytes_read_uncompressed) as avg_size,
    sum(rows_read) as total_rows
FROM system.merges
WHERE event_date = today()
GROUP BY table
ORDER BY merge_count DESC;

Merge Queue Size

-- Tables with too many parts (need optimization)
SELECT
    database,
    table,
    count() as parts,
    sum(rows) as total_rows
FROM system.parts
WHERE active = 1
GROUP BY database, table
HAVING parts > 1000
ORDER BY parts DESC;

Common Merge Issues

Mutation Debugging

Show Mutations

-- All mutations
SELECT * FROM system.mutations;

-- Active mutations
SELECT
    database,
    table,
    command,
    is_done,
    parts_to_do,
    parts_to_do_names
FROM system.mutations
WHERE is_done = 0;

Mutation Progress

-- Mutation progress details
SELECT
    table,
    command,
    is_done,
    parts_to_do - parts_to_do_names as parts_remaining,
    elapsed,
    formatReadableSize(bytes_read_uncompressed) as bytes_processed
FROM system.mutations
WHERE is_done = 0
ORDER BY parts_to_do DESC;

Common Mutation Issues

Replication Debugging

Replication Status

-- All replicas
SELECT
    database,
    table,
    is_leader,
    is_readonly,
    queue_size,
    absolute_delay
FROM system.replicas
ORDER BY absolute_delay DESC;

-- Replication queue
SELECT * FROM system.replication_queue
WHERE delay > 5
ORDER BY delay DESC;

ZooKeeper Health

-- ZooKeeper connection
SELECT * FROM system.zookeeper WHERE path = '/';

-- Check ZK queue
SELECT count() FROM system.zookeeper
WHERE path = '/clickhouse/queues';

-- ZK exceptions
SELECT * FROM system.zookeeper_log
WHERE type = 'ERROR'
ORDER BY event_time DESC
LIMIT 10;

Common Replication Issues

Data Issues

Check Parts Count

-- Parts per partition
SELECT
    partition,
    count() as parts,
    sum(rows) as total_rows
FROM system.parts
WHERE active = 1
  AND table = 'my_table'
GROUP BY partition
HAVING parts > 1000
ORDER BY parts DESC;

Check for Duplicates

-- Check for duplicates in ReplacingMergeTree
SELECT
    user_id,
    count() as cnt
FROM ReplacingMergeTree_table
GROUP BY user_id
HAVING cnt > 1;

Data Distribution

-- Data skew
SELECT
    user_id,
    count() as cnt
FROM events
GROUP BY user_id
ORDER BY cnt DESC
LIMIT 10;

Common Issues & Solutions

Issue: Too Many Parts

Symptoms:

• Slow queries
• High memory usage
• Large merge queue

Solutions:

-- Force merge
OPTIMIZE TABLE table FINAL;

-- Check partitioning
SELECT partition, count() as parts
FROM system.parts
WHERE active = 1 AND table = 'my_table'
GROUP BY partition;

-- Consider larger partitions
-- Monthly instead of daily

Issue: Mutation Stuck

Symptoms:

• ALTER UPDATE/DELETE not completing
• is_done = 0 in system.mutations

Solutions:

-- Check mutation progress
SELECT * FROM system.mutations WHERE is_done = 0;

-- Cancel and retry
KILL MUTATION WHERE mutation_id = '...';

-- Force with OPTIMIZE
OPTIMIZE TABLE table FINAL;

Issue: OOM on Queries

Symptoms:

• Query killed with memory limit
• "Memory limit exceeded" error

Solutions:

-- Increase memory limit
SET max_memory_usage = 10000000000;

-- Optimize query
-- - Select fewer columns
-- - Add filters
-- - Use SAMPLE

-- Check query memory usage
SELECT
    query,
    formatReadableSize(memory_usage) as memory
FROM system.query_log
WHERE type = 'QueryFinish'
ORDER BY memory_usage DESC
LIMIT 10;

Issue: Replication Lag

Symptoms:

• Data not replicating
• High absolute_delay in system.replicas

Solutions:

-- Check replication queue
SELECT * FROM system.replication_queue;

-- Check ZooKeeper
SELECT * FROM system.zookeeper WHERE path = '/clickhouse/tables';

-- Check network/disk
-- Verify connectivity between replicas
-- Check disk I/O performance

Debugging Workflow

1. Identify the problem

• Query slow? Check system.processes
• Data missing? Check replication status
• High memory? Check system.query_log

2. Gather information

• Use EXPLAIN for query plans
• Check system.merges for merge issues
• Check system.mutations for mutation progress

3. Apply fixes

• OPTIMIZE TABLE for merge issues
• Adjust settings for performance
• Fix network/ZooKeeper for replication

4. Verify

• Re-run query
• Check metrics again
• Monitor for recurrence

See Also

• ../SKILL.md - Main skill entry point
• monitoring.md - Health checks and monitoring queries
• cluster-management.md - Replication setup and configuration

External References

Altinity Knowledge Base (200+ articles) and official ClickHouse documentation.

Altinity Knowledge Base

Schema Design

Query Optimization

Table Engines

Operations

Advanced Features

Integrations

Troubleshooting

Official ClickHouse Documentation

Getting Started

SQL Reference

Table Engines

System Tables

Operations

Cluster Management

Backup & Restore

Monitoring

Integration

Kubernetes

Community Resources

Training & Certification

See Also

• ../SKILL.md - Main skill entry point
• best-practices.md - Production-ready checklist
• All reference files - Topic-specific deep dives

ClickHouse Integrations

Kafka, S3, PostgreSQL, MySQL, RabbitMQ, and BI tools.

⚠️ Credential Security: All credentials in examples below (password123, AKIAIOSFODNN7EXAMPLE, etc.) are placeholders only. Never use these in production. Use proper secret management:

- Environment variables

- Secret managers (AWS Secrets Manager, HashiCorp Vault, etc.)

- Kubernetes secrets (for K8s deployments)

- ClickHouse named collections with external configuration

Kafka Integration

Kafka Table Engine

CREATE TABLE kafka_queue (
    timestamp UInt64,
    level String,
    message String
)
ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'broker1:9092,broker2:9092',
    kafka_topic_list = 'logs',
    kafka_group_name = 'clickhouse_consumer',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers = 2,
    kafka_max_block_size = 65536,
    kafka_skip_broken_messages = 100;

Virtual Columns

Kafka tables provide virtual columns:

-- Virtual columns available
SELECT
    _topic,      -- Kafka topic
    _key,        -- Message key
    _offset,     -- Message offset
    _partition,  -- Partition number
    _timestamp   -- Message timestamp
FROM kafka_queue;

Materialized View for Streaming

-- Create target table
CREATE TABLE logs (
    timestamp DateTime,
    level String,
    message String
)
ENGINE = MergeTree()
ORDER BY (timestamp, level);

-- Create materialized view to consume continuously
CREATE MATERIALIZED VIEW consumer TO logs
AS SELECT
    toDateTime(timestamp) as timestamp,
    level,
    message
FROM kafka_queue;

-- Data automatically flows from Kafka → MV → target table

Kafka Settings

Consumer Lag Monitoring

SELECT
    topic,
    partition,
    max_offset,
    lag
FROM system.kafka_consumers;

S3 Integration

S3 Table Engine

CREATE TABLE s3_table (
    id UInt32,
    data String,
    timestamp DateTime
)
ENGINE = S3(
    'https://my-bucket.s3.amazonaws.com/data/*.parquet',
    'AWS_ACCESS_KEY',
    'AWS_SECRET_KEY',
    'Parquet'
);

S3 Table Function

-- Query S3 directly
SELECT * FROM s3(
    'https://bucket.s3.amazonaws.com/data/*.csv',
    'access_key',
    'secret_key',
    'CSV'
);

-- With wildcards
SELECT * FROM s3(
    'https://bucket.s3.amazonaws.com/data/file-{000..999}.csv',
    'CSV'
);

-- With compression
SELECT * FROM s3(
    'https://bucket.s3.amazonaws.com/data/*.csv.gz',
    'CSV',
    'access_key',
    'secret_key'
);

S3 Disk (Tiered Storage)

-- Configuration in config.xml
<storage_configuration>
    <disks>
        <hot_ssd>
            <path>/mnt/ssd/clickhouse/</path>
        </hot_ssd>
        <s3_cold>
            <type>s3</type>
            <endpoint>https://bucket.s3.amazonaws.com/clickhouse/</endpoint>
            <!-- ⚠️ Use AWS IAM roles or environment variables in production -->
            <access_key_id>AKIAIOSFODNN7EXAMPLE</access_key_id>
            <secret_access_key>wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY</secret_access_key>
        </s3_cold>
    </disks>

    <policies>
        <hot_cold>
            <volumes>
                <hot>
                    <disk>hot_ssd</disk>
                </hot>
                <cold>
                    <disk>s3_cold</disk>
                </cold>
            </volumes>
        </hot_cold>
    </policies>
</storage_configuration>

-- Usage with TTL
CREATE TABLE events (
    timestamp DateTime,
    data String
)
ENGINE = MergeTree()
ORDER BY timestamp
SETTINGS storage_policy = 'hot_cold'
TTL timestamp + INTERVAL 7 DAY TO DISK 's3_cold';

Production S3 Configuration: Use IAM roles or externalized secrets:

<s3_cold>
    <type>s3</type>
    <endpoint>https://bucket.s3.amazonaws.com/clickhouse/</endpoint>
    <!-- Use environment variable: ${S3_ACCESS_KEY} -->
    <access_key_id>${S3_ACCESS_KEY}</access_key_id>
    <secret_access_key>${S3_SECRET_KEY}</secret_access_key>
</s3_cold>

S3 Caching

-- Enable filesystem cache
SELECT * FROM s3(...)
SETTINGS
    filesystem_cache_name = 's3_cache',
    enable_filesystem_cache = 1;

PostgreSQL Integration

PostgreSQL Table Engine

CREATE TABLE pg_table (
    id Int32,
    name String,
    created_at DateTime
)
ENGINE = PostgreSQL(
    'localhost:5432',
    'mydb',
    'users',
    'postgres',
    'password'
);

PostgreSQL Table Function

-- Query PostgreSQL directly
SELECT * FROM postgresql(
    'localhost:5432',
    'mydb',
    'users',
    'postgres',
    'password'
)
WHERE active = true;

PostgreSQL Dictionary

CREATE DICTIONARY pg_users (
    user_id UInt32,
    email String,
    name String,
    created_at DateTime
)
PRIMARY KEY user_id
SOURCE(POSTGRESQL(
    port 5432
    host 'localhost'
    db 'mydb'
    table 'users'
    user 'postgres'
    password 'password'
))
LIFETIME(60)
LAYOUT(HASHED());

Named Collections

<named_collections>
    <pg_connection>
        <host>localhost</host>
        <port>5432</port>
        <user>postgres</user>
        <password>secret</password>
    </pg_connection>
</named_collections>

-- Use named collection
SELECT * FROM postgresql(
    'pg_connection',
    'mydb',
    'users'
);

MySQL Integration

MySQL Table Engine

CREATE TABLE mysql_table (
    id Int32,
    name String
)
ENGINE = MySQL(
    'localhost:3306',
    'mydb',
    'users',
    'mysql_user',
    'mysql_pass'
);

MySQL Table Function

SELECT * FROM mysql(
    'localhost:3306',
    'mydb',
    'users',
    'mysql_user',
    'mysql_pass'
);

MySQL Dictionary

CREATE DICTIONARY mysql_users (
    user_id UInt32,
    email String
)
PRIMARY KEY user_id
SOURCE(MYSQL(
    port 3306
    host 'localhost'
    db 'mydb'
    table 'users'
    user 'mysql_user'
    password 'mysql_pass'
))
LIFETIME(60)
LAYOUT(HASHED());

RabbitMQ Integration

CREATE TABLE rabbitmq_queue (
    timestamp DateTime,
    data String
)
ENGINE = RabbitMQ()
SETTINGS
    amqp_exchange = 'events',
    amqp_exchange_type = 'fanout',
    amqp_routing_key_list = 'events',
    amqp_queue_base = 'clickhouse_consumer',
    amqp_format = 'JSONEachRow',
    amqp_host = 'localhost',
    amqp_port = 5672,
    amqp_user = 'guest',
    amqp_password = 'guest';

MongoDB Integration

SELECT * FROM mongodb(
    'localhost:27017',
    'mydb',
    'mycollection',
    'user',
    'password'
);

Redis Integration

Redis Dictionary

CREATE DICTIONARY redis_dict (
    key String,
    value String
)
PRIMARY KEY key
SOURCE(REDIS(
    host 'localhost'
    port 6379
    db 0
    password 'password'
))
LIFETIME(60)
LAYOUT(HASHED());

Redis Table Function

SELECT * FROM redis(
    'localhost',
    6379,
    'mykey'
);

BI Tools Integration

Tableau

# Install ODBC driver
# Download from: https://github.com/ClickHouse/clickhouse-odbc

# Configure DSN in /etc/odbcinst.ini
[ClickHouse]
Description = ClickHouse ODBC Driver
Driver = /usr/lib/libclickhouseodbc.so
Setup = /usr/lib/libclickhouseodbc_s.so

Grafana

# Install ClickHouse data source plugin
grafana-cli plugins install clickhouse-datasource

# Configure in Grafana UI:
# - Host: http://clickhouse-server:8123
# - Database: default
# - User: default
# - Password: (empty)

Metabase

-- Native ClickHouse support
-- Add connection in Metabase UI:
-- - Database type: ClickHouse
# - Host: clickhouse-server
# - Port: 8123
# - Database: mydb
# - Username: default

Superset

# Install ClickHouse SQLAlchemy dialect
pip install clickhouse-sqlalchemy

# Configure in Superset:
# Connection string:
# clickhouse+native://default:@clickhouse-server:9000/mydb

Looker

# Looker connection configuration
connection:
  dialect: clickhouse
  host: clickhouse-server
  port: 8123
  database: analytics
  username: default
  password: ""
  ssl: false

Data Import Tools

clickhouse-import

# Import from CSV
clickhouse-import --query="INSERT INTO table FORMAT CSV" < data.csv

# Import from JSON
clickhouse-import --query="INSERT INTO table FORMAT JSONEachRow" < data.json

clickhouse-copier

# Copy data between clusters
clickhouse-copier \
  --config=config.xml \
  --base-dir=/tmp/ \
  --src-cluster=production \
  --dst-cluster=staging \
  --tables=events

See Also

• ../SKILL.md - Main skill entry point
• cluster-management.md - Distributed queries and sharding
• monitoring.md - Data pipeline monitoring

ClickHouse Monitoring

Query monitoring, health checks, system queries, and metrics.

Current Query Monitoring

Running Queries

-- All running queries
SELECT
    query_id,
    user,
    query,
    elapsed,
    formatReadableSize(memory_usage) as memory,
    formatReadableQuantity(read_rows) as rows_read,
    formatReadableSize(read_bytes) as bytes_read
FROM system.processes
ORDER BY elapsed DESC;

-- Long-running queries (> 1 minute)
SELECT
    query_id,
    user,
    query,
    elapsed / 60 as minutes_elapsed,
    formatReadableSize(memory_usage) as memory
FROM system.processes
WHERE elapsed > 60
ORDER BY elapsed DESC;

-- Queries by user
SELECT
    user,
    count() as query_count,
    sum(memory_usage) as total_memory
FROM system.processes
GROUP BY user
ORDER BY total_memory DESC;

Kill Query

-- Kill specific query
KILL QUERY WHERE query_id = 'query-id';

-- Kill all queries from user
KILL QUERY WHERE user = 'username';

-- Kill long-running queries
KILL QUERY WHERE elapsed > 3600;

Query History

Recent Queries

-- Recent queries (last hour)
SELECT
    type,
    substring(query, 1, 100) as query_preview,
    query_duration_ms / 1000 as seconds,
    formatReadableSize(memory_usage) as memory,
    event_time
FROM system.query_log
WHERE event_time > now() - INTERVAL 1 HOUR
ORDER BY event_time DESC
LIMIT 100;

Slow Queries

-- Slow queries (> 5 seconds)
SELECT
    query,
    query_duration_ms / 1000 as seconds,
    formatReadableSize(memory_usage) as memory,
    formatReadableQuantity(read_rows) as rows_read,
    event_time
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query_duration_ms > 5000
ORDER BY query_duration_ms DESC
LIMIT 20;

Failed Queries

-- Failed queries with errors
SELECT
    query,
    exception_code,
    exception_text,
    event_time
FROM system.query_log
WHERE type = 'Exception'
  AND event_time > now() - INTERVAL 1 DAY
ORDER BY event_time DESC
LIMIT 50;

Table Usage

Most Accessed Tables

-- Most queried tables
SELECT
    database,
    table,
    count() as query_count,
    sum(read_rows) as total_rows,
    avg(query_duration_ms) as avg_duration_ms
FROM system.query_log
WHERE type = 'QueryFinish'
  AND event_date = today()
GROUP BY database, table
ORDER BY query_count DESC
LIMIT 20;

Table Sizes

-- Tables by size
SELECT
    database,
    table,
    formatReadableSize(sum(bytes)) as size,
    sum(rows) as total_rows,
    count() as parts
FROM system.parts
WHERE active = 1
GROUP BY database, table
ORDER BY sum(bytes) DESC
LIMIT 20;

Column Compression

-- Column compression ratios
SELECT
    database,
    table,
    column,
    type,
    formatReadableSize(sum(data_uncompressed_bytes)) as uncompressed,
    formatReadableSize(sum(data_compressed_bytes)) as compressed,
    sum(data_compressed_bytes) / sum(data_uncompressed_bytes) as ratio
FROM system.columns
WHERE database != 'system'
GROUP BY database, table, column, type
ORDER BY sum(data_uncompressed_bytes) DESC
LIMIT 50;

Merge Monitoring

Active Merges

-- Current merges
SELECT
    database,
    table,
    elapsed,
    progress,
    formatReadableSize(bytes_read_uncompressed) as bytes_read,
    rows_read,
    is_mutation,
    merge_type
FROM system.merges
ORDER BY elapsed DESC;

Merge Performance

-- Merge performance by day
SELECT
    table,
    count() as merge_count,
    avg(bytes_read_uncompressed) as avg_size,
    sum(rows_read) as total_rows
FROM system.merges
WHERE event_date = today()
GROUP BY table
ORDER BY merge_count DESC;

Merge Queue

-- Tables needing optimization
SELECT
    database,
    table,
    count() as parts,
    sum(rows) as total_rows
FROM system.parts
WHERE active = 1
GROUP BY database, table
HAVING parts > 1000
ORDER BY parts DESC;

Mutation Monitoring

-- Active mutations
SELECT
    database,
    table,
    command,
    is_done,
    parts_to_do,
    parts_to_do_names,
    formatReadableSize(bytes_read_uncompressed) as bytes_processed
FROM system.mutations
WHERE is_done = 0;

-- Mutation history
SELECT
    table,
    command,
    is_done,
    created_at,
    finished_at
FROM system.mutations
ORDER BY created_at DESC
LIMIT 50;

Cluster Health

Basic Health

-- Quick health check
SELECT
    'uptime' as metric,
    toString(uptime()) as value
UNION ALL
SELECT 'version', version()
UNION ALL
SELECT 'running_queries', toString(count())
FROM system.processes
UNION ALL
SELECT 'active_merges', toString(count())
FROM system.merges
UNION ALL
SELECT 'mutations_running', toString(count())
FROM system.mutations WHERE is_done = 0;

Cluster Status

-- Cluster nodes
SELECT
    cluster,
    shard_num,
    replica_num,
    host_name,
    port,
    user
FROM system.clusters
WHERE cluster = 'my_cluster';

Replica Status

-- All replicas
SELECT
    database,
    table,
    is_leader,
    is_readonly,
    queue_size,
    absolute_delay
FROM system.replicas
ORDER BY absolute_delay DESC;

Replication Lag

-- Lagging replicas
SELECT
    database,
    table,
    replica_name,
    is_leader,
    is_readonly,
    queue_size,
    delay * 1000 as lag_ms
FROM system.replication_queue
WHERE delay > 5
ORDER BY delay DESC;

Disk and Memory Health

Disk Usage

-- All disks
SELECT
    name,
    path,
    formatReadableSize(free_space) as free,
    formatReadableSize(total_space) as total,
    formatReadableSize(keep_free_space) as keep_free,
    (free_space / total_space) * 100 as percent_free
FROM system.disks;

Memory Usage

-- Memory by dictionaries
SELECT
    formatReadableSize(sum(bytes_allocated)) as allocated,
    formatReadableSize(sum(bytes_used)) as used
FROM system.dictionaries;

-- Current memory metrics
SELECT
    formatReadableSize(os_userspace_memory) as userspace,
    formatReadableSize(os_committed_memory) as committed
FROM system.asynchronous_metrics
WHERE metric LIKE '%memory%';

Performance Metrics

Query Statistics

-- Query statistics by type
SELECT
    type,
    count() as count,
    avg(query_duration_ms) as avg_duration_ms
FROM system.query_log
WHERE event_date = today()
GROUP BY type
ORDER BY count DESC;

Top Consumers

-- Most memory-intensive queries
SELECT
    query,
    formatReadableSize(memory_usage) as memory,
    query_duration_ms / 1000 as seconds
FROM system.query_log
WHERE type = 'QueryFinish'
  AND event_date = today()
ORDER BY memory_usage DESC
LIMIT 10;

-- Most row-intensive queries
SELECT
    query,
    formatReadableQuantity(read_rows) as rows_read,
    query_duration_ms / 1000 as seconds
FROM system.query_log
WHERE type = 'QueryFinish'
  AND event_date = today()
ORDER BY read_rows DESC
LIMIT 10;

Alerting Queries

Critical Alerts

-- Replicas lagging > 30s
SELECT
    'CRITICAL: Replication lag' as alert,
    database,
    table,
    delay as lag_seconds
FROM system.replication_queue
WHERE delay > 30;

-- Too many parts
SELECT
    'WARNING: Too many parts' as alert,
    database,
    table,
    count() as parts
FROM system.parts
WHERE active = 1
GROUP BY database, table
HAVING parts > 5000;

-- Long-running queries
SELECT
    'WARNING: Long-running query' as alert,
    query_id,
    elapsed / 60 as minutes,
    substring(query, 1, 100) as query_preview
FROM system.processes
WHERE elapsed > 1800;  -- 30 minutes

Asynchronous Metrics

-- System metrics
SELECT
    metric,
    formatReadableSize(value) as value
FROM system.asynchronous_metrics
WHERE metric LIKE '%bytes%'
ORDER BY metric;

-- Load average
SELECT
    metric,
    value
FROM system.asynchronous_metrics
WHERE metric LIKE '%load%';

Prometheus Integration

Export Metrics

-- ClickHouse exposes metrics on port 9363
# Enable in config.xml
<prometheus>
    <endpoint>/metrics</endpoint>
    <port>9363</port>
</prometheus>

# Scrape config
scrape_configs:
  - job_name: 'clickhouse'
    static_configs:
      - targets: ['localhost:9363']

Key Metrics

# Query metrics
clickhouse_queries_total
clickhouse_query_duration_seconds
clickhouse_query_memory_usage_bytes

# Merge metrics
clickhouse_merges_total
clickhouse_merge_duration_seconds

# Replication metrics
clickhouse_replication_queue_size
clickhouse_replication_lag_seconds

# Table metrics
clickhouse_table_rows{table="events"}
clickhouse_table_size_bytes{table="events"}

Grafana Dashboard

Recommended Panels

1. Query Performance

• Queries per second
• Average query duration
• Memory usage

2. Merge Health

• Active merges
• Merge queue size
• Parts count

3. Replication

• Replication lag
• Queue size
• Replica status

4. Table Sizes

• Top 10 tables by size
• Growth rate

5. System Resources

• CPU usage
• Memory usage
• Disk usage

See Also

• ../SKILL.md - Main skill entry point
• debugging.md - Troubleshooting issues
• system-queries.md - Ready-to-use monitoring queries

ClickHouse Query Optimization

Techniques for optimizing ClickHouse query performance.

EXPLAIN

Use EXPLAIN to understand query execution:

-- Basic explain
EXPLAIN SELECT * FROM events WHERE user_id = 123;

-- Readable query plan
EXPLAIN PLAN SELECT * FROM events WHERE user_id = 123;

-- Pipeline details
EXPLAIN PIPELINE SELECT * FROM events WHERE user_id = 123;

-- With estimates
EXPLAIN ESTIMATE SELECT * FROM events WHERE user_id = 123;

Key indicators:

• "Filter" → Not using index (full scan)
• "Index" → Using index marks
• "Projection" → Pre-computed data
• "Filter with index" → Using skip index

JOIN Optimization

Rule: Smaller Table on RIGHT

-- ✅ Good: Smaller table on RIGHT
SELECT *
FROM large_table lt
RIGHT JOIN small_table st ON lt.id = st.id;

-- ❌ Bad: Large table on RIGHT
SELECT *
FROM small_table st
LEFT JOIN large_table lt ON lt.id = st.id;

Why? ClickHouse sends RIGHT table to all shards. Smaller RIGHT = less network transfer.

GLOBAL JOIN for Distributed Queries

-- For distributed queries, use GLOBAL
SELECT *
FROM large_table lt
GLOBAL RIGHT JOIN small_table st ON lt.id = st.id;

-- GLOBAL ensures right table is sent once to each shard

ASOF JOIN for Time-Series

-- As-of join for time-series (latest value before timestamp)
SELECT *
FROM ticks t
ASOF LEFT JOIN quotes q
ON t.symbol = q.symbol
AND t.time <= q.time;

GROUP BY Optimization

WITH ROLLUP

-- Hierarchical aggregation
SELECT
    user_id,
    event_type,
    count() as cnt
FROM events
GROUP BY user_id, event_type WITH ROLLUP;
-- Results: (user, type), (user, NULL), (NULL, NULL)

WITH CUBE

-- All combinations
SELECT
    user_id,
    event_type,
    count() as cnt
FROM events
GROUP BY user_id, event_type WITH CUBE;
-- Results: All 4 combinations

Projections (Pre-computed Aggregations)

Projections automatically speed up common aggregations:

-- Create projection
ALTER TABLE events ADD PROJECTION pr_user_daily (
    SELECT
        user_id,
        toDate(timestamp) as date,
        count() as events,
        sum(revenue) as total_revenue
    GROUP BY user_id, date
);

-- Query automatically uses projection (fast!)
SELECT
    user_id,
    date,
    count() as events,
    sum(revenue) as total_revenue
FROM events
GROUP BY user_id, date;

-- Drop projection
ALTER TABLE events DROP PROJECTION pr_user_daily;

When projections are used:

• Query GROUP BY matches projection GROUP BY
• Query SELECT matches projection SELECT
• Automatic (no query changes needed)

Skip Indexes

Skip indexes allow data skipping during reads:

-- Bloom filter for exact match
CREATE TABLE events (
    user_id UInt32,
    email String,
    timestamp DateTime
)
ENGINE = MergeTree()
ORDER BY (user_id, timestamp)
INDEX idx_email email TYPE bloom_filter GRANULARITY 1;

-- Minmax for range queries
INDEX idx_time timestamp TYPE minmax GRANULARITY 4;

-- Set for IN queries
INDEX idx_user user_id TYPE set(1000) GRANULARITY 1;

-- Tokenbf for string search
INDEX idx_msg message TYPE tokenbf_v1(512, 3, 0) GRANULARITY 1;

Skip index types:

• minmax: Min/max values per granule
• set: Set of values (good for IN queries)
• bloom_filter: Probabilistic exact match
• tokenbf_v1: Token-based bloom filter for text search

Check if index used:

EXPLAIN SELECT * FROM events WHERE email = 'user@example.com';
-- Look for "Index" or "Filter with index" in output

Common Query Pitfalls

-- ❌ SELECT * reads all columns (expensive in columnar store)
SELECT * FROM events;

-- ✅ Select only needed columns
SELECT user_id, timestamp, event_type FROM events;

-- ❌ Suboptimal WHERE (function on column)
SELECT * FROM events WHERE toDate(timestamp) = today();

-- ✅ Use date functions on literal
SELECT * FROM events
WHERE timestamp >= today() AND timestamp < tomorrow();

-- ❌ String comparison for dates
SELECT * FROM events WHERE toString(timestamp) LIKE '2024-01-01%';

-- ✅ Use date range
SELECT * FROM events
WHERE timestamp >= toDateTime('2024-01-01 00:00:00')
  AND timestamp < toDateTime('2024-01-02 00:00:00');

Window Functions

-- Ranking
SELECT
    user_id,
    event_timestamp,
    row_number() OVER (PARTITION BY user_id ORDER BY event_timestamp) as rn
FROM events;

-- Running totals
SELECT
    date,
    revenue,
    sum(revenue) OVER (ORDER BY date) as running_total
FROM daily_revenue;

-- Lag/Lead
SELECT
    date,
    revenue,
    lag(revenue, 1) OVER (ORDER BY date) as prev_revenue,
    lead(revenue, 1) OVER (ORDER BY date) as next_revenue
FROM daily_revenue;

Performance Tuning Settings

-- Increase memory limit
SET max_memory_usage = 10000000000;

-- Parallel processing
SET max_threads = 8;

-- Block size
SET max_block_size = 65536;

-- Disable query cache
SET use_uncompressed_cache = 0;

Query Profiling Checklist

• [ ] Used EXPLAIN to verify index usage
• [ ] Selected only needed columns (no SELECT *)
• [ ] Used date range filters instead of functions on columns
• [ ] Smaller table on RIGHT side of JOIN
• [ ] Used GLOBAL JOIN for distributed queries
• [ ] Added skip indexes for frequent filters
• [ ] Considered projections for common aggregations
• [ ] Set appropriate max_memory_usage

See Also

• ../SKILL.md - Main skill entry point
• sql-reference.md - Complete SQL dialect
• table-design.md - ORDER BY and indexing strategies

ClickHouse Schema Design

Database engines, schema organization, and migration strategies.

Database Engines

ClickHouse supports multiple database engines for different use cases:

Ordinary (Default)

CREATE DATABASE my_db ENGINE = Ordinary;

Simple database with no special features. Default engine for basic use.

Atomic (Recommended for Production)

CREATE DATABASE my_db ENGINE = Atomic;

Features:

• Supports non-blocking DDL operations
• Atomic table exchange (zero-downtime schema changes)
• Transactional DDL

Table Exchange:

-- Zero-downtime schema change
EXCHANGE TABLES events AND events_v2;

-- Atomic swap - instant, no downtime

Lazy

CREATE DATABASE my_db ENGINE = Lazy
SETTINGS lazy_database_ttl = 60;

Features:

• Loaded on first access
• Unloaded after timeout
• Useful for rarely-used databases

Replicated

CREATE DATABASE my_db ENGINE = Replicated(
    'zk_path',
    'replica_name'
);

Features:

• Multi-datacenter setups
• Automatic replication
• ZooKeeper-based coordination

Dictionary

CREATE DATABASE my_db ENGINE = Dictionary(dictionaries_db_name);

For in-memory dictionary tables.

PostgreSQL/MySQL (Proxy)

CREATE DATABASE my_db ENGINE = PostgreSQL(
    'postgres-host:5432',
    'postgres_db',
    'postgres_user',
    'postgres_password'
);

Proxy to external databases for queries.

Schema Organization

Organize databases by environment and purpose:

-- Environment structure
CREATE DATABASE analytics_raw;      -- Staging tables
CREATE DATABASE analytics_staging;  -- Cleaned data
CREATE DATABASE analytics_prod;     -- Production tables
CREATE DATABASE analytics_mvs;      -- Materialized views
CREATE DATABASE analytics_dicts;    -- Dictionary definitions

Naming Conventions

-- Tables: snake_case, plural
events, user_sessions, daily_metrics

-- Columns: snake_case
event_timestamp, user_id, session_id

-- Partitions: YYYYMM format
PARTITION BY toYYYYMM(timestamp)

-- Engines: Explicit ENGINE = clause
ENGINE = MergeTree()

Schema Migration Strategy

Safe Operations (No Data Rewrite)

-- Add column (safe, metadata-only)
ALTER TABLE events ADD COLUMN new_column UInt32 DEFAULT 0;

-- Add index (safe, background)
ALTER TABLE events ADD INDEX idx_new_column new_column TYPE bloom_filter GRANULARITY 1;

-- Add projection (safe, background)
ALTER TABLE events ADD PROJECTION pr_summary (
    SELECT user_id, count() as cnt
    GROUP BY user_id
);

Unsafe Operations (Require Data Rewrite)

-- These trigger mutations (expensive!)
ALTER TABLE events DROP COLUMN old_column;
ALTER TABLE events MODIFY COLUMN col Type;
ALTER TABLE events RENAME COLUMN old_name TO new_name;
ALTER TABLE events DELETE WHERE expr;
ALTER TABLE events UPDATE col = expr WHERE expr;

Zero-Downtime Schema Change

-- 1. Create new table with new schema
CREATE TABLE events_v2 (
    timestamp DateTime,
    user_id UInt32,
    new_column String
)
ENGINE = MergeTree()
ORDER BY (user_id, timestamp);

-- 2. Backfill data
INSERT INTO events_v2 SELECT * FROM events;

-- 3. Verify data
SELECT count() FROM events;
SELECT count() FROM events_v2;

-- 4. Swap tables (atomic, instantaneous)
EXCHANGE TABLES events AND events_v2;

-- 5. Drop old table after validation
DROP TABLE events_v2;

Schema Version Control

Best practices:

• Store all DDL in version control (Git)
• Use migration tools: clickhouse-migrate, golang-migrate
• Document table relationships and dependencies
• Track migration order to support rollbacks
• Test migrations on staging first

-- Example migration tracking table
CREATE TABLE schema_migrations (
    version UInt32,
    description String,
    applied_at DateTime,
    checksum String
)
ENGINE = MergeTree()
ORDER BY version;

Database Configuration

Settings

-- Database-level settings
ALTER DATABASE my_db MODIFY SETTING max_bytes = 10000000000;

-- Check database settings
SELECT * FROM system.databases WHERE name = 'my_db';

Quotas

CREATE QUOTA my_quota
KEYED BY user_name
FOR INTERVAL 1 hour
MAX queries = 1000
MAX errors = 100
MAX result_rows = 1000000000
MAX result_bytes = 10000000000
MAX read_rows = 10000000000
MAX read_bytes = 100000000000
MAX execution_time = 60
TO user_name;

See Also

• ../SKILL.md - Main skill entry point
• core-concepts.md - Architecture and data model
• table-design.md - ORDER BY and partitioning strategies

ClickHouse SQL Reference

Complete SQL dialect reference for ClickHouse.

Data Types

Numeric Types

-- Unsigned integers
UInt8   -- 0 to 255
UInt16  -- 0 to 65,535
UInt32  -- 0 to 4,294,967,295
UInt64  -- 0 to 18,446,744,073,709,551,615
UInt128 -- Very large unsigned
UInt256 -- Extremely large unsigned

-- Signed integers
Int8    -- -128 to 127
Int16   -- -32,768 to 32,767
Int32   -- -2,147,483,648 to 2,147,483,647
Int64   -- -9,223,372,036,854,775,808 to 9,223,372,036,854,775,807

-- Floating point
Float32 -- Single precision (7 decimal digits)
Float64 -- Double precision (16 decimal digits)

-- Decimal (precise decimal arithmetic)
Decimal32(S)  -- 1-9 significant digits
Decimal64(S)  -- 10-18 significant digits
Decimal128(S) -- 19-38 significant digits
Decimal256(S) -- 39-76 significant digits
Decimal(P, S) -- P total digits, S after decimal point

-- Examples
CREATE TABLE examples (
    price Decimal(18, 2),       -- Currency (up to 999 trillion)
    percentage Decimal(5, 2),   -- Percentages (0.00 to 999.99)
    metric Float64              -- Approximate metrics
)

Temporal Types

-- Date (2 bytes, days since Unix epoch)
Date              -- Range: 1970-01-01 to 2149-06-06

-- DateTime (4 bytes, seconds since Unix epoch)
DateTime          -- Range: 1970-01-01 00:00:00 to 2106-02-07 06:28:15
DateTime('UTC')   -- With timezone

-- DateTime64 (8 bytes, sub-second precision)
DateTime64(3)                -- Millisecond precision
DateTime64(6, 'UTC')         -- Microsecond precision with timezone
DateTime64(9, 'America/New_York') -- Nanosecond precision

-- Functions
toDate('2024-01-01')                    -- String → Date
toDateTime('2024-01-01 12:00:00')       -- String → DateTime
toDateTime64('2024-01-01 12:00:00.123', 3) -- String → DateTime64

now()                                   -- Current DateTime
today()                                 -- Current Date
yesterday()                             -- Yesterday's Date
tomorrow()                              -- Tomorrow's Date

-- Date arithmetic
now() + INTERVAL 1 DAY
now() - INTERVAL 1 HOUR
date_diff('day', timestamp, now())      -- Days between dates

String Types

-- Variable-length string
String          -- No length limit, stores any string

-- Fixed-length string
FixedString(N)  -- Fixed N bytes, pads with zeros
-- Example: UUID as FixedString(16)

-- Common operations
length(string)           -- String length
concat(s1, s2, ...)      -- Concatenate
substring(s, offset, length)  -- Substring
splitByChar(separator, s)    -- Split into array
arrayJoin(splitByChar(' ', s))  -- Split and explode

Advanced Types

-- Arrays
Array(UInt32)                    -- Array of unsigned integers
Array(String)                    -- Array of strings
['a', 'b', 'c']                  -- Array literal
arrayJoin([1, 2, 3])             -- Explode array into rows

-- Tuples
Tuple(UInt32, String, Float64)   -- Mixed-type tuple
tuple(123, 'abc', 45.6)          -- Tuple literal
t.1, t.2                         -- Access tuple fields

-- Maps (key-value pairs)
Map(String, UInt64)              -- String → UInt64 map
map('key1', 1, 'key2', 2)        -- Map literal

-- Enums (efficient string storage)
Enum8('action1'=1, 'action2'=2)     -- 1 byte per value
Enum16('status1'=1, 'status2'=2)    -- 2 bytes per value

-- Nullable (allows NULL values)
Nullable(UInt32)           -- Can be NULL or UInt32
-- Nullable has overhead (special NULL marker)

-- LowCardinality (compression for low-cardinality strings)
LowCardinality(String)     -- Efficient for < 10k distinct values

-- UUID
UUID                       -- 16-byte UUID
generateUUIDv4()           -- Generate random UUID

-- IPv4/IPv6
IPv4                       -- 4-byte IPv4 address
IPv6                       -- 16-byte IPv6 address
toIPv4('192.168.1.1')      -- String → IPv4

CREATE TABLE

-- Basic syntax
CREATE TABLE [IF NOT EXISTS] [db.]table_name
(
    column1 Type [DEFAULT|ALIAS expr] [COMMENT 'description'],
    column2 Type [DEFAULT|ALIAS expr] [COMMENT 'description'],
    ...
)
ENGINE = MergeTree()
PARTITION BY expr
ORDER BY expr
PRIMARY KEY expr
SAMPLE BY expr
TTL expr
SETTINGS name=value, ...;

-- Example with all features
CREATE TABLE events (
    timestamp DateTime,
    user_id UInt32,
    event_type LowCardinality(String) DEFAULT 'unknown',
    session_id UUID DEFAULT generateUUIDv4(),
    metadata String DEFAULT '',
    revenue Decimal(18, 2) DEFAULT 0.00 COMMENT 'Revenue in USD'
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (user_id, timestamp)
PRIMARY KEY (user_id)
SAMPLE BY user_id
TTL timestamp + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;

INSERT

-- Insert values
INSERT INTO events VALUES
    (now(), 123, 'login', generateUUIDv4(), '{}', 0.00),
    (now(), 456, 'logout', generateUUIDv4(), '{}', 0.00);

-- Insert with columns
INSERT INTO events (timestamp, user_id, event_type) VALUES
    (now(), 789, 'purchase');

-- Insert from SELECT
INSERT INTO events_archive
SELECT * FROM events
WHERE timestamp < today() - INTERVAL 90 DAY;

-- Insert with format
INSERT INTO events FORMAT JSONEachRow
{"timestamp":"2024-01-01 00:00:00","user_id":123,"event_type":"login"}
{"timestamp":"2024-01-01 00:01:00","user_id":456,"event_type":"logout"}

SELECT

-- Basic SELECT
SELECT [DISTINCT] [TOP n] expr
FROM table
[FINAL]  -- Apply mutations (deduplication)
[WHERE expr]
[GROUP BY expr] [WITH ROLLUP|WITH CUBE]
[HAVING expr]
[ORDER BY expr]
[LIMIT [offset, ]n]
[UNION ALL]
[SETTINGS name=value, ...];

-- Examples
SELECT * FROM events WHERE user_id = 123;
SELECT DISTINCT user_id FROM events;
SELECT TOP 10 user_id, count() as cnt FROM events GROUP BY user_id ORDER BY cnt DESC;

-- With FINAL (apply mutations)
SELECT * FROM ReplacingMergeTree_table FINAL WHERE user_id = 123;

-- WITH ROLLUP (hierarchical aggregation)
SELECT
    user_id,
    event_type,
    count() as cnt
FROM events
GROUP BY user_id, event_type WITH ROLLUP;

JOIN

-- All JOINs are RIGHT JOINs internally
SELECT *
FROM t1
[GLOBAL] [ANY|ALL|ASOF] [INNER|LEFT|RIGHT|FULL|CROSS] JOIN t2
ON t1.key = t2.key

-- USING for same column names
SELECT *
FROM t1
[INNER|LEFT|RIGHT|FULL] JOIN t2 USING (common_key);

-- GLOBAL: Sends right table to each shard (for distributed queries)
SELECT *
FROM distributed_table lt
GLOBAL RIGHT JOIN small_table st ON lt.id = st.id;

-- ANY: First match only
SELECT * FROM t1 ANY LEFT JOIN t2 ON t1.id = t2.id;

-- ASOF: As-of join with inequality (time-series)
SELECT *
FROM ticks t1
ASOF LEFT JOIN trades t2
ON t1.symbol = t2.symbol
AND t1.time <= t2.time;

ALTER TABLE

-- Add column (safe, metadata-only)
ALTER TABLE table ADD COLUMN col Type DEFAULT expr;

-- Drop column (mutation, expensive)
ALTER TABLE table DROP COLUMN col;

-- Modify column (mutation, expensive)
ALTER TABLE table MODIFY COLUMN col Type;

-- Rename column (metadata-only)
ALTER TABLE table RENAME COLUMN old_name TO new_name;

-- Comment column
ALTER TABLE table COMMENT COLUMN col 'description';

-- Delete data (mutation, expensive)
ALTER TABLE table DELETE WHERE expr;

-- Update data (mutation, expensive)
ALTER TABLE table UPDATE col = expr WHERE expr;

-- Add index
ALTER TABLE table ADD INDEX idx_name col TYPE bloom_filter GRANULARITY 1;

-- TTL
ALTER TABLE table MODIFY TTL timestamp + INTERVAL 90 DAY;

DROP/TRUNCATE

-- Drop table
DROP TABLE [IF EXISTS] table;

-- Truncate table
TRUNCATE TABLE [IF EXISTS] table;

-- Drop partition (instant, no mutation)
ALTER TABLE table DROP PARTITION '202401';

-- Detach/Attach
DETACH TABLE table;  -- Keeps data on disk
ATTACH TABLE table;

OPTIMIZE

-- Force merge of parts
OPTIMIZE TABLE table [PARTITION partition] [FINAL];

-- FINAL: Apply mutations, deduplicate
-- Without FINAL: Just merges parts

SYSTEM Commands

-- Stop/start merges
SYSTEM STOP MERGES [ON CLUSTER cluster];
SYSTEM START MERGES [ON CLUSTER cluster];

-- Stop/start replication
SYSTEM STOP REPLICATION QUEUES;
SYSTEM START REPLICATION QUEUES;

-- Flush logs
SYSTEM FLUSH LOGS;

Common Functions

Date Functions

now()                          -- Current DateTime
today()                        -- Current Date
yesterday()                    -- Yesterday
tomorrow()                     -- Tomorrow
toDate(expr)                   -- Convert to Date
toDateTime(expr)               -- Convert to DateTime
toStartOfMonth(date)           -- First day of month
toStartOfWeek(date)            -- First day of week
toStartOfDay(date)             -- Start of day
date_diff(unit, start, end)    -- Difference between dates

Array Functions

array(x1, x2, ...)             -- Create array
arrayJoin(arr)                 -- Explode array into rows
length(arr)                    -- Array length
has(arr, elem)                -- Check if element exists
arrayConcat(arr1, arr2)       -- Concatenate arrays
arrayMap(func, arr)           -- Apply function to each element
arrayFilter(func, arr)        -- Filter array
arraySort(func, arr)          -- Sort array

String Functions

length(str)                    -- String length
substring(str, offset, length) -- Substring
concat(s1, s2, ...)            -- Concatenate
splitByChar(sep, str)          -- Split into array
join(arr, sep)                 -- Join array into string
lower(str)                     -- Lowercase
upper(str)                     -- Uppercase
trim(str)                      -- Trim whitespace
replaceOne(str, pattern, replacement)  -- Replace first
replaceAll(str, pattern, replacement)  -- Replace all

Aggregation Functions

count()                        -- Count rows
sum(expr)                      -- Sum
avg(expr)                      -- Average
min(expr)                      -- Minimum
max(expr)                      -- Maximum
quantile(level)(expr)          -- Quantile
median(expr)                   -- Median
uniq(expr)                     -- Approximate unique count
uniqCombined(expr)             -- Better approximate unique
topK(K)(expr)                  -- Top K values
histogram(K)(expr)             -- Histogram

See Also

• ../SKILL.md - Main skill entry point
• query-optimization.md - Query performance and EXPLAIN
• table-design.md - Schema design and ORDER BY

ClickHouse System Queries

Useful queries for monitoring, debugging, and managing ClickHouse databases.

---

Table Information

Table Sizes

-- All tables with sizes
SELECT
    database,
    table,
    formatReadableSize(sum(bytes)) as size,
    sum(rows) as total_rows,
    count() as parts
FROM system.parts
WHERE active = 1
GROUP BY database, table
ORDER BY sum(bytes) DESC;

-- Top 20 largest tables
SELECT
    database,
    table,
    formatReadableSize(sum(bytes)) as size,
    sum(rows) as total_rows
FROM system.parts
WHERE active = 1
GROUP BY database, table
ORDER BY sum(bytes) DESC
LIMIT 20;

-- Tables by row count
SELECT
    database,
    table,
    sum(rows) as total_rows,
    formatReadableQuantity(sum(rows)) as readable_rows
FROM system.parts
WHERE active = 1
GROUP BY database, table
ORDER BY sum(rows) DESC
LIMIT 20;

Column Information

-- Column sizes and compression
SELECT
    database,
    table,
    column,
    type,
    formatReadableSize(sum(data_uncompressed_bytes)) as uncompressed,
    formatReadableSize(sum(data_compressed_bytes)) as compressed,
    sum(data_compressed_bytes) / sum(data_uncompressed_bytes) as ratio
FROM system.columns
WHERE database != 'system'
GROUP BY database, table, column, type
ORDER BY sum(data_uncompressed_bytes) DESC
LIMIT 50;

-- All columns in a table
SELECT
    name,
    type,
    default_kind,
    default_expression,
    comment
FROM system.columns
WHERE database = 'my_database'
  AND table = 'my_table'
ORDER BY position;

Partition Information

-- Partition details
SELECT
    partition,
    sum(rows) as total_rows,
    count() as parts,
    formatReadableSize(sum(bytes)) as size,
    min(min_timestamp) as min_time,
    max(max_timestamp) as max_time
FROM system.parts
WHERE active = 1
  AND table = 'my_table'
  AND database = currentDatabase()
GROUP BY partition
ORDER BY partition DESC;

-- Large partitions (many parts)
SELECT
    partition,
    count() as parts,
    sum(rows) as total_rows
FROM system.parts
WHERE active = 1
  AND table = 'my_table'
GROUP BY partition
HAVING parts > 100
ORDER BY parts DESC;

---

Query Monitoring

Running Queries

-- All running queries
SELECT
    query_id,
    user,
    query,
    elapsed,
    formatReadableSize(memory_usage) as memory,
    formatReadableQuantity(read_rows) as rows_read,
    formatReadableSize(read_bytes) as bytes_read
FROM system.processes
ORDER BY elapsed DESC;

-- Long-running queries (> 1 minute)
SELECT
    query_id,
    user,
    query,
    elapsed / 60 as minutes_elapsed,
    formatReadableSize(memory_usage) as memory
FROM system.processes
WHERE elapsed > 60
ORDER BY elapsed DESC;

-- Queries by user
SELECT
    user,
    count() as query_count,
    sum(memory_usage) as total_memory
FROM system.processes
GROUP BY user
ORDER BY total_memory DESC;

Query History

-- Recent queries (last hour)
SELECT
    type,
    substring(query, 1, 100) as query_preview,
    query_duration_ms / 1000 as seconds,
    formatReadableSize(memory_usage) as memory,
    event_time
FROM system.query_log
WHERE event_time > now() - INTERVAL 1 HOUR
ORDER BY event_time DESC
LIMIT 100;

-- Slow queries (> 5 seconds)
SELECT
    query,
    query_duration_ms / 1000 as seconds,
    formatReadableSize(memory_usage) as memory,
    formatReadableQuantity(read_rows) as rows_read,
    event_time
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query_duration_ms > 5000
ORDER BY query_duration_ms DESC
LIMIT 20;

-- Most expensive by memory
SELECT
    query,
    formatReadableSize(memory_usage) as memory,
    query_duration_ms / 1000 as seconds
FROM system.query_log
WHERE type = 'QueryFinish'
ORDER BY memory_usage DESC
LIMIT 10;

-- Failed queries
SELECT
    query,
    exception_code,
    exception_text,
    event_time
FROM system.query_log
WHERE type = 'Exception'
  AND event_time > now() - INTERVAL 1 DAY
ORDER BY event_time DESC
LIMIT 50;

Query Statistics by Table

-- Most accessed tables
SELECT
    database,
    table,
    count() as query_count,
    sum(read_rows) as total_rows_read,
    avg(query_duration_ms) as avg_duration_ms
FROM system.query_log
WHERE type = 'QueryFinish'
  AND event_date = today()
GROUP BY database, table
ORDER BY query_count DESC
LIMIT 20;

-- Queries by type
SELECT
    type,
    count() as count,
    avg(query_duration_ms) as avg_duration_ms
FROM system.query_log
WHERE event_date = today()
GROUP BY type
ORDER BY count DESC;

---

Merge Monitoring

Active Merges

-- Current merges
SELECT
    database,
    table,
    elapsed,
    progress,
    formatReadableSize(bytes_read_uncompressed) as bytes_read,
    rows_read,
    is_mutation,
    merge_type
FROM system.merges
ORDER BY elapsed DESC;

-- Merge performance by day
SELECT
    table,
    count() as merge_count,
    avg(bytes_read_uncompressed) as avg_size,
    sum(rows_read) as total_rows
FROM system.merges
WHERE event_date = today()
GROUP BY table
ORDER BY merge_count DESC;

-- Merge queue size
SELECT
    database,
    table,
    count() as parts_to_merge,
    sum(bytes) as total_bytes
FROM system.parts
WHERE active = 1
  AND rows > 0
GROUP BY database, table
HAVING parts_to_merge > 100
ORDER BY parts_to_merge DESC;

Merge Statistics

-- Merge performance over time
SELECT
    toStartOfHour(event_time) as hour,
    count() as merge_count,
    avg(bytes_read_uncompressed) as avg_size,
    sum(rows_read) as total_rows
FROM system.part_log
WHERE event_time > now() - INTERVAL 1 DAY
  AND event_type = 'MergeParts'
GROUP BY hour
ORDER BY hour;

---

Mutation Monitoring

Active Mutations

-- Current mutations
SELECT
    database,
    table,
    command,
    is_done,
    parts_to_do,
    parts_to_do_names,
    formatReadableSize(bytes_read_uncompressed) as bytes_processed
FROM system.mutations
WHERE is_done = 0;

-- Mutation progress
SELECT
    table,
    command,
    is_done,
    parts_to_do - parts_to_do_names as parts_remaining,
    elapsed
FROM system.mutations
ORDER BY parts_to_do DESC;

-- All mutations history
SELECT
    database,
    table,
    command,
    is_done,
    created_at,
    finished_at
FROM system.mutations
ORDER BY created_at DESC
LIMIT 50;

---

Replication Monitoring

Replica Status

-- All replicas
SELECT
    database,
    table,
    is_leader,
    is_readonly,
    queue_size,
    absolute_delay,
    delay * 1000 as lag_ms
FROM system.replicas
ORDER BY absolute_delay DESC;

-- Replication queue
SELECT * FROM system.replication_queue
WHERE delay > 5
ORDER BY delay DESC;

-- Replication lag by table
SELECT
    database,
    table,
    replica_name,
    is_leader,
    is_readonly,
    queue_size,
    absolute_delay
FROM system.replicas
WHERE absolute_delay > 10
ORDER BY absolute_delay DESC;

ZooKeeper Status

-- ZooKeeper connection
SELECT * FROM system.zookeeper WHERE path = '/';

-- Check ZooKeeper path
SELECT
    name,
    value,
    data
FROM system.zookeeper
WHERE path = '/clickhouse/tables';

---

Disk and Storage

Disk Usage

-- All disks
SELECT
    name,
    path,
    formatReadableSize(free_space) as free,
    formatReadableSize(total_space) as total,
    formatReadableSize(keep_free_space) as keep_free,
    (free_space / total_space) * 100 as percent_free
FROM system.disks;

-- Disk usage by table
SELECT
    disk_name,
    database,
    table,
    formatReadableSize(sum(bytes_on_disk)) as size
FROM system.parts
WHERE active = 1
GROUP BY disk_name, database, table
ORDER BY sum(bytes_on_disk) DESC
LIMIT 20;

Storage Policies

-- All storage policies
SELECT * FROM system.storage_policies;

-- Volumes in storage policies
SELECT * FROM system.disks;

---

Cluster Information

Cluster Status

-- All clusters
SELECT * FROM system.clusters;

-- Cluster nodes
SELECT
    cluster,
    shard_num,
    replica_num,
    host_name,
    port,
    user
FROM system.clusters
WHERE cluster = 'my_cluster';

-- Cluster health
SELECT
    cluster,
    sum(error_count) as errors,
    sum(num_requests) as requests
FROM system.clusters
GROUP BY cluster;

---

Database Information

All Databases

-- All databases
SELECT
    name,
    engine,
    data_path,
    metadata_path
FROM system.databases
ORDER BY name;

-- Database engine types
SELECT
    engine,
    count() as count
FROM system.databases
GROUP BY engine
ORDER BY count DESC;

---

User and Security

User Information

-- All users
SELECT * FROM system.users;

-- User quotas
SELECT * FROM system.quotas;

-- Current roles
SELECT * FROM system.current_roles;

Access Control

-- Row policies
SELECT * FROM system.row_policies;

-- Grants
SELECT * FROM system.grants;

---

System Health

General Health

-- Server info
SELECT version(), uptime(), now() as current_time;

-- Memory usage
SELECT
    formatReadableSize(sum(bytes_allocated)) as allocated,
    formatReadableSize(sum(bytes_used)) as used
FROM system.dictionaries;

-- Load average
SELECT * FROM system.asynchronous_metrics
WHERE metric LIKE '%load%';

Settings

-- All settings
SELECT * FROM system.settings
ORDER BY name;

-- Current settings
SELECT * FROM system.settings
WHERE changed == 1;

---

Dictionary Information

All Dictionaries

-- Dictionary status
SELECT
    name,
    status,
    origin,
    type,
    element_count,
    bytes_allocated
FROM system.dictionaries
ORDER BY name;

-- Loading dictionaries
SELECT
    name,
    status,
    loading_start_time,
    loading_duration_seconds
FROM system.dictionaries
WHERE status != 'LOADED';

---

Formatted Queries

Human-Readable Output

-- Size formatted
SELECT
    database,
    table,
    formatReadableSize(sum(bytes)) as size,
    formatReadableQuantity(sum(rows)) as rows,
    formatReadableSize(sum(bytes_on_disk)) as on_disk
FROM system.parts
WHERE active = 1
GROUP BY database, table
ORDER BY sum(bytes) DESC
LIMIT 20;

-- Time formatted
SELECT
    query,
    formatDateTime(query_start_time, '%Y-%m-%d %H:%M:%S') as started,
    formatDuration(query_duration_ms / 1000) as duration
FROM system.query_log
WHERE type = 'QueryFinish'
ORDER BY query_start_time DESC
LIMIT 10;

---

Quick Diagnostic Queries

One-Liner Health Checks

-- Quick health check
SELECT 'uptime' as metric, toString(uptime()) as value
UNION ALL SELECT 'version', version()
UNION ALL SELECT 'running_queries', toString(count())
FROM system.processes
UNION ALL SELECT 'active_merges', toString(count())
FROM system.merges
UNION ALL SELECT 'replication_lag', toString(count())
FROM system.replication_queue WHERE delay > 5;

-- Table count by database
SELECT database, count() as tables
FROM system.tables
WHERE database != 'system'
GROUP BY database
ORDER BY tables DESC;

-- Parts count (health check)
SELECT
    database,
    table,
    count() as parts,
    sum(rows) as total_rows
FROM system.parts
WHERE active = 1
GROUP BY database, table
HAVING parts > 1000
ORDER BY parts DESC;

---

Useful Subqueries

Find Problematic Queries

-- Queries with exceptions
SELECT
    query,
    exception_text,
    count() as error_count
FROM system.query_log
WHERE type = 'Exception'
  AND event_date = today()
GROUP BY query, exception_text
ORDER BY error_count DESC
LIMIT 20;

-- Queries reading too much data
SELECT
    query,
    read_rows,
    formatReadableQuantity(read_rows) as readable_rows,
    read_bytes,
    formatReadableSize(read_bytes) as readable_bytes
FROM system.query_log
WHERE type = 'QueryFinish'
  AND event_date = today()
  AND read_rows > 1000000000
ORDER BY read_rows DESC
LIMIT 20;

Table Growth Over Time

-- Daily table growth
SELECT
    toStartOfDay(event_time) as day,
    database,
    table,
    formatReadableSize(sum(bytes)) as size
FROM system.part_log
WHERE event_type = 'NewPart'
  AND event_time > now() - INTERVAL 7 DAY
GROUP BY day, database, table
ORDER BY day DESC, sum(bytes) DESC;

---

Tips:

• Use formatReadableSize() for human-readable byte sizes
• Use formatReadableQuantity() for human-readable row counts
• Filter WHERE active = 1 to see only active parts
• Use currentDatabase() to refer to current database
• Most system.* tables have event_date for partitioning

ClickHouse Table Design

ORDER BY design, partitioning strategies, column selection, and sampling.

ORDER BY Design (CRITICAL)

The ORDER BY clause defines data layout on disk - the most important schema decision!

ORDER BY Design Principles

1. Match query WHERE patterns: Most selective filter first 2. Time-series: Timestamp as second column (after dimension) 3. High-cardinality first: user_id before event_type 4. Limit to 3-4 columns: More columns = larger index

Examples

-- ✅ GOOD: ORDER BY matches query pattern
CREATE TABLE events (
    user_id UInt32,
    event_timestamp DateTime,
    event_type String
)
ENGINE = MergeTree()
ORDER BY (user_id, event_timestamp);

-- Query leverages index (reads contiguous data)
SELECT * FROM events
WHERE user_id = 123
  AND event_timestamp >= now() - INTERVAL 1 DAY;

-- ❌ BAD: ORDER BY doesn't match queries
CREATE TABLE events (
    user_id UInt32,
    event_timestamp DateTime,
    event_type String
)
ENGINE = MergeTree()
ORDER BY (event_timestamp);

-- Query scans all data (timestamp-ordered, but filtering by user_id)
SELECT * FROM events WHERE user_id = 123;

Primary Key vs Sorting Key

-- Sorting key: defines data layout on disk
ORDER BY (user_id, event_timestamp, event_type)

-- Primary key: subset of ORDER BY for row-level index
PRIMARY KEY (user_id, event_timestamp)

-- Effect:
-- - Data on disk: sorted by (user_id, timestamp, event_type)
-- - Primary index: only (user_id, timestamp)
-- - Queries filtering by user_id + timestamp use index
-- - Queries scanning event_type read more data

When to use different PRIMARY KEY:

• Want to optimize for queries that filter on subset of ORDER BY
• Reduce primary key size (faster index scans)
• Note: PRIMARY KEY must be prefix of ORDER BY

Common ORDER BY Patterns

-- Time-series by user
ORDER BY (user_id, timestamp)

-- Time-series by sensor
ORDER BY (sensor_id, timestamp)

-- Multi-tenant time-series
ORDER BY (tenant_id, user_id, timestamp)

-- Event log
ORDER BY (timestamp, event_type)

-- Metrics
ORDER BY (metric_name, timestamp, labels_hash)

Partition Strategy

Partitions enable fast data drop and TTL operations:

Time-Based Partitioning (Most Common)

-- Monthly partitions (recommended for 10GB+ per day)
PARTITION BY toYYYYMM(timestamp);
-- Results: partitions like '202401', '202402', ...

-- Daily partitions (for high-volume, 1GB+ per day)
PARTITION BY toDate(timestamp);
-- Results: one partition per day

Custom Partitioning

-- By tenant (multi-tenant cleanup)
PARTITION BY tenant_id;
-- Drop all data for tenant: ALTER TABLE DROP PARTITION 'tenant_123';

-- Composite (use carefully - too many partitions!)
PARTITION BY (tenant_id, toYYYYMM(timestamp));
-- Results: 'tenant_123_202401', 'tenant_123_202402', ...

Partitioning Guidelines

Key principles:

• Aim for 100-1000 parts total across all partitions
• Monthly partitions: Good for 10GB+ per day
• Daily partitions: Good for 1GB+ per day
• Too many partitions = slow queries, high merge overhead

Partition Operations

-- Drop partition (instant, no mutation)
ALTER TABLE events DROP PARTITION '202401';

-- Detach/Attach partition
ALTER TABLE events DETACH PARTITION '202401';
ALTER TABLE events ATTACH PARTITION '202401';

-- Freeze partition (for backup)
ALTER TABLE events FREEZE PARTITION '202401';

-- Check partition sizes
SELECT
    partition,
    sum(rows) as total_rows,
    formatReadableSize(sum(bytes)) as size
FROM system.parts
WHERE active = 1 AND table = 'events'
GROUP BY partition
ORDER BY partition DESC;

Column Selection

Choose smallest sufficient type:

Numeric Types

-- Use smallest integer type
UInt8   -- 0-255
UInt16  -- 0-65,535
UInt32  -- 0-4,294,967,295
UInt64  -- 0-18,446,744,073,709,551,615

-- Decimal for currency
Decimal(18, 2)  -- Up to 999 trillion
Decimal(10, 2)  -- Smaller range, faster

Temporal Types

timestamp Date           -- 2 bytes (days since epoch)
timestamp DateTime       -- 4 bytes (seconds since epoch)
timestamp DateTime64(3)  -- 8 bytes (milliseconds)

String Types

-- LowCardinality for enum-like strings (< 10k distinct values)
event_type LowCardinality(String)

-- Nullable vs default values
status Nullable(String)   -- Has overhead (special NULL marker)
status String DEFAULT ''  -- Better: Use default value

Sampling Key

Enable SAMPLE queries for approximate analytics:

-- Enable sampling
CREATE TABLE events (
    user_id UInt32,
    event_timestamp DateTime,
    data String
)
ENGINE = MergeTree()
ORDER BY (user_id, event_timestamp)
SAMPLE BY user_id;  -- Must be column in ORDER BY

-- Query with sampling (10% of data)
SELECT * FROM events SAMPLE 0.1;

-- Sampling with interpolation
SELECT
    user_id,
    count() * 10 as estimated_count  -- Multiply by 1/sampling_rate
FROM events
SAMPLE 0.1
GROUP BY user_id;

Use cases for sampling:

• Exploratory analytics (fast approximate results)
• Testing queries on large datasets
• Dashboard previews (refresh faster)

Codecs

Apply compression to individual columns:

-- ZSTD compression (good default)
CREATE TABLE events (
    data String CODEC(ZSTD)
)
ENGINE = MergeTree()
ORDER BY user_id;

-- Compression levels
CODEC(ZSTD(3))   -- Fast, less compression
CODEC(ZSTD(15))  -- Slower, more compression

-- No compression (for already compressed data)
CODEC(NONE)

-- Multiple codecs
CODEC(Delta, ZSTD)

-- LZ4 (faster than ZSTD, less compression)
CODEC(LZ4)

Codec guidelines:

• Use ZSTD for most data (good balance)
• Use NONE for already-compressed data (images, encrypted data)
• Use Delta for monotonically increasing values (timestamps, IDs)
• Higher compression levels = slower queries

Skip Indexes

Data skipping indexes allow ClickHouse to skip data during reads:

-- Bloom filter for exact match
CREATE TABLE events (
    user_id UInt32,
    email String,
    timestamp DateTime
)
ENGINE = MergeTree()
ORDER BY (user_id, timestamp)
INDEX idx_email email TYPE bloom_filter GRANULARITY 1;

-- Minmax for range queries
INDEX idx_time timestamp TYPE minmax GRANULARITY 4;

-- Set for IN queries
INDEX idx_user user_id TYPE set(1000) GRANULARITY 1;

-- Tokenbf for string search
INDEX idx_msg message TYPE tokenbf_v1(512, 3, 0) GRANULARITY 1;

Index types:

• minmax: Min/max values per granule
• set: Set of values (good for IN queries)
• bloom_filter: Probabilistic exact match
• tokenbf_v1: Token-based bloom filter for text search

Granularity:

• Lower = more precise index, larger index size
• Higher = less precise index, smaller index size
• Default: 1 (most precise)

Table Settings

CREATE TABLE events (
    timestamp DateTime,
    user_id UInt32
)
ENGINE = MergeTree()
ORDER BY (user_id, timestamp)
SETTINGS
    index_granularity = 8192,        -- Rows per mark (default)
    index_granularity_bytes = 10485760, -- Bytes per mark
    enable_mixed_granularity_parts = 1,  -- Adaptive marks
    min_rows_for_wide_part = 0,       -- Always use wide parts
    merge_max_block_size = 1048544,   -- Block size for merges
    storage_policy = 'default';       -- Storage policy

See Also

• ../SKILL.md - Main skill entry point
• core-concepts.md - MergeTree internals
• table-engines.md - Complete table engine reference
• schema-design.md - Database engines and migrations

Sections

This file defines all sections, their ordering, impact levels, and descriptions. The section ID (in parentheses) is the filename prefix used to group rules.

---

1. Schema Design (schema)

Impact: CRITICAL

Description: Proper schema design is foundational to ClickHouse performance. ORDER BY is immutable after table creation; wrong choices require full data migration. Includes primary key selection, data types, partitioning strategy, and JSON usage. Column types and ordering can impact query speed by orders of magnitude.

2. Query Optimization (query)

Impact: CRITICAL

Description: Query patterns dramatically affect performance. JOIN algorithms, filtering strategies, skipping indices, and materialized views can reduce query time from minutes to milliseconds. Pre-computed aggregations read thousands of rows instead of billions.

3. Insert Strategy (insert)

Impact: CRITICAL

Description: Each INSERT creates a data part. Single-row inserts overwhelm the merge process. Proper batching (10K-100K rows), async inserts for high-frequency writes, mutation avoidance, and letting background merges work are essential for stable cluster performance.

Rule Title Here

Impact: CRITICAL (optional description)

Brief explanation of the rule and why it matters. This should be clear and concise, explaining the performance implications.

Incorrect (description of what's wrong):

-- Bad: description
SELECT * FROM table;

Correct (description of what's right):

-- Good: description
SELECT * FROM table;

Reference: Official Docs

Use Async Inserts for High-Frequency Small Batches

Impact: HIGH

When client-side batching isn't practical, async inserts buffer server-side and create larger parts automatically.

Incorrect (small batches without async):

# Small batches without async_insert - creates too many parts
for batch in chunks(events, 100):
    client.execute("INSERT INTO events VALUES", batch)

Correct (enable async inserts):

# Enable async_insert with safe defaults
client.execute("SET async_insert = 1")
client.execute("SET wait_for_async_insert = 1")  # Confirms durability

for batch in chunks(events, 100):
    client.execute("INSERT INTO events VALUES", batch)
# Server buffers and creates larger parts automatically

-- Configure server-side for specific users
ALTER USER my_app_user SETTINGS
    async_insert = 1,
    wait_for_async_insert = 1,
    async_insert_max_data_size = 10000000,  -- Flush at 10MB
    async_insert_busy_timeout_ms = 1000;    -- Flush after 1s

Flush conditions (whichever occurs first):

• Buffer reaches async_insert_max_data_size
• Time threshold async_insert_busy_timeout_ms elapses
• Maximum insert queries accumulate

Return modes:

Reference: Selecting an Insert Strategy

Use Native Format for Best Insert Performance

Impact: MEDIUM

Data format affects insert performance. Native format is column-oriented with minimal parsing overhead.

Performance Ranking (fastest to slowest):

Example:

# Use Native format for best performance
client.execute("INSERT INTO events VALUES", data, settings={'input_format': 'Native'})

Reference: Selecting an Insert Strategy