README.md

@@ -374,10 +374,12 @@ Airflow закреплён на 3.3.0. Состав обязательных п
 [архитектурой Airflow 3.3](https://airflow.apache.org/docs/apache-airflow/3.3.0/core-concepts/overview.html),
 [публичным интерфейсом](https://airflow.apache.org/docs/apache-airflow/3.3.0/public-airflow-interface.html)
 и [описанием здоровья](https://airflow.apache.org/docs/apache-airflow/3.3.0/administration-and-deployment/logging-monitoring/check-health.html).
-Для пробников проверены публичный `Connection.get` из `airflow.sdk` и клиент
+Для ClickHouse установлен официальный провайдер Airflow: SQL-файлы исполняет
+общий `SQLExecuteQueryOperator`, программные запросы идут через
+`ClickHouseHook`. Провайдер работает поверх закреплённого
 `clickhouse-connect`. Официальный провайдер Kafka сам использует
-`confluent-kafka`; отдельное подключение и обёртки провайдера здесь не нужны,
+`confluent-kafka`; отдельное подключение и его обёртки здесь не нужны, поэтому
-поэтому прямой клиент оставляет образ и пример короче.
+клиент Kafka добавлен в образ напрямую.
 Superset закреплён на 6.1.0; драйвер `clickhouse-connect`, форма
 `clickhousedb://` и драйвер Postgres сверены с
 [документацией подключений Superset](https://superset.apache.org/user-docs/6.1.0/databases/)

compose.yaml

@@ -79,8 +79,8 @@ x-airflow-common: &airflow-common
     KAFKA_ORDERS_TOPIC: *orders-topic
   volumes:
     - ./dags:/opt/airflow/dags:ro
-    # Запросы дагов: даг называет файл, а текст читает Airflow при исполнении
+    # Запросы дагов: даг называет файл, а обработчик DAG при разборе загружает
-    # задачи. Каталог тот же, что применяет DDL, — вся SQL стенда живёт в sql/.
+    # его текст. Каталог тот же, что применяет DDL, — вся SQL стенда живёт в sql/.
     - ./sql:/opt/airflow/sql:ro
     # Инициализация мира собирает канонический образ из того же контекста, что
     # ручной generator. Docker SDK архивирует эти файлы внутри задачи.

dags/orders_ingest.py

@@ -18,9 +18,9 @@
 from __future__ import annotations
 
 import datetime
-import logging
 
-from airflow.sdk import Connection, dag, get_current_context, task
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
+from airflow.sdk import dag
 
 START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
 
@@ -43,67 +43,25 @@ SQL_ROOT = "/opt/airflow/sql"
 )
 def orders_ingest():
     """Забрать приехавший слепок заказов и разложить его по слоям."""
-
+    pull_batch = SQLExecuteQueryOperator(
-    def clickhouse_client():
+        task_id="pull_batch",
-        # Импорт при выполнении задачи, а не при разборе файла: обработчик DAG
+        conn_id="clickhouse_default",
-        # разбирает его снова и снова, и импорт наверху оплачивался бы каждым
+        sql="stg/orders_raw_load.sql",
-        # разбором.
+        parameters={"load_id": "{{ run_id }}"},
-        import clickhouse_connect
+        do_xcom_push=False,
-
+    )
-        connection = Connection.get("clickhouse_default")
+    parse_batch = SQLExecuteQueryOperator(
-        return clickhouse_connect.get_client(
+        task_id="parse_batch",
-            host=connection.host,
+        conn_id="clickhouse_default",
-            port=connection.port,
+        sql="ods/order_parse.sql",
-            username=connection.login,
+        parameters={"load_id": "{{ run_id }}"},
-            password=connection.password,
+        split_statements=True,
-            database=connection.schema or "default",
+        do_xcom_push=False,
-            connect_timeout=5,
-            send_receive_timeout=30,
     )
 
-    # Аргумент с расширением из templates_exts Airflow подменяет текстом файла:
+    # ClickHouse не даёт транзакции между двумя вставками parse_batch. Один
-    # берёт его из template_searchpath и прогоняет через Jinja при исполнении
+    # task_id сохраняет их порядок и одну точку повтора для неизменного load_id.
-    # задачи, а не при разборе дага. В задачу приезжает готовый запрос — вместе
+    pull_batch >> parse_batch
-    # с тем, что файл подключил через 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(sql, parameters={"load_id": load_id})
-        finally:
-            client.close()
-        # Размер порции — read_rows: written_rows у вставки в Distributed
-        # считает не приехавшее.
-        logging.info(
-            "порция принята: строк %s, _load_id %s",
-            summary.summary["read_rows"],
-            load_id,
-        )
-
-    @task(templates_exts=(".sql",))
-    def parse_batch(good_rows_sql: str, bad_rows_sql: str) -> None:
-        """Разобрать срез сырья в версии заказов и в брак.
-
-        Обе вставки в одном task_id: транзакции между ними ClickHouse не даёт,
-        а повтор задачи безопасен — срез читается по тому же неизменному
-        _load_id (docs/architecture/orders/ingestion.md, «Поток данных»).
-        """
-        load_id = get_current_context()["run_id"]
-        client = clickhouse_client()
-        try:
-            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("stg/orders_raw_load.sql") >> parse_batch(
-        "ods/order_load.sql",
-        "ods/order_errors_load.sql",
-    )
 
 
 orders_ingest()

dags/test_clickhouse.py

@@ -19,6 +19,7 @@ from __future__ import annotations
 import datetime
 import uuid
 
+from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
 from airflow.sdk import Connection, dag, get_current_context, task
 
 CLUSTER = "clickstream_cluster"
@@ -54,27 +55,12 @@ def _remote_parameters() -> dict[str, str]:
     }
 
 
-def _clickhouse_client():
+def _clickhouse_hook() -> ClickHouseHook:
-    # clickhouse_connect импортируется внутри функции, а не наверху файла:
+    return ClickHouseHook(clickhouse_conn_id="clickhouse_default")
-    # обработчик DAG разбирает этот файл снова и снова, и импорт наверху
-    # оплачивался бы каждым разбором. Тяжёлые импорты Airflow советует
-    # держать внутри задач.
-    import clickhouse_connect
-
-    connection = Connection.get("clickhouse_default")
-    return clickhouse_connect.get_client(
-        host=connection.host,
-        port=connection.port,
-        username=connection.login,
-        password=connection.password,
-        database=connection.schema or "default",
-        connect_timeout=5,
-        send_receive_timeout=30,
-    )
 
 
-def _table_engines(client, source: str) -> list[tuple[str, str]]:
+def _table_engines(hook: ClickHouseHook, source: str) -> list[tuple[str, str]]:
-    result = client.query(
+    return hook.get_records(
         f"""
         SELECT name, engine
         FROM {source}
@@ -84,23 +70,23 @@ def _table_engines(client, source: str) -> list[tuple[str, str]]:
         """,
         parameters=_remote_parameters(),
     )
-    return result.result_rows
 
 
-def _drop_tables(client) -> None:
+def _drop_tables(hook: ClickHouseHook) -> None:
-    client.command(
+    hook.run(
-        f"DROP TABLE IF EXISTS default.{DISTRIBUTED_TABLE} ON CLUSTER {CLUSTER} SYNC"
+        [
-    )
+            f"DROP TABLE IF EXISTS default.{DISTRIBUTED_TABLE} "
-    client.command(
+            f"ON CLUSTER {CLUSTER} SYNC",
-        f"DROP TABLE IF EXISTS default.{LOCAL_TABLE} ON CLUSTER {CLUSTER} SYNC"
+            f"DROP TABLE IF EXISTS default.{LOCAL_TABLE} ON CLUSTER {CLUSTER} SYNC",
+        ],
     )
 
 
 # Отсутствие таблиц проверяют обе задачи — перед созданием и после уборки,
 # — поэтому у этой проверки своё имя, а остальные живут прямо в теле задач.
-def _assert_tables_absent(client) -> None:
+def _assert_tables_absent(hook: ClickHouseHook) -> None:
     for node_name, source in NODES:
-        remaining = _table_engines(client, source)
+        remaining = _table_engines(hook, source)
         if remaining:
             raise RuntimeError(
                 f"служебные таблицы остались на {node_name}: {remaining}"
@@ -118,11 +104,11 @@ def _assert_tables_absent(client) -> None:
 def test_clickhouse():
     @task
     def prepare_tables() -> None:
-        client = _clickhouse_client()
+        hook = _clickhouse_hook()
-        try:
+        _drop_tables(hook)
-            _drop_tables(client)
+        _assert_tables_absent(hook)
-            _assert_tables_absent(client)
+        hook.run(
-            client.command(
+            [
                 f"""
                 CREATE TABLE default.{LOCAL_TABLE} ON CLUSTER {CLUSTER}
                 (
@@ -133,9 +119,7 @@ def test_clickhouse():
                     '{{replica}}'
                 )
                 ORDER BY marker
-                """
+                """,
-            )
-            client.command(
                 f"""
                 CREATE TABLE default.{DISTRIBUTED_TABLE} ON CLUSTER {CLUSTER}
                 AS default.{LOCAL_TABLE}
@@ -145,46 +129,41 @@ def test_clickhouse():
                     '{LOCAL_TABLE}',
                     cityHash64(marker)
                 )
-                """
+                """,
+            ],
         )
         for node_name, source in NODES:
-                actual_tables = _table_engines(client, source)
+            actual_tables = _table_engines(hook, source)
             if actual_tables != EXPECTED_TABLES:
                 raise RuntimeError(
                     f"неверный набор таблиц на {node_name}: {actual_tables}"
                 )
-        finally:
-            client.close()
 
     @task
     def write_marker() -> dict[str, str]:
-        client = _clickhouse_client()
+        hook = _clickhouse_hook()
-        try:
         marker = f"{get_current_context()['run_id']}:{uuid.uuid4()}"
-            client.insert(
+        hook.bulk_insert_rows(
             f"default.{LOCAL_TABLE}",
             [[marker]],
             column_names=["marker"],
         )
-            local_rows = client.query(
+        local_rows = hook.get_records(
             f"""
             SELECT hostName(), marker
             FROM default.{LOCAL_TABLE}
             WHERE marker = {{marker:String}}
             """,
             parameters={"marker": marker},
-            ).result_rows
+        )
         if len(local_rows) != 1 or local_rows[0][1] != marker:
             raise RuntimeError(f"маркер не найден в локальной таблице: {marker}")
         return {"marker": marker, "hostname": local_rows[0][0]}
-        finally:
-            client.close()
 
     @task
     def read_from_node_2(written: dict[str, str]) -> None:
-        client = _clickhouse_client()
+        hook = _clickhouse_hook()
-        try:
+        node_2_rows = hook.get_records(
-            node_2_rows = client.query(
             """
             SELECT hostName()
             FROM remote(
@@ -193,10 +172,10 @@ def test_clickhouse():
             )
             """,
             parameters=_remote_parameters(),
-            ).result_rows
+        )
         if len(node_2_rows) != 1:
             raise RuntimeError(f"не удалось определить имя ноды 2: {node_2_rows}")
-            distributed_rows = client.query(
+        distributed_rows = hook.get_records(
             f"""
             SELECT _shard_num, hostName(), marker
             FROM remote(
@@ -212,19 +191,15 @@ def test_clickhouse():
                 "marker": written["marker"],
                 **_remote_parameters(),
             },
-            ).result_rows
-            if written["hostname"] == node_2_rows[0][0]:
-                raise RuntimeError(
-                    "запись и чтение маркера должны выполняться с разных нод"
         )
+        if written["hostname"] == node_2_rows[0][0]:
+            raise RuntimeError("запись и чтение маркера должны идти с разных нод")
         expected_rows = [(1, written["hostname"], written["marker"])]
         if distributed_rows != expected_rows:
             raise RuntimeError(
                 "нода 2 не прочитала маркер первого шарда через Distributed: "
                 f"{written['marker']}, получено {distributed_rows}"
             )
-        finally:
-            client.close()
 
     # Уборка идёт только после успеха: упавший пробник оставляет кластер таким,
     # каким сломался, а остатки сносит начало следующего запуска. Правило
@@ -233,12 +208,9 @@ def test_clickhouse():
     # с упавшей проверкой (ADR 0003).
     @task
     def cleanup_tables() -> None:
-        client = _clickhouse_client()
+        hook = _clickhouse_hook()
-        try:
+        _drop_tables(hook)
-            _drop_tables(client)
+        _assert_tables_absent(hook)
-            _assert_tables_absent(client)
-        finally:
-            client.close()
 
     prepared = prepare_tables()
     written = write_marker()

dags/world_lifecycle.py

@@ -18,10 +18,12 @@ import time
 from pathlib import Path
 
 from airflow.exceptions import AirflowException
+from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
 from airflow.providers.docker.hooks.docker import DockerHook
 from airflow.providers.docker.operators.docker import DockerOperator
 from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
-from airflow.sdk import Connection, Variable, dag, task
+from airflow.sdk import Variable, chain, dag, task
 
 START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
 TAGS = ["жизненный цикл мира"]
@@ -34,7 +36,8 @@ KAFKA_BOOTSTRAP_SERVERS = os.environ["KAFKA_BOOTSTRAP_SERVERS"]
 GENERATOR_IMAGE = os.environ["GENERATOR_IMAGE"]
 STAND_NETWORK = os.environ["STAND_NETWORK"]
 
-SQL_ROOT = Path("/opt/airflow/sql/ddl")
+SQL_ROOT = Path("/opt/airflow/sql")
+DDL_ROOT = SQL_ROOT / "ddl"
 WORLD_ROOT = Path("/opt/airflow/world")
 INVENTORY = WORLD_ROOT / "data/world-inventory.json"
 CLICKHOUSE_CONNECTION = "clickhouse_lifecycle"
@@ -44,48 +47,9 @@ CONSUMER_GROUPS = ("clickstream_hits", "clickstream_orders")
 TOPICS = {HITS_TOPIC: 2, ORDERS_TOPIC: 1}
 
 
-def _clickhouse_client():
+def _clickhouse_hook() -> ClickHouseHook:
-    """Открыть соединение владельца жизненного цикла."""
+    """Подключить владельца жизненного цикла."""
-    import clickhouse_connect
+    return ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION)
-
-    connection = Connection.get(CLICKHOUSE_CONNECTION)
-    return clickhouse_connect.get_client(
-        host=connection.host,
-        port=connection.port,
-        username=connection.login,
-        password=connection.password,
-        database=connection.schema or "default",
-        connect_timeout=5,
-        # `ON CLUSTER` ждёт хосты до 180 с. Транспорт живёт дольше, чтобы
-        # ClickHouse сам назвал незавершённый хост вместо сетевого тайм-аута.
-        send_receive_timeout=300,
-    )
-
-
-def _ddl_statements(text: str) -> list[str]:
-    """Разделить канонический DDL по его явному правилу файлов.
-
-    ClickHouse Connect исполняет один запрос за вызов. В `sql/ddl` конец
-    выражения — точка с запятой в конце строки; это проверяемая конвенция
-    наших файлов, а не попытка написать общий SQL-парсер.
-    """
-    statements: list[str] = []
-    lines: list[str] = []
-    for line in text.splitlines():
-        lines.append(line)
-        stripped = line.rstrip()
-        if stripped.endswith(";") and not stripped.lstrip().startswith("--"):
-            statement = "\n".join(lines).strip()
-            statements.append(statement[:-1].rstrip())
-            lines = []
-
-    remainder = "\n".join(lines).strip()
-    if remainder and any(
-        line.strip() and not line.lstrip().startswith("--")
-        for line in remainder.splitlines()
-    ):
-        raise AirflowException("DDL-файл оканчивается незавершённым выражением")
-    return statements
 
 
 def _generator(task_id: str, command: list[str]) -> DockerOperator:
@@ -118,6 +82,7 @@ def _kafka_error_code(error: Exception) -> int:
     start_date=START_DATE,
     is_paused_upon_creation=False,
     max_active_runs=1,
+    template_searchpath=str(SQL_ROOT),
     tags=TAGS,
 )
 def world_initialize():
@@ -194,23 +159,6 @@ def world_initialize():
                     "выполните make rebuild-storage"
                 )
 
-    @task
-    def apply_ddl() -> None:
-        """Применить канонические файлы DDL с ноды 1 по порядку имён."""
-        files = sorted(SQL_ROOT.glob("*.sql"))
-        if not files:
-            raise AirflowException(f"в {SQL_ROOT} нет файлов DDL")
-
-        client = _clickhouse_client()
-        try:
-            for path in files:
-                statements = _ddl_statements(path.read_text(encoding="utf-8"))
-                logging.info("применяем %s: выражений %s", path.name, len(statements))
-                for statement in statements:
-                    client.command(statement)
-        finally:
-            client.close()
-
     @task
     def build_generator_image() -> None:
         """Собрать генератор из текущего дерева, а не из старого образа."""
@@ -248,11 +196,10 @@ def world_initialize():
               + (SELECT count() FROM ods.event_errors_dist)
         """
 
-        client = _clickhouse_client()
+        hook = _clickhouse_hook()
-        try:
         arrived = 0
         for attempt in range(1, 101):
-                row = client.query(query).first_row
+            row = hook.get_first(query)
             if row is None:
                 raise AirflowException("ClickHouse не вернул счётчик приёма")
             arrived = int(row[0])
@@ -262,8 +209,6 @@ def world_initialize():
             if attempt % 5 == 0:
                 logging.info("принято %s из %s строк", arrived, expected)
             time.sleep(3)
-        finally:
-            client.close()
         raise AirflowException(
             f"стартовый мир не принят за 300 с: строк {arrived} из {expected}; "
             "проверьте журнал send_initial_events и "
@@ -279,7 +224,19 @@ def world_initialize():
     route = choose_path()
     ready = world_already_exists()
     topics = create_topics()
-    ddl = apply_ddl()
+    ddl_files = sorted(DDL_ROOT.glob("*.sql"))
+    if not ddl_files:
+        raise AirflowException(f"в {DDL_ROOT} нет файлов DDL")
+    ddl_tasks = [
+        SQLExecuteQueryOperator(
+            task_id=f"apply_{path.stem.replace('-', '_')}",
+            conn_id=CLICKHOUSE_CONNECTION,
+            sql=f"ddl/{path.name}",
+            split_statements=True,
+            do_xcom_push=False,
+        )
+        for path in ddl_files
+    ]
     image = build_generator_image()
     events = _generator(
         "send_initial_events",
@@ -307,7 +264,7 @@ def world_initialize():
     position = remember_starting_position()
 
     route >> [ready, topics]
-    topics >> ddl >> image >> events >> arrived >> snapshots >> orders >> position
+    chain(topics, *ddl_tasks, image, events, arrived, snapshots, orders, position)
 
 
 @dag(
@@ -328,16 +285,13 @@ def world_recreate():
 
     @task
     def drop_application_databases() -> None:
-        client = _clickhouse_client()
+        hook = _clickhouse_hook()
-        try:
         for database in APPLICATION_DATABASES:
             logging.info("удаляем базу %s", database)
-                client.command(
+            hook.run(
                 f"DROP DATABASE IF EXISTS {database} "
                 f"ON CLUSTER {CLICKHOUSE_CLUSTER} SYNC"
             )
-        finally:
-            client.close()
 
     @task
     def delete_consumer_groups() -> None:

docs/adr/0001-stand-services.md

@@ -32,13 +32,18 @@ JSON-файл отдельного тома `airflow_auth`. Новый имен
 
 Airflow получает два подготовленных подключения к `clickhouse-01`: рабочее
 `clickhouse_default` и административное `clickhouse_lifecycle` для дагов
-жизненного цикла мира ([ADR 0013](0013-world-lifecycle.md)). Superset получает
+жизненного цикла мира ([ADR 0013](0013-world-lifecycle.md)). Оба имеют тип
-подключение `clickhousedb://` к `clickhouse-02`. Провайдер ClickHouse для
+`clickhouse`, который регистрирует официальный провайдер
-Airflow не нужен: локальный образ содержит прямой клиент.
+`apache-airflow-providers-clickhousedb`. SQL-файлы исполняет общий
-Пробник создаёт служебные таблицы `ON CLUSTER`, пишет на первой ноде и через
+`SQLExecuteQueryOperator`, а программные запросы идут через `ClickHouseHook`.
-`remote` читает `Distributed` на второй. Разные ноды сохраняют учебную
+Сам провайдер работает через `clickhouse-connect`; его версия закреплена
-ловушку: забытый `ON CLUSTER` проявится в Superset, даже если операция Airflow
+отдельно, чтобы сборка не меняла транспорт незаметно.
-на первой ноде прошла успешно.
+
+Superset получает подключение `clickhousedb://` к `clickhouse-02`. Пробник
+Airflow создаёт служебные таблицы `ON CLUSTER`, пишет на первой ноде и через
+`remote` читает `Distributed` на второй. Разные ноды сохраняют учебную ловушку:
+забытый `ON CLUSTER` проявится в Superset, даже если операция Airflow на первой
+ноде прошла успешно.
 
 Клиент Kafka также добавлен прямо в образ Airflow. Официальный провайдер
 использует тот же `confluent-kafka`, а пробнику не нужны его подключение,
@@ -84,10 +89,18 @@ Prometheus читает встроенные точки метрик двух с
 Оттуда взяты `airflow.sdk`, обязательный отдельный обработчик DAG, возможность
 не запускать triggerer и проверка конкретных компонентов здоровья.
 
-Для интеграционных пробников проверены `Connection.get` в публичном
+Для ClickHouse через Context7 и официальную документацию Airflow проверены
-`airflow.sdk`, запросы через `clickhouse-connect` и состав официального
+`ClickHouseHook`, тип подключения `clickhouse` и работа
-провайдера Kafka. Выбран прямой `confluent-kafka`: провайдер строит свои
+`SQLExecuteQueryOperator` с SQL-файлом, параметрами и несколькими выражениями.
-подключения и обёртки поверх него, которые двум коротким пробникам не нужны.
+Закреплён `apache-airflow-providers-clickhousedb==1.0.0`: на 21 августа
+2026 года это последний стабильный выпуск; `1.0.1` опубликован только как
+предварительная версия. Зависимость провайдера допускает любой
+`clickhouse-connect>=1.3.0`, поэтому прежний `clickhouse-connect==1.6.0`
+оставлен прямой зависимостью образа.
+
+Для Kafka проверен состав официального провайдера. Выбран прямой
+`confluent-kafka`: провайдер строит свои подключения и обёртки поверх него,
+которые двум коротким пробникам не нужны.
 
 Драйверы и строки подключения проверены по документации Superset 6.1.0:
 [подключения к базам](https://superset.apache.org/user-docs/6.1.0/databases/),

docs/architecture/orders/ingestion.md

@@ -39,24 +39,27 @@
 конца прогона. Все строки получают `_load_id`, равный `run_id` Airflow.
 `_load_ts` вычисляется при этой записи и дальше переносится без пересчёта.
 Запрос забора лежит в [`sql/stg/orders_raw_load.sql`](../../../sql/stg/orders_raw_load.sql):
-даг задаёт порядок и параметры, а преобразование остаётся в SQL своего слоя.
+`SQLExecuteQueryOperator` задаёт подключение и передаёт `run_id` как параметр
+ClickHouse `load_id`, а преобразование остаётся в SQL своего слоя.
 
 Один следующий `task_id` отвечает за весь переход STG → ODS. Внутри него два
 последовательных `INSERT SELECT` читают неизменный срез по `_load_id`: первый
 пишет годные строки в `ods.order`, второй — брак в `ods.order_errors`.
 Транзакции между запросами нет. При частичном сбое Airflow повторяет весь
 `task_id`; одинаковые исходные строки и служебные метки не вычисляются заново.
-Запросы лежат рядом с целями:
+Обе вставки лежат в одном самодостаточном файле
-[`order_load.sql`](../../../sql/ods/order_load.sql) и
+[`order_parse.sql`](../../../sql/ods/order_parse.sql). Airflow сначала
-[`order_errors_load.sql`](../../../sql/ods/order_errors_load.sql).
+шаблонизирует файл, затем делит получившийся SQL на выражения и исполняет их по
+порядку в одной задаче `parse_batch`.
 
-Условия запросов взаимодополняющие: один общий предикат определяет брак, а
+Условия запросов взаимодополняющие: один общий предикат определяет годную
-годная ветвь использует его буквальное отрицание. Все функции предиката
+строку, а ветвь брака использует его буквальное отрицание. Все функции предиката
 возвращают результат без исключения, а сам предикат всегда заканчивается в
 `true` или `false`, не в `NULL`. Постоянный классификатор между STG и ODS для
-этого не нужен. Обе ветви включают один файл
+этого не нужен. Предикат объявлен локальным макросом Jinja внутри
-[`_order_wire_contract.sql`](../../../sql/ods/_order_wire_contract.sql), поэтому
+`order_parse.sql` и разворачивается в оба выражения. Макрос убирает
-предикат нельзя случайно исправить только в одной из них.
+дублирование, но не вводит отдельный фрагмент, который нельзя исполнить
+самостоятельно.
 
 ## Граница строгого приёма
 
@@ -178,6 +181,11 @@ JSON-объектом с точным набором ключей: `order_id`, `
 `20260608-0007`, который прежде приходил с отрицательным итогом, теперь
 `5490.00 − 823.50 + 299.00 = 4965.50`.
 
+При исполнении #124 21 августа 2026 года адресная пересборка мира прошла через
+`SQLExecuteQueryOperator`: `pull_batch` и `parse_batch` завершились успешно. В
+`system.query_log` на первой ноде осталось ровно по одному завершённому запросу
+пользователя `etl` для забора в STG, записи годных строк и записи брака.
+
 Забор из Kafka в STG снят на живом стенде 18 августа 2026 года при исполнении
 #93: одно прямое чтение приносит весь слепок дня, метаданные доставки доступны,
 офсеты коммитятся, а сбой забора не двигает позицию мира. Числа — [ADR

docs/architecture/storage.md

@@ -450,16 +450,18 @@ kafka_offset)`: смотрят такую таблицу от класса, а
 
 Исполняемые дагами запросы лежат по правилу
 `sql/<слой>/<объект>_<роль>.sql`. Слой — слой цели запроса: забор заказов в
-сырьё живёт в `sql/stg/orders_raw_load.sql`, две ветви разбора — в `sql/ods/`.
+сырьё живёт в `sql/stg/orders_raw_load.sql`, разбор — в
-Даг называет файлы, передаёт параметры и задаёт порядок выполнения, но не
+`sql/ods/order_parse.sql`. Даг называет файлы, передаёт параметры и задаёт
-хранит текст преобразований. Общий фрагмент начинается с подчёркивания: файл
+порядок выполнения, но не хранит текст преобразований. Два выражения разбора
-`sql/ods/_order_wire_contract.sql` сам не исполняется, его включают обе ветви
+делят локальный макрос Jinja в том же файле; подключаемого SQL-фрагмента между
-разбора.
+ними нет.
 
-Airflow читает и собирает эти файлы штатным шаблонизатором при исполнении
+SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. При разборе Python-кода
-задачи. Каталог `sql/` смонтирован во все его службы только для чтения. Поэтому
+обработчик DAG читает список DDL-файлов, загружает содержимое каждого
-обработчик дагов не зависит от наличия SQL на машине при разборе Python, а
+названного SQL-файла и кладёт его в сериализованный граф. При исполнении Airflow
-планировщик видит те же файлы при исполнении задачи.
+разворачивает в этом тексте Jinja; параметры ClickHouse передаются отдельно и
+не подставляются строкой. Каталог `sql/` смонтирован во все службы Airflow
+только для чтения.
 
 ### DDL
 
@@ -489,8 +491,10 @@ Airflow читает и собирает эти файлы штатным шаб
 отладке.
 
 Файлы применяет даг `world_initialize` с ноды 1 отдельной учётной записью
-`lifecycle`: всё `ON CLUSTER`, кроме чтеца заказов — он объявлен только на этой
+`lifecycle`: один файл — одна последовательная задача
-ноде. Перед DDL даг создаёт `hits` с двумя партициями и `orders` с одной
+`SQLExecuteQueryOperator`, а выражения внутри разделяет сам оператор. Всё
+`ON CLUSTER`, кроме чтеца заказов — он объявлен только на этой ноде. Перед DDL
+даг создаёт `hits` с двумя партициями и `orders` с одной
 ([ADR 0008](../adr/0008-order-ingestion.md)). Этот порядок не даёт `hits`
 автоматически создаться с одной партицией. Брокер автосоздание разрешает, а
 потребитель librdkafka внутри ClickHouse его не просит. Оба конца измерены;
@@ -532,8 +536,8 @@ Airflow читает и собирает эти файлы штатным шаб
 | `dic` | `dic.products` | словарь товаров поверх подложки |
 
 Матвью разбора у заказов нет: срез сырья раскладывают по этим двум целям два
-`INSERT SELECT` шага `parse_batch` из файлов `sql/ods/order_load.sql` и
+`INSERT SELECT` одной задачи `parse_batch` из файла
-`sql/ods/order_errors_load.sql`.
+`sql/ods/order_parse.sql`.
 
 Слой DDS и слой DM появляются на следующих этапах; их состав задан разделом
 7 мастер-спеки и переносится сюда по мере постройки.

docs/architecture/testing.md

@@ -211,9 +211,11 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`.
 Особняком стоит `scripts/run-dag.sh`: это не проверка, а общий REST-переход от
 `make` к Airflow. `make up` запускает через него `world_initialize`, а
 `make rebuild-storage` — `world_recreate`; сценарий ждёт появления дага в API,
-печатает идентификаторы дага и прогона и возвращает вызывающему его итог.
+а затем до минуты повторяет только стартовый ответ `Dag with dag_id not found`.
-Ожидание доставки событий и запуск `orders_ingest` теперь видны задачами
+Прочие ответы и сетевые ошибки он не повторяет: при потерянном ответе запуск мог
-`world_initialize`, а не спрятаны в shell.
+уже создаться. Сценарий печатает идентификаторы дага и прогона и возвращает
+вызывающему его итог. Ожидание доставки событий и запуск `orders_ingest` теперь
+видны задачами `world_initialize`, а не спрятаны в shell.
 
 Общее у смоука и `check-services` — счёт проверок, обращение к Compose и две
 проверки — вынесено в `scripts/stand-common.sh`; сам он не запускается.
@@ -235,6 +237,21 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`.
 
 ## Что проверено
 
+**Перезамер 21 августа 2026 года при исполнении #124.** Образ Airflow собран с
+`apache-airflow-providers-clickhousedb==1.0.0`,
+`apache-airflow-providers-common-sql==2.0.1` и
+`clickhouse-connect==1.6.0`; `pip check` не нашёл несовместимых зависимостей.
+Адресная пересборка мира через `make rebuild-storage` заняла 3 м 5 с: все шесть
+задач DDL, `pull_batch` и `parse_batch` завершились успешно. После неё прошли
+смоук 21/21 за 17 с, проверка ClickHouse 10/10 и межсервисная проверка 7/7 за
+100 с.
+
+Первый `make up` после замены образа попал в короткий рассинхрон запуска
+Airflow: API уже сообщил о здоровье и отдал `world_initialize` по GET, но POST
+нового запуска ещё ответил `Dag with dag_id not found`. Ошибок импорта не было;
+через несколько секунд повторный `make up` завершился успешно и не менял
+готовый мир.
+
 **Перезамер 21 августа 2026 года при исполнении #114.** После добавления
 `kafka-exporter` и источника ClickHouse смоук прошёл 21/21 за 16–17 секунд,
 `check-services` — 7/7 за 93 секунды. При остановленном экспортёре смоук дал

docs/specs/2026-08-21-world-lifecycle.md

@@ -51,8 +51,8 @@ Airflow держит прикладные топики Kafka, каноничес
 ([ADR 0007](../adr/0007-clickhouse-access.md)). Ей разрешено администрировать
 прикладные базы — заводить и удалять их и объекты внутри, — читать их и
 выполнять кластерный DDL. Ей разрешено чтение `system.settings`: без него
-`clickhouse-connect` не открывает соединение. Административные запросы к
+драйвер `clickhouse-connect` под официальным провайдером Airflow не открывает
-системным базам ей запрещены.
+соединение. Административные запросы к системным базам ей запрещены.
 
 `etl` при этом не расширяется. Право удаления баз нужно редкой операции
 восстановления и остаётся у `lifecycle`. Граница проверяема: при лишнем запросе
@@ -69,9 +69,10 @@ Airflow держит прикладные топики Kafka, каноничес
 Меняющиеся API сверены через Context7 и затем с установленными пакетами
 стенда. Выбраны публичный REST API v2 и ждущий `TriggerDagRunOperator` Airflow,
 сборка образа через `DockerHook.get_conn().build()`, futures `AdminClient` для
-Kafka и по одному `clickhouse-connect.command()` на выражение DDL. Последний
+Kafka, `SQLExecuteQueryOperator` для исполняемых SQL-файлов ClickHouse и
-выбор требует явного разбора файлов: клиент сам многозапросный файл не делит,
+`ClickHouseHook` для запросов, чей результат обрабатывает Python. Оператор сам
-а в каноническом SQL конец выражения уже задан точкой с запятой в конце строки.
+шаблонизирует файл и при `split_statements=True` исполняет выражения по порядку;
+собственный разбор SQL дагу больше не нужен.
 
 ## Контракты дагов
 
@@ -90,8 +91,11 @@ Kafka и по одному `clickhouse-connect.command()` на выражени
    матвью, и топик, созданный брокером сам, приехал бы с одним разделом
    ([ADR 0008](../adr/0008-order-ingestion.md)).
 3. Применяет канонический DDL из `sql/ddl/` по порядку имён файлов, с ноды 1,
-   учётной записью `lifecycle`. Правило «матвью приёма создаётся последней»
+   учётной записью `lifecycle`. Каждый файл — отдельная последовательная
-   остаётся за раскладкой файлов и не переезжает в даг.
+   задача `SQLExecuteQueryOperator`; выражения внутри неё исполняются по
+   порядку с `split_statements=True`. Красная задача называет файл и не пускает
+   последующие. Правило «матвью приёма создаётся последней» остаётся за
+   раскладкой файлов и не переезжает в даг.
 4. Собирает канонический образ генератора. Иначе мир мог бы приехать из старого
    кода: сборкой образа при подъёме больше никто не занимается.
 5. Играет стартовые дни одним прогоном `batch` того же образа.

infra/airflow/Dockerfile

@@ -2,14 +2,19 @@ FROM apache/airflow:3.3.0
 
 USER root
 
+# SQL-файлы ClickHouse исполняет официальный провайдер Airflow. Он разрешает
+# любую версию clickhouse-connect >= 1.3.0, поэтому драйвер закреплён отдельно:
+# сборка стенда не должна незаметно менять транспорт при обновлении зависимостей.
+#
 # Официальный провайдер Kafka использует тот же confluent-kafka. Пробникам
 # не нужны его подключения и обёртки, поэтому в образ добавлен сам клиент.
 #
 # Провайдера docker, которым пульт мира зовёт генератор, ставить не нужно:
 # база несёт его сама — 4.5.7 к Airflow 3.3.0, замер 13 августа 2026 года.
 RUN uv pip install --python /home/airflow/.local/bin/python --no-cache \
+        "apache-airflow-providers-clickhousedb==1.0.0" \
         "clickhouse-connect==1.6.0" \
         "confluent-kafka==2.15.0" \
-    && python -c "import clickhouse_connect, confluent_kafka"
+    && python -c "import airflow.providers.clickhousedb, clickhouse_connect, confluent_kafka"
 
 USER airflow

infra/airflow/init.sh

@@ -22,21 +22,26 @@ PY
 
 runuser -u airflow -- airflow db migrate
 runuser -u airflow -- airflow connections delete clickhouse_default >/dev/null 2>&1 || true
+# Прикладные запросы короткие: 30 с отличают зависший транспорт от работы запроса.
 runuser -u airflow -- airflow connections add clickhouse_default \
-    --conn-type generic \
+    --conn-type clickhouse \
     --conn-host clickhouse-01 \
     --conn-port 8123 \
     --conn-login etl \
     --conn-password "$CLICKHOUSE_ETL_PASSWORD" \
     --conn-schema default \
-    --conn-description "ClickHouse, нода 1; типизированный провайдер появится на этапе ETL"
+    --conn-extra '{"connect_timeout": 5, "send_receive_timeout": 30}' \
+    --conn-description "ClickHouse, нода 1; запросы прикладных дагов"
 
 runuser -u airflow -- airflow connections delete clickhouse_lifecycle >/dev/null 2>&1 || true
+# `ON CLUSTER` ждёт хосты до 180 с. Транспорт живёт дольше, чтобы ClickHouse
+# успел назвать незавершённый хост вместо сетевого тайм-аута.
 runuser -u airflow -- airflow connections add clickhouse_lifecycle \
-    --conn-type generic \
+    --conn-type clickhouse \
     --conn-host clickhouse-01 \
     --conn-port 8123 \
     --conn-login lifecycle \
     --conn-password "$CLICKHOUSE_LIFECYCLE_PASSWORD" \
     --conn-schema default \
+    --conn-extra '{"connect_timeout": 5, "send_receive_timeout": 300}' \
     --conn-description "ClickHouse, нода 1; создание и пересоздание прикладного мира"

scripts/run-dag.sh

@@ -8,6 +8,7 @@ readonly DAG_ID="${1:-}"
 readonly ATTEMPTS=400
 readonly PAUSE_SECONDS=3
 readonly REPORT_EVERY=5
+readonly TRIGGER_ATTEMPTS=20
 
 read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}"
 
@@ -68,14 +69,43 @@ jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \
     >/dev/null 2>&1 <<<"$dag" || \
     fail "даг ${DAG_ID} не появился за $((ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)"
 
-if ! response="$(curl -sS --fail-with-body --max-time 10 -X POST \
+# GET читает DagModel по идентификатору, а POST запуска принимает только запись
+# с `is_stale = false`. На холодном старте GET уже может видеть даг, пока POST
+# ещё отвечает этим 404. Повторяем только его: после сетевой ошибки запрос мог
+# успеть создать запуск, и повтор с автоматическим run_id создал бы второй.
+expected_missing_detail="Dag with dag_id: '${DAG_ID}' not found"
+triggered=0
+for ((attempt = 1; attempt <= TRIGGER_ATTEMPTS; attempt++)); do
+    if ! response_with_status="$(curl -sS --max-time 10 -w $'\n%{http_code}' -X POST \
         -H "Authorization: Bearer ${token}" \
         -H 'Content-Type: application/json' \
         -d '{"logical_date":null}' \
         "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns" 2>/dev/null)"; then
+        fail "Airflow не ответил при запуске дага ${DAG_ID}; проверьте список запусков перед повтором команды"
+    fi
+    http_status="${response_with_status##*$'\n'}"
+    response="${response_with_status%$'\n'*}"
     detail="$(jq -r '.detail // .message // empty' <<<"$response" 2>/dev/null || true)"
-    fail "Airflow не запустил даг ${DAG_ID}${detail:+: ${detail}}; $(next_step)"
+    case "$http_status" in
-fi
+        2??)
+            triggered=1
+            break
+            ;;
+        404)
+            if [[ "$detail" == "$expected_missing_detail" ]]; then
+                if ((attempt == 1 || attempt % REPORT_EVERY == 0)); then
+                    printf 'Airflow: даг %s ещё не готов к запуску; ждём (%d с).\n' \
+                        "$DAG_ID" "$((attempt * PAUSE_SECONDS))"
+                fi
+                ((attempt < TRIGGER_ATTEMPTS)) && sleep "$PAUSE_SECONDS"
+                continue
+            fi
+            ;;
+    esac
+    fail "Airflow не запустил даг ${DAG_ID}: HTTP ${http_status}${detail:+: ${detail}}; $(next_step)"
+done
+((triggered == 1)) || \
+    fail "даг ${DAG_ID} не стал доступен для запуска за $((TRIGGER_ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)"
 run_id="$(jq -r '.dag_run_id // empty' <<<"$response")"
 state="$(jq -r '.state // "queued"' <<<"$response")"
 [[ -n "$run_id" ]] || fail "Airflow не вернул идентификатор запуска ${DAG_ID}"

scripts/stand-smoke.sh

@@ -204,6 +204,7 @@ check_airflow() {
     local dag
     local dag_id
     local health
+    local lifecycle_connection
     local password
     local response
     local user
@@ -253,21 +254,33 @@ check_airflow() {
         -H "Authorization: Bearer ${airflow_token}" \
         "http://127.0.0.1:${airflow_port}/api/v2/connections/clickhouse_default" \
         2>/dev/null || true)"
+    lifecycle_connection="$(curl -sf --max-time 10 \
+        -H "Authorization: Bearer ${airflow_token}" \
+        "http://127.0.0.1:${airflow_port}/api/v2/connections/clickhouse_lifecycle" \
+        2>/dev/null || true)"
     if jq -e '
         .connection_id == "clickhouse_default" and
+        .conn_type == "clickhouse" and
         .host == "clickhouse-01" and
         .port == 8123 and
         .login == "etl" and
         .schema == "default"
-    ' >/dev/null 2>&1 <<<"$connection" && \
+    ' >/dev/null 2>&1 <<<"$connection" && jq -e '
+        .connection_id == "clickhouse_lifecycle" and
+        .conn_type == "clickhouse" and
+        .host == "clickhouse-01" and
+        .port == 8123 and
+        .login == "lifecycle" and
+        .schema == "default"
+    ' >/dev/null 2>&1 <<<"$lifecycle_connection" && \
         timeout 20s "${COMPOSE_CMD[@]}" --project-directory "$ROOT_DIR" \
             exec -T airflow-scheduler \
             curl -sf -u "etl:${clickhouse_password}" \
             'http://clickhouse-01:8123/?query=SELECT%201' \
             2>/dev/null | grep -qx '1'; then
-        pass 'подготовленное подключение Airflow указывает на clickhouse-01, нода доступна из контейнера'
+        pass 'оба подключения Airflow имеют тип clickhouse, нода 1 доступна из контейнера'
     else
-        fail 'подключение Airflow не указывает на clickhouse-01 или нода недоступна из контейнера'
+        fail 'тип или адрес подключений Airflow отличается от ожидаемого либо нода 1 недоступна из контейнера'
     fi
 }
 

sql/ods/_order_wire_contract.sql

@@ -1,61 +0,0 @@
--- Форма провода одной строки слепка: общая часть обеих вставок ODS.
---
--- Строгий приём: проверяется форма провода и ничего сверх неё.
---
--- Ключей одиннадцать, и проверяются все: десять скалярных значений прямо
--- образуют типизированную строку заказа. Внутрь items приём не смотрит —
--- содержимое позиций, переходы статуса и равенства сумм остаются ниже границы
--- (docs/architecture/orders/ingestion.md, «Граница строгого приёма»).
---
--- Файл включают обе вставки нарочно: годная ветвь берёт row_is_valid, брак —
--- буквальное NOT. Разойдись условия хоть на символ — строка либо задвоится,
--- либо исчезнет молча.
---
--- Отсюда два запрета на выражения предиката, и оба серьёзные. Первый: ни одно
--- не возвращает NULL — трёхзначная логика дала бы строку, которую не берёт ни
--- условие, ни его отрицание. Поэтому сравнения дают 0 или 1, а обнуляемый
--- разбор заканчивается IS NOT NULL. Второй: ни одно не бросает исключений на
--- произвольном raw — иначе одна грязная строка роняет весь переход, ради
--- отсутствия чего таблица брака и заведена.
-
-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

sql/ods/order_errors_load.sql

@@ -1,25 +0,0 @@
--- Брак: тот же срез и буквальное отрицание того же предиката.
---
--- Классы перекрываются, поэтому проверяются по порядку, а в error_class идёт
--- первый совпавший: скаляр проваливает и проверку на объект, и сверку ключей —
--- без объявленного порядка он попал бы то в один класс, то в другой.
-
-INSERT INTO ods.order_errors_dist
-{% include "ods/_order_wire_contract.sql" %}
-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/ods/order_load.sql

@@ -1,33 +0,0 @@
--- Годные версии заказа. Срез читается по _load_id — тому же, что проставил
--- забор: разбор идёт по неизменной порции, а не по «всему, что появилось».
---
--- Разобранные значения берёт assumeNotNull: обнуляемый разбор стоит за
--- предикатом, который NULL уже отсёк, и приведение здесь не может упасть.
-
-INSERT INTO ods.order_dist
-{% include "ods/_order_wire_contract.sql" %}
-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

sql/ods/order_parse.sql

@@ -0,0 +1,121 @@
+-- Разбор одного среза заказов: сначала годные версии, затем брак.
+--
+-- Обе вставки должны делить один предикат. Локальный макрос 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
+SETTINGS distributed_foreground_insert = 1;
+
+-- Брак: тот же срез и буквальное отрицание того же предиката.
+--
+-- Классы перекрываются, поэтому проверяются по порядку, а в 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
+SETTINGS distributed_foreground_insert = 1;