refactor(orders): SQL приёма вынесен из DAG

- Зачем:
  - преобразования хранилища должны читаться рядом с целевым слоем, а DAG должен показывать оркестрацию.
- Что:
  - запросы забора и разбора заказов разложены по каталогам STG и ODS.
  - общий контракт провода подключён в обе ветви штатным шаблонизатором Airflow.
  - SQL смонтирован во все службы Airflow, а правило раскладки записано в архитектуре.
- Проверка:
  - make lint config-test smoke check-services.
  - airflow tasks render для pull_batch и parse_batch; airflow tasks test для parse_batch.

dags/orders_ingest.py

@@ -24,159 +24,9 @@ from airflow.sdk import Connection, dag, get_current_context, task
 
 START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
 
-# Забор — один прямой SELECT, без цикла до пустоты: одна порция ClickHouse
-# берёт десятки тысяч сообщений, а слепок дня — порядка полутора тысяч строк.
-# Короткая порция оставит хвост до следующего прогона, а отказ после чтения
-# унесёт прочитанное с собой: офсеты коммитятся в момент чтения. Граница
-# целиком — ADR 0008, «Следствия».
-#
-# stream_like_engine_allow_direct_select разрешает читать чтеца запросом; вторая
-# половина пары объявлена на самой таблице (sql/ddl/10-stg-tables.sql).
-# distributed_foreground_insert = 1 — конвенция ETL-вставок стенда: задача не
-# должна зеленеть раньше, чем строки легли на шарды.
-TAKE_ONE_BATCH = """
-INSERT INTO stg.orders_raw_dist
-SELECT
-    raw,
-    _topic AS kafka_topic,
-    _partition AS kafka_partition,
-    _offset AS kafka_offset,
-    _timestamp_ms AS kafka_timestamp,
-    hostName() AS consumer_host,
-    {load_id:String} AS _load_id,
-    now64(3) AS _load_ts
-FROM stg.orders_raw_kafka
-SETTINGS
-    stream_like_engine_allow_direct_select = 1,
-    distributed_foreground_insert = 1
-"""
-
-# Строгий приём: проверяется форма провода и ничего сверх неё.
-#
-# Ключей одиннадцать, и проверяются все: десять скалярных значений прямо
-# образуют типизированную строку заказа. Внутрь items приём не смотрит —
-# содержимое позиций, переходы статуса и равенства сумм остаются ниже границы
-# (docs/architecture/orders/ingestion.md, «Граница строгого приёма»).
-#
-# Блок общий у обеих вставок нарочно: годная ветвь берёт row_is_valid, брак —
-# буквальное NOT. Разойдись условия хоть на символ — строка либо задвоится,
-# либо исчезнет молча.
-#
-# Отсюда два запрета на выражения предиката, и оба серьёзные. Первый: ни одно
-# не возвращает NULL — трёхзначная логика дала бы строку, которую не берёт ни
-# условие, ни его отрицание. Поэтому сравнения дают 0 или 1, а обнуляемый
-# разбор заканчивается IS NOT NULL. Второй: ни одно не бросает исключений на
-# произвольном raw — иначе одна грязная строка роняет весь переход, ради
-# отсутствия чего таблица брака и заведена.
-WIRE_CONTRACT = r"""
-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
-"""
-
-# Годные версии заказа. Срез читается по _load_id — тому же, что проставил
-# забор: разбор идёт по неизменной порции, а не по «всему, что появилось».
-#
-# Разобранные значения берёт assumeNotNull: обнуляемый разбор стоит за
-# предикатом, который NULL уже отсёк, и приведение здесь не может упасть.
-PARSE_GOOD_ROWS = (
-    "INSERT INTO ods.order_snapshot_dist"
-    + WIRE_CONTRACT
-    + r"""
-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
-SETTINGS distributed_foreground_insert = 1
-"""
-)
-
-# Брак: тот же срез и буквальное отрицание того же предиката.
-#
-# Классы перекрываются, поэтому проверяются по порядку, а в error_class идёт
-# первый совпавший: скаляр проваливает и проверку на объект, и сверку ключей —
-# без объявленного порядка он попал бы то в один класс, то в другой.
-PARSE_BAD_ROWS = (
-    "INSERT INTO ods.order_snapshot_errors_dist"
-    + WIRE_CONTRACT
-    + r"""
-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
-SETTINGS distributed_foreground_insert = 1
-"""
-)
+# Запросы лежат файлами в sql/ репозитория, разложенные по слоям хранилища;
+# сюда их монтирует compose — всем службам Airflow сразу.
+SQL_ROOT = "/opt/airflow/sql"
 
 
 @dag(
@@ -188,6 +38,7 @@ SETTINGS distributed_foreground_insert = 1
     # дрались бы за неё, а слепок разъехался бы по двум _load_id; второй
     # прогон подождёт своей очереди.
     max_active_runs=1,
+    template_searchpath=SQL_ROOT,
     tags=["заказы"],
 )
 def orders_ingest():
@@ -210,12 +61,16 @@ def orders_ingest():
             send_receive_timeout=30,
         )
 
-    @task
-    def pull_batch() -> None:
+    # Аргумент с расширением из templates_exts Airflow подменяет текстом файла:
+    # берёт его из template_searchpath и прогоняет через Jinja при исполнении
+    # задачи, а не при разборе дага. В задачу приезжает готовый запрос — вместе
+    # с тем, что файл подключил через include.
+    @task(templates_exts=(".sql",))
+    def pull_batch(sql: str) -> None:
         load_id = get_current_context()["run_id"]
         client = clickhouse_client()
         try:
-            summary = client.command(TAKE_ONE_BATCH, parameters={"load_id": load_id})
+            summary = client.command(sql, parameters={"load_id": load_id})
         finally:
             client.close()
         # Размер порции — read_rows: written_rows у вставки в Distributed
@@ -226,8 +81,8 @@ def orders_ingest():
             load_id,
         )
 
-    @task
-    def parse_batch() -> None:
+    @task(templates_exts=(".sql",))
+    def parse_batch(good_rows_sql: str, bad_rows_sql: str) -> None:
         """Разобрать срез сырья в версии заказов и в брак.
 
         Обе вставки в одном task_id: транзакции между ними ClickHouse не даёт,
@@ -237,15 +92,18 @@ def orders_ingest():
         load_id = get_current_context()["run_id"]
         client = clickhouse_client()
         try:
-            client.command(PARSE_GOOD_ROWS, parameters={"load_id": load_id})
-            client.command(PARSE_BAD_ROWS, parameters={"load_id": load_id})
+            client.command(good_rows_sql, parameters={"load_id": load_id})
+            client.command(bad_rows_sql, parameters={"load_id": load_id})
         finally:
             client.close()
         # Счётчиков строк нет: у запроса с WHERE read_rows считает прочитанное
         # с диска, а не подошедшее (storage.md, «Что проверено»).
         logging.info("срез разобран: _load_id %s", load_id)
 
-    pull_batch() >> parse_batch()
+    pull_batch("stg/orders_raw_load.sql") >> parse_batch(
+        "ods/order_snapshot_load.sql",
+        "ods/order_snapshot_errors_load.sql",
+    )
 
 
 orders_ingest()