Metadata-Version: 2.5
Name: cdc-1c
Version: 0.1.16
Summary: Change data capture (CDC) from 1C:Enterprise to your data warehouse
Project-URL: Homepage, https://github.com/pavel-v-sobolev/cdc_1C
Project-URL: Repository, https://github.com/pavel-v-sobolev/cdc_1C
Project-URL: Issues, https://github.com/pavel-v-sobolev/cdc_1C/issues
Author-email: Pavel Sobolev <pavel-v-sobolev@yandex.ru>
License-Expression: MIT
License-File: LICENSE
Keywords: 1c,1c-enterprise,cdc,data-engineering,dwh,etl,postgres,python
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Information Technology
Classifier: Intended Audience :: Science/Research
Classifier: Intended Audience :: System Administrators
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: dbmerge>=1.0.22
Requires-Dist: requests>=2.33.0
Requires-Dist: sqlalchemy>=2.0.49
Requires-Dist: xmltodict>=1.0.4
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == 'dev'
Provides-Extra: postgres
Requires-Dist: psycopg2-binary>=2.9.12; extra == 'postgres'
Description-Content-Type: text/markdown

[![PyPI version](https://img.shields.io/pypi/v/cdc-1c.svg)](https://pypi.org/project/cdc-1c/)
[![Python versions](https://img.shields.io/pypi/pyversions/cdc-1c.svg)](https://pypi.org/project/cdc-1c/)


**cdc-1c** is a docker container and a Python library, that provides 1C system data loading to data warehouse using Change Data Capture apporach. \
It engages standard ODATA mechanism and standard 1C exchange plan mechanism to extract data from 1C system and upsert changes to the target DB.

**cdc-1c** - это докер контейнер и python-библиотека, предназначенные для получения данных из 1С, использующий подход CDC (загрузка изменений данных). \
Продукт использует стандартный интерфейс ODATA и механизм планов обмена для выгрузки изменений данных из системы 1С и обновления данных в целевой БД.

# Что нужно для работы
1) опубликовать базу 1с на web
2) создать пользователя odata и дать ему необходимые права
3) настроить план обмена в конфигураторе и включить в его состав нужные объекты 1с
3) настроить узел обмена с использованием внешней обработки `cdc-1c.odt`
4) запустить загрузку: python-библиотекой `cdc-1c` (см. ниже) — готового docker-образа пока нет, он в планах

# Использование библиотеки python

Библиотека даёт оркестратор `Replicator1C`, который читает изменения из 1С (OData + план обмена) и
складывает их в целевую БД, подтверждая приём пакета только после успешного сохранения. БД
передаётся готовым SQLAlchemy `engine`.

## Установка

```bash
pip install cdc-1c

# вместе с драйвером PostgreSQL:
pip install "cdc-1c[postgres]"
```

Python 3.10+ и **PostgreSQL**. Другие СУБД не поддерживаются: обработчики пишут обычный SQL, а
рекомендуемые для них приёмы опираются на возможности Postgres — массивы и GIN-индексы по ним,
`array_agg(DISTINCT ... ) FILTER`, `ARRAY(SELECT ...)`, `NOT MATERIALIZED` у CTE. Сама запись идёт
через `dbmerge`, который умеет и другие СУБД, поэтому базовая репликация на них, скорее всего,
заработает, но не проверяется и не поддерживается.

## Быстрый старт

```python
from sqlalchemy import create_engine
from cdc_1c import Replicator1C

# pool_size >= full_load_workers + 3 — почему столько, см. «Сколько нужно соединений к БД»
engine = create_engine("postgresql+psycopg2://user:pass@localhost:5432/dwh", pool_size=5)

rep = Replicator1C(
    odata_url="http://host/base/odata/standard.odata",
    odata_auth=("odata", "secret"),        # (user, password) либо None без авторизации
    exchange_name="ДляODATA",              # имя плана обмена в 1С
    queue_guid="a9bc23c5-3689-11f1-926c-0800270bc6cb",  # Ref_Key узла обмена (очереди)
    engine=engine,
    db_schema="cdc_1c",                    # None → схема БД по умолчанию (public у Postgres)
    request_timeout=60,                    # таймаут HTTP-запросов к 1С, сек (по умолчанию 60 на коннект, 900 на ответ)
    full_load_workers=2,                   # число фоновых потоков полной выгрузки
)

rep.run_forever(interval=60)               # цикл опроса раз в 60 секунд
```

