Главы

  1. Введение в документы
  2. Базовые возможности JSON
  3. JSON в таблицах
  4. Индексирование JSON
  5. Ограничения в документах
  6. Язык путей JSONPath
  7. Отчеты, функции, расписание
  8. Функции на языке Python
  9. Версии и патчи
  10. Релевантный поиск

Содержание

В этой главе мы поговорим о том, как хранить разные версии одного документа. Обсудим, как организовать историческую таблицу, перемещать в нее документы и восстанавливать их. Также мы затронем тему разности JSON и познакомимся со стандартом JSON Patch (не путать с JSON Path).

Как мы упоминали, документ – это большая структура данных. Некоторые документы содержат сотни полей, и они часто меняются. При этом изменения фиксируют: создают резервные копии документов. Эти копии открывают полезные возможности, например:

  • откатить ошибочную операцию, что важно в финансах и банковском секторе.
  • Пересчитать какие-то показатели, если в прошлом это сделали неверно.
  • Вести аналитику: смотреть как менялись сущности во времени, строить графики, отслеживать рост и падение показателей.

Некоторые разработчики уверены, что ответ на все перечисленное – бекапы. Чтобы получить документ во вчерашнем состоянии, нужно восстановить бекап на тестовую базу и прочитать его оттуда. Этот подход справедлив лишь отчасти, потому что восстановление бекапа – дорогая операция. Выполнять ее ради одного документа неэффективно. Иногда один и тот же документ нужен в нескольких срезах, и приложение не может ждать, пока восстановятся все бекапы.

Рассмотрим, как устроено версионирование документов. Оно подразумевает, что перед обновлением документ сохраняется в особую таблицу. Часто ее называют исторической или архивной. В простом случае на каждое обновление документа создается копия. Позже мы рассмотрим сценарии, когда копию создают по условию, например не чаще чем раз в пять минут.

Иногда версии хранятся неограниченно долго, а иногда их “подрезают”. Записи старше определенного срока считаются устаревшими, и спецальная задача переносит их в другое место, например хранилище S3. Очистка запускается регулярно по расписанию.

В особых случаях таблица хранит не сам документ, а разницу (патч) между прошлой и новой версиями. Имея патч, можно определить, какие именно поля документа затронули изменения. Эти и другие техники мы рассмотрим по тексту ниже.

Историческая таблица

Подумаем, какие поля понадобится для исторической таблицы. Кроме своего первичного ключа (id), таблица должна хранить первичный ключ документа. Назовем его pk или doc_id. Это поле не будет уникальным, потому что один и тот же документ может иметь несколько версий.

Поле doc типом jsonb хранит документ в состоянии до того, как к нему применили изменения.

Вспомогательное поле entity содержит имя таблицы, которой принадлежит документ: application, organization, user и так далее. С его помощью мы определим, на какую таблицу ссылается поле pk. В некоторых случаях (например, в триггерах) поле entity заполняется из локальной переменной.

Колонка operation пригодится, чтобы знать тип операции над документом: обновление или удаление. Будем хранить в ней значения “update” и “delete”. Как и в случае с entity, иногда имя операции доступно в переменной.

Необходимо поле, по которому различают версии в рамках одного документа. Наивный подход в том, чтобы использовать счетчик: версия 1, 2, 3 и так далее. Недостаток счетчиков в том, что произвести новое значение может только база данных. Приходится выполнять запрос вида “взять последнюю версию и добавить к ней единицу”, что неудобно и чревато конфликтами.

Кроме того, даже имея число, необходимы дата и время события. Поэтому желательно хранить версию как метку времени, потому что они упорядочены и могут быть произведены в приложении. Еще лучшим выбором будет UUID версии 7, который, как мы помним, совмещает время и упорядоченность и тоже же может быть сгенерирован в приложении.

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

С учетом сказанного таблица выглядит так:

create table history(
    id uuid not null default gen_random_uuid(),
    pk uuid not null,
    entity text not null,
    operation text not null,
    doc jsonb not null,
    created_at timestamptz not null default current_timestamp,
    user_id uuid null,
    comment text null
);

Эта таблица хранит версии всех сущностей: заявок, организаций и других. Ранее мы упоминали, что сущности хранят в отдельных таблицах: applications, organizations и так далее. Причина в том, каждая таблица требует своих индексов. Однако для исторических данных это не так: независимо от типа они подчиняются общим правилам, поэтому индексы будут общими.

Подумаем теперь, за какой информацией мы будем обращаться к таблице history. В основном это два запроса:

  • Зная ключ документа и дату, определить актуальную на тот момент версию.
  • По ключу документа получить версии по убыванию даты, возможно за какой-то период.

Обе задачи решаются составным уникальным индексом (pk, created_at desc). Поскольку pk – лидирующий компонент, он сразу находит все версии документа. Они упорядочены по убыванию даты, и это полезно по двум критериям. Во-первых, чаще всего нас интересуют недавние версии, и они будут в начале выборки. Во-вторых, чтобы получить актуальную версию на дату, ограничим выборку по created_at и добавим limit 1. Уникальность индекса не позволит иметь несколько версий с одинаковой датой.

Демонстрация

Опробуем историческую таблицу в действии. Запишем в нее сто документов по десять версий на каждый, что в дает тысячу записей. Для этого подготовим функцию для генерации UUID:

create or replace function gen_uuid(x integer)
returns uuid
language sql immutable strict parallel safe
return to_char(x, 'FM00000000-0000-0000-0000-000000000000')::uuid;

Далее выполним запрос:

insert into history (pk, entity, operation, doc, created_at, user_id, comment)
select
    gen_uuid(x % 100),
    ((array['applications', 'organizations', 'users', 'events'])[ceil(random() * 4)]),
    ((array['update', 'delete'])[ceil(random() * 2)]),
    jsonb_build_object(
        'foo', x,
        'bar', format('some field %s', x),
        'test', 'hello'
    ),
    now() - interval '1 year' * random(),
    gen_random_uuid(),
    format('comment %s', x)
from
    generate_series(1, 1000) as seq(x);

Проверим случайные записи из середины таблицы:

select * from history
limit 10 offset 855;

┌─[ RECORD 1 ]────────────────────────────────────────────────────────┐
│ id         │ cc896171-b22e-443e-9f1e-d31cc1bf7d0b                   │
│ pk         │ 00000000-0000-0000-0000-000000000056                   │
│ entity     │ events                                                 │
│ operation  │ delete                                                 │
│ doc        │ {"bar": "some field 856", "foo": 856, "test": "hello"} │
│ created_at │ 2025-10-03 08:32:13.55007+03                           │
│ user_id    │ 6a7b7910-8baa-4cd3-8b7a-e2337e3430a8                   │
│ comment    │ comment 856                                            │
├─[ RECORD 2 ]────────────────────────────────────────────────────────┤
│ id         │ b878b211-11bf-4ce6-b3ab-dd23246cde6e                   │
│ pk         │ 00000000-0000-0000-0000-000000000057                   │
│ entity     │ organizations                                          │
│ operation  │ delete                                                 │
│ doc        │ {"bar": "some field 857", "foo": 857, "test": "hello"} │
│ created_at │ 2026-01-09 16:44:54.44607+03                           │
│ user_id    │ 2bec5043-f61e-4dfe-97ad-75fc95588174                   │
│ comment    │ comment 857                                            │
├─[ RECORD 3 ]────────────────────────────────────────────────────────┤
│ id         │ c92ee4a1-f634-4d94-bf59-0b9e727c17c5                   │
│ pk         │ 00000000-0000-0000-0000-000000000058                   │
│ entity     │ events                                                 │
│ operation  │ delete                                                 │
│ doc        │ {"bar": "some field 858", "foo": 858, "test": "hello"} │
│ created_at │ 2026-03-15 23:42:29.06367+03                           │
│ user_id    │ baa57871-0134-4b66-8c73-bfa6772f2b84                   │
│ comment    │ comment 858                                            │
├─[ RECORD 4 ]────────────────────────────────────────────────────────┤

Убедимся, что конкретный документ содержит десять версий:

select count(*) from history
where pk = '00000000-0000-0000-0000-000000000056';

┌───────┐
│ count │
├───────┤
│    10 │
└───────┘

Добавим составной индекс:

create unique index idx_history_pg_created_at
on history (pk, created_at desc);

Следующий запрос вернет версии документа по убыванию даты. Для краткости мы выбираем только ключ и дату:

select id, created_at from history
where pk = '00000000-0000-0000-0000-000000000056'
order by created_at desc;

┌──────────────────────────────────────┬──────────────────────────────┐
│                  id                  │          created_at          │
├──────────────────────────────────────┼──────────────────────────────┤
│ 13aad2e6-dc8d-4506-a502-4a906f5098ae │ 2026-05-25 13:51:35.42847+03 │
│ 61e69328-cf7d-4c08-b1e4-4348ec90a355 │ 2026-03-06 13:49:14.76927+03 │
│ 2114a3bd-8227-4fc2-8d19-792b67efb24c │ 2026-02-28 22:41:54.82047+03 │
│ 41e96a49-dddc-45c2-acec-440476fcbae7 │ 2026-01-26 15:56:18.70527+03 │
│ 9d0bb1e6-31a5-4c15-b860-56ca36bb2bf1 │ 2026-01-03 04:48:32.26047+03 │
│ 44c8ddf0-0a24-4f98-b05f-7cd920bf06f0 │ 2025-11-27 12:14:12.45567+03 │
│ af7004de-ac56-4263-818f-b1de9d07dcaf │ 2025-11-08 01:02:17.17887+03 │
│ cc896171-b22e-443e-9f1e-d31cc1bf7d0b │ 2025-10-03 08:32:13.55007+03 │
│ 2e903e63-65e8-4b78-89f2-938415d5e762 │ 2025-09-20 21:43:27.49887+03 │
│ 8b6d1d09-568e-481a-a96f-a8e3cd9626d7 │ 2025-07-19 18:00:57.48927+03 │
└──────────────────────────────────────┴──────────────────────────────┘

