ClickHouse — Конспект

Правило: ClickHouse = много данных + быстрое чтение + пакетная запись

Таблица: id | name | balance
В файловой системе:
  - id.bin    (только id)
  - name.bin  (только name)
  - balance.bin (только balance)

Пользователь
    ↓
Прокси / Load Balancer (порт 8123)
    ↓             ↓
ClickHouse1    ClickHouse2    (порт 9000 — нативный)
  [Shard 1]    [Shard 2]
  [Replica]    [Replica]

-- Посмотреть состав кластера
SELECT * FROM system.clusters;

-- Создать базу данных на всём кластере
CREATE DATABASE IF NOT EXISTS mydb ON CLUSTER '{cluster}';
 
-- Создать таблицу на всём кластере
CREATE TABLE IF NOT EXISTS mydb.events_local ON CLUSTER '{cluster}'
(
    id     UInt64,
    dt     DateTime,
    event  String
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(dt)
ORDER BY (id, dt);

events_local  — локальная таблица (MergeTree), хранит данные
events        — distributed-таблица (ссылка на events_local)

-- Целые числа (со знаком): Int8, Int16, Int32, Int64, Int128, Int256
i  Int8    -- -128 до 127
i  Int32   -- -2 млрд до 2 млрд
 
-- Беззнаковые: UInt8, UInt16, UInt32, UInt64, UInt128, UInt256
ui UInt8   -- 0 до 255
ui UInt64  -- 0 до 18 квинтиллионов
 
-- Вещественные (для математики, НЕ для финансов!)
fl Float32
fl Float64
 
-- Точная дробная арифметика (для финансов)
dc Decimal(9, 4)    -- 9 цифр всего, 4 после запятой
dc Decimal32(4)     -- алиас
dc Decimal64(4)

st  String          -- переменная длина, UTF-8
fst FixedString(5)  -- ровно N байт, быстрее для коротких фиксированных строк

UID UUID            -- генерация: generateUUIDv4()
ip4 IPv4            -- '192.168.1.1'
ip6 IPv6            -- '2001:db8::1'

dt  Date            -- только дата: '2025-04-30'
dtm DateTime        -- дата + время (с секундами)
dtm64 DateTime64(3) -- дата + время с миллисекундами (параметр — точность)
 
-- Полезные функции
toDate('2025-01-01')
toDateTime('2025-01-01 12:00:00')
now()               -- текущее время
today()             -- сегодняшняя дата
toYYYYMM(dt)        -- числовое представление год+месяц: 202501

-- Массив
ar Array(UInt32)
-- Обращение: ar[1], ar.size0 (размер)
 
-- Кортеж (Tuple)
tu Tuple(UInt32, String)
 
-- Вложенная структура
ns Nested(col1 String, col2 UInt32, col3 Float32)
-- Обращение: ns.col1, ns.col2
 
-- Map
mp Map(String, UInt32)
-- Обращение: mp['key1']
 
-- Enum (экономит место и обеспечивает валидацию)
en Enum('оплачено' = 1, 'не оплачено' = 0)

-- Nullable — разрешает NULL-значения (добавляет накладные расходы)
a Nullable(UInt32)
 
-- LowCardinality — словарное кодирование для низкой кардинальности (статусы, коды, категории)
b LowCardinality(String)  -- быстрее и компактнее для повторяющихся значений

-- Синтаксис ::
SELECT '42'::Int64
 
-- CAST
SELECT CAST('42' AS Int8)
 
-- Функции с обработкой ошибок
toInt64OrNull('abc')      -- вернёт NULL при ошибке
toInt8OrZero('abc')       -- вернёт 0 при ошибке
toInt8OrDefault('abc', -1) -- вернёт -1 при ошибке
 
-- Беззнаковые: выход за пределы → значение по модулю диапазона
toUInt8(-1)  -- вернёт 255
toUInt8(256) -- вернёт 0

CREATE TABLE IF NOT EXISTS mydb.my_table ON CLUSTER '{cluster}'
(
    id   UInt64,
    name String,
    dt   DateTime
)
ENGINE = MergeTree              -- ОБЯЗАТЕЛЬНО указывать движок
PRIMARY KEY (id)                -- необязательно; по умолчанию = ORDER BY
ORDER BY (id, dt)               -- ОБЯЗАТЕЛЬНО для MergeTree; определяет физический порядок данных
PARTITION BY toYYYYMM(dt)       -- необязательно; разбивает данные на секции
TTL dt + INTERVAL 1 MONTH DELETE; -- необязательно; автоудаление устаревших данных

CREATE TABLE mydb.test_fields
(
    col_default    UInt64  DEFAULT 42,                   -- значение по умолчанию при вставке
    col_materialized UInt64 MATERIALIZED col_default*33, -- вычисляется автоматически, нельзя вставить явно
    col_alias      UInt64  ALIAS col_default + 1,        -- не хранится физически; вычисляется при чтении
    col_codec      String  CODEC(ZSTD(10)),              -- алгоритм сжатия (ZSTD, LZ4, Delta, Gorilla...)
    col_comment    Date    COMMENT 'Дата создания',      -- комментарий к колонке
    col_ttl        UInt64  DEFAULT 10 TTL col_comment + INTERVAL 5 DAY -- TTL для отдельной колонки
)
ENGINE = MergeTree
ORDER BY (col_default);

-- Ошибка!
INSERT INTO product_sales (sale_id, total) VALUES (1, 100);
-- Cannot insert column total, because it is MATERIALIZED column.

-- TTL для всей строки (на уровне таблицы)
ENGINE = MergeTree
ORDER BY id
TTL dt + INTERVAL 1 MONTH DELETE;
 
-- TTL для отдельной колонки (значение сбрасывается до DEFAULT)
col_ttl UInt64 DEFAULT 0 TTL dt + INTERVAL 10 DAY
 
-- Принудительное применение TTL (без ожидания фонового merge)
OPTIMIZE TABLE my_table FINAL;

-- Данные вставляются в порядке insert (рандомный порядок)
ORDER BY tuple()

DESCRIBE TABLE mydb.my_table;  -- показывает колонки, типы и комментарии

CREATE TABLE mydb.table_range (...)
ENGINE = MergeTree
PARTITION BY
    CASE
        WHEN id < 10000 THEN 'range_1'
        WHEN id < 20000 THEN 'range_2'
        ELSE 'range_3'
    END
ORDER BY id;

CREATE TABLE mydb.table_interval (...)
ENGINE = MergeTree
PARTITION BY toYYYYMM(sale_date)  -- партиция на каждый месяц
ORDER BY id;

CREATE TABLE mydb.table_hash (...)
ENGINE = MergeTree
PARTITION BY cityHash64(user_id) % 10  -- 10 равномерных партиций
ORDER BY user_id;

CREATE TABLE mydb.table_composite (...)
ENGINE = MergeTree
PARTITION BY (toYYYYMM(order_date), customer_id % 10)
ORDER BY order_id;

-- Через виртуальную колонку _part
SELECT _part, count(*) FROM mydb.table_interval GROUP BY _part;
 
-- Через системную таблицу
SELECT * FROM system.parts WHERE table = 'my_table';

OPTIMIZE TABLE mydb.my_table FINAL;
-- FINAL — дожидается полного слияния (без него MergeTree делает одно слияние шага)

20250101_1_1_0/
├── primary.idx       — разреженный первичный индекс (подгружается в RAM)
├── id.bin            — сжатые данные колонки id
├── id.mrk2           — marks-файл колонки id
├── name.bin          — сжатые данные колонки name
├── name.mrk2         — marks-файл колонки name
├── dt.bin            — сжатые данные колонки dt
├── dt.mrk2           — marks-файл колонки dt
└── checksums.txt     — контрольные суммы

-- Посмотреть все активные парты таблицы
SELECT name, rows, bytes_on_disk, active
FROM system.parts
WHERE table = 'my_table' AND active = 1
ORDER BY name;

Данные:            [строки 1-8192] [строки 8193-16384] [строки 16385-24576] ...
Индекс:               ↑ метка 0        ↑ метка 1             ↑ метка 2

CREATE TABLE ... ENGINE = MergeTree
SETTINGS index_granularity = 8192;  -- значение по умолчанию

ORDER BY (region, user_id, dt)
 
-- Эффективный запрос (использует индекс):
WHERE region = 'Moscow'
 
-- Менее эффективный (частично использует индекс):
WHERE user_id = 12345
 
-- НЕ использует первичный индекс (нет первого поля):
WHERE dt > '2024-01-01'

-- Добавить индекс к существующей таблице
ALTER TABLE mydb.events
    ADD INDEX idx_event_type (event_type) TYPE set(100) GRANULARITY 4;
 
-- Применить индекс к уже существующим данным (без этого индекс работает только для новых данных)
ALTER TABLE mydb.events MATERIALIZE INDEX idx_event_type;
 
-- Удалить индекс
ALTER TABLE mydb.events DROP INDEX idx_event_type;
 
-- Объявить индекс при создании таблицы
CREATE TABLE mydb.events (
    id       UInt64,
    dt       DateTime,
    event    String,
    user_id  UInt64,
    amount   Float32,
    INDEX idx_event event TYPE set(100) GRANULARITY 4,
    INDEX idx_amount amount TYPE minmax GRANULARITY 2
)
ENGINE = MergeTree
ORDER BY (id, dt);

INDEX idx_amount amount TYPE minmax GRANULARITY 4
 
-- Эффективно использует minmax:
WHERE amount > 1000
WHERE amount BETWEEN 100 AND 500
WHERE dt >= '2024-01-01'

INDEX idx_event event TYPE set(100) GRANULARITY 4
-- 100 — максимальное количество уникальных значений; 0 = без ограничений
 
-- Эффективно использует set:
WHERE event = 'click'
WHERE event IN ('click', 'view')

-- bloom_filter: для равенства и IN
INDEX idx_user_id user_id TYPE bloom_filter(0.01) GRANULARITY 4
-- 0.01 — допустимая вероятность ложного срабатывания (1%)
 
WHERE user_id = 12345          -- использует индекс
WHERE user_id IN (1, 2, 3)    -- использует индекс
 
-- ngrambf_v1: для LIKE-запросов (n=3 — триграммы)
INDEX idx_name name TYPE ngrambf_v1(3, 256, 2, 0) GRANULARITY 4
 
WHERE name LIKE '%москва%'     -- использует индекс
WHERE hasToken(name, 'click')  -- использует индекс
 
-- tokenbf_v1: для полнотекстового поиска по словам
INDEX idx_desc description TYPE tokenbf_v1(256, 2, 0) GRANULARITY 4
 
WHERE hasToken(description, 'payment')  -- использует индекс

CREATE TABLE mydb.events_local
(
    id    UInt64,
    dt    DateTime,
    event String,
    value Float64
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(dt)
ORDER BY (id, dt);

CREATE TABLE mydb.daily_stats
(
    user_id  UInt32,
    dt       DateTime,
    sessions UInt32,      -- будет суммироваться
    clicks   UInt64,      -- будет суммироваться
    extra    String       -- не суммируется (оставляется произвольное значение)
)
ENGINE = SummingMergeTree(sessions, clicks)  -- если не указать — все числовые
ORDER BY (user_id)
PARTITION BY toYYYYMM(dt);

-- Без FINAL — данные могут быть несуммированы
SELECT user_id, SUM(sessions) FROM mydb.daily_stats GROUP BY user_id;
 
-- FINAL — принудительное логическое слияние перед выдачей
SELECT user_id, sessions FROM mydb.daily_stats FINAL;
 
-- Ручное физическое слияние
OPTIMIZE TABLE mydb.daily_stats FINAL;

CREATE TABLE mydb.simple_agg
(
    id      UInt64,
    val_sum SimpleAggregateFunction(sum, UInt64),
    val_max SimpleAggregateFunction(max, UInt32)
)
ENGINE = AggregatingMergeTree
ORDER BY id;
 
-- Вставка обычная
INSERT INTO mydb.simple_agg SELECT 1, number, number FROM numbers(10);
 
-- Чтение: нужно явно применить агрегат
SELECT id, sum(val_sum), max(val_max) FROM mydb.simple_agg GROUP BY id;
-- или через FINAL
SELECT id, val_sum, val_max FROM mydb.simple_agg FINAL;

CREATE TABLE mydb.complex_agg
(
    id         UInt64,
    val_uniq   AggregateFunction(uniq, UInt64),
    val_quant  AggregateFunction(quantiles(0.5, 0.9), UInt64)
)
ENGINE = AggregatingMergeTree
ORDER BY id;
 
-- Вставка через State-комбинатор
INSERT INTO mydb.complex_agg
SELECT 1, uniqState(number), quantilesState(0.5, 0.9)(number)
FROM numbers(100);
 
-- Чтение через Merge-комбинатор
SELECT id, uniqMerge(val_uniq), quantilesMerge(0.5, 0.9)(val_quant)
FROM mydb.complex_agg
GROUP BY id;

-- Без версии (оставляется любая из дублей)
CREATE TABLE mydb.users
(
    id  UInt32,
    dt  Date,
    val UInt32
)
ENGINE = ReplacingMergeTree
ORDER BY (id, dt);
 
-- С версией (оставляется запись с максимальным значением dt)
CREATE TABLE mydb.users_versioned
(
    id  UInt32,
    dt  Date,   -- версионная колонка (может быть числовой)
    val UInt32
)
ENGINE = ReplacingMergeTree(dt)
ORDER BY (id)
PARTITION BY toYYYYMM(dt);

CREATE TABLE mydb.books_progress
(
    user_id UInt64,
    book_id UInt64,
    page    UInt8,
    Sign    Int8   -- только +1 или -1
)
ENGINE = CollapsingMergeTree(Sign)
ORDER BY (user_id, book_id);

-- Пользователь прочитал 10 страниц
INSERT INTO mydb.books_progress VALUES (1, 42, 10, 1);
 
-- Пользователь прочитал ещё 5 страниц (сначала отменяем, потом вставляем новое)
INSERT INTO mydb.books_progress VALUES (1, 42, 10, -1);  -- отмена
INSERT INTO mydb.books_progress VALUES (1, 42, 15, 1);   -- новое состояние

ENGINE = ReplicatedReplacingMergeTree(
    '/clickhouse/tables/{cluster}/{shard}/events',  -- ZK-путь
    '{replica}',                                    -- имя реплики
    dt                                              -- версионная колонка ReplacingMergeTree
)
ORDER BY (id);

-- ПРАВИЛЬНО: очищает данные И метаданные ZooKeeper
DROP TABLE mydb.events_local ON CLUSTER '{cluster}' SYNC;
 
-- НЕПРАВИЛЬНО: данные удалятся, но ZooKeeper-путь останется
DROP TABLE mydb.events_local ON CLUSTER '{cluster}';

-- Рекомендация для аналитиков (работа днём):
SELECT * FROM mydb.events FINAL;
 
-- Рекомендация для ETL (ночная загрузка, перед началом рабочего дня):
OPTIMIZE TABLE mydb.events FINAL;

-- Вставляем 500 млн строк, соединение обрывается
INSERT INTO mydb.events SELECT ... FROM numbers(500000000);
-- Результат: в таблице может оказаться, например, 337 млн строк

-- Паттерн идемпотентной загрузки через ReplacingMergeTree:
INSERT INTO mydb.events ...;  -- запускаем снова при сбое
OPTIMIZE TABLE mydb.events FINAL;  -- убираем дубли

CREATE TABLE mydb.events (
    id       UInt64,
    -- ... 100 других колонок ...
    row_hash UInt64 MATERIALIZED cityHash64(id, col1, col2, ...)  -- хэш всей строки
)
ENGINE = ReplacingMergeTree
ORDER BY (id, row_hash);  -- row_hash обеспечивает уникальность всей строки

-- При вставке вычисляем хэш заранее и передаём как колонку
INSERT INTO mydb.events (id, ..., row_hash) VALUES (..., cityHash64(...));

ENGINE = Log

ENGINE = File(CSV)  -- или TSV, JSON, Parquet и др.

ENGINE = Memory

CREATE TABLE mydb.events_buffer (...)
ENGINE = Buffer(
    mydb,         -- база данных целевой таблицы
    events_local, -- целевая таблица
    16,           -- параллелизм (рекомендуется 16)
    30,           -- min время сброса (секунды)
    60,           -- max время сброса (секунды)
    5,            -- min строк для сброса
    100,          -- max строк для сброса
    10000,        -- min байт для сброса
    1000000       -- max байт для сброса
);

CREATE TABLE mydb.blocked_users (id UInt32) ENGINE = Set SETTINGS persistent = 1;
INSERT INTO mydb.blocked_users SELECT number FROM numbers(100);
 
-- Использование: только в WHERE ... IN
SELECT * FROM mydb.orders WHERE user_id IN mydb.blocked_users;

CREATE TABLE mydb.test_data
(
    id  UInt32,
    val String,
    dt  Date,
    a   Float32,
    b   UUID,
    g   Array(UInt32)
)
ENGINE = GenerateRandom;
 
-- Каждый SELECT генерирует случайные данные
SELECT * FROM mydb.test_data LIMIT 100;

CREATE TABLE mydb.pg_orders (...)
ENGINE = PostgreSQL(
    'postgres_host:5432',  -- хост:порт
    'prod_db',             -- имя БД в Postgres
    'orders',              -- таблица в Postgres
    'user',                -- логин
    'password',            -- пароль
    'public'               -- схема (необязательно)
);
 
-- SELECT автоматически пушится в Postgres (pushdown)
SELECT * FROM mydb.pg_orders WHERE status = 'paid';

-- 1. Таблица-приёмник (куда реально пишутся данные)
CREATE TABLE mydb.events_local (...) ENGINE = MergeTree ...;
 
-- 2. Таблица-источник Kafka
CREATE TABLE mydb.kafka_events (id UInt32, event String)
ENGINE = Kafka
SETTINGS
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list  = 'mydb.events.topic',
    kafka_group_name  = 'clickhouse_consumer_group',
    kafka_format      = 'JSONEachRow';
 
-- 3. Материализованное представление — связывает Kafka с реальной таблицей
CREATE MATERIALIZED VIEW mydb.kafka_mv TO mydb.events_local AS
SELECT id, event FROM mydb.kafka_events;
 
-- Отправка данных в Kafka
INSERT INTO mydb.kafka_events SELECT number, 'click' FROM numbers(30);

-- 1. Принимаем сырой JSON как одну строку
CREATE TABLE mydb.kafka_stage (json_kafka String)
ENGINE = Kafka
SETTINGS
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list  = 'source.public.order_events',
    kafka_group_name  = 'clickhouse_consumer_group',
    kafka_format      = 'JSONAsString';   -- весь JSON-объект → одна строка
 
-- 2. Таблица-хранилище (промежуточная или финальная)
CREATE TABLE mydb.orders_stage (json_kafka String) ENGINE = Log;
 
-- 3. Materialized View: данные из Kafka → хранилище
CREATE MATERIALIZED VIEW mydb.kafka_mv TO mydb.orders_stage AS
    SELECT json_kafka FROM mydb.kafka_stage;
 
-- 4. Чтение и разбор JSON при SELECT
SELECT
    JSONExtractInt(JSONExtractString(json_kafka, 'before'), 'id')           AS before_id,
    JSONExtractString(JSONExtractString(json_kafka, 'before'), 'status')    AS before_status,
    JSONExtractInt(JSONExtractString(json_kafka, 'after'),  'id')           AS after_id,
    JSONExtractString(JSONExtractString(json_kafka, 'after'),  'status')    AS after_status
FROM mydb.orders_stage;

-- JSONExtract<Type>(<json_string>, <key>)
JSONExtractInt(json, 'id')           -- Int
JSONExtractString(json, 'name')      -- String
JSONExtractFloat(json, 'amount')     -- Float
JSONExtractUInt(json, 'ts')          -- UInt
JSONExtractBool(json, 'active')      -- Bool
JSONExtractRaw(json, 'nested_obj')   -- сырой JSON-фрагмент (строка)
 
-- Вложенные поля: сначала вытащить объект, потом поле
JSONExtractString(JSONExtractRaw(json, 'address'), 'city')
 
-- simpleJSON* — более быстрая, но менее строгая альтернатива
simpleJSONExtractString(json, 'name')
simpleJSONExtractRaw(json, 'nested')

-- По умолчанию SELECT из Kafka-таблицы запрещён (данные потребляются)
-- Разрешить временно для отладки:
SET stream_like_engine_allow_direct_select = 1;
SELECT * FROM mydb.kafka_stage LIMIT 5;

-- Локальная таблица (реальное хранение)
CREATE TABLE mydb.events_local ON CLUSTER '{cluster}'
(
    id    UInt64,
    dt    DateTime,
    event String
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(dt)
ORDER BY (id, dt);
 
-- Distributed-таблица (ссылка)
CREATE TABLE mydb.events ON CLUSTER '{cluster}'
(
    id    UInt64,
    dt    DateTime,
    event String
)
ENGINE = Distributed(
    '{cluster}',      -- имя кластера
    mydb,             -- база данных локальной таблицы
    events_local,     -- имя локальной таблицы
    rand()            -- ключ шардирования (как распределять данные)
);

-- rand() — случайное равномерное распределение (для тестов, плохо для JOIN)
ENGINE = Distributed('{cluster}', db, table_local, rand())
 
-- По колонке — данные с одинаковым user_id попадут на один шард (хорошо для JOIN)
ENGINE = Distributed('{cluster}', db, table_local, user_id)
 
-- halfMD5 — хеш от нескольких колонок (рекомендуется на production)
ENGINE = Distributed('{cluster}', db, table_local, halfMD5(user_id, region))

-- Через distributed-таблицу (данные автоматически распределяются по шардам)
INSERT INTO mydb.events VALUES (...);
 
-- НЕ делать: прямая вставка в локальную таблицу попадёт только на один сервер
INSERT INTO mydb.events_local VALUES (...);  -- данные будут только на этом сервере

-- Через distributed — агрегирует данные со всех шардов
SELECT count() FROM mydb.events;
 
-- hostname() — видно, с каких серверов пришли данные
SELECT hostname(), count() FROM mydb.events GROUP BY hostname();
 
-- _part — видно партиции
SELECT _part, count() FROM mydb.events GROUP BY _part;

-- Порядок строк непредсказуем — данные с первого шарда, потом со второго (или наоборот)
SELECT * FROM mydb.events;
 
-- Правильно: если важен порядок — добавить ORDER BY
SELECT * FROM mydb.events ORDER BY dt DESC LIMIT 100;

CREATE TABLE mydb.events_local ON CLUSTER '{cluster}'
(
    id    UInt64,
    dt    DateTime,
    event LowCardinality(String)
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{cluster}/{shard}/mydb/events',  -- путь в ZooKeeper
    '{replica}'                                           -- имя реплики
)
PARTITION BY toDate(dt)
ORDER BY (id);

-- Шаг 1: локальная реплицируемая таблица
CREATE TABLE mydb.events ON CLUSTER '{cluster}'
(
    time DateTime,
    uid  Int64,
    type LowCardinality(String)
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{cluster}/{shard}/events', '{replica}')
PARTITION BY toDate(time)
ORDER BY (uid);
 
-- Шаг 2: distributed-таблица с таким же именем в другой БД (или с суффиксом _distr)
-- Удобный синтаксис: AS <table> копирует схему колонок
CREATE TABLE mydb.events_distr ON CLUSTER '{cluster}'
AS mydb.events                                           -- копирует структуру колонок
ENGINE = Distributed('{cluster}', mydb, events, uid);   -- шардирование по uid

Пользователь/Аналитик
        ↓
  events (Distributed)        ← единая точка входа
   /              \
events_local     events_local  ← физические данные на шардах
[Shard 1]        [Shard 2]
[Replica 1]      [Replica 1]
[Replica 2]      [Replica 2]

SELECT
    count(),                    -- количество строк
    count(id),                  -- количество непустых значений
    sum(amount),                -- сумма
    avg(price),                 -- среднее
    min(dt), max(dt),           -- минимум/максимум
    uniq(user_id),              -- приближённое число уникальных (аналог COUNT DISTINCT)
    countDistinct(user_id),     -- точное число уникальных
    groupArray(event),          -- собрать все значения в массив
    quantiles(0.5, 0.9)(price)  -- квантили
FROM mydb.events;

SELECT
    countIf(user_id, device = 'mobile')         -- количество мобильных пользователей
    sumIf(amount, status = 'paid')               -- сумма только оплаченных
    avgIf(view_time, page = 'home')              -- среднее время только на главной
    uniqIf(user_id, device = 'web')              -- уникальные пользователи с веба
    groupArrayIf(page, device = 'web')           -- список страниц только с веба
    countDistinctIf(user_id, device = 'mobile')  -- уникальные мобильные
FROM mydb.page_views;

SELECT
    device,
    uniq(user_id) AS unique_users,
    avgIf(view_time, page = 'home') AS avg_home_time,
    groupArrayIf(page, user_id = 2 AND device = 'web') AS web_pages
FROM mydb.page_views
GROUP BY device;

-- INNER JOIN
SELECT a.id, a.name, b.amount
FROM mydb.users a
INNER JOIN mydb.orders b ON a.id = b.user_id;
 
-- LEFT JOIN
SELECT a.id, a.name, b.amount
FROM mydb.users a
LEFT JOIN mydb.orders b ON a.id = b.user_id;
 
-- FULL OUTER JOIN
SELECT ...
FROM table1 FULL OUTER JOIN table2 ON ...;
 
-- CROSS JOIN
SELECT ... FROM table1 CROSS JOIN table2;

-- Правильно: маленький справочник справа
SELECT events.*, users.name
FROM mydb.events
LEFT JOIN mydb.users ON events.user_id = users.id;  -- users меньше events

-- Обе таблицы шардированы по user_id
-- JOIN по user_id будет выполнен локально на каждом шарде без передачи данных по сети
ENGINE = Distributed('{cluster}', db, users_local, user_id)
ENGINE = Distributed('{cluster}', db, orders_local, user_id)

-- Создать таблицу-джойн для многократного использования
CREATE TABLE mydb.user_dict (id UInt32, name String)
ENGINE = Join(ANY, LEFT, id);  -- тип джойна и ключ
 
-- Заполнить
INSERT INTO mydb.user_dict VALUES ...;
 
-- Использовать в запросе
SELECT id, joinGet('mydb.user_dict', 'name', user_id)
FROM mydb.events;

-- Обе таблицы Distributed с РАЗНЫМИ ключами шардирования
-- tabl_join_1 шардирована по id1
-- tabl_join_2 шардирована по id2
 
-- ЭТО ВЕРНЁТ ОШИБКУ по умолчанию:
SELECT *
FROM mydb.tabl_join_1 AS t1
JOIN mydb.tabl_join_2 AS t2 ON t1.id1 = t2.id1;
-- Double-distributed IN/JOIN subqueries is denied...

SELECT *
FROM mydb.tabl_join_1 AS t1
GLOBAL JOIN mydb.tabl_join_2 AS t2 ON t1.id1 = t2.id1;

SET distributed_product_mode = 'local';  -- по умолчанию 'deny'
 
SELECT *
FROM mydb.tabl_join_1 AS t1
JOIN mydb.tabl_join_2 AS t2 ON t1.id1 = t2.id1;

-- Таблица 1: шардирование по user_id
ENGINE = Distributed('{cluster}', db, orders_local, user_id)
 
-- Таблица 2: шардирование по user_id (тому же!)
ENGINE = Distributed('{cluster}', db, users_local, user_id)
 
-- Теперь JOIN по user_id выполняется локально на каждом шарде
SET distributed_product_mode = 'local';
SELECT * FROM mydb.orders t1
JOIN mydb.users t2 ON t1.user_id = t2.user_id;

SELECT T_A.k, T_A.ts, T_B.ts, T_A.a, T_B.b
FROM
    -- Левая таблица: события с временными метками
    (
        SELECT number AS k,
               toDateTime('2020-10-10 10:30:00') + number * 100 AS ts,
               number * 10 AS a
        FROM system.numbers LIMIT 5
    ) AS T_A
ASOF JOIN
    -- Правая таблица: снапшоты/цены/состояния с временными метками
    (
        SELECT number AS k,
               toDateTime('2020-10-10 10:00:00') + number * 100 AS ts,
               number * 100 AS b
        FROM system.numbers LIMIT 5
        UNION ALL
        SELECT number AS k,
               toDateTime('2020-10-10 11:00:00') + number * 100 AS ts,
               number * 1000 AS b
        FROM system.numbers LIMIT 5
    ) AS T_B
ON T_A.k = T_B.k AND T_A.ts >= T_B.ts;
-- Для каждой строки T_A найти строку T_B с максимальным ts <= T_A.ts

SELECT
    e.event_name,
    count() AS login_count
FROM user_login AS l
ASOF JOIN event AS e
    ON l.login_time >= e.start_time
WHERE l.login_time <= e.end_time
GROUP BY e.event_name;
 
-- Альтернатива без ASOF:
SELECT e.event_name, count() AS logins
FROM user_login l
JOIN event e ON l.login_time BETWEEN e.start_time AND e.end_time
GROUP BY e.event_name;

-- Хорошо: часто используемые в WHERE поля первыми
ORDER BY (region, user_id, dt)
 
-- Плохо: слишком много полей — замедляет вставки и мержи
ORDER BY (a, b, c, d, e, f)  -- не более 4 полей

-- Хорошо: по дате с помощью toYYYYMM — стандарт для временных рядов
PARTITION BY toYYYYMM(dt)
 
-- Осторожно: слишком много мелких партиций = много файлов
PARTITION BY toYYYYMMDD(dt)  -- ОК только если данных много в сутки
 
-- Плохо: партиция по высококардинальному полю
PARTITION BY user_id  -- миллионы партиций — катастрофа

-- Production-рекомендация: halfMD5 от поля, по которому чаще всего GROUP BY / JOIN
ENGINE = Distributed('{cluster}', db, tbl_local, halfMD5(user_id))

Всегда создавайте пару таблиц:
  - events_local  (MergeTree / ReplicatedMergeTree)  — хранение
  - events        (Distributed)                        — доступ

Вставка → events (distributed)
Чтение  → events (distributed)
Прямой доступ к events_local — только для отладки

-- FINAL — логическое слияние при чтении (медленнее, не меняет данные)
SELECT * FROM mydb.table FINAL;
 
-- OPTIMIZE TABLE — физическое слияние на диске (нужна осторожность на production)
OPTIMIZE TABLE mydb.table FINAL;

-- Текущие запросы
SELECT * FROM system.processes;
 
-- Части таблицы (партиции, размеры)
SELECT * FROM system.parts WHERE table = 'my_table' AND active = 1;
 
-- Состав кластера
SELECT * FROM system.clusters;
 
-- Реплики
SELECT * FROM system.replicas;
 
-- Словари
SELECT * FROM system.dictionaries;
 
-- Метрики
SELECT * FROM system.metrics;
SELECT * FROM system.asynchronous_metrics;

-- Хостнейм сервера (удобно в distributed-запросах)
SELECT hostname(), count() FROM events GROUP BY hostname();
 
-- Генератор строк (для тестов)
SELECT * FROM numbers(100);         -- 0..99
SELECT * FROM numbers(1000000);     -- 0..999999
 
-- Информация о типе колонки
SELECT toTypeName(col) FROM table LIMIT 1;
 
-- Форматирование чисел
SELECT formatReadableSize(1073741824);  -- '1.00 GiB'
SELECT formatReadableQuantity(1500000); -- '1.50 million'

-- Типичный (и почти единственный) нормальный сценарий MV:
CREATE MATERIALIZED VIEW kafka_mv TO events_local AS
SELECT
    JSONExtractString(message, 'user_id') AS user_id,
    JSONExtractUInt(message, 'ts')        AS ts
FROM kafka_events;
-- Всё остальное (витрины, агрегаты) — лучше делать через обычные батч-джобы

-- Плохо (оконная функция):
SELECT user_id, event, row_number() OVER (PARTITION BY user_id ORDER BY ts) AS rn
FROM events;
 
-- Лучше (подзапрос с GROUP BY):
SELECT e.user_id, e.event
FROM events e
JOIN (
    SELECT user_id, min(ts) AS first_ts
    FROM events
    GROUP BY user_id
) g ON e.user_id = g.user_id AND e.ts = g.first_ts;

Нужно хранить обычные данные?
    → MergeTree

Нужна автоматическая агрегация при записи (счётчики, суммы)?
    → SummingMergeTree (простые суммы)
    → AggregatingMergeTree (сложные агрегаты: uniq, quantile, any)

Нужна дедупликация по ключу?
    → ReplacingMergeTree

Нужно отслеживать изменения состояния (CDC)?
    → CollapsingMergeTree

Нужна интеграция с Kafka?
    → Kafka Engine + Materialized View → MergeTree

Нужна интеграция с PostgreSQL?
    → PostgreSQL Engine

Нужны быстрые вставки из множества маленьких батчей?
    → Buffer Engine → MergeTree

Нужен оптимизированный IN-запрос?
    → Set Engine

Нужны тестовые данные?
    → GenerateRandom Engine

Нужно временно хранить данные в памяти?
    → Memory Engine