ClickHouse: колоночная OLAP СУБД

Client (clickhouse-client, HTTP API, приложение)
  → ClickHouse Server
    → Database
      → Table (MergeTree engine)
        → Parts (куски данных на диске, отсортированные по ORDER BY)
          → Columns (каждый столбец хранится отдельно, сжатый)

OLTP (PostgreSQL, MySQL)          OLAP (ClickHouse)
─────────────────────────         ─────────────────────
Много мелких операций             Мало тяжёлых аналитических запросов
INSERT/UPDATE/DELETE по одной      Batch INSERT, миллионы строк за раз
Строковое хранение                Колоночное хранение
Индексы на каждый запрос          Sparse index + сканирование столбцов
Нормализация (3NF)                Денормализация (широкие таблицы)
Транзакции (ACID)                 Eventual consistency
Миллисекунды на точечный запрос   Миллисекунды на агрегацию миллиардов строк

# Ubuntu / Debian
sudo apt-get install -y apt-transport-https ca-certificates curl gnupg
curl -fsSL 'https://packages.clickhouse.com/rpm/lts/repodata/repomd.xml.key' | sudo gpg --dearmor -o /usr/share/keyrings/clickhouse-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/clickhouse-keyring.gpg] https://packages.clickhouse.com/deb stable main" | sudo tee /etc/apt/sources.list.d/clickhouse.list
sudo apt-get update
sudo apt-get install -y clickhouse-server clickhouse-client
sudo systemctl enable --now clickhouse-server

# macOS
brew install clickhouse

# Docker (самый простой способ для разработки)
docker run -d --name clickhouse \
  -p 8123:8123 \          # HTTP-интерфейс
  -p 9000:9000 \          # Native TCP
  -v ch_data:/var/lib/clickhouse \
  -v ch_logs:/var/log/clickhouse-server \
  clickhouse/clickhouse-server

# Docker Compose

# compose.yaml
services:
  clickhouse:
    image: clickhouse/clickhouse-server
    ports:
      - "8123:8123"
      - "9000:9000"
    volumes:
      - ch_data:/var/lib/clickhouse
      - ch_logs:/var/log/clickhouse-server
      - ./config.xml:/etc/clickhouse-server/config.d/custom.xml
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    ulimits:
      nofile:
        soft: 262144
        hard: 262144

volumes:
  ch_data:
  ch_logs:

# Встроенный клиент
clickhouse-client                              # localhost:9000
clickhouse-client -h host -u user --password pass -d mydb
clickhouse-client --query "SELECT 1"           # одноразовый запрос
clickhouse-client --multiquery < script.sql    # выполнить SQL-файл

# HTTP-интерфейс
curl 'http://localhost:8123/?query=SELECT+1'
curl 'http://localhost:8123/' --data-binary "SELECT count() FROM system.tables"
curl 'http://localhost:8123/?user=default&password=pass' --data-binary @query.sql

# Из Docker
docker exec -it clickhouse clickhouse-client

-- Базовая аналитическая таблица
CREATE TABLE events
(
    event_date  Date,
    event_time  DateTime,
    user_id     UInt64,
    event_type  LowCardinality(String),
    page_url    String,
    duration_ms UInt32,
    metadata    Map(String, String)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_date)        -- партиция по месяцу
ORDER BY (event_type, user_id, event_time)  -- сортировка и sparse index
TTL event_date + INTERVAL 6 MONTH       -- автоудаление через 6 месяцев
SETTINGS index_granularity = 8192;       -- строк в грануле (по умолчанию)