Последнюю версию мы получим, снабдив запрос выражением LIMIT 1:

select id, created_at from history
where pk = '00000000-0000-0000-0000-000000000056'
order by created_at desc
limit 1;

┌──────────────────────────────────────┬──────────────────────────────┐
│                  id                  │          created_at          │
├──────────────────────────────────────┼──────────────────────────────┤
│ 13aad2e6-dc8d-4506-a502-4a906f5098ae │ 2026-05-25 13:51:35.42847+03 │
└──────────────────────────────────────┴──────────────────────────────┘

Актуальную версию на дату мы получим, ограничив поле created_at параметром:

select id, created_at from history
where pk = '00000000-0000-0000-0000-000000000056'
    and created_at <= '2025-11-08 00:00:00+03'::timestamptz
order by created_at desc
limit 1;

┌──────────────────────────────────────┬──────────────────────────────┐
│                  id                  │          created_at          │
├──────────────────────────────────────┼──────────────────────────────┤
│ cc896171-b22e-443e-9f1e-d31cc1bf7d0b │ 2025-10-03 08:32:13.55007+03 │
└──────────────────────────────────────┴──────────────────────────────┘

Если предварить запросы командой EXPLAIN, план подтвердит, что мы попали в индекс.

Создание версий

Итак, хранище версий готово. Обсудим теперь, как перемещать документы между основной и исторической таблицами.

Уточним, что версию создают только обновлении или удалении документа. Это значит, мы заинтересованы в запросах UPDATE и DELETE. При создании документа (INSERT) порождать версию бессмысленно: она будет точной копией документа.

Когда речь заходит об автоматическом действиях, часто упоминают триггеры. Напомним, триггер – это реакция на событие в базе, например вставку записи, обновление или удаление. Триггер связывает событие с функцией, которая выполняет побочный эффект. Триггеры вызываются на разных этапах: до или после события, при этом им доступны прежняя и новая версии записи.

Хотя автор не поощряет триггеры, наше первое решение будет основано на них.

Версии на триггерах

Чтобы при обновлении и удалении заявки в таблице history появлялась версия документа, проделаем следующее. Сперва подготовим триггерную функцию, которая выполняет вставку в history. Триггерные функции отличаются от обычных тем, что им доступны переменные OLD, NEW, TG_OP и другие. OLD и NEW Содержат прежнюю и новую версии строки, связанной с событием. TG_OP хранит имя операции: UPDATE или DELETE. Функция:

create or replace function fn_applications_history()
returns trigger as $$
begin
    insert into history(pk, entity, operation, doc, created_at)
    values (
        OLD.id,
        'applications',
        TG_OP,
        old.doc,
        current_timestamp
    );
    return OLD;
end;
$$ language plpgsql;

Команда create trigger связывает таблицу, событие и функцию, которую следует вызывать. Вызовем команду дважды для обновления и удаления. Мы заинтересованы в прошлой версии документа (до события), поэтому указываем before delete/update:

create trigger trg_application_before_delete
before delete on applications
for each row execute function fn_applications_history();

create trigger trg_application_before_update
before update on applications
for each row execute function fn_applications_history();

Обновим одну из заявок и проверим таблицу версий:

update applications
set doc['extra'] = to_jsonb(42)
where id = '00000000-0000-0000-0000-000000123999';

select
    id,
    pk,
    entity,
    operation,
    created_at,
    jsonb_pretty(doc) as doc
from history where pk = '00000000-0000-0000-0000-000000123999';