Параметры присваиваются явно, по одному — и это единственный способ их задать: объекта настроек нет.
В python-приложении на их месте литералы, при запуске из окружения — `os.environ`; сборка в обоих
случаях выглядит одинаково.

## Запуск из окружения

Если хочется не писать код вовсе, есть готовый entrypoint — `python -m cdc_1c` (он же команда
`cdc-1c`). Он читает те же параметры из переменных окружения:

| Переменная | Обязательна | Значение |
|---|---|---|
| `CDC1C_ODATA_URL` | да | адрес OData-интерфейса базы 1С |
| `CDC1C_EXCHANGE_NAME` | да | имя плана обмена |
| `CDC1C_QUEUE_GUID` | да | `Ref_Key` узла обмена (очереди) |
| `CDC1C_DB_URL` | да | строка подключения SQLAlchemy к целевой БД |
| `CDC1C_ODATA_USER` / `CDC1C_ODATA_PASSWORD` | нет | без пользователя запросы идут без авторизации |
| `CDC1C_DB_SCHEMA` | нет | схема целевой БД; не задана — схема по умолчанию |
| `CDC1C_FULL_LOAD_WORKERS` | нет | число фоновых потоков полной выгрузки (по умолчанию 2) |
| `CDC1C_POLL_INTERVAL` | нет | период опроса в секундах (по умолчанию 60) |
| `CDC1C_MODE` | нет | `loop` (по умолчанию) или `once` |
| `CDC1C_LOG_LEVEL` | нет | уровень логирования (по умолчанию `INFO`) |

Свой код (обработчики, о них ниже) через переменные окружения не подключить — он объявляется кодом.
Для этого случая есть готовый шаблон точки входа: [example_config/](example_config/) — каталог с
`runner.py` и пакетом `handlers/` рядом. Запускается как обычный скрипт:

```bash
python example_config/runner.py
```

Python сам кладёт каталог скрипта в `sys.path`, поэтому `from handlers import ...` внутри `runner.py`
находит соседний пакет. Скопируйте этот каталог себе, положите туда свои обработчики — и то же самое
станет содержимым тома, монтируемого в контейнер.


## Режимы: `run_once` и `run_forever`

```python
rep.run_once()                 # один цикл: read → save → notify (подтверждение только после save)
rep.run_forever(interval=60)   # бесконечный цикл run_once с паузой; фоном — полные выгрузки
```

- `run_once(notify_changes=False)` — не подтверждать приём (пакет останется в очереди 1С; сделано для отладки).
- `run_forever(interval, max_iterations=0)` — `max_iterations>0` ограничивает число итераций.

## Полная (первоначальная) выгрузка

При работе `run_forever` объекты, впервые встреченные в пакете изменений, автоматически ставятся в
очередь на полную выгрузку и грузятся фоновыми потоками. Можно запустить выгрузку и вручную:

Полная выгрузка объекта реализована на стороне python, чтобы поддержать выгрузку объектов больших размеров, 
т.к. если инициировать полную выгрузку по плану обмена в 1С, то данные поступят без возможности постраничной загрузки.
(Поэтому данная функция специально убрана из модуля 1С).

Полная выгрузка спроектирована так, чтобы работать параллельно с получением изменений объекта.

```python
rep.list_objects()             # имена объектов 1С, доступных для выгрузки (Catalog_…, Document_…, …)

rep.full_load("Catalog_Номенклатура")              
rep.full_load("Document_РеализацияТоваровУслуг", batch_size=500)
```

`batch_size` — верхняя граница, а не жёсткий размер страницы: реальный размер подбирается по весу
выданных страниц, потому что одна запись 1С может тянуть за собой и одну строку, и тысячи (все
табличные части документа, весь набор движений регистратора). Как именно устроена постраничная
выгрузка и почему keyset-курсор в 1С неприменим к ссылочным ключам — см.
[DOCUMENTATION.md](DOCUMENTATION.md).


### Фильтр по периоду

Для ручной догрузки за нужный период укажите поле даты/времени и границы (включительно):

```python
from datetime import date, datetime

# весь месяц: date-граница включает последний день целиком (даже для поля дата-время)
rep.full_load("Document_РеализацияТоваровУслуг",
              date_field="Date", date_from=date(2026, 6, 1), date_to=date(2026, 6, 30))

# точная граница по времени — передайте datetime
rep.full_load("Document_РеализацияТоваровУслуг",
              date_field="Date", date_from=datetime(2026, 6, 1, 9, 0, 0))
```

`date_field` — имя поля 1С (`Date` у документов, `Period` у регистров). Границы транслируются в OData
`$filter` и объединяются с курсором пагинации.

## Что появляется в целевой БД

- На каждый объект 1С — таблица (имя транслитерируется, длинные имена усекаются с хэшем под лимит СУБД).
- Служебные поля строк: `merged_on`/`inserted_on` (момент merge/первой вставки), `is_deleted_or_empty`
  (строку не учитывать, см. ниже), `exchange_message_no` (номер пакета обмена — диагностическое поле,
  логика загрузки на нём не построена).
- Служебные таблицы `replicator_1c_log`, `metadata_objects_1c` и `handlers_1c` (см. ниже).

`merged_on` — момент **последнего реального изменения** строки, а не последнего пакета, в котором она
приехала. 1С регулярно переписывает объекты, не меняя реквизитов; такие записи строку не трогают,
иначе инкрементальная материализация пересчитывала бы группы впустую. Поля, которые 1С меняет при
каждой записи (`exchange_message_no`, `DataVersion`), сами по себе изменением не считаются — они
записываются, только когда строку обновило что-то ещё.

### `is_deleted_or_empty` — универсальный признак «строку не учитывать»

Данные из 1С приходят инкрементально, поэтому строку нельзя просто выбросить: её исчезновение —
такое же событие, как изменение, и оно должно быть видно потребителям. Вместо удаления строка
остаётся с поднятым флагом.

Флаг сводит в одно булево поле все причины, по которым строку не следует учитывать в расчётах.
Их пять, и приходят они из разных мест:

| Причина | Откуда | Что означает |
|---|---|---|
| Пометка удаления | поле 1С `DeletionMark` | объект помечен на удаление в 1С |
| Неактивная запись | поле 1С `Active` = `false` | движение не участвует в итогах 1С |
| Строка выпала из набора | проставляется при merge | строки больше нет в наборе движений / табличной части |
| Пустой набор движений | запись сформирована при загрузке | все движения регистратора удалены |
| Опустевшая табличная часть | запись сформирована при загрузке | в табличной части не осталось строк |

Первые две — факты самой 1С, они просто переносятся во флаг.

Третья — про строки, пропавшие из набора. Набор движений регистратора и табличная часть приходят
целиком и целиком заменяют сохранённые: строки, которой в наборе больше нет, в 1С больше не
существует. Такая строка **не удаляется, а помечается**, и вместе с флагом ей поднимается
`merged_on`, а числовые ресурсы гасятся в `NULL`. Причина: инкрементальная материализация ищет
изменившиеся группы по `merged_on`, а исчезнувшая строка следа не оставляет — витрина навсегда
сохранила бы удалённое движение. Обнуление ресурсов — вторая линия обороны: `SUM` игнорирует `NULL`,
поэтому итог остаётся верным даже в запросе, забывшем фильтр по флагу. Остальные поля надгробия
сохраняются, так что видно, что это была за строка.

Последние две 1С сообщает отсутствием строк, а отсутствие в инкрементальный пакет не помещается:
чтобы «набор опустел» вообще доехало, формируется одна фиктивная запись с реальным ключом набора
(регистратор или `Ref_Key` владельца) и поднятым флагом. Она же вводит группу в пакет, без чего
пометка выпавших строк не сработала бы. Номер строки у неё — 1: как только набор снова наполнится,
первая настоящая строка перезапишет фиктивную, и та не осядет в таблице навсегда.

Флаг сбрасывается сам: если объект сняли с пометки удаления, запись снова стала активной или строка
вернулась в набор — приходит обычное изменение с `false`, и строка возвращается в расчёты вместе с
восстановленными значениями ресурсов.

Что это значит для запросов: **любой расчёт по сырым таблицам обязан учитывать флаг**, иначе в сумму
попадут удалённые, неактивные и выбывшие из наборов строки — они физически остаются в таблицах. В примерах материализации это сделано множителем
`* (NOT "is_deleted_or_empty")::int`, обнуляющим значение погашенной строки. Поля 1С `DeletionMark`
и `Active` при этом сохраняются как есть — если нужно различать причины, они рядом.

## Служебные таблицы

### `replicator_1c_log` — журнал загрузок

Строка на каждую загрузку объекта: пакет изменений или полная выгрузка.

| Колонка | Назначение |
|---|---|
| `id` | суррогатный ключ |
| `exchange` | имя плана обмена |
| `object` | имя объекта 1С |
| `type` | `changes` (пакет изменений) или `full` (полная выгрузка) |
| `message_no` | номер пакета обмена; `NULL` для полной выгрузки |
| `started_at` / `finished_at` | начало и конец загрузки; `finished_at IS NULL` — не завершена (упала) |
| `inserted_row_count` / `updated_row_count` / `deleted_row_count` | счётчики строк merge |
| `total_time` | суммарное время merge, сек |

Предназначено для мониторинга: незавершённые строки (`finished_at IS NULL`) — упавшие загрузки; по `type` и
`object` видно, что и когда грузилось.

### `metadata_objects_1c` — реестр объектов и состояние полной выгрузки

Синхронизируется с метаданными 1С — строка на каждый объект, встреченный в обмене. Ключ таблицы — полное
имя объекта (регистр и документ могут иметь одинаковое короткое имя).

| Колонка | Назначение |
|---|---|
| `object_full_name` | полное имя объекта 1С (ключ), например `Catalog_Номенклатура` |
| `object_full_name_en` | транслит = имя таблицы объекта в БД |
| `object_name` / `object_type` | имя и тип объекта (`Catalog` / `Document` / `AccumulationRegister` / …) |
| `fields` / `fields_en` | список полей объекта: имена 1С и их транслит (= колонки в БД) |
| `full_load_is_required` | объект ожидает полной выгрузки |
| `last_full_load_dt` | когда объект был полностью выгружен; `NULL` — ни разу |
| `last_full_load_rows_modified` | сколько строк выгрузка реально изменила |
| `last_full_load_minutes` | сколько она заняла, минуты (дробное) |
| `merged_on` | момент синхронизации записи реестра |

Новый объект оркестратор помечает `full_load_is_required=true`, фоновый воркер выгружает его целиком и
проставляет `last_full_load_dt`. Отсюда же удобно посмотреть список доступных объектов и имена их таблиц.

**`last_full_load_rows_modified` — это проверка самого CDC.** Полная выгрузка читает объект из 1С
целиком и сравнивает с тем, что уже лежит в БД. Если изменения доезжают исправно, менять ей нечего и
значение должно быть **0**. Ненулевое — значит часть изменений в обмен не попала, и стоит разобраться,
что именно: перевыгрузить объект и посмотреть, повторится ли.

```sql
SELECT object_full_name, last_full_load_dt, last_full_load_rows_modified, last_full_load_minutes
FROM cdc_1c.metadata_objects_1c
WHERE last_full_load_rows_modified > 0
ORDER BY last_full_load_rows_modified DESC;
```

### `handlers_1c` — состояние обработчиков

Появляется, только если вы подключили свой код по событию изменения — описана
[ниже](#таблица-состояния-handlers_1c) вместе с самим механизмом.

## Логирование

Из коробки библиотека вешает вывод на логгер `cdc_1c` (INFO), если приложение не настроило логирование
само. Настроили своё — библиотека молчит и пишет через стандартный `logging`.

## Свой код по событию изменения: обработчики (handlers)

Витрина (или отправка изменений во внешнюю систему) сама не знает, что данные приехали: она либо
опрашивает БД вхолостую, либо ждёт, пока её запустят руками. Раннер эту информацию имеет — он и
сохраняет данные, — поэтому он же и зовёт ваш код.

Обработчик — класс, унаследованный от `Handler1C`. Готовые примеры — в
[example_config/](example_config/).

```python
from cdc_1c import Handler1C

class ZakazyKlientov(Handler1C):
    # имена ТАБЛИЦ в целевой БД, а не имена объектов 1С
    ON = ["AccumulationRegister_ZakazyKlientov", "Catalog_Nomenklatura"]
    ON_FULL_LOAD = True     # звать ли на страницах полной выгрузки (по умолчанию да)
    MIN_INTERVAL = 0        # не чаще раза в N секунд

    def setup(self, context):   # один раз за процесс: DDL вьюшек и целевых таблиц
        self.execute(context, DDL)

    def handle(self, context):  # полезная работа за окно
        ...
```

Подключается явным списком — никакого сканирования каталогов:

```python
from handlers import ZakazyKlientov, OtpravkaVOchered

replicator.run_forever(interval=60, handlers=[ZakazyKlientov(), OtpravkaVOchered(queue="cdc")])
```

Порядок списка = порядок вызова в пределах прохода (витрина поверх витрины идёт после базовой).
Никакого особого каталога обработчикам не нужно: это обычные python-модули, поэтому лежат где
угодно, лишь бы импортировались, а общий код между ними подключается обычным `import`. Обновление
библиотеки этот код не трогает.

В список передаётся **экземпляр**, а не класс: так обработчик можно параметризовать конструктором
(одна логика на две схемы — два экземпляра). Имя (по умолчанию — имя класса) служит ключом состояния
в `handlers_1c`, поэтому у параметризованных экземпляров оно обязано различаться:
`ZakazyKlientov(name='ZakazyKlientov_mart2')`. Одноимённые раннер отвергнет на старте.

Наследование не обязательно: раннеру достаточно `ON` и `handle` — годится и модуль целиком, и
функция с этими атрибутами. `Handler1C` даёт `setup()`, `since(context)`, `changed_since(context, *колонки)` и
`execute(context, sql)` — то, что иначе копируется из обработчика в обработчик.

В `ON` перечисляются имена **таблиц в целевой БД** (транслит), а не имена объектов 1С: обработчик
пишет SQL по таблицам, имя 1С он в глаза не видит. Имя 1С в `ON` не совпало бы ни с чем и обработчик
молча никогда бы не сработал — поэтому такой список отвергается на старте с подсказкой, как это имя
выглядит в базе.

Это не только «на что реагировать», но и «что я читаю»: по тому же списку считается верхняя граница
окна, поэтому перечислять надо **все** таблицы, из которых обработчик выбирает данные.

**Данные в обработчик не передаются** — он делает свой `SELECT`. Ему передаётся окно времени, за
которое надо отработать:

```sql
WHERE merged_on > :last_run_at
```

- `context.last_run_at` — отметка предыдущего **успешного** запуска; `NULL` — с начала времён;
- `context.boundary` — верхняя граница окна: то, что обработчик получит как `last_run_at` в следующий раз;
- `context.full_rebuild` — витрину просят собрать заново (см. ниже);
- `context.engine`, `context.schema`, `context.objects` (что изменилось), `context.sources` (`changes` / `full_load`),
  `context.logger`.

Обе границы — по часам БД (тем же `now()`, которым `dbmerge` штампует `merged_on`), поэтому
расхождение часов между хостами роли не играет.

В `WHERE` верхняя граница не нужна, хотя окно ею закрывается. От пропуска строк защищает не условие
выборки, а само значение `context.boundary`: оно прижато к старту незавершённого merge, чьи строки
обработчику всё равно не видны. Видимую строку правее границы обработчик посчитает раньше срока — и
посчитает ещё раз в следующем окне, потому что `last_run_at` станет `boundary`. Это лишняя работа,
а не пропуск, и взамен витрина получается свежее. Добавить `merged_on <= :boundary` имеет смысл
только там, где повтор дорог сам по себе — например, при отправке во внешнюю систему.

### Когда обработчик зовут

Только когда merge **реально что-то изменил** (вставил, обновил или пометил удалённой хоть одну
строку). 1С регистрирует изменение объекта на любую перезапись, и в пакет приезжает масса записей,
идентичных тому, что уже лежит в БД; шумные поля (`DataVersion`, номер сообщения обмена) при
сравнении не учитываются. Звать на таком пакете незачем — `SELECT` по окну всё равно вернёт пусто.

Полная выгрузка сигналит **на каждую сохранённую страницу**, а не один раз в конце: очередь
схлопывающая (сигнал — это «объект стал грязным», а не «запусти»), поэтому лишних вызовов это не
даёт, зато витрина начинает наполняться после первой же страницы, а не через часы. Обработчику,
которому бэкфилл не нужен (рассылка уведомлений, отправка в очередь), ставьте `ON_FULL_LOAD = False`.

По той же причине DDL живёт в `setup()`, а не в `handle()`: `setup` вызывается один раз за процесс,
иначе `CREATE OR REPLACE VIEW` выполнялся бы на каждую страницу выгрузки.

Исполнение — **один поток**, отдельный от цикла изменений: тяжёлая витрина не тормозит приём пакетов,
а два обработчика не дерутся за одну целевую таблицу. Пользовательский код никогда не выполняется в
потоках полной выгрузки — они только кладут сигнал.

### Сколько нужно соединений к БД

Потоков три сорта, и пулы у них раздельные: цикл изменений (главный поток), `full_load_workers`
потоков полной выгрузки и один поток обработчиков. Занять чужие потоки они не могут, но `engine`
у них общий, и одновременно держать соединение могут все сразу:

```
pool_size >= full_load_workers + 3
```

Дефолтный для SQLAlchemy `pool_size=5` при `full_load_workers=2` покрывает это впритык. Если
поднимаете число воркеров, поднимайте и пул: не хватает соединений — кто-то встаёт в ожидание, и
затормозить может как обработчик, так и сама выгрузка.

### Таблица состояния `handlers_1c`

| колонка | смысл |
|---|---|
| `name` | `NAME`, либо имя класса обработчика |
| `enabled` | выключенный обработчик не зовут (и окно за время простоя не копят) |
| `last_run_at` | граница последнего успешного запуска |
| `last_error` | traceback последнего падения |
| `full_rebuild_is_required` | заказ на пересборку витрины |
| `last_full_rebuild_dt` | когда пересборка отработала в последний раз |
| `last_full_rebuild_minutes` | сколько она заняла, минуты (дробное) |

Попросить собрать витрину заново:

```sql
UPDATE cdc_1c.handlers_1c SET full_rebuild_is_required = true WHERE name = 'ZakazyKlientov';
```

Заказ сам ставит обработчик в очередь — ждать изменений по подписанным объектам не нужно. Он
получит окно с начала времён (`context.last_run_at` = `None`) и `context.full_rebuild` = `True`.
Первый в жизни прогон — тоже пересборка, флаг для этого ставить не нужно. Обработчику, который
считает только по окну, `full_rebuild` не нужен вовсе; он пригодится, если пересборка устроена
иначе, чем обычный прогон — скажем, идёт по частям, чтобы не делать один гигантский merge.

После успеха требование снимается, а в `last_full_rebuild_dt` записывается время. `last_run_at`
пересборка не обнуляет: если она упадёт, прежняя граница останется на месте.

Тот же заказ раннер ставит сам, когда в таблице объекта появляется **новая колонка** (новый реквизит
в 1С). Инкремент её не увидел бы никогда: окно строится по `merged_on`, а `merged_on` двигается
только у строк, у которых изменились значения — добавление колонки не меняет ни одного значения, и
все уже лежащие строки остались бы левее окна навсегда.

### Что нужно помнить

- Доставка **at-least-once**: упали после работы обработчика, но до записи `last_run_at` — окно
  повторится. Для витрины это безразлично (merge идемпотентен), для отправки во внешнюю систему
  потребитель должен быть идемпотентным.
- Упавший обработчик границу не двигает и остаётся «грязным» — повтор произойдёт сам, без нового
  изменения.
- Удаления в окно попадают: из целевых таблиц ничего не исчезает физически (у справочников и
  документов пометка приезжает из 1С, у регистров выпавшие из набора строки помечаются
  `is_deleted_or_empty` с обнулением ресурсов), а пометка — это `UPDATE`, который двигает
  `merged_on`. Поэтому `SELECT` обработчика **не должен** отфильтровывать `is_deleted_or_empty`:
  для витрины это строка с нулём, для очереди — событие удаления.

## Дальнейшая материализация и сборка денормализованных таблиц

1С хранит данные в нормализованном виде: чтобы дотянуться, например, из регистра заказов до кода
товара, нужен `JOIN` со справочником номенклатуры по guid. Для задач DWH обычно нужна менее строгая
нормализация, поэтому в [example_config/handlers/](example_config/handlers/) лежат два разобранных
примера инкрементальной витрины:

- [zakazy_klientov.py](example_config/handlers/zakazy_klientov.py) — ключ таблицы фактов сохраняется
  (строки регистра как есть, плюс артикул из справочника);
- [zakazy_klientov_grouped.py](example_config/handlers/zakazy_klientov_grouped.py) — ключ меняется
  (`GROUP BY` по номеру документа, году и артикулу), пересчёт остаётся инкрементальным, а ключ
  группы составной — видно, как это отражается на фильтрах (`tuple_(a, b).in_(...)`).

Оба построены на одном правиле: **вьюшка отдаёт `merged_on` каждого участника `JOIN` отдельной
колонкой** (`merged_on`, `Nomenklatura_merged_on`, …), и в инкремент попадает всё, что стало свежее
хотя бы по одному источнику.

Так надо потому, что объекты 1С приезжают в обмене независимо и в произвольном порядке: справочник
номенклатуры может доехать (или измениться) позже регистра. Если ориентироваться только на
`merged_on` регистра, такая строка уже не попадёт в инкремент и витрина навсегда останется с `NULL`
или старым артикулом. Отдельные колонки заодно не смешивают несвязанные «часы» в одну.

Собирать эти отметки в один `WHERE ... OR ...` на объёме нельзя, и дело не в отсутствии индексов:
ветки `OR` живут на **разных** таблицах соединения, поэтому ни по одной нельзя отфильтровать до
`JOIN` — условие проверяется на готовой паре и вырождается в `Join Filter` поверх полного соединения.
`BitmapOr` из индексных сканов Postgres строит только когда все ветки `OR` на одной таблице, а
переписать `OR` в `UNION` через границу джойна он не умеет (тип соединения ни при чём: `INNER`
вместо `LEFT` плана не меняет).

Поэтому «какие группы изменились» спрашивается через `UNION ALL` по той же вьюшке — по ветке на
отметку:

```sql
SELECT "Number", "Year" FROM "..._rows_view" WHERE "merged_on" > :since
UNION ALL
SELECT "Number", "Year" FROM "..._rows_view" WHERE "ZakazKlienta_merged_on" > :since
UNION ALL
SELECT "Number", "Year" FROM "..._rows_view" WHERE "Nomenklatura_merged_on" > :since
```

В каждой ветке остаётся предикат ровно по одной базовой таблице, планировщик опускает его внутрь
вьюшки, в скан этой таблицы, и стоимость начинает зависеть от размера окна, а не от размера таблиц.
Переписывать `JOIN`-ы руками не нужно — логика соединений остаётся в одном месте. На синтетике
(регистр 2 млн строк, окно ~0.5%): `OR` — 620 мс с полным сканом регистра, `UNION ALL` — 98 мс без
единого `Seq Scan`.

Двух вещей это требует: `random_page_cost` под SSD (`1.1` вместо дефолтных `4` — иначе планировщик
всё равно предпочтёт полный скан, и выигрыш будет вдвое скромнее) и индекса по колонке соединения
со справочником: репликатор создаёт только индексы `merged_on`, а что с чем соединяется, знает
витрина.

Пересчитывать при этом нужно **группу целиком** (тот же ключ, по которому идёт удаление), а не
отдельные «свежие» строки — иначе `delete_condition` снесёт соседние строки группы, которые в
инкремент не попали.

Хранить границу обработки в самой витрине (по отметке на источник) больше не нужно — она приходит
готовой в `context.last_run_at`, одна на все источники сразу.