-- Таблица логов
CREATE TABLE logs
(
    timestamp   DateTime64(3),
    level       Enum8('DEBUG' = 0, 'INFO' = 1, 'WARN' = 2, 'ERROR' = 3),
    service     LowCardinality(String),
    message     String,
    trace_id    UUID,
    attrs       Map(String, String)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (service, level, timestamp)
TTL toDate(timestamp) + INTERVAL 30 DAY
SETTINGS
    index_granularity = 8192,
    ttl_only_drop_parts = 1;             -- удалять целые parts, а не строки

-- Правило: слева — столбцы с низкой кардинальностью, фильтры по равенству
--          справа — столбцы с высокой кардинальностью, диапазоны
ORDER BY (event_type, user_id, event_time)
--        ^low card     ^filter by =   ^range filter

-- Если большинство запросов: WHERE user_id = X AND event_time BETWEEN ...
ORDER BY (user_id, event_time)

-- Если запросы разные — компромисс или несколько materialized views

PARTITION BY toYYYYMM(event_date)        -- по месяцу (самый частый вариант)
PARTITION BY toYYYYMMDD(event_date)      -- по дню (если очень много данных)
PARTITION BY (region, toYYYYMM(date))    -- составной ключ

-- Управление партициями
SELECT partition, name, rows, bytes_on_disk
FROM system.parts
WHERE table = 'events' AND active;

-- Удалить партицию (мгновенно — удаляет файлы)
ALTER TABLE events DROP PARTITION '202401';

-- Отсоединить / присоединить (для бэкапа или переноса)
ALTER TABLE events DETACH PARTITION '202401';
ALTER TABLE events ATTACH PARTITION '202401';

-- Добавить столбец
ALTER TABLE events ADD COLUMN browser LowCardinality(String) DEFAULT '';

-- Удалить столбец
ALTER TABLE events DROP COLUMN browser;

-- Переименовать столбец
ALTER TABLE events RENAME COLUMN page_url TO url;

-- Изменить тип
ALTER TABLE events MODIFY COLUMN duration_ms UInt64;

-- Изменить TTL
ALTER TABLE events MODIFY TTL event_date + INTERVAL 1 YEAR;

-- Добавить/изменить дефолтное значение
ALTER TABLE events MODIFY COLUMN browser LowCardinality(String) DEFAULT 'unknown';

-- Комментарий
ALTER TABLE events COMMENT COLUMN user_id 'Internal user identifier';

-- Заморозить партицию (бэкап)
ALTER TABLE events FREEZE PARTITION '202501';

-- Дедупликация по ORDER BY ключу
-- При merge оставляет строку с максимальным ver
CREATE TABLE users
(
    id        UInt64,
    name      String,
    email     String,
    updated_at DateTime,
    ver       UInt64            -- версия
)
ENGINE = ReplacingMergeTree(ver)
ORDER BY id;

-- INSERT перезаписывает "логически" — физически старые строки удаляются при merge
INSERT INTO users VALUES (1, 'Alice', 'alice@old.com', now(), 1);
INSERT INTO users VALUES (1, 'Alice', 'alice@new.com', now(), 2);

-- Гарантированная дедупликация при чтении:
SELECT * FROM users FINAL WHERE id = 1;
-- FINAL — медленнее, применяет merge логику в query time

-- Или через подзапрос (часто эффективнее FINAL):
SELECT * FROM users
WHERE (id, ver) IN (SELECT id, max(ver) FROM users GROUP BY id);

-- Для materialized views с агрегатами
CREATE MATERIALIZED VIEW events_daily_mv
ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, event_type)
AS SELECT
    event_date,
    event_type,
    countState() AS events_count,          -- не count(), а countState()!
    uniqState(user_id) AS unique_users,
    avgState(duration_ms) AS avg_duration
FROM events
GROUP BY event_date, event_type;

-- Чтение — через Merge-функции
SELECT
    event_date,
    event_type,
    countMerge(events_count) AS events,
    uniqMerge(unique_users) AS users,
    avgMerge(avg_duration) AS avg_dur
FROM events_daily_mv
GROUP BY event_date, event_type
ORDER BY event_date DESC;

-- Значения
INSERT INTO events (event_date, event_time, user_id, event_type, page_url, duration_ms)
VALUES ('2025-03-15', '2025-03-15 14:30:00', 42, 'pageview', '/home', 1500);

-- Несколько строк (рекомендуемый способ — batch insert)
INSERT INTO events VALUES
    ('2025-03-15', '2025-03-15 14:30:00', 42, 'pageview', '/home', 1500, {}),
    ('2025-03-15', '2025-03-15 14:31:00', 42, 'click', '/products', 200, {}),
    ('2025-03-15', '2025-03-15 14:32:00', 43, 'pageview', '/home', 800, {});

-- INSERT из SELECT
INSERT INTO events_archive
SELECT * FROM events WHERE event_date < '2024-01-01';

-- INSERT из файла через HTTP
curl 'http://localhost:8123/?query=INSERT+INTO+events+FORMAT+CSV' --data-binary @events.csv
curl 'http://localhost:8123/?query=INSERT+INTO+events+FORMAT+JSONEachRow' --data-binary @events.jsonl

-- INSERT из stdin через клиент
clickhouse-client --query "INSERT INTO events FORMAT CSV" < events.csv
cat events.jsonl | clickhouse-client --query "INSERT INTO events FORMAT JSONEachRow"

-- Мутация UPDATE (перезаписывает parts)
ALTER TABLE events UPDATE duration_ms = 0 WHERE event_type = 'error';

-- Мутация DELETE
ALTER TABLE events DELETE WHERE event_date < '2023-01-01';

-- Лёгкое удаление (помечает, не перезаписывает — быстрее)
DELETE FROM events WHERE event_date < '2023-01-01';

-- Проверить прогресс мутаций
SELECT * FROM system.mutations WHERE is_done = 0;