┌─[ RECORD 1 ]───────────────────────────────────────────────────────────────────┐
│ id         │ 009cc4ce-ea6b-4ea1-87b1-bed7825c8781                              │
│ pk         │ 00000000-0000-0000-0000-000000123999                              │
│ entity     │ applications                                                      │
│ operation  │ UPDATE                                                            │
│ created_at │ 2026-06-21 15:52:54.03302+03                                      │
│ doc        │ {                                                                ↵│
│            │     "id": "00000000-0000-0000-0000-000000123999",                ↵│
│            │     "status": "approved",                                        ↵│
│            │     "amounts": [                                                 ↵│
│            │         {                                                        ↵│
│            │             "amount": 41513124,                                  ↵│
│            │             "period": {                                          ↵│
│            │                 "d": 4,                                          ↵│
│            │                 "m": 8,                                          ↵│
│            │                 "w": 7,                                          ↵│
│            │                 "y": 1                                           ↵│
│            │             },                                                   ↵│
│            │             "currency": "USD"                                    ↵│

Видим, что версия с таким ключом появилась. То же самое с удалением: перед тем как запись исчезнет из основной таблицы, ее копия окажется в архивной таблице. Быстрая проверка:

delete from applications
where id = '00000000-0000-0000-0000-000000012345';

select
    id,
    pk,
    entity,
    operation,
    created_at,
    jsonb_pretty(doc) as doc
from history where pk = '00000000-0000-0000-0000-000000012345';

┌─[ RECORD 1 ]───────────────────────────────────────────────────────────────────┐
│ id         │ e0437f5d-19b6-4850-a2ae-51b21e626a0e                              │
│ pk         │ 00000000-0000-0000-0000-000000012345                              │
│ entity     │ applications                                                      │
│ operation  │ DELETE                                                            │
│ created_at │ 2026-06-21 15:48:55.118652+03                                     │
│ doc        │ {                                                                ↵│
│            │     "id": "00000000-0000-0000-0000-000000012345",                ↵│
│            │     "status": "archived",                                        ↵│
│            │     "amounts": [                                                 ↵│
│            │         {                                                        ↵│
│            │             "amount": 40528181,                                  ↵│
│            │             "period": {                                          ↵│
│            │                 "d": 7,                                          ↵│
│            │                 "m": 7,                                          ↵│

В целом задача выполнена: версии создаются автоматически, и теперь мы обсудим детали.

Доводы против триггеров

Выше мы упомянули, что, возможно, триггеры – не лучшее решение для версий. Дело в том, что триггеры слишком буквальны: они выполняются строго на все операции UPDATE и DELETE. У этой строгости недостаток: она не допускает исключений, а порой они необходимы. Например, при изменении некоторых полей создавать версии не нужно. Также они не требуются в служебных миграциях: исправлении дат, часовых поясов, корректировки ссылок. Наоборот, будет хуже, если мы исправили миллион документов и получили лишний миллион версий.

Триггеры замедляют тесты. Как правило, каждый тест начинается с того, что создает в базе документы. В процессе они меняются, а в конце сессии – удаляются. Триггеры, хоть и немного, но замедляют весь процесс.

Можно временно отключить триггеры для тех тестов, которые в них не нуждаются. Для этого служит команда alter table ... disable trigger и ее обратное действие enable. Эти запросы помещают в фикстуру, которая выполняется однажды для блока тестов:

alter table applications disable trigger trg_application_before_delete;
/* run all tests */
alter table applications disable trigger trg_application_before_delete;

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

Еще один довод не использовать триггеры в том, что некоторые документы меняются слишком часто. Поминутные версий избыточны и вдобавок мешают пользователям: им трудно отличить важное изменение от косметического. Скорее всего, вас попросят сделать так, чтобы версия сохранялась не чаще чем раз в десять минут и около. Перед тем как записать новую версию, нужно узнать дату последней и рассчитать разницу во времени. Если она меньше порога, версию не создавать.

Подобную логику можно выразить триггерами. Однако чем больше условий, тем менее триггеры очевидны для команды. Со временем их поддержка становится дорогой.

Версии в приложении

Другое возможное решение – поручить создание версий приложению. Если оно написано на Python, Java и другом высокоуровневом языке, скорее всего для работы с базой используется ORM. Почти все ORM поддерживают события, которые вызываются до или после операции над сущностью. В зависимости от языка они называются сигналами, хуками или иначе, но принцип одинаков.

Идея в том, чтобы назначить модели событие before update или before delete, а в обработчике записать документ в историческую таблицу. Приведем пример для ORM из фреймворка Django:

from django.db.models.signals import pre_save
from django.dispatch import receiver

from project.models import Application, History

@receiver(pre_save, sender=Application)
def save_history(sender, instance, **kwargs):
    if not instance.pk:
        return
    History.objects.create(
        pk=instance.id,
        entity='application',
        operation='UPDATE',
        doc=instance.doc
    )

Декоратор receiver объединяет событие pre_save (до записи в базу), модель Application и обработчик save_history. Последний сработает до того, как вызван метод .save() модели. Параметр instance содержит саму модель (экземпляр Application), которую мы намерены изменить. Если она новая (первичный ключ не заполнен), версия не создается.

Обработчиков, которые прослушивают одни и те же событие и модель, может быть несколько. В этом случае они вызываются в порядке регистрации.

По аналогии устроена реакция на удаление модели: разница лишь в типе сигнала (pre_delete).

Недостаток сигналов в том, что они работают только для одного экземпляра модели. При пакетных операциях (bulk_update, bulk_delete) они не вызываются. Кроме того, сигналы выполняются силами приложения, и база данных ничего не знает о них. Если подключиться к базе из другого фреймворка, сигналы первого не будут работать.

Чтобы создание версий работало везде, составим специальные запросы. Они могут храниться в отдельном репозитории или быть частью базы, например вызываться из процедур.

Версии в Postgres

Опишем удаление документа на SQL. Это подготовленный оператор с одним параметром – ключом документа. В нем две операции: удаление из основной таблицы и вставка в историческую. Обе операции выполняются атомарно, им доступен один и тот же снимок данных.

prepare delete_application as
with
OLD as (
    delete from applications where id = $1::uuid
    returning *
)
insert into history(pk, entity, operation, doc, created_at)
select
    OLD.id,
    'applications',
    'DELETE',
    old.doc,
    current_timestamp
from
    OLD;

Сперва запрос с псевдонимом OLD удаляет документ из основной таблицы, при этом удаленные строки возвращаются. В главном запросе OLD выступает источником данных: строки из него вставляются в таблицу версий. Данные словно перетекают из одной таблицы в другую. Процесс происходит атомарно и без нашего участия: клиент ничего не передает через себя, а только командует.

Убедимся, что вызов оператора порождает копии:

execute delete_application('00000000-0000-0000-0000-000000100999'::uuid);

select id from applications
where id = '00000000-0000-0000-0000-000000100999';
-- (0 rows)

В таблице заявок таковой больше нет. Документ переехал в history:

select
    id,
    pk,
    entity,
    operation,
    created_at,
    jsonb_pretty(doc) as doc
from history where pk = '00000000-0000-0000-0000-000000100999';

┌─[ RECORD 1 ]───────────────────────────────────────────────────────────────────┐
│ id         │ 78ce995b-c865-489c-8c24-0ab26d62caef                              │
│ pk         │ 00000000-0000-0000-0000-000000100999                              │
│ entity     │ applications                                                      │
│ operation  │ DELETE                                                            │
│ created_at │ 2026-06-21 16:16:23.510292+03                                     │
│ doc        │ {                                                                ↵│
│            │     "id": "00000000-0000-0000-0000-000000100999",                ↵│
│            │     "status": "archived",                                        ↵│
│            │     "amounts": [                                                 ↵│
│            │         {                                                        ↵│
│            │             "amount": 11520867,                                  ↵│
│            │             "period": {                                          ↵│
│            │                 "d": 2,                                          ↵│
│            │                 "m": 8,                                          ↵│
│            │                 "w": 6,                                          ↵│
│            │                 "y": 1                                           ↵│
│            │             },                                                   ↵│
│            │             "currency": "EUR"                                    ↵│

Подчеркнем, что логика не зависит от того, из какого языка или фреймворка обращается клиент.

Обновление документа несколько сложнее. Сначала документ копируют в таблицу версий, а затем обновляют. Оператор update_application ниже ожидает два параметра: ключ документа $1::uuid и его тело $2::jsonb:

prepare update_application as
with
OLD as (
    select * from applications
    where id = $1
),
NEW as (
    update applications
    set doc = $2::jsonb
    where id = $1::uuid
    returning *
)
insert into history(pk, entity, operation, doc, created_at)
select
    OLD.id,
    'applications',
    'UPDATE',
    OLD.doc,
    current_timestamp
from
    OLD;

Очевидны три этапа запроса:

  • OLD читает прежний документ;
  • NEW обновляет текущий;
  • insert записывает OLD в историческую таблицу.

Обновим одну из заявок и убедимся, что в таблице applications ее новая версия:

execute update_application(
    '00000000-0000-0000-0000-000000100321'::uuid,
    $$
{
    "application_id": 100321,
    "some_field": "test"
}
    $$::jsonb
);

select doc from applications
where id = '00000000-0000-0000-0000-000000100321';

┌─[ RECORD 1 ]───────────────────────────────────────────┐
│ doc │ {"some_field": "test", "application_id": 100321} │
└─────┴──────────────────────────────────────────────────┘

А исторической таблице – прежняя:

select
    id,
    pk,
    entity,
    operation,
    created_at,
    jsonb_pretty(doc) as doc
from history where pk = '00000000-0000-0000-0000-000000100321';

┌─[ RECORD 1 ]───────────────────────────────────────────────────────────────────┐
│ id         │ 47e6cdba-e1fe-4320-acdb-ef193ef7f3e6                              │
│ pk         │ 00000000-0000-0000-0000-000000100321                              │
│ entity     │ applications                                                      │
│ operation  │ UPDATE                                                            │
│ created_at │ 2026-06-21 16:27:53.384049+03                                     │
│ doc        │ {                                                                ↵│
│            │     "id": "00000000-0000-0000-0000-000000100321",                ↵│
│            │     "status": "archived",                                        ↵│
│            │     "amounts": [                                                 ↵│
│            │         {                                                        ↵│
│            │             "amount": 50222648,                                  ↵│
│            │             "period": {                                          ↵│

Ограничение по времени

Усложним требования: предположим, версию документа нужно создавать не чаще чем раз пять минут. Изучите новый код, а ниже мы разберем его:

prepare update_application_throttled as
with
upsert as (
    insert into applications(id, doc)
    values ($1::uuid, $2::jsonb)
    on conflict (id) do update set
    doc = excluded.DOC,
    updated_at = now()
    returning NEW.id, OLD.doc as doc_old
)
insert into history(pk, entity, operation, doc, created_at)
select
    id,
    'applications',
    'UPDATE',
    doc_old,
    current_timestamp
from
    upsert
where
        doc_old is not null
    and not exists(
      select id from history
      where
          pk = upsert.id
          and created_at > now() - interval '1 minute'
);

Ключевая деталь кроется в самом конце в выражении not exists. Подзапрос ищет в исторической таблице запись таким же ключом и датой создания позже порога. Если запись нашлась, not exists вернет ложь. В результате insert получит пустой набор записей, и версия не будет создана.

Еще одно улучшение кроется в части upsert. Как ясно из названия, она не только обновляет документ, но и создает в случае отсутствия. При этом возвращается прежнее значение документа – поле OLD.doc под именем doc_old. За счет него мы определим тип события: вставку или обновление. Значение doc_old помещается в историческую таблицу, если оно не пустое. В противном случае документ был создан, и порождать версию нет смысла.

Заметим, что в выражении on conflict переменные OLD и NEW доступны с версии 18. Для младших версий мы бы переписали запрос так, чтобы OLD был подзапросом:

with
OLD as (
    select * from applications
    where id = $1
)

Теперь проверим наш код. Для начала вставим простейший документ {"foo": "1"}. Оператор вернул INSERT 0 0, что означает, что вставка в историческую таблицу не состоялась. Это ожидаемое поведение, потому что для нового документа версия не создается.

execute update_application_throttled(
    '00000000-0000-0000-0000-000010123123'::uuid,
    $$
{
    "foo": "1"
}
    $$::jsonb
);
-- INSERT 0 0

Обновим тот же документ и увидим INSERT 0 1 – версия создана, потому что в пределах пяти минут не нашлось других версий.

execute update_application_throttled(
    '00000000-0000-0000-0000-000010123123'::uuid,
    $$
{
    "foo": "2"
}
    $$::jsonb
);
-- INSERT 0 1

Если обновить документ еще раз без ожидания, версия создана не будет:

execute update_application_throttled(
    '00000000-0000-0000-0000-000010123123'::uuid,
    $$
{
    "foo": "3"
}
    $$::jsonb
);
-- INSERT 0 0

Проверим таблицы: в application хранится текущий документ {"foo": "3"}, а в history – версия {"foo": "1"}. Версию {"foo": "2"} мы пропустили, потому что с момента {"foo": "1"} прошло меньше пяти минут.

table applications;
┌─[ RECORD 1 ]──────────────────────────────────────┐
│ id         │ 00000000-0000-0000-0000-000010123123 │
│ doc        │ {"foo": "3"}                         │
│ created_at │ 2026-08-18 09:32:46.966243+03        │
│ updated_at │ 2026-08-18 09:33:00.341051+03        │
└────────────┴──────────────────────────────────────┘

table history;
┌─[ RECORD 1 ]──────────────────────────────────────┐
│ id         │ 8e6b2261-76b1-4aa2-8e4f-5771faaf1e8f │
│ pk         │ 00000000-0000-0000-0000-000010123123 │
│ entity     │ applications                         │
│ operation  │ UPDATE                               │
│ doc        │ {"foo": "1"}                         │
│ created_at │ 2026-08-18 09:32:54.304452+03        │
│ user_id    │ <null>                               │
│ comment    │ <null>                               │
└────────────┴──────────────────────────────────────┘

Подождем 5 минут и обновим документ еще раз:

execute update_application_throttled(
    '00000000-0000-0000-0000-000010123123'::uuid,
    $$
{
    "foo": "4"
}
    $$::jsonb
);
-- INSERT 0 1

На этот раз в исторической таблице окажется версия {"foo": "3"}:

table history;
┌─[ RECORD 1 ]──────────────────────────────────────┐
│ id         │ 8e6b2261-76b1-4aa2-8e4f-5771faaf1e8f │
│ pk         │ 00000000-0000-0000-0000-000010123123 │
│ entity     │ applications                         │
│ operation  │ UPDATE                               │
│ doc        │ {"foo": "1"}                         │
│ created_at │ 2026-08-18 09:32:54.304452+03        │
│ user_id    │ <null>                               │
│ comment    │ <null>                               │
├─[ RECORD 2 ]──────────────────────────────────────┤
│ id         │ 5fa063d6-5e43-4cb3-b1ef-8d12d82657fa │
│ pk         │ 00000000-0000-0000-0000-000010123123 │
│ entity     │ applications                         │
│ operation  │ UPDATE                               │
│ doc        │ {"foo": "3"}                         │
│ created_at │ 2026-08-18 09:44:15.368447+03        │
│ user_id    │ <null>                               │
│ comment    │ <null>                               │
└────────────┴──────────────────────────────────────┘

Временной порог (число минут, часов и так далее) можно вынести в параметр или переменную сеанса.

Восстановление версии

Итак, у нас есть все, чтобы создавать версии документов. Однако мы не рассмотрели, как восстанавливать их. Без восстановления версии не имеют смысла, ведь иначе зачем создавать их?

Восстановление может быть устроено по-разному. Наш алгоритм ожидает ключ документа и дату версии. Его шаги следующе:

  • текущий документ удаляется и помещается в переменную;
  • из нее создается историческая запись (версия);
  • целевая версия удаляется из исторической таблицы…;
  • …и добавляется в основную.

Получаются две перестановки: из главной таблицы в историческую и наоборот. Приведем подготовленный оператор restore_application, выполняющий эти действия. Ключ документа передается первым параметром ($1::uuid), версия – вторым ($2::timestamptz).

prepare restore_application as
with
del_current_doc as (
    delete from applications where id = $1::uuid
    returning *
),
create_version as (
    insert into history(pk, entity, operation, doc, created_at)
    select
        id,
        'applications',
        'RESTORE',
        doc,
        current_timestamp
    from
        del_current_doc
),
delete_version as (
    delete from history
    where
            pk = $1::uuid
        and created_at = $2::timestamptz
    returning *
)
insert into applications (id, doc, updated_at)
select
    id, doc, current_timestamp
from
    delete_version;

Если вызвать оператор для одной из версий, сработают указанные перестановки. Для экономии места мы опустим демонстрацию; предлагаем читателю опробовать ее на документе {"foo": "4"} из прошлого параграфа.

Архивация версий

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

Архивация включает несколько шагов:

  • выбрать документы старше определенного срока;
  • записать их в файл, сжать и передать в долгосрочное хранилище;
  • удалить из базы обработанные документы.

Задача выполняется регулярно, например раз в месяц.

Архивация решается разными способами, в том числе силами Java, Python и других языков. Ниже мы рассмотрим, как сделать это в Postgres без привлечения других технологий. Изучите следующий запрос:

copy (delete from history where created_at < now() - interval '1 year' returning *)
to program 'gzip > /path/to/history.csv.gzip'
with (format csv, header on);

Несмотря на краткость, он выполняет все указанное выше. Во-первых, версии, устаревшие на год, удаляются из таблицы (внутренний запрос DELETE). Данные, что он удалил, возвращаются оператором RETURNING. Выражение COPY приводит записи к формату CSV и передает процессу gzip, а тот сжимает результат в файл.

Вместо to program можно указать to stdout, и тогда данные получит клиент. Уточним, что в терминах Postgres под stdout понимается не стандартный канал Unix, а клиентская сторона. Postgres передает серию сообщений CopyData; клиент принимает их и как-то обрабатывает, например записывает в файл.

Тонкости оператора COPY мы рассмотрели в седьмой главе про отчетность. Обратитесь к ней, если что-то непонятно.

Напомним, что CSV-файлы подлежат сжатию. Даже обычные алгоритмы zip и gzip будут хорошим выбором. Еще большего выигрыша можно добиться, выбрав lz4 (утилита доступна во всех менеджерах пакетов). Алгоритмы LZMA2 и ZPAQ предлагают своего рода компромисс: крайне эффективное сжатие при медленной распаковке. Как правило, именно их выбирают для долгосрочного хранения версий, логов и подобной информации. Распаковка случается редко, а остальное время мы платим за объем.

Для полноты картины приведем процедуру truncate_and_dump_history, которая записывает удаленные версии в файл. Особенность в том, что выражение copy строится функцией format и выполняется оператором execute. Форматирование необходимо, чтобы путь к файлу включал в себя год, месяц и день. Для этого параметр %s внутри пути заменяется значением to_char(now(), 'yyyy_mm_dd').

create or replace procedure truncate_and_dump_history()
language plpgsql as $$
begin
    execute format($sql$
copy (delete from history where created_at < now() - interval '1 year' returning *)
to program 'gzip > /Users/ivan/work/pg-json-book-code/history_%s.csv.gzip' with (format csv, header on);
    $sql$, to_char(now(), 'yyyy_mm_dd'));
end;
$$;

Мы идем на подобные ухищрения, потому что copy не принимает параметров: все значения должны находиться в теле запроса. Процедуру легко расширить так, чтобы срок давности (interval '1 year') передавался параметром, ровно как и директория, куда записывать файл.

Как и в случае отчетами, задачу ставят на расписание при помощи расширения pg_cron. Тело задачи сводится к вызову процедуры:

call truncate_and_dump_history().

Стандарт JSON Patch

Последний вопрос этой главы касается стандарта JSON Patch (не путайте с JSON Path). Прежде чем рассказать, что это такое, рассмотрим, какую задачу мы надеемся решить с его помощью.

Когда мы сохраняем версию документа, то копируем его в историческую таблицу. При этом разница между документами может быть мала: скажем, поле status изменилось с pending на active. Это всего лишь одно поле, однако функция ничего не знает о характере обновления. Ей все равно, изменилось одно поле или десятки – документ копируется целиком.

Мы храним версии документа, но не можем ответить на вопрос, что именно в нем изменилось. Порой это необходимо: клиенты просят, чтобы при просмотре документа в боковом виджете были недавние изменения. Программистам тоже удобно, когда система предлагает быстрый способ сравнить документы. Это полезно при расследовании инцидентов или ошибок операторов. Сравнение документов вручную в программе Kdiff3 и аналогах занимает много времени. Здесь и приходит на помощь JSON Patch.

JSON Patch – это соглашение о том, как описать изменения в JSON-документе. Стандарт доступен на официальном сайте по одноименному адресу. JSON Patch – это даже не язык, а данные: массив объектов с полями op, path, value и некоторые другие. Каждый объект описывает одну операцию над документом, а массив определяет порядок.

Предположим, имеются два документа:

{
  "a": 1,
  "c": [1, 3],
  "d": true
}

{
  "b": 2,
  "c": [1, 2, 3]
}

Какие действия необходимы, чтобы перейти от первого ко второму? Вот как выглядит ответ на языке JSON Patch:

[
    {
        "op": "remove",
        "path": "/d"
    },
    {
        "op": "replace",
        "path": "/c/1",
        "value": 2
    },
    {
        "op": "add",
        "path": "/c/2",
        "value": 3
    },
    {
        "op": "remove",
        "path": "/a"
    },
    {
        "op": "add",
        "path": "/b",
        "value": 2
    }
]

Каждый шаг легко выразить на человеческом языке.

  • Сначала из документа удаляется поле d (путь /d).
  • В массиве "c" второй элемент меняется с 3 на 2.
  • К списку добавляется элемент 3.
  • Удаляется поле a.
  • Добавляется поле b со значением 2.

Если применить эти действия к первому документу, получим второй.

Не путайте JSON Patch с JSON Path (англ. patch – “заплатка”, path – “путь”). Это разные стандарты, причем первый описывает данные, а второй – язык. В разговорной речи JSON Patch называют патчем, заплаткой или “дифом” от английского difference – “разница”.

Из описания видно, что JSON Patch довольно прост. Он подразумевает всего шесть операций над документом: add, remove, replace, copy, move, test. Имея исходный документ и список изменений, легко применить их в цикле. Вычисление разницы между документами – напротив, сложная задача, и чуть ниже мы ее рассмотрим.

Замечания заслуживает поле “path” – путь к элементу. С этим полем связан еще один стандарт – Jsonpointer, который определяет, как строится путь к произвольному JSON-элементу. Стандарт имеет код RFC 6901 и может быть прочитан на сайте ietf.org.

Теперь когда мы знакомы с JSON Patch, объясним, как он поможет в версионировании документов. Если пользователи часто смотрят историю изменений, их можно материализовать – хранить в отдельной колонке. Для этого добавим в таблицу history поле patch с типом jsonb:

alter table history
add column patch jsonb null;

Ожидается, что в нем хранится патч – JSON-массив объектов. Если клиент хочет знать, какие изменения произошли в указанной версии, он передает ключ документа и версию. В ответ ему высылают содержимое колонки patch, например:

[
    {"op": "add", "path": "/status", "value": "pending"},
    {"op": "add", "path": "/accounts", "value": [1, 2, 3]},
    {"op": "replace", "path": "/some_field", "value": "foo"}
]

Добавьте к этим данным другие поля из таблицы history: пользователя, который произвел версию, комментарий, дату. У клиента окажется все необходимое, чтобы вывести изменения в удобном виде.

Патчи в приложении

Подумаем теперь, как заполнить колонку patch. Можно сделать это двумя способами: в приложении и в базе данных. Если приложение написано на Python, для работы с патчами понадобится пакет jsonpatch. Установите его командой:

pip3 install jsonpatch

Пользоваться им просто: статичный метод JsonPatch.from_diff принимает две структуры данных и возвращает экземпляр JsonPatch. Приведение его к строке дает патч в виде JSON. Метод apply принимает документ и применяет к нему патч, возвращая новый документ. Этих средств достаточно для нашей работы. Небольшая демонстрация:

import jsonpatch

# declare two documents
doc_old = {
  "a": 1,
  "c": [1, 3],
  "d": True
}

doc_new = {
  "c": [1, 2, 3],
  "b": 2
}

# create a patch object
patch = jsonpatch.JsonPatch.from_diff(doc_old, doc_new)

# get JSON representation of the patch
str(patch)
# [{"op": "remove", "path": "/a"}, {"op": "remove", "path": "/d"}, {"op": "add", "path": "/b", "value": 2}, {"op": "add", "path": "/c/1", "value": 2}]

# Apply the patch to the old document
doc_new2 = patch.apply(doc_old)

# Both new docs are equal
doc_new == doc_new2
# True

Когда приложение создает версию документа, заполняется поле patch. Необходимы старый и новый экземпляры документа, чтобы передать их в метод from_diff и этот самый патч получить. При этом вопрос, в какой момент вам доступны оба экземпляра, зависит от фреймворка и библиотек.

Django ORM не предлагает сигнала, который принимал бы оба экземпляра модели (условные instance_old и instance_new). Вместо этого регистрируют сигнал pre_save, где instance содержит новый экземпляр модели, а прежний мы получим, обратившись к базе методом .get. При этом учитываем, что экземпляр может быть новым (не заполнено поле .pk) либо прежнего документа почему-то не существует (исключение DoesNotExist). В обоих случаях сигнал прекращает работу. Если все идет по плану, из полей doc обеих моделей вычисляется патч, который сохраняется в таблицу версий.

import jsonpatch
from django.db.models.signals import pre_save
from django.forms.models import model_to_dict

@receiver(pre_save, sender=Application)
def save_patch(sender, instance, **kwargs):

    if not instance.pk:
        return

    try:
        app_old = Application.objects.get(pk=instance.pk)
    except sender.DoesNotExist:
        return

    patch = jsonpatch.JsonPatch.from_diff(app_old.doc, instance.doc)

    History.objects.create(
        pk=instance.id,
        entity='application',
        ...
        patch=patch.to_string()
    )

Еще одно средство Django – класс FieldTracker из пакета Model Utils. Это служебное поле, которое не сохраняется в базе, а только отслеживает изменения других полей. Вот как выглядит с ним модель:

from django.db import models
from model_utils import FieldTracker

class Application(models.Model):
    doc = JSONField(...)
    ...
    tracker = FieldTracker()


app = Application.objects.get(pk=...)
app.doc["extra"] = 123
app.save()

В сигнале pre_save мы получим прежнее значение doc методом previous. За счет этого не понадобится запрос к базе:

@receiver(pre_save, sender=Application)
def save_patch(sender, instance, **kwargs):

    if not instance.pk:
        return

    doc_old = instance.tracker.previous("doc")
    if not doc_old:
        return

    patch = jsonpatch.JsonPatch.from_diff(doc_old, instance.doc)

    History.objects.create(
        pk=instance.id,
        entity='application',
        ...
        patch=patch.to_string()
    )

На официальном сайте JSON Patch собраны к библиотеки к популярным языкам программирования. Скорее всего, среди них найдется и тот, на котором написано ваше приложение.

Патчи в базе данных

Рассмотрим другую сторону вопроса: как работать с JSON Patch на уровне базы данных? По умолчанию Postgres не предлагает для этого никаких инструментов. Однако мы умеем работать с патчами на языке Python, а последний поддерживается в Postgres. Отсюда решение: напишем функцию на языке plpython3, которая сделает все необходимое.

Для начала установите пакет jsonpatch в тот интерпретатор, который используется расширением plpython3. Напомним, что одна из частых ошибок в том, что пакет установлен в другой интерпретатор и не находится при запуске кода в Postgres.

Создайте и опробуйте функцию py_make_patch. Она довольно проста: принимает два документа и возвращает разницу:

create or replace function py_make_patch(src jsonb, dst jsonb)
returns jsonb
language plpython3u strict as $$
    import json
    import jsonpatch
    doc_src = json.loads(src)
    doc_dst = json.loads(dst)
    patch = jsonpatch.JsonPatch.from_diff(doc_src, doc_dst)
    return patch.to_string()
$$;

Мы не использовали автоматический вывод типов директивой transform for type jsonb. Дело в том, что приведение класса JsonPatch к jsonb силами Postgres влечет некоторые сложности, в которые мы не будем углубляться. Заинтересованному читателю предлагаем домашнюю работу: добавить в функцию transform for type jsonb и избавиться от вызовов json.loads. Вам понадобится знание о внутреннем устройстве класса JsonPatch и чтение исходного кода.

Проверим функцию на минимальном примере:

select py_make_patch('{"a": 1}'::jsonb, '{"a": 2}'::jsonb) as patch;

┌───────────────────────────────────────────────┐
│                     patch                     │
├───────────────────────────────────────────────┤
│ [{"op": "replace", "path": "/a", "value": 2}] │
└───────────────────────────────────────────────┘

Добавим теперь вторую функцию, которая принимает документ и патч и возвращает новый документ:

create or replace function py_apply_patch(doc jsonb, patch jsonb)
returns jsonb
transform for type jsonb
language plpython3u immutable strict as $$
    import jsonpatch
    patch_obj = jsonpatch.JsonPatch(patch)
    return patch_obj.apply(doc)
$$;

Опробуем ее в действии:

select py_apply_patch(
    $$
{"some_field": "test", "application_id": 100321}
    $$::jsonb,
    $$
[
    {"op": "add", "path": "/status", "value": "pending"},
    {"op": "add", "path": "/accounts", "value": [1, 2, 3]},
    {"op": "replace", "path": "/some_field", "value": "foo"}

]
    $$::jsonb) as doc_new;

┌─────────────────────────────────────────────────────────────────────────────────────────────┐
│                                           doc_new                                           │
├─────────────────────────────────────────────────────────────────────────────────────────────┤
│ {"status": "pending", "accounts": [1, 2, 3], "some_field": "foo", "application_id": 100321} │
└─────────────────────────────────────────────────────────────────────────────────────────────┘

Видим, что все три операции применились корректно.

Версии с патчами

Теперь когда Postgres работает с патчами, перепишем версионирование. Вот как выглядит оператор, который создает или обновляет документ; во втором случае создается версия документа с патчем:

prepare update_application as
with
upsert as (
    insert into applications(id, doc)
    values ($1::uuid, $2::jsonb)
    on conflict (id) do update set
    doc = excluded.DOC,
    updated_at = now()
    returning NEW.id, OLD.doc as doc_old, NEW.doc as doc_new
)
insert into history(pk, entity, operation, doc, patch, created_at)
select
    id,
    'applications',
    'UPDATE',
    doc_old,
    py_make_patch(doc_old, doc_new),
    current_timestamp
from
    upsert
where
    doc_old is not null;

Предположим, исходный документ выглядит следующим образом:

{"some_field": "test", "application_id": 100321}

Обновим его и посмотрим на содержимое таблиц:

execute update_application(
    '00000000-0000-0000-0000-000000100321'::uuid,
    $$
{
    "application_id": 100321,
    "status": "pending",
    "some_field": "foo",
    "accounts": [1, 2, 3]
}
    $$::jsonb
);

select
    id, pk, entity, doc,
    jsonb_pretty(patch) as patch
from
    history;

┌─[ RECORD 1 ]──────────────────────────────────────────────┐
│ id     │ dcbb3da3-5f2c-4ae7-861c-5c401c550905             │
│ pk     │ 00000000-0000-0000-0000-000000100321             │
│ entity │ applications                                     │
│ doc    │ {"some_field": "test", "application_id": 100321} │
│ patch  │ [                                               ↵│
│        │     {                                           ↵│
│        │         "op": "add",                            ↵│
│        │         "path": "/status",                      ↵│
│        │         "value": "pending"                      ↵│
│        │     },                                          ↵│
│        │     {                                           ↵│
│        │         "op": "add",                            ↵│
│        │         "path": "/accounts",                    ↵│
│        │         "value": [                              ↵│
│        │             1,                                  ↵│
│        │             2,                                  ↵│
│        │             3                                   ↵│
│        │         ]                                       ↵│
│        │     },                                          ↵│
│        │     {                                           ↵│
│        │         "op": "replace",                        ↵│
│        │         "path": "/some_field",                  ↵│
│        │         "value": "foo"                          ↵│
│        │     }                                           ↵│
│        │ ]                                                │
└────────┴──────────────────────────────────────────────────┘

В колонке patch оказались следующие операции:

  • {"op": "add", "path": "/status", "value": "pending"}
  • {"op": "add", "path": "/accounts", "value": [1, 2, 3]}
  • {"op": "replace", "path": "/some_field", "value": "foo"}

Если применить к версии операции из поля patch, получим текущий документ.

Прием патчей от клиента

На базе JSON Patch можно построить еще одно решение. Если клиент хочет изменить документ, можно принять от него только патч. Его применяют к текущему документу, и дополнительно мы создаем версию. Вычислять патч при этом не нужно, потому что он уже есть. Подход оправдывает себя на крупных документах, которые дорого передавать по сети. Патч практически всегда меньше документа.

Напомним, мы уже написали функцию py_apply_patch, которая принимает документ, патч и возвращает новый документ:

create or replace function py_apply_patch(doc jsonb, patch jsonb)
returns jsonb
transform for type jsonb
language plpython3u immutable strict as $$
    import jsonpatch
    patch_obj = jsonpatch.JsonPatch(patch)
    return patch_obj.apply(doc)
$$;

Добавим оператор patch_application_with_history от двух параметров: ключа документа ($1::uuid) и патча ($2::jsonb). Стадия step_old запоминает документ до того, как мы наложим на него патч. Стадия step_update обновляет документ, задействуя функцию py_apply_patch. В основной части мы записываем версию документа, используя старое значение из step_old и переданный патч.

prepare patch_application_with_history as
with
step_old as (
  select * from applications where id = $1::uuid
),
step_update as (
  update applications set doc = py_apply_patch(doc, $2::jsonb)
  where id = $1::uuid
)
insert into history(pk, entity, operation, doc, patch, created_at)
  select
      step_old.id,
      'applications',
      'UPDATE',
      step_old.doc,
      $2::jsonb,
      current_timestamp
  from
      step_old;

Вызовем оператор для одного из документов:

execute patch_application_with_history(
    '00000000-0000-0000-0000-000000100321'::uuid,
    $$
    [
        {"op": "add", "path": "/comment", "value": "updated using a JSON patch"}
    ]
    $$::jsonb
);

Проверим историю: в ней окажется прошлый документ и патч.

select
    jsonb_pretty(doc) as doc,
    jsonb_pretty(patch) as patch
from
    history
where
    pk = '00000000-0000-0000-0000-000000100321';

┌──────────────────────────────┬───────────────────────────────────────────────┐
│             doc              │                     patch                     │
├──────────────────────────────┼───────────────────────────────────────────────┤
│ {                           ↵│ [                                            ↵│
│     "some_field": "test",   ↵│     {                                        ↵│
│     "application_id": 100321↵│         "op": "add",                         ↵│
│ }                            │         "path": "/comment",                  ↵│
│                              │         "value": "updated using a JSON patch"↵│
│                              │     }                                        ↵│
│                              │ ]                                             │
└──────────────────────────────┴───────────────────────────────────────────────┘

Нам удалось изменить документ, не передавая его целиком. Очевидный плюс в экономии трафика: патч на порядок меньше документа, изменения которого описывает. На этом преимущества патчей не заканчиваются: в следующем параграфе мы рассмотрим еще одну технику.

Коротко об event-sourcing

Event-sourcing – это подход, когда хранятся не только данные, и но и события, которые эти данные формируют. При этом сущность можно пересобрать, применив события в том же порядке. Последнее особенно важно: события хранятся не просто для истории, они равноценны данным. Сушности, построенные из событий, по-другому называют агрегатами. Говорят, что агрегат равен сумме его событий.

Когда сущность строится из событий, это дает интересные возможности. В любой момент ее можно “переиграть” – собрать с нуля. Если ограничить события датой (например, не позже 2026 года), то получим версию документа в прошлом. В редких случаях событие можно удалить и собрать агрегат без него. События полезны в отладке и по-своему заменяют логи.

Не путайте event-sourcing с версионированием. Хоть мы и рассматриваем их в одной главе, это не одно и то же. Версии хранят полноценные документы, а события описывают отдельные изменения. Версия самодостаточна, а событию требуются вся цепочка. Рассматривайте версии как бекапы, а события – как миграции отдельных документов.

Версии и события не противоречат, а дополняют друг друга. В проекте, где работал автор, использовались оба подхода для разных нужд. Система событий позволяла в любой момент собрать агрегат заново. Этим пользовались, когда определенное событие признавали ошибочным. Версии были необходимы, потому что от них отталкивались некоторые операции, и вычислять версию через события было долго.

Простым примером event-sourcing служит история банковского счета. Когда клиент вносит деньги на счет, мы не просто увеличиваем баланс. Пополнение счета записывается в таблицу операций и хранится вместе с другими фактами пополнения, снятия, переводов, комиссий и других. Таким образом баланс – это сумма всех операций. Баланс материализуют (хранят в отдельном поле), потому что рассчитывать его каждый раз дорого. Однако важно понимать, что совокупность событий важнее результата. Если отдельную операцию признали ошибочной или незаконной, баланс пересчитают.

Еще одно уточнение касается слова “сумма”. Под ним мы имеем в виду результат применения операций. Какие именно действия выполняет событие – зависит от его типа. В случае с банковским счетом баланс нельзя вычислить арифметически, потому что иные события включают процент на остаток или конвертацию валют.

В прошлом параграфе мы научились хранить патч для каждого изменения. Легко заметить, что получился своего рода event-sourcing. Действительно: если взять исходный документ и применить к нему серию патчей, получим текущий (последний) документ. Если результаты отличаются, то имеет место нарушение: либо события теряются, либо кто-то исправил документ в обход нашей системы.

Сбор агрегатов из событий

Рассмотрим, как применить набор патчей к исходному документу. У нас уже достаточный опыт работы с jsonpatch, поэтому приведем минимальные примеры. Если doc – начальный документ, а patches – список патчей, код выглядит так:

import jsonpatch

doc = {}
patches = [
    [{"op": "add", "path": "/a", "value": 1}],
    [{"op": "add", "path": "/b", "value": 2}],
]

for patch in patches:
    p = jsonpatch.JsonPatch(patch)
    doc = p.apply(doc)

print(doc)
# {'a': 1, 'b': 2}

Программисты, знакомые с функциональным программированием, заметят, что это свертка, известная также как fold или reduce. Чтобы выполнить свертку, достаточно функции приращения – такой, что принимает аккумулятор и очередной элемент и возвращает новый аккумулятор. Перепишем код в виде свертки:

import jsonpatch
from functools import reduce

def step_fn(acc, patch):
    p = jsonpatch.JsonPatch(patch)
    return p.apply(acc)

doc = reduce(step_fn, patches, {})

print(doc)
# {'a': 1, 'b': 2}

Мы не случайно упомянули свертку. Дело в том, что агрегатные функции Postgres строятся на том же принципе: им необходим начальный элемент и функция приращения. Зная это, мы напишем агрегатную функцию, которая, если передать ей столбец патчей, применит их один за другим. С ее помощью мы получим документ одним запросом к таблице событий.

Объявим агрегатную функцию json_patch_agg:

create aggregate json_patch_agg (jsonb) (
    sfunc = py_apply_patch,
    stype = jsonb,
    initcond = '{}'
);

Из многих параметров мы указали три:

  • sfunc (state function): функция приращения. Принимает два параметра, где первый – аккумулятор, который накапливает значение. Второй аргумент функции – очередное значение колонки. Функция возвращает новый аккумулятор.
  • stype: тип, с которым работает агрегатная функция. В нашем случае это jsonb.
  • initcond: начальный (пустой) элемент. Используется в качестве аккумулятора при первом вызове функции приращения.

Напомним, что функцию приращения мы уже объявляли – это py_apply_patch. Она принимает документ, патч и возвращает новый документ. Приведем ее здесь для контекста:

create or replace function py_apply_patch(doc jsonb, patch jsonb)
returns jsonb
transform for type jsonb
language plpython3u immutable strict as $$
    import jsonpatch
    patch_obj = jsonpatch.JsonPatch(patch)
    return patch_obj.apply(doc)
$$;

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

select
    json_patch_agg(patch) as doc
from (values
    (1, '[{"op": "add", "path": "/a", "value": 1}]'::jsonb),
    (2, '[{"op": "add", "path": "/b", "value": 2}]'::jsonb),
    (3, '[{"op": "replace", "path": "/a", "value": 3}]'::jsonb)
) as vals(id, patch);

┌─[ RECORD 1 ]───────────┐
│ doc │ {"a": 3, "b": 2} │
└─────┴──────────────────┘

Конечно, подобный запрос дорог в плане вычислений. Однако с ним мы получим документ мгновенно: не понадобится приложение, его настройка и запуск. Достаточно минимального клиента к базе вроде psql или PGAdmin.

Махинации с патчами

Чем дольше вы работаете с JSON Patch, тем больше откроете трюков и техник. В этом параграфе мы расскажем о двух из них.

Несколько патчей можно свести к одному, банально объединих их. Действительно: если патч p1 содержит операции o1, o2 и o3, а p2 – o4, o5 и o6, то легко получить патч p3 с операциями o1 – o6, и они применятся в том же порядке. Напомним, что в Python списки объединяются оператором + (плюс), а в Postgres – || (палочки). Полученный патч называют супер- или пакетным патчем.

В примере ниже оба патча применяются как один, при этом порядок сохраняется: в итоговом списке числа идут в верной последовательности. Дефис на конце пути "/nums/-" означает несуществующий элемент списка – его конец, куда и добавляются числа.

p1 = [
    {"op": "add", "path": "/nums/-", "value": 1},
    {"op": "add", "path": "/nums/-", "value": 2},
    {"op": "add", "path": "/nums/-", "value": 3},
]

p2 = [
    {"op": "add", "path": "/nums/-", "value": 4},
    {"op": "add", "path": "/nums/-", "value": 5},
    {"op": "add", "path": "/nums/-", "value": 6},
]

doc = {"nums": []}

print(jsonpatch.JsonPatch(p1 + p2).apply(doc))
# {'nums': [1, 2, 3, 4, 5, 6]}

Перенесемся в Postgres. Предположим, имеется таблица событий с колонкой patch. Как получить пакетный патч всех событий? Для этого напишем агрегатную функцию, которая объединяет jsonb-массивы в один:

create aggregate jsonb_concat_agg (jsonb) (
    sfunc = jsonb_concat,
    stype = jsonb,
    initcond = '[]'
);

В результате агрегации получим единый патч:

select
    jsonb_pretty(jsonb_concat_agg(patch)) as doc
from (values
    (1, '[{"op": "add", "path": "/a", "value": 1}]'::jsonb),
    (2, '[{"op": "add", "path": "/b", "value": 2}]'::jsonb),
    (3, '[{"op": "add", "path": "/c", "value": 3}]'::jsonb)
) as vals(id, patch);

┌─[ RECORD 1 ]────────────────┐
│ doc │ [                    ↵│
│     │     {                ↵│
│     │         "op": "add", ↵│
│     │         "path": "/a",↵│
│     │         "value": 1   ↵│
│     │     },               ↵│
│     │     {                ↵│
│     │         "op": "add", ↵│
│     │         "path": "/b",↵│
│     │         "value": 2   ↵│
│     │     },               ↵│
│     │     {                ↵│
│     │         "op": "add", ↵│
│     │         "path": "/c",↵│
│     │         "value": 3   ↵│
│     │     }                ↵│
│     │ ]                     │
└─────┴───────────────────────┘

По умолчанию JSON Patch предлагает шесть типов операций: add, replace и другие. Однажды вы обнаружите, что для комфортной работы их не хватает. Например, чтобы описать пополнение счета, необходима операция приращения (increase):

{"op": "inc", "path": "/amount", "value": 200}

Использование replace для этой цели будет ошибкой. Если до операции на счету было 1000 рублей и клиент добавил 500, патч с replace выглядит так:

[{"op": "replace", "path": "/amount", "value": 1500}]

Нарушается важный принцип: событие не должно знать о состоянии документа, потому что иначе его нельзя удалить без последствий. Например, если клиент внес 200 и 300 рублей подряд, события выглядят как в примере ниже:

[
    {"op": "replace", "path": "/amount", "value": 1200},
    {"op": "replace", "path": "/amount", "value": 1500}
]

Если удалить первое событие (200 рублей), то после второго на счету снова окажется окажется 1500 рублей, что неверно. Предположим теперь, что события созданы при помощи операции inc:

[
    {"op": "inc", "path": "/amount", "value": 200},
    {"op": "inc", "path": "/amount", "value": 300}
]

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

Библиотеку jsonpatch легко расширить. Встроенные операции унаследованы от класса PatchOperation и содержатся в приватном словаре. Расширение сводится к двум шагам:

  • на базе PatchOperation определить класс с частным методом apply;
  • добавить его в словарь операций.

Опишем класс операции:

from jsonpatch import JsonPatch, PatchOperation

class IncrementOperation(PatchOperation):

    def apply(self, obj):
        subobj, part = self.pointer.to_last(obj)
        try:
            val = subobj[part]
        except (KeyError, IndexError):
            raise

        if not isinstance(val, (int, float)):
            raise TypeError(f"Value is not a number: {val}")

        try:
            value = self.operation["value"]
        except KeyError as ex:
            raise InvalidJsonPatch("The operation does not contain a 'value' member")

        subobj[part] += value
        return obj

Большую часть метода apply занимают проверки и внятные сообщения об ошибках. Регистрация несколько неуклюжа: поле JsonPatch.operations – неизменяемый словарь, так что мы замещаем его новым.

ops = JsonPatch.operations | {
    'inc': IncrementOperation
}

JsonPatch.operations = ops

Опробуем новую операцию в действии:

doc = {"amount": 1000}

p = JsonPatch([
    {"op": "inc", "path": "/amount", "value": 200},
    {"op": "add", "path": "/foo", "value": "a"},
    {"op": "inc", "path": "/amount", "value": 300},
    {"op": "add", "path": "/bar", "value": "b"},
])

print(p.apply(doc))
# {'amount': 1500, 'foo': 'a', 'bar': 'b'}

По аналогии пишутся другие операции: это классы и регистрация под уникальным именем. Схожим образом расширяются библиотеки для Java или JavaScript.

Чтобы инкремент был доступен в Postgres, скопируйте файл json_patch_ops.py на сервер. Он должен быть в директории, которую Python просматривает при загрузке модулей. Простое решение в том, чтобы скопировать файл в директорию site-packages или указать к нему путь в переменной PYTHONPATH. Более грамотный подход – оформить код в пакет и устанавливать при помощи pip3. Рассмотрите вариант, чтобы вашу работу включили в официальную библиотеку, например в виде расширения.

Бесконфликтное редактирование

Концепция патчей решает и другие проблемы, связанные с документами. Одна из них касается конфликтов при совместном редактировании. Мы уже упоминали ее в третьей главе (параграф “Обновление”), но на тот момент у нас не хватало знаний. Настало время это исправить.

Предположим, сотрудники Иванов и Петров редактируют один и тот же документ. Наша задача — сделать так, чтобы изменения одного сотрудника не затерли изменения другого. Например, если Иванов изменил поле amount, а Петров — comment, эти правки не конфликтуют друг с другом и можно применить их в любом порядке. Если Петров тоже исправил amount, перед нами классический конфликт.

Наивное решение в том, чтобы при открытии документа пометить его в базе занятым. Для других сотрудников такой документ открывается в режиме чтения. Однако сотрудник может заблокировать документ и уйти на совещание. Все это время документ либо недоступен для правок другим, либо через какое-то время система сбросит блокировку, и локальные изменения потеряются.

Можно заблокировать документ на уровне базы командой select for update. Однако выражение for update требует транзакции, и мы не можем держать ее неограниченно долго.

Желательно, чтобы сотрудник, который отправил amount вторым, получил негативный ответ и новое значение amount, чтобы пересмотреть его. Для этого изменения отслеживают на уровне отдельных полей документа, и JSON Patch выступает помощником.

Техника, что мы рассматриваем, опирается на три версии документа: исходную (А) и две локальных (B и C). Имея эти три версии, легко найти патч для пар (А → В) и (А → С). Далее мы проверяем, есть у патчей общие поля. Если нет, патчи не конфликтуют, и можно применить их в любом порядке. В противном случае мы не только уведомим пользователя о конфликте, но и скажем, какое поле стало его причиной.

Мы привели самое общее решение; у него может быть множество реализаций. В зависимости от условий изменения могут применяться на клиентской части или серверной. Клиент может передавать документ целиком или только патч. Ниже мы рассмотрим вариант с приоритетом на бэкенд (серверную часть).

Решение конфликтов

Итак, Иванов и Петров одновременно открыли документ на редактирование. Изначально документ выглядел как в примере ниже. Назовем его ревизией А:

{
  "id": 100500,
  "version": 1,
  "amount": 100,
  "title": "New document"
}

Иванов исправил поле amount на 110 и нажал “Сохранить”. На сервер ушел документ ревизии B:

{
  "id": 100500,
  "version": 1,
  "amount": 110,
  "title": "New document"
}

Выше мы указываем версию нарастающим число: 1, 2 и так далее. Это небезопасно: ничто не мешает злоумышленнику передать версию 999, и серверный алгоритм сработает с ошибкой. Для версий лучше использовать идентификатор UUID, подобрать который вручную невозможно. В нашем случае мы используем числа только для краткости.

По полю version сервер определяет, что изменения отталкиваются от версии 1. Выполняется проверка: 1 — это текущий документ? Для этого мы читаем версию из таблицы applications. Если в ней записано 1, это значит, что никто не успел внести изменения до нас, и мы вправе сделать это сейчас. Замещение документа происходит в два шага:

  • документ А (версии 1, исходный) уходит в историческую таблицу;
  • новый документ В замещает текущий, его версия становится 2.

Изменения Иванова приняты. А что с Петровым? Бедняга просидел весь день на митингах и вернулся к компьютеру в конце дня. Петров нашел силы только на то, чтобы изменить заголовок с “New document” на “My document”. На сервер ушел документ, который мы назовем ревизией C:

{
  "id": 100500,
  "version": 1,
  "amount": 100,
  "title": "My document"
}

Сервер выполнит те же самые действия, что и в прошлый раз. Он прочитает текущий документ и выяснит, что его версия не 1, а 2. Это значит, за время отсутствия Петрова документ изменился. Его нельзя переписать: если сделать текущей версию Петрова, то изменения Иванова (amount 110) пропадут.

Можно сообщить Петрову, что документ изменился, и вынудить его повторить действия, отталкиваясь от последней версии. Однако прежде чем так поступить, подумаем, можно ли решить проблему автоматически.

Вопрос, который мы намереваемся решить, звучит так: действительно ли изменения Иванова и Петрова конфликтуют? Если да, то ничего поделать нельзя, и мы вынуждены делегировать вопрос людям. Пусть Иванов и Петров встретятся лично и решат, чьи изменения важнее. Если же конфликта нет, можно применить изменения каждого сотрудника по отдельности без необходимости их беспокоить.

Легко заметить, что в нашем примере конфликта нет. Иванов исправил поле amount, а Петров — title, поэтому изменения Петрова все-таки можно принять даже с учетом того, что они опираются на устаревший документа. Вот как проверить конфликт программным способом:

  • найти изменения (патч) между версиями A и B (исходной и Иванова);
  • то же самое для A и C (исходной и Петрова);
  • определить, есть ли в патчах элементы с одинаковым полем path.

В нашем случае патчи выглядят следующим образом. От А к В:

[
    {
        "op": "replace",
        "path": "/amount",
        "value": 110
    }
]

От А к С:

[
    {
        "op": "replace",
        "path": "/title",
        "value": "My document"
    }
]

Видим, что пути /amount и /title не пересекаются, поэтому изменения не конфликтуют. Чтобы применить версию Петрова, мы берем последний актуальный документ (версию Иванова) и накладываем на него второй патч (A → C). Результат становится новым текущим документом с версией 3.

Предположим теперь, Петров изменил не только заголовок документа, но и поле amount. Назовем эту ревизию C2:

{
  "id": 100500,
  "version": 1,
  "amount": 115,
  "title": "My document"
}

Патч А → C2 выглядит так:

[
    {
        "op": "replace",
        "path": "/title",
        "value": "My document"
    },
    {
        "op": "replace",
        "path": "/amount",
        "value": 115
    }
]

Очевидно, теперь он пересекается с патчем А → В: в обоих есть элемент с полем /amount. Решить конфликт программа не в состоянии, потому что только авторы знают, какое значение верное. Со стороны сервера будет правильным сделать следующее:

  • в ответ на запрос Петрова сообщить, что имеются конфликты с последней версией документа;
  • выслать сам документ;
  • сообщить набор полей и их значений, которые привели к конфликту. В примере выше это элемент
{
    "op": "replace",
    "path": "/amount",
    "value": 115
}

На стороне клиента программа должна выделить конфликтные поля, в идеале — показать их значение и автора изменений, чтобы было понятно, к кому обратиться.

Когда конфликт исправлен (скажем, стороны договорились, что поле amount должно быть вообще 120), Петров высылает документ ревизии D:

{
  "id": 100500,
  "version": 2,
  "amount": 120,
  "title": "My document"
}

Это изменение отталкивается от текущей версии 2. Когда сервер проделает шаги, что мы уже рассмотрели, новый документ заменит прежний, а последний уйдет в таблицу версий.

Интересно, что наш подход работает для множественных изменений. Предположим, пока Петров сидел на совещании, сразу несколько сотрудников изменили документ. В результате текущим стал документ версии 9 (обозначим его F):

{
  "id": 100500,
  "version": 9,
  "amount": 100,
  "category": "risk",
  "email": "john@test.com",
  "title": "Customer #123",
  "history": [
      {
          "event": "edited",
          "user": 92323
       }
  ]
}

Патч от исходного документа к текущему (A → F) выглядит так:

[
    {
        "op": "replace",
        "path": "/title",
        "value": "Customer #123"
    },
    {
        "op": "replace",
        "path": "/version",
        "value": 9
    },
    {
        "op": "add",
        "path": "/category",
        "value": "risk"
    },
    {
        "op": "add",
        "path": "/email",
        "value": "john@test.com"
    },
    {
        "op": "add",
        "path": "/history",
        "value": [
            {
                "event": "edited",
                "user": 92323
            }
        ]
    }
]

Среди измененных полей нет /amount, следовательно патчи A → F и A → C не пересекаются. Изменения Петрова можно по-прежнему применить к последнему документу.

Технические детали

Рассмотрим функцию сравнения двух патчей. Она принимает два jsonb-документа (массивы объектов) и определяет, есть ли в них одинаковые поля path. Функция довольно проста: это пересечение (intersect) двух вызовов jsonb_path_query. Если пересечение не пустое (exists нашел хотя бы одну запись), имел место конфликт.

create or replace function find_conflicts(patch1 jsonb, patch2 jsonb)
returns boolean
language sql immutable strict parallel safe
return exists(
  select jsonb_path_query(patch1, 'strict $[*].path')
  intersect
  select jsonb_path_query(patch2, 'strict $[*].path')
);

Возможно, вам понадобится не просто логический флаг, а сводная информация: какие именно поля затронуты и с какими значениями. Подойдет вторая функцию, которая оперирует таблицами. Каждый патч она приводит к нативной таблице Postgres с полями op, path, value и другими. Далее эти таблицы соединяются оператором JOIN. Если результат не пустой, мы увидим, что именно привело к конфликту. Функция:

create or replace function find_conflicts(patch1 jsonb, patch2 jsonb)
returns table (
    i1 integer,
    op1 text,
    path1 text,
    value1 jsonb,
    from1 text,
    i2 integer,
    op2 text,
    path2 text,
    value2 jsonb,
    from2 text
)
language sql immutable strict parallel safe as $$
with
p1 as (
    select jt.*
    from json_table(patch1, '$[*]' columns(
        i     for ordinality,
        op    text  path '$.op',
        path  text  path '$.path',
        value jsonb path '$.value',
        "from"  jsonb path '$.from'
    )) as jt
),
p2 as (
    select jt.*
    from json_table(patch2, '$[*]' columns(
        i     for ordinality,
        op    text  path '$.op',
        path  text  path '$.path',
        value jsonb path '$.value',
        "from"  jsonb path '$.from'
    )) as jt
)
select *
from p1, p2
where
        p1.op = p2.op
    and p1.path = p2.path
    and p1.value != p2.value;
$$;

Вызовем ее с патчами без конфликтов:

select * from find_conflicts(
$$
[
    {
        "op": "replace",
        "path": "/amount",
        "value": 120
    },
    {
        "op": "add",
        "path": "/comment",
        "value": "edited"
    }
]
$$::jsonb,
$$
[
    {
        "op": "add",
        "path": "/user_ids/3",
        "value": 40
    },
    {
        "op": "replace",
        "path": "/title",
        "value": "Test 2"
    }
]
$$::jsonb
);

┌────┬─────┬───────┬────────┬───────┬────┬─────┬───────┬────────┬───────┐
│ i1 │ op1 │ path1 │ value1 │ from1 │ i2 │ op2 │ path2 │ value2 │ from2 │
├────┼─────┼───────┼────────┼───────┼────┼─────┼───────┼────────┼───────┤
└────┴─────┴───────┴────────┴───────┴────┴─────┴───────┴────────┴───────┘
(0 rows)

А теперь – с конфликтами:

select * from find_conflicts(
$$
[
    {
        "op": "replace",
        "path": "/amount",
        "value": 120
    },
    {
        "op": "add",
        "path": "/comment",
        "value": "edited"
    }
]
$$::jsonb,
$$
[
    {
        "op": "replace",
        "path": "/amount",
        "value": 150
    },
    {
        "op": "replace",
        "path": "/title",
        "value": "Test 2"
    }
]
$$::jsonb
);

┌────┬─────────┬─────────┬────────┬────────┬────┬─────────┬─────────┬────────┬────────┐
│ i1 │   op1   │  path1  │ value1 │ from1  │ i2 │   op2   │  path2  │ value2 │ from2  │
├────┼─────────┼─────────┼────────┼────────┼────┼─────────┼─────────┼────────┼────────┤
│  1 │ replace │ /amount │ 120    │ <null> │  1 │ replace │ /amount │ 150    │ <null> │
└────┴─────────┴─────────┴────────┴────────┴────┴─────────┴─────────┴────────┴────────┘

Из отчета видно, что причиной стало поле /amount, значения сторон B и C – 120 и 150. Отправим эту информацию клиенту; ожидается, что приложение выделит проблемные поля и поможет пользователю их исправить.

На примере JSON Patch мы рассмотрели, как отслеживать изменения в документах с точностью до поля. Не всегда бизнес нуждается в подобной технике, но в редких случаях она снимает множество проблем.

Важно, что алгоритм полностью реализуется силами базы: роль приложения минимальна. Базе требуется только функция, которая по двум документам строит патч. Проще всего делегировать эту задачу расширению plpython3u и пакету jsonpatch. Искушенные читатели могут написать код на на Си и оформить его в расширение Postgres.

На этом мы закончим главу о версионировании документов. Добавим, что хранение версий, их откат и восстановление – та задача, которую трудно перенести на другие проекты. В каждой области свои требования и правовые нормы. Надеемся, мы рассмотрели достаточно техник, чтобы читатель собрал из них свое решение.

Приглашаем читателя к последней главе, где мы закроем нерешенный вопрос с поиском.

← Русская речь в фильмах