Files
clickstream-data-platform/sql/ods/order_parse.sql
T
ddadminandClaude Opus 5 8c76a7a0d8 fix(ods): заказы разложены по своему ключу шардирования
- Зачем:
  - заводское parallel_distributed_insert_select = 2 исполняло вставку из
    Distributed в Distributed локально на каждом шарде, минуя ключ цели:
    1189 из 1662 заказов задвоились на обоих шардах, а ReplacingMergeTree
    через границу шарда дубли не схлопывает.
- Что:
  - в обе вставки sql/ods/order_parse.sql добавлено явное
    parallel_distributed_insert_select = 0 с доводом комментарием.
  - в docs/architecture/storage.md описан капкан «INSERT из Distributed в
    Distributed при несовпадающих ключах» и записана проверка (#137).
- Проверка:
  - межшардовые дубли по ods.order_rep дают 0 (было 1189), каждый шард
    держит только свои ключи; uniqExact(order_id) не изменился.
  - после прожитого дага world_next_day новый слепок принят, дублей
    по-прежнему 0.
  - make smoke, make check-clickhouse.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-22 04:02:52 -04:00

136 lines
8.8 KiB
SQL
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
-- Разбор одного среза заказов: сначала годные версии, затем брак.
--
-- Обе вставки должны делить один предикат. Локальный макрос Jinja
-- разворачивает его в каждое выражение до того, как оператор разделит файл по
-- точкам с запятой. Так контракт остаётся рядом с запросами и не становится
-- третьим SQL-файлом, который нельзя исполнить самостоятельно.
--
-- Строгий приём проверяет форму провода и ничего сверх неё. Ключей одиннадцать,
-- и проверяются все: десять скалярных значений прямо образуют типизированную
-- строку заказа. Внутрь items приём не смотрит — содержимое позиций, переходы
-- статуса и равенства сумм остаются ниже этой границы
-- (docs/architecture/orders/ingestion.md, «Граница строгого приёма»).
--
-- У общего предиката два обязательных свойства. Он всегда даёт 0 или 1, не
-- NULL: иначе строку не возьмут ни условие, ни его отрицание. И он не бросает
-- исключений на произвольном raw: одна грязная строка не должна ронять весь
-- переход, ради этого и существует таблица брака.
{% macro order_wire_contract() %}
WITH
-- Каноническое время провода: UTC, ровно три знака долей секунды.
'yyyy-MM-dd\'T\'HH:mm:ss.SSS\'Z\'' AS ts_mask,
JSONType(raw) = 'Object' AS is_object,
arraySort(JSONExtractKeys(raw)) = arraySort([
'order_id', 'user_id', 'status', 'created_at', 'updated_at',
'items_total', 'discount', 'delivery', 'total', 'items',
'snapshot_date'
]) AS keys_match,
JSONType(raw, 'order_id') = 'String'
AND JSONType(raw, 'status') = 'String'
-- Целое у JSONType зовётся двумя именами, и нужны оба: с одним
-- Int64 законный идентификатор за 2^63 уехал бы в брак.
AND JSONType(raw, 'user_id') IN ('Int64', 'UInt64')
AND JSONExtract(raw, 'user_id', 'Nullable(UInt64)') IS NOT NULL
AND JSONType(raw, 'items') = 'Array'
-- Форму держит регулярка, реальность — разбор, и порознь они дырявы:
-- маска берёт «2026-6-3T…» без ведущих нулей, а регулярка пропускает
-- 30 февраля. Обе половины измерены — docs/architecture/storage.md,
-- «Что проверено».
AND arrayAll(k ->
JSONType(raw, k) = 'String'
AND match(JSONExtractString(raw, k),
'^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z$')
AND parseDateTime64InJodaSyntaxOrNull(
JSONExtractString(raw, k), ts_mask, 'UTC') IS NOT NULL,
['created_at', 'updated_at'])
-- Деньги — строка с ровно двумя знаками после точки. Регулярка держит
-- форму, разбор — вместимость: у строки из двадцати девяток форма
-- хороша, а в Decimal(18, 2) она не влезает.
AND arrayAll(k ->
JSONType(raw, k) = 'String'
AND match(JSONExtractString(raw, k), '^\\d+\\.\\d{2}$')
AND toDecimal64OrNull(JSONExtractString(raw, k), 2) IS NOT NULL,
['items_total', 'discount', 'delivery', 'total'])
AND JSONType(raw, 'snapshot_date') = 'String'
AND match(JSONExtractString(raw, 'snapshot_date'), '^\\d{4}-\\d{2}-\\d{2}$')
AND parseDateTimeInJodaSyntaxOrNull(
JSONExtractString(raw, 'snapshot_date'), 'yyyy-MM-dd', 'UTC') IS NOT NULL
AS fields_valid,
is_object AND keys_match AND fields_valid AS row_is_valid
{% endmacro %}
-- Годные версии заказа. Срез читается по _load_id — тому же, что проставил
-- забор: разбор идёт по неизменной порции, а не по «всему, что появилось».
--
-- Разобранные значения берёт assumeNotNull: обнуляемый разбор стоит за
-- предикатом, который NULL уже отсёк, и приведение здесь не может упасть.
INSERT INTO ods.order_dist
{{ order_wire_contract() }}
SELECT
JSONExtractString(raw, 'order_id') AS order_id,
JSONExtract(raw, 'user_id', 'UInt64') AS user_id,
JSONExtractString(raw, 'status') AS status,
assumeNotNull(parseDateTime64InJodaSyntaxOrNull(
JSONExtractString(raw, 'created_at'), ts_mask, 'UTC')) AS created_at,
assumeNotNull(parseDateTime64InJodaSyntaxOrNull(
JSONExtractString(raw, 'updated_at'), ts_mask, 'UTC')) AS updated_at,
assumeNotNull(toDecimal64OrNull(
JSONExtractString(raw, 'items_total'), 2)) AS items_total,
assumeNotNull(toDecimal64OrNull(JSONExtractString(raw, 'discount'), 2)) AS discount,
assumeNotNull(toDecimal64OrNull(JSONExtractString(raw, 'delivery'), 2)) AS delivery,
assumeNotNull(toDecimal64OrNull(JSONExtractString(raw, 'total'), 2)) AS total,
-- items кладётся сырым фрагментом JSON, а не разобранной структурой.
JSONExtractRaw(raw, 'items') AS items,
toDate(assumeNotNull(parseDateTimeInJodaSyntaxOrNull(
JSONExtractString(raw, 'snapshot_date'),
'yyyy-MM-dd', 'UTC'))) AS snapshot_date,
-- Метки запуска и прибытия переносятся как есть. Поставь здесь now64(3) —
-- и _load_ts молча ответила бы на другой вопрос: «когда разобрали».
_load_id,
_load_ts
FROM stg.orders_raw_dist
WHERE _load_id = {load_id:String} AND row_is_valid
-- Заводское значение parallel_distributed_insert_select — 2: вставку вида
-- INSERT ... SELECT из Distributed в Distributed на одном кластере ClickHouse
-- исполняет локально на каждом шарде, и строки минуют ключ шардирования цели.
-- Ключи здесь разные — сырьё разложено по хешу строки, заказы по
-- cityHash64(order_id), — так версии одного заказа и расползлись по обоим
-- шардам, где ReplacingMergeTree их уже не схлопывает. Ноль возвращает вставку
-- инициатору: он читает срез со всех шардов и пишет через ods.order_dist,
-- то есть по её ключу.
SETTINGS distributed_foreground_insert = 1,
parallel_distributed_insert_select = 0;
-- Брак: тот же срез и буквальное отрицание того же предиката.
--
-- Классы перекрываются, поэтому проверяются по порядку, а в error_class идёт
-- первый совпавший: скаляр проваливает и проверку на объект, и сверку ключей —
-- без объявленного порядка он попал бы то в один класс, то в другой.
INSERT INTO ods.order_errors_dist
{{ order_wire_contract() }}
SELECT
raw,
multiIf(
NOT is_object, 'not_an_object',
NOT keys_match, 'keyset_mismatch',
'field_invalid'
) AS error_class,
kafka_topic,
kafka_partition,
kafka_offset,
kafka_timestamp,
consumer_host,
_load_id,
_load_ts
FROM stg.orders_raw_dist
WHERE _load_id = {load_id:String} AND NOT row_is_valid
-- Тот же ноль, хотя здесь заводское значение раскладку не ломало: ключ у
-- ods.order_errors_dist — cityHash64(raw), тот же, что у сырья, и локальная
-- запись попадала бы в свой шард сама. Совпадение ключей не гарантия:
-- поменяется ключ любой из двух таблиц — и строки разъедутся молча.
SETTINGS distributed_foreground_insert = 1,
parallel_distributed_insert_select = 0;