-- Отменить мутацию
KILL MUTATION WHERE mutation_id = 'mutation_0.txt';

-- 1. TTL (лучший вариант для данных с ограниченным сроком жизни)
ALTER TABLE logs MODIFY TTL toDate(timestamp) + INTERVAL 90 DAY;

-- 2. DROP PARTITION (мгновенно, если данные партиционированы по нужному ключу)
ALTER TABLE events DROP PARTITION '202301';

-- 3. TRUNCATE (очистить всю таблицу)
TRUNCATE TABLE events;

SELECT * FROM events LIMIT 10;
SELECT event_type, count() FROM events GROUP BY event_type;

-- LIMIT BY — TOP N для каждой группы (уникальная фича ClickHouse)
SELECT user_id, event_type, event_time
FROM events
ORDER BY event_time DESC
LIMIT 3 BY user_id;                      -- последние 3 события для каждого пользователя

-- WITH — CTE (как в PostgreSQL)
WITH daily AS (
    SELECT event_date, count() AS cnt
    FROM events
    GROUP BY event_date
)
SELECT avg(cnt) FROM daily;

-- SAMPLE — случайная выборка (быстрый approximate на больших данных)
SELECT count() FROM events SAMPLE 0.1;   -- ~10% данных
SELECT avg(duration_ms) FROM events SAMPLE 1000000;  -- ~1M строк

-- Стандартные
SELECT
    count(),                              -- количество строк (без аргумента!)
    countIf(duration_ms > 1000),          -- условный count
    sum(duration_ms),
    avg(duration_ms),
    min(event_time),
    max(event_time),
    any(page_url),                        -- любое значение из группы (быстро)
    anyLast(page_url)                     -- последнее вставленное
FROM events;

-- Приближённые уникальные (быстро, ~2% погрешность)
SELECT uniq(user_id) FROM events;         -- HyperLogLog (основной)
SELECT uniqExact(user_id) FROM events;    -- точный (медленнее, больше памяти)
SELECT uniqCombined(user_id) FROM events; -- adaptive

-- Квантили
SELECT
    quantile(0.5)(duration_ms) AS median,
    quantile(0.95)(duration_ms) AS p95,
    quantile(0.99)(duration_ms) AS p99,
    quantiles(0.5, 0.95, 0.99)(duration_ms) AS quantiles_arr
FROM events;

-- Массивы из групп
SELECT
    event_type,
    groupArray(user_id) AS user_ids,                     -- все значения в массив
    groupArray(10)(user_id) AS first_10_users,           -- первые 10
    groupUniqArray(user_id) AS unique_user_ids,
    groupArraySorted(5)(duration_ms) AS top5_durations   -- топ 5 отсортированных
FROM events
GROUP BY event_type;

-- Гистограмма
SELECT histogram(10)(duration_ms) FROM events;

-- TopK — приближённые топ значения
SELECT topK(10)(page_url) FROM events;    -- топ 10 URL

-- -If комбинатор — условные агрегаты (работает с любой агрегатной функцией)
SELECT
    countIf(event_type = 'pageview') AS pageviews,
    countIf(event_type = 'click') AS clicks,
    uniqIf(user_id, event_type = 'purchase') AS buyers,
    avgIf(duration_ms, duration_ms > 0) AS avg_nonzero
FROM events;

-- -Array комбинатор — агрегат по массиву
SELECT sumArray([1, 2, 3]);               -- 6
SELECT groupUniqArrayArray(tags) FROM posts;  -- все уникальные теги

SELECT
    today(),                              -- текущая дата
    now(),                                -- текущее время
    yesterday(),

    -- Извлечение
    toYear(event_time),
    toMonth(event_time),
    toDayOfWeek(event_time),              -- 1=пн, 7=вс
    toHour(event_time),
    toStartOfDay(event_time),
    toStartOfWeek(event_time),
    toStartOfMonth(event_time),
    toStartOfInterval(event_time, INTERVAL 15 MINUTE),

    -- Форматирование
    formatDateTime(event_time, '%Y-%m-%d %H:%M:%S'),
    toString(event_date),

    -- Парсинг
    parseDateTimeBestEffort('2025-03-15T14:30:00Z'),
    toDateTime('2025-03-15 14:30:00'),
    toDate('2025-03-15'),

    -- Арифметика
    event_time + INTERVAL 1 DAY,
    dateDiff('day', start_date, end_date),
    dateAdd(DAY, 7, event_date),
    dateSub(MONTH, 1, event_date)
FROM events;

-- Генерация временного ряда (для заполнения пропусков)
WITH toDate('2025-01-01') AS start, toDate('2025-03-31') AS end
SELECT arrayJoin(
    arrayMap(x -> addDays(start, x), range(toUInt32(dateDiff('day', start, end) + 1)))
) AS date;

