undefined

undefined

-- p95 latency (replace 0.95 for p99 = 0.99)
SELECT quantile(0.95)(query_duration_ms) as p95_latency_ms
FROM system.query_log
WHERE event_date >= today()
  AND type = 'QueryFinish'
  AND query NOT LIKE '%system.%';

-- Queries per second (current)
SELECT COUNT() / max(CAST(elapsed as Float64)) as qps
FROM system.processes
WHERE query NOT LIKE '%system.%';

-- Insert throughput (rows/second)
SELECT
    table,
    SUM(rows) / SUM(CAST(query_duration_ms as Float64)) * 1000 as rows_per_second
FROM system.query_log
WHERE query LIKE 'INSERT%'
  AND type = 'QueryFinish'
  AND event_date >= today()
GROUP BY table;

-- Memory usage
SELECT
    formatReadableSize(value) as current_memory,
    formatReadableSize(80000000000) as target_limit  -- Adjust to 80% of your RAM
FROM system.metrics
WHERE metric = 'MemoryTracking';

-- Disk utilization
SELECT
    formatReadableSize(free_space) as free,
    formatReadableSize(total_space) as total,
    round(free_space / total_space * 100, 2) as free_percent
FROM system.disks;

-- Merge pressure (high pending merges = slow inserts)
SELECT database, table, COUNT() as parts_count
FROM system.parts
WHERE active = 1
GROUP BY database, table
HAVING parts_count > 1000
ORDER BY parts_count DESC;

-- Slowest queries (last 24 hours)
SELECT
    event_time,
    query_duration_ms,
    read_rows,
    formatReadableSize(read_bytes) as bytes_read,
    formatReadableSize(memory_usage) as peak_memory,
    query
FROM system.query_log
WHERE type = 'QueryFinish'
  AND event_date >= today() - 1
  AND query NOT LIKE '%system.%'
ORDER BY query_duration_ms DESC
LIMIT 20;

-- Most resource-intensive queries
SELECT
    query,
    COUNT() as exec_count,
    AVG(query_duration_ms) as avg_duration_ms,
    MAX(memory_usage) / 1024 / 1024 as peak_memory_mb,
    SUM(read_bytes) / 1024 / 1024 / 1024 as total_gb_read
FROM system.query_log
WHERE event_date >= today() - 7
  AND type = 'QueryFinish'
GROUP BY query
ORDER BY total_gb_read DESC
LIMIT 20;

-- Error breakdown
SELECT
    exception,
    COUNT() as count,
    MAX(event_time) as last_error
FROM system.query_log
WHERE type = 'ExceptionWhileProcessing'
  AND event_date >= today() - 1
GROUP BY exception
ORDER BY count DESC;

-- Largest tables and growth rate
SELECT
    database,
    name as table,
    SUM(rows) as total_rows,
    formatReadableSize(SUM(bytes_on_disk)) as disk_size,
    formatReadableSize(SUM(data_compressed_bytes)) as compressed,
    round(SUM(data_uncompressed_bytes) / SUM(data_compressed_bytes), 1) as compression_ratio
FROM system.parts
WHERE active = 1
  AND database NOT IN ('system', 'information_schema')
GROUP BY database, name
ORDER BY bytes_on_disk DESC;

-- Check current thread setting
SELECT value FROM system.settings WHERE name = 'max_threads';

-- Optimize for your hardware
SET max_threads = 16;  -- Set to CPU core count (or slightly less)

-- Permanent setting in /etc/clickhouse-server/config.xml
-- <max_threads>16</max_threads>

-- Current server memory limit
SELECT value FROM system.settings WHERE name = 'max_server_memory_usage';

-- Set per-query limit (10 GB example)
SET max_memory_usage = 10000000000;

-- Production config (/etc/clickhouse-server/config.xml):
-- <max_server_memory_usage>64000000000</max_server_memory_usage>  <!-- 64 GB = 80% of 80GB RAM -->
-- <max_concurrent_queries>100</max_concurrent_queries>

-- Monitor current usage
SELECT formatReadableSize(value) as memory_usage
FROM system.metrics
WHERE metric = 'MemoryTracking';

<!-- In /etc/clickhouse-server/config.xml -->
<merge_tree>
    <!-- Enable direct I/O for large table scans -->
    <min_bytes_to_use_direct_io>10485760</min_bytes_to_use_direct_io>  <!-- 10 MB -->
</merge_tree>

<!-- Filesystem recommendations: Use NVMe SSD for best performance -->
<!-- Separate data and logs to different disks if possible -->

<!-- Add to /etc/clickhouse-server/config.xml on all nodes -->
<remote_servers>
    <production_cluster>
        <shard>
            <replica>
                <host>node1.example.com</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>node2.example.com</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <replica>
                <host>node3.example.com</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>node4.example.com</host>
                <port>9000</port>
            </replica>
        </shard>
    </production_cluster>
</remote_servers>

-- Step 1: Create local table on each shard
CREATE TABLE events_local ON CLUSTER 'production_cluster' (
    user_id UInt32,
    event_type String,
    timestamp DateTime,
    properties String
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}')
ORDER BY (user_id, timestamp)
PARTITION BY toDate(timestamp);

-- Step 2: Create distributed table (insert point)
CREATE TABLE events ON CLUSTER 'production_cluster' AS events_local
ENGINE = Distributed('production_cluster', default, events_local, rand());

-- Step 3: Use distributed table for inserts
INSERT INTO events VALUES (123, 'purchase', now(), '{"amount": 99.99}');

-- Step 4: Query across all shards
SELECT COUNT() as total_events FROM events;
SELECT topK(10)(event_type) FROM events;  -- Top 10 event types

-- Check replication status
SELECT
    database,
    table,
    total_replicas,
    active_replicas,
    absolute_delay
FROM system.replicas
ORDER BY absolute_delay DESC;

-- Verify all nodes responding
SELECT hostname(), version(), uptime() as uptime_seconds
FROM clusterAllReplicas('production_cluster', system.one);

#!/bin/bash

**Native BACKUP Command (ClickHouse 22.8+):**
```sql
-- Backup single table
BACKUP TABLE events TO Disk('backups', 'events_' || toString(today()) || '.zip');

-- Backup entire database
BACKUP DATABASE default TO Disk('backups', 'db_' || toString(today()) || '.zip');

-- Restore table
RESTORE TABLE events FROM Disk('backups', 'events_20240115.zip');

-- Restore database
RESTORE DATABASE default FROM Disk('backups', 'db_20240115.zip');

#!/bin/bash

**Test Backups Monthly:**
```bash

undefined

-- For analytical queries on large datasets
SET max_threads = 16;           -- Use all cores
SET max_memory_usage = 50000000000;  -- 50 GB for large aggregations
SET max_execution_time = 300;   -- 5 minutes timeout

-- For batch operations
SET async_insert = 1;           -- Queue inserts asynchronously
SET wait_for_async_insert = 0;  -- Don't wait for async confirmation

-- For complex GROUP BY operations
SET group_by_overflow_mode = 'any';  -- Switch to approximate aggregation if limit exceeded

-- Enable query results cache
SET use_query_cache = 1;

<clickhouse>
    <!-- Memory: Set to ~80% of available RAM -->
    <max_server_memory_usage>64000000000</max_server_memory_usage>

    <!-- Concurrency -->
    <max_concurrent_queries>100</max_concurrent_queries>
    <max_connections>4096</max_connections>

    <!-- Background operations (merges, mutations) -->
    <background_pool_size>16</background_pool_size>
    <background_schedule_pool_size>8</background_schedule_pool_size>

    <!-- Compression: LZ4 is fast (default), ZSTD is smaller -->
    <compression>
        <case>
            <method>lz4</method>
        </case>
    </compression>

    <!-- Network: Keep connections alive efficiently -->
    <keep_alive_timeout>3</keep_alive_timeout>
    <socket_receive_timeout_sec>300</socket_receive_timeout_sec>
    <socket_send_timeout_sec>300</socket_send_timeout_sec>

    <!-- Logging: Use warning level in production to reduce I/O -->
    <logger>
        <level>warning</level>
        <log>/var/log/clickhouse-server/clickhouse-server.log</log>
        <errorlog>/var/log/clickhouse-server/clickhouse-server.err.log</errorlog>
        <size>100M</size>
        <count>10</count>
    </logger>
</clickhouse>

-- Use appropriate codecs for different column types
CREATE TABLE events (
    user_id UInt32,
    timestamp DateTime CODEC(DoubleDelta, LZ4),  -- Delta encoding for timestamps
    event_type LowCardinality(String),           -- Dictionary encoding for low-cardinality
    price Decimal(10, 2) CODEC(T64, ZSTD(3)),   -- Better compression for decimals
    json_data String CODEC(ZSTD(3))              -- ZSTD for larger strings
)
ENGINE = MergeTree()
ORDER BY (user_id, timestamp)
PARTITION BY toDate(timestamp)
SETTINGS
    index_granularity = 8192,
    min_bytes_for_wide_part = 10485760,     -- 10 MB
    min_rows_for_wide_part = 100000;

-- Enable TTL for automatic data lifecycle management
ALTER TABLE events MODIFY TTL timestamp + INTERVAL 90 DAY;

-- Force optimization (merges all parts, applies TTL)
OPTIMIZE TABLE events FINAL;

SELECT query, query_duration_ms, read_rows, read_bytes / 1024 / 1024 as read_mb
FROM system.query_log
WHERE query_duration_ms > 30000  -- > 30 seconds
ORDER BY query_duration_ms DESC
LIMIT 10;

-- Option 1: Increase timeout for specific query
SET max_execution_time = 300;  -- 5 minutes
SELECT * FROM events WHERE timestamp > '2024-01-01';

-- Option 2: Add index for faster filtering
ALTER TABLE events ADD INDEX idx_timestamp timestamp TYPE minmax GRANULARITY 4;

-- Option 3: Use SAMPLE to estimate results faster
SELECT COUNT() FROM events SAMPLE 0.1;  -- Estimate from 10% of data

-- Option 4: Reduce scanned data with appropriate WHERE clauses
SELECT COUNT() FROM events WHERE timestamp >= today() - 7;  -- Last 7 days only

SELECT
    query_id,
    user,
    memory_usage / 1024 / 1024 as memory_mb,
    query
FROM system.processes
ORDER BY memory_usage DESC;

-- Check recent OOM errors
SELECT exception FROM system.query_log
WHERE exception LIKE '%Memory%'
  AND event_date >= today() - 1;

-- Option 1: Increase memory limit per query
SET max_memory_usage = 30000000000;  -- 30 GB (if server has capacity)

-- Option 2: Use approximate aggregations instead of exact
SELECT uniq(user_id) FROM events;       -- Faster approximate distinct count
SELECT topK(1000)(category) FROM events;  -- Top 1000 without full GROUP BY

-- Option 3: Process in smaller batches
SELECT * FROM events WHERE date = toDate(now()) LIMIT 1000000;

-- Option 4: Increase server memory limit if within hardware capacity
-- Edit config.xml: <max_server_memory_usage>128000000000</max_server_memory_usage>

SELECT
    table,
    COUNT() as insert_count,
    SUM(rows) as total_rows,
    SUM(rows) / SUM(CAST(query_duration_ms as Float64)) * 1000 as rows_per_second
FROM system.query_log
WHERE query LIKE 'INSERT%'
  AND type = 'QueryFinish'
  AND event_date >= today()
GROUP BY table;

-- Check number of parts (high = slow inserts)
SELECT database, table, COUNT() as parts_count
FROM system.parts
WHERE active = 1
GROUP BY database, table
ORDER BY parts_count DESC;

-- Option 1: Increase batch size (1000 -> 100000 rows per insert)
INSERT INTO events
SELECT * FROM source_table
LIMIT 100000;  -- Larger batches = better throughput

-- Option 2: Enable async inserts for non-critical data
SET async_insert = 1;
SET wait_for_async_insert = 0;  -- Fire and forget
INSERT INTO events VALUES (row1), (row2), ...;

-- Option 3: Reduce merge pressure
ALTER TABLE events MODIFY SETTING max_parts_in_total = 10000;
SYSTEM STOP MERGES events;  -- Temporarily disable merges during bulk load
-- ... run inserts ...
SYSTEM START MERGES events;

-- Option 4: Force merge after bulk load to consolidate
OPTIMIZE TABLE events;

SELECT
    database,
    table,
    total_replicas,
    active_replicas,
    absolute_delay
FROM system.replicas
ORDER BY absolute_delay DESC;

-- On problematic replica, check queue
SELECT * FROM system.replication_queue;

undefined

---

**Issue: Kafka Table Not Consuming (No new data from Kafka)**

Diagnosis:
```sql
-- Check Kafka consumer status
SELECT * FROM system.kafka_consumers;

-- Check for Kafka-related errors
SELECT exception, COUNT() as count
FROM system.query_log
WHERE query LIKE '%Kafka%'
  AND exception != ''
  AND event_date >= today() - 1
GROUP BY exception;

undefined

undefined

-- Monitor disk usage
SELECT
    formatReadableSize(free_space) as free,
    formatReadableSize(total_space) as total,
    round(free_space / total_space * 100, 2) as free_percent
FROM system.disks;

-- Drop old partitions (instant, no merge)
ALTER TABLE events DROP PARTITION '202301';
ALTER TABLE events DROP PARTITION '202302';

-- Set TTL for automatic cleanup
ALTER TABLE events MODIFY TTL timestamp + INTERVAL 90 DAY;

-- Manual optimization (applies TTL, merges parts)
OPTIMIZE TABLE events FINAL;

-- Check size reduction from optimization
SELECT
    formatReadableSize(SUM(bytes_on_disk)) as disk_size,
    COUNT() as parts_count
FROM system.parts
WHERE table = 'events' AND active = 1;

-- View all current connections
SELECT
    query_id,
    user,
    client_hostname,
    elapsed as elapsed_seconds,
    query
FROM system.processes
ORDER BY elapsed DESC;

-- Kill specific long-running query
KILL QUERY WHERE query_id = 'abc123';

-- Kill all queries from a user (e.g., runaway job)
KILL QUERY WHERE user = 'batch_user';

-- Persistent limit in config.xml
-- <max_connections>4096</max_connections>

-- Check merge queue
SELECT
    database,
    table,
    elapsed,
    progress,
    num_parts,
    formatReadableSize(total_size_bytes_compressed) as size
FROM system.merges
ORDER BY elapsed DESC;

-- Pause merges temporarily (e.g., for urgent queries)
SYSTEM STOP MERGES events;

-- Resume merges
SYSTEM START MERGES events;

-- Force immediate merge
OPTIMIZE TABLE events;

-- Prevent excessive merges for new tables
CREATE TABLE events_bulk (...)
ENGINE = MergeTree()
SETTINGS
    max_parts_in_total = 10000,  -- Allow more parts before forcing merge
    parts_to_throw_insert = 20000;  -- Fail inserts at this threshold

-- 1. Identify the problem query
SELECT
    query_id,
    user,
    elapsed,
    memory_usage / 1024 / 1024 / 1024 as memory_gb,
    query
FROM system.processes
WHERE memory_usage > 10000000000  -- > 10 GB
ORDER BY memory_usage DESC;

-- 2. Kill it immediately if necessary
KILL QUERY WHERE query_id = 'user_batch_12345';

-- 3. Investigate what went wrong in query log
SELECT
    query_duration_ms,
    memory_usage / 1024 / 1024 / 1024 as memory_gb,
    query
FROM system.query_log
WHERE user = 'batch_user'
  AND event_date >= today()
ORDER BY event_time DESC
LIMIT 10;

-- 4. Optimize the problematic query
-- Instead of: SELECT * FROM events GROUP BY user_id, event_type, device, os, browser, location
-- Use approximate: SELECT topK(1000)(user_id) FROM events

-- 1. Check disk space immediately
SELECT
    formatReadableSize(free_space) as free,
    formatReadableSize(total_space) as total,
    round(free_space / total_space * 100, 2) as free_percent
FROM system.disks;

-- 2. Identify largest tables
SELECT
    database,
    name as table,
    formatReadableSize(SUM(bytes_on_disk)) as size,
    COUNT() as parts_count
FROM system.parts
WHERE active = 1
  AND database NOT IN ('system', 'information_schema')
GROUP BY database, name
ORDER BY bytes_on_disk DESC
LIMIT 10;

-- 3. Drop old partitions from largest table
ALTER TABLE events DROP PARTITION '202301';
ALTER TABLE events DROP PARTITION '202302';
ALTER TABLE events DROP PARTITION '202303';

-- 4. Trigger optimization to merge parts and reduce size
OPTIMIZE TABLE events FINAL;

-- 5. Monitor recovery
SELECT formatReadableSize(free_space) as free
FROM system.disks;

-- 6. For permanent fix: add more disk capacity and adjust TTL
ALTER TABLE events MODIFY TTL timestamp + INTERVAL 90 DAY;

global:
  scrape_interval: 30s

scrape_configs:
  - job_name: 'clickhouse-metrics'
    static_configs:
      - targets: ['clickhouse1:9090', 'clickhouse2:9090', 'clickhouse3:9090']
      - labels: {job: 'clickhouse-cluster'}
    metrics_path: '/metrics'

-- Dashboard: Query Latency
SELECT
    avg(query_duration_ms) as avg_latency_ms,
    quantile(0.95)(query_duration_ms) as p95_latency_ms,
    quantile(0.99)(query_duration_ms) as p99_latency_ms
FROM system.query_log
WHERE event_date >= today()
  AND type = 'QueryFinish'
  AND query NOT LIKE '%system.%';

-- Dashboard: Insert Rate
SELECT
    toStartOfMinute(event_time) as minute,
    SUM(rows) as total_rows
FROM system.query_log
WHERE query LIKE 'INSERT%'
  AND event_date >= today()
GROUP BY minute
ORDER BY minute DESC;

-- Dashboard: Replication Health
SELECT
    database,
    table,
    absolute_delay as lag_seconds,
    active_replicas,
    total_replicas
FROM system.replicas;