SELECT
    length('hello'),
    lower('Hello'),
    upper('hello'),
    trim('  hello  '),
    concat('a', 'b', 'c'),              -- или 'a' || 'b' || 'c'
    substring('hello world', 1, 5),
    replace('hello world', 'world', 'CH'),
    splitByChar(',', 'a,b,c'),           -- ['a', 'b', 'c']
    splitByString(', ', 'a, b, c'),
    extractAll('abc 123 def 456', '\\d+'),  -- ['123', '456']
    match(url, '/products/\\d+'),         -- regex match (boolean)
    extract(url, '/products/(\\d+)'),     -- regex extract (группа)
    like(email, '%@gmail.com'),
    notEmpty(name),
    multiSearchAnyCaseInsensitive(message, ['error', 'fail', 'exception']);

SELECT
    [1, 2, 3],                           -- литерал массива
    array(1, 2, 3),                       -- то же самое
    arrayJoin([1, 2, 3]),                 -- развернуть массив в строки (UNNEST)
    length([1, 2, 3]),                    -- 3
    has([1, 2, 3], 2),                    -- true
    indexOf([10, 20, 30], 20),            -- 2 (1-based)
    arrayConcat([1, 2], [3, 4]),          -- [1, 2, 3, 4]
    arrayDistinct([1, 1, 2, 3]),          -- [1, 2, 3]
    arraySort([3, 1, 2]),                 -- [1, 2, 3]
    arrayFilter(x -> x > 1, [1, 2, 3]),  -- [2, 3]
    arrayMap(x -> x * 2, [1, 2, 3]),     -- [2, 4, 6]
    arrayReduce('sum', [1, 2, 3]),        -- 6
    arrayZip([1, 2], ['a', 'b']);         -- [(1,'a'), (2,'b')]

-- arrayJoin — развернуть массив (одна из ключевых операций)
SELECT user_id, arrayJoin(tags) AS tag FROM posts;
-- user_id=1, tags=['sql','db'] → две строки: (1,'sql'), (1,'db')

SELECT
    map('key1', 'val1', 'key2', 'val2'),
    metadata['browser'],                  -- доступ по ключу
    mapKeys(metadata),                    -- все ключи
    mapValues(metadata),                  -- все значения
    mapContains(metadata, 'browser');     -- ключ существует?

-- Фильтр по Map
SELECT * FROM events WHERE metadata['browser'] = 'Chrome';

-- ClickHouse по умолчанию использует Hash Join
-- Правая таблица загружается в память!

-- Обычный JOIN
SELECT e.event_type, u.name
FROM events e
JOIN users u ON e.user_id = u.id;

-- LEFT JOIN
SELECT e.*, u.name
FROM events e
LEFT JOIN users u ON e.user_id = u.id;

-- JOIN с подсказкой алгоритма
SELECT *
FROM big_table a
JOIN small_table b ON a.key = b.key
SETTINGS join_algorithm = 'hash';        -- hash (default), partial_merge, full_sorting_merge, grace_hash

-- GLOBAL JOIN — для Distributed таблиц (правая таблица рассылается на все шарды)
SELECT *
FROM distributed_events e
GLOBAL JOIN users u ON e.user_id = u.id;

-- ASOF JOIN — присоединить ближайшее по времени значение
SELECT
    t.timestamp, t.symbol, t.price,
    q.bid, q.ask
FROM trades t
ASOF LEFT JOIN quotes q
    ON t.symbol = q.symbol AND t.timestamp >= q.timestamp;

CREATE DICTIONARY user_dict
(
    id        UInt64,
    name      String,
    email     String,
    role      String
)
PRIMARY KEY id
SOURCE(CLICKHOUSE(TABLE 'users' DB 'default'))
LAYOUT(HASHED())                         -- вся таблица в хешмапе в RAM
LIFETIME(MIN 300 MAX 600);               -- обновление каждые 5-10 минут

-- Использование (вместо JOIN)
SELECT
    user_id,
    dictGet('user_dict', 'name', user_id) AS user_name,
    dictGet('user_dict', 'role', user_id) AS user_role
FROM events;

-- Источники словарей
SOURCE(CLICKHOUSE(TABLE 'users' DB 'default'))
SOURCE(MYSQL(HOST 'db' PORT 3306 USER 'reader' PASSWORD 'pass' TABLE 'users' DB 'app'))
SOURCE(POSTGRESQL(HOST 'db' PORT 5432 USER 'reader' TABLE 'users' DB 'app'))
SOURCE(HTTP(URL 'https://api.example.com/users.csv' FORMAT 'CSV'))
SOURCE(FILE(PATH '/data/users.csv' FORMAT 'CSV'))

-- Layouts
LAYOUT(FLAT())                           -- массив (если ключ UInt64 и не разреженный)
LAYOUT(HASHED())                         -- хешмап (основной)
LAYOUT(CACHE(SIZE_IN_CELLS 100000))      -- кеш (для больших справочников, lazy loading)
LAYOUT(RANGE_HASHED())                   -- диапазонный (для данных с date_from/date_to)
LAYOUT(COMPLEX_KEY_HASHED())             -- составной ключ

-- Исходная таблица
CREATE TABLE raw_events
(
    timestamp DateTime,
    user_id   UInt64,
    event     LowCardinality(String),
    value     Float64
)
ENGINE = MergeTree()
ORDER BY (event, timestamp);

-- Целевая таблица для агрегатов
CREATE TABLE events_hourly
(
    hour      DateTime,
    event     LowCardinality(String),
    count     SimpleAggregateFunction(sum, UInt64),
    users     AggregateFunction(uniq, UInt64),
    total     SimpleAggregateFunction(sum, Float64),
    avg_val   AggregateFunction(avg, Float64)
)
ENGINE = AggregatingMergeTree()
ORDER BY (event, hour);

-- Materialized View — триггер на INSERT в raw_events
CREATE MATERIALIZED VIEW events_hourly_mv
TO events_hourly                         -- записывает результат в events_hourly
AS SELECT
    toStartOfHour(timestamp) AS hour,
    event,
    count() AS count,
    uniqState(user_id) AS users,
    sum(value) AS total,
    avgState(value) AS avg_val
FROM raw_events
GROUP BY hour, event;

-- Чтение из целевой таблицы
SELECT
    hour,
    event,
    sum(count) AS total_events,
    uniqMerge(users) AS unique_users,
    sum(total) AS total_value,
    avgMerge(avg_val) AS avg_value
FROM events_hourly
WHERE hour >= today()
GROUP BY hour, event
ORDER BY hour DESC;

INSERT INTO raw_events → MV автоматически агрегирует → INSERT INTO events_hourly
                         (происходит при каждом INSERT)

-- Вывод в разных форматах
SELECT * FROM events FORMAT JSON;
SELECT * FROM events FORMAT JSONEachRow;     -- одна строка JSON на запись (для streaming)
SELECT * FROM events FORMAT CSV;
SELECT * FROM events FORMAT CSVWithNames;
SELECT * FROM events FORMAT TSV;
SELECT * FROM events FORMAT Parquet;
SELECT * FROM events FORMAT Arrow;
SELECT * FROM events FORMAT Pretty;          -- для человека
SELECT * FROM events FORMAT Vertical;        -- столбцами

-- Вставка из разных форматов
INSERT INTO events FORMAT CSV
2025-03-15,2025-03-15 14:30:00,42,pageview,/home,1500,{}
;

INSERT INTO events FORMAT JSONEachRow
{"event_date": "2025-03-15", "event_time": "2025-03-15 14:30:00", "user_id": 42, "event_type": "pageview", "page_url": "/home", "duration_ms": 1500, "metadata": {}}
;

-- Файлы (на сервере)
SELECT * FROM file('data.csv', CSVWithNames);
SELECT * FROM file('logs/*.jsonl', JSONEachRow);

-- URL
SELECT * FROM url('https://example.com/data.csv', CSVWithNames);

-- S3
SELECT * FROM s3('https://bucket.s3.amazonaws.com/data/*.parquet', 'Parquet');

-- PostgreSQL
SELECT * FROM postgresql('pg:5432', 'mydb', 'users', 'reader', 'pass');

-- MySQL
SELECT * FROM mysql('mysql:3306', 'mydb', 'users', 'reader', 'pass');

-- Вставка из внешнего источника
INSERT INTO events
SELECT * FROM s3('s3://bucket/events/*.parquet', 'Parquet');

-- Вставка из PostgreSQL
INSERT INTO users
SELECT * FROM postgresql('pg:5432', 'mydb', 'users', 'reader', 'pass');

-- План запроса
EXPLAIN SELECT count() FROM events WHERE event_type = 'click';

-- Подробный план с pipeline
EXPLAIN PIPELINE SELECT event_type, count() FROM events GROUP BY event_type;

-- План с оценкой строк
EXPLAIN ESTIMATE SELECT * FROM events WHERE event_date = today();

-- Реальные метрики после выполнения
SELECT
    query,
    read_rows,
    read_bytes,
    result_rows,
    memory_usage,
    query_duration_ms
FROM system.query_log
WHERE type = 'QueryFinish'
ORDER BY event_time DESC
LIMIT 10;

-- Активные запросы
SELECT query_id, user, elapsed, query FROM system.processes;

-- Убить запрос
KILL QUERY WHERE query_id = 'xxx';

-- Размер таблиц
SELECT
    table,
    formatReadableSize(sum(bytes_on_disk)) AS size,
    sum(rows) AS rows,
    count() AS parts
FROM system.parts
WHERE active AND database = 'default'
GROUP BY table
ORDER BY sum(bytes_on_disk) DESC;

-- Размер столбцов (какой столбец сколько занимает)
SELECT
    column,
    formatReadableSize(sum(column_data_compressed_bytes)) AS compressed,
    formatReadableSize(sum(column_data_uncompressed_bytes)) AS uncompressed,
    round(sum(column_data_uncompressed_bytes) / sum(column_data_compressed_bytes), 2) AS ratio
FROM system.parts_columns
WHERE active AND table = 'events' AND database = 'default'
GROUP BY column
ORDER BY sum(column_data_compressed_bytes) DESC;

-- Статистика merge-ей
SELECT * FROM system.merges WHERE database = 'default';

-- Прогресс мутаций
SELECT * FROM system.mutations WHERE NOT is_done;

-- Логи запросов (медленные запросы)
SELECT
    query_duration_ms,
    read_rows,
    formatReadableSize(read_bytes) AS read,
    formatReadableSize(memory_usage) AS memory,
    query
FROM system.query_log
WHERE type = 'QueryFinish' AND query_duration_ms > 1000
ORDER BY query_duration_ms DESC
LIMIT 20;

-- Ошибки
SELECT * FROM system.query_log WHERE type = 'ExceptionWhileProcessing' ORDER BY event_time DESC LIMIT 10;

-- 1. Фильтр по ORDER BY ключу (использует sparse index)
-- ХОРОШО: event_type — первый столбец в ORDER BY
SELECT count() FROM events WHERE event_type = 'click';

-- ПЛОХО: duration_ms не в ORDER BY — полный скан столбца
SELECT count() FROM events WHERE duration_ms > 1000;

-- 2. PREWHERE — автоматическая оптимизация (или явная)
-- Сначала читает столбец фильтра, потом остальные только для подходящих строк
SELECT * FROM events PREWHERE event_type = 'click' WHERE duration_ms > 1000;
-- (обычно ClickHouse делает это автоматически)

-- 3. Избегай SELECT * — читай только нужные столбцы
SELECT event_type, count() FROM events GROUP BY event_type;  -- быстро
SELECT * FROM events;                                         -- медленно (все столбцы)

-- 4. Используй LowCardinality
-- До: String → 60 байт на значение
-- После: LowCardinality(String) → ~1-2 байта на значение (словарная кодировка)

-- 5. FINAL — избегай на больших таблицах
-- Вместо: SELECT * FROM replacing_table FINAL
-- Лучше:
SELECT * FROM replacing_table
WHERE (key, ver) IN (SELECT key, max(ver) FROM replacing_table GROUP BY key);

-- Дополнительные индексы (не заменяют ORDER BY, а дополняют)
ALTER TABLE events ADD INDEX idx_url page_url TYPE tokenbf_v1(32768, 3, 0) GRANULARITY 4;
ALTER TABLE events MATERIALIZE INDEX idx_url;

-- Типы индексов пропуска
-- minmax  — диапазон значений в грануле
-- set(N)  — множество уникальных значений (до N)
-- bloom_filter — вероятностный (подходит для равенства и IN)
-- tokenbf_v1  — bloom filter по токенам (для поиска подстрок)
-- ngrambf_v1  — bloom filter по n-граммам

CREATE TABLE logs
(
    timestamp DateTime,
    message   String,
    level     LowCardinality(String),
    INDEX idx_level level TYPE set(10) GRANULARITY 4,
    INDEX idx_message message TYPE tokenbf_v1(32768, 3, 0) GRANULARITY 4
)
ENGINE = MergeTree()
ORDER BY timestamp;

# CSV
clickhouse-client --query "INSERT INTO events FORMAT CSVWithNames" < events.csv

# JSON Lines
clickhouse-client --query "INSERT INTO events FORMAT JSONEachRow" < events.jsonl

# Parquet
clickhouse-client --query "INSERT INTO events FORMAT Parquet" < events.parquet

# С gzip на лету
gunzip -c events.csv.gz | clickhouse-client --query "INSERT INTO events FORMAT CSVWithNames"

# Экспорт
clickhouse-client --query "SELECT * FROM events FORMAT CSVWithNames" > events.csv
clickhouse-client --query "SELECT * FROM events FORMAT Parquet" > events.parquet
clickhouse-client --query "SELECT * FROM events FORMAT JSONEachRow" | gzip > events.jsonl.gz

# Вставка CSV
curl 'http://localhost:8123/?query=INSERT+INTO+events+FORMAT+CSVWithNames' --data-binary @events.csv

# Вставка JSON
curl 'http://localhost:8123/' --data-binary "INSERT INTO events FORMAT JSONEachRow $(cat events.jsonl)"

# Экспорт
curl 'http://localhost:8123/?query=SELECT+*+FROM+events+FORMAT+CSVWithNames' > events.csv

# Gzip
curl -H 'Content-Encoding: gzip' 'http://localhost:8123/?query=INSERT+INTO+events+FORMAT+CSV' --data-binary @events.csv.gz
curl -H 'Accept-Encoding: gzip' 'http://localhost:8123/?query=SELECT+*+FROM+events+FORMAT+CSV' | gunzip > events.csv

-- Из PostgreSQL
INSERT INTO events
SELECT * FROM postgresql('pg-host:5432', 'source_db', 'events', 'reader', 'pass');

-- Из MySQL
INSERT INTO users
SELECT * FROM mysql('mysql-host:3306', 'source_db', 'users', 'reader', 'pass');

-- Из S3
INSERT INTO events
SELECT * FROM s3('s3://bucket/data/events_*.parquet', 'key', 'secret', 'Parquet');

-- В S3
INSERT INTO FUNCTION s3('s3://bucket/export/events.parquet', 'key', 'secret', 'Parquet')
SELECT * FROM events WHERE event_date = today();

-- Между ClickHouse кластерами
INSERT INTO events
SELECT * FROM remote('other-ch:9000', 'default', 'events', 'reader', 'pass');

<!-- /etc/clickhouse-server/config.d/custom.xml -->
<clickhouse>
    <max_concurrent_queries>100</max_concurrent_queries>
    <max_connections>4096</max_connections>

    <merge_tree>
        <max_suspicious_broken_parts>5</max_suspicious_broken_parts>
    </merge_tree>
</clickhouse>

-- На уровне запроса
SELECT count() FROM events
SETTINGS max_threads = 4, max_memory_usage = 10000000000;

-- На уровне сессии
SET max_threads = 4;
SET max_memory_usage = 10000000000;       -- 10GB
SET max_execution_time = 60;              -- секунды
SET max_rows_to_read = 1000000000;
SET join_algorithm = 'grace_hash';        -- для больших JOIN

-- Полезные настройки
SET optimize_read_in_order = 1;           -- использовать ORDER BY таблицы
SET optimize_aggregation_in_order = 1;    -- GROUP BY по ORDER BY ключу
SET allow_experimental_lightweight_delete = 1;  -- лёгкие DELETE
SET input_format_allow_errors_ratio = 0.01;     -- допустить 1% ошибок при импорте

-- Retention (когортный анализ)
WITH first_events AS (
    SELECT user_id, min(event_date) AS cohort_date
    FROM events
    WHERE event_type = 'signup'
    GROUP BY user_id
)
SELECT
    cohort_date,
    count() AS cohort_size,
    countIf(dateDiff('day', cohort_date, return_date) BETWEEN 1 AND 1) AS day1,
    countIf(dateDiff('day', cohort_date, return_date) BETWEEN 7 AND 7) AS day7,
    countIf(dateDiff('day', cohort_date, return_date) BETWEEN 30 AND 30) AS day30
FROM first_events fe
LEFT JOIN (
    SELECT DISTINCT user_id, event_date AS return_date
    FROM events
) ret ON fe.user_id = ret.user_id
GROUP BY cohort_date
ORDER BY cohort_date;

-- Funnel (воронка)
SELECT
    count() AS step1_view,
    countIf(step2_time > 0) AS step2_cart,
    countIf(step3_time > 0) AS step3_purchase
FROM (
    SELECT
        user_id,
        min(if(event_type = 'page_view', event_time, NULL)) AS step1_time,
        min(if(event_type = 'add_to_cart' AND event_time > step1_time, event_time, NULL)) AS step2_time,
        min(if(event_type = 'purchase' AND event_time > step2_time, event_time, NULL)) AS step3_time
    FROM events
    WHERE event_date = today()
    GROUP BY user_id
);

-- windowFunnel — встроенная функция для воронок
SELECT
    level,
    count() AS users
FROM (
    SELECT
        user_id,
        windowFunnel(86400)(event_time, event_type = 'page_view', event_type = 'add_to_cart', event_type = 'purchase') AS level
    FROM events
    WHERE event_date >= today() - 7
    GROUP BY user_id
)
GROUP BY level
ORDER BY level;

-- Заполнение пропусков (WITH FILL)
SELECT
    toStartOfHour(event_time) AS hour,
    count() AS events
FROM events
WHERE event_date = today()
GROUP BY hour
ORDER BY hour WITH FILL
    FROM toStartOfDay(now())
    TO now()
    STEP INTERVAL 1 HOUR;

-- Скользящее среднее
SELECT
    event_date,
    events_count,
    avg(events_count) OVER (ORDER BY event_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS ma_7d
FROM (
    SELECT event_date, count() AS events_count
    FROM events
    GROUP BY event_date
)
ORDER BY event_date;

-- Rate of change (изменение в процентах)
SELECT
    event_date,
    events_count,
    events_count / lagInFrame(events_count) OVER (ORDER BY event_date) - 1 AS change_pct
FROM (
    SELECT event_date, count() AS events_count FROM events GROUP BY event_date
);

-- ReplacingMergeTree + FINAL
SELECT * FROM users FINAL;

-- argMax — взять значение столбца для строки с максимальным другим столбцом
SELECT
    user_id,
    argMax(name, updated_at) AS latest_name,
    argMax(email, updated_at) AS latest_email,
    max(updated_at) AS last_update
FROM users
GROUP BY user_id;

<!-- keeper_config.xml -->
<clickhouse>
    <keeper_server>
        <tcp_port>9181</tcp_port>
        <server_id>1</server_id>
        <raft_configuration>
            <server><id>1</id><hostname>keeper1</hostname><port>9234</port></server>
            <server><id>2</id><hostname>keeper2</hostname><port>9234</port></server>
            <server><id>3</id><hostname>keeper3</hostname><port>9234</port></server>
        </raft_configuration>
    </keeper_server>
</clickhouse>

-- Реплицированная таблица
CREATE TABLE events ON CLUSTER 'mycluster'
(
    event_date Date,
    event_time DateTime,
    user_id    UInt64,
    event_type LowCardinality(String),
    page_url   String
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}')
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_type, user_id, event_time);

-- Распределённая таблица (прокси ко всем шардам)
CREATE TABLE events_distributed ON CLUSTER 'mycluster'
AS events
ENGINE = Distributed('mycluster', 'default', 'events', user_id);
--                    cluster      database   table      sharding_key

-- INSERT в distributed → раскидывает по шардам по sharding_key
-- SELECT из distributed → запрос ко всем шардам → агрегация результатов

# Встроенный бэкап (ClickHouse 22.8+)
clickhouse-client --query "BACKUP TABLE events TO Disk('backups', 'events_backup')"
clickhouse-client --query "RESTORE TABLE events FROM Disk('backups', 'events_backup')"

# Бэкап в S3
clickhouse-client --query "BACKUP TABLE events TO S3('https://bucket.s3.amazonaws.com/backup/', 'key', 'secret')"

# clickhouse-backup (утилита)
clickhouse-backup create daily_backup
clickhouse-backup upload daily_backup
clickhouse-backup restore_remote daily_backup

DB::Exception: Too many parts (N). Merges are processing significantly slower than inserts.

-- Async insert — ClickHouse сам батчит мелкие вставки
SET async_insert = 1;
SET wait_for_async_insert = 0;

-- Ограничить память на запрос
SET max_memory_usage = 10000000000;       -- 10GB

-- Использовать внешнюю сортировку/агрегацию (на диске, медленнее)
SET max_bytes_before_external_sort = 5000000000;
SET max_bytes_before_external_group_by = 5000000000;

-- 1. Проверить, сколько строк читается
SELECT count() FROM events WHERE <conditions>;

-- 2. Проверить, попадает ли фильтр в ORDER BY ключ
EXPLAIN indexes = 1 SELECT ... FROM events WHERE ...;
-- Ищи "Selected Parts" и "Selected Granules" — чем меньше, тем лучше

-- 3. Посмотреть, какие столбцы самые тяжёлые
SELECT column, formatReadableSize(sum(column_data_compressed_bytes)) AS size
FROM system.parts_columns
WHERE table = 'events' AND active
GROUP BY column ORDER BY sum(column_data_compressed_bytes) DESC;

-- 4. Не используй SELECT * — укажи только нужные столбцы
-- 5. Добавь data skipping index, если фильтр не по ORDER BY
-- 6. Создай materialized view с преагрегацией

SELECT * FROM system.mutations WHERE NOT is_done AND table = 'events';
-- Если мутация висит — подождать или:
OPTIMIZE TABLE events FINAL;              -- форсировать merge (тяжёлая операция!)

-- Для ReplicatedMergeTree — встроенная дедупликация блоков
-- Для обычного MergeTree — нет автоматической дедупликации

-- Ручная дедупликация через ReplacingMergeTree + FINAL
-- Или через OPTIMIZE ... DEDUPLICATE
OPTIMIZE TABLE events DEDUPLICATE BY user_id, event_time;