Временные паттерны в SQL: хранение истории изменений данных

Данные меняются. Цена товара росла, пользователь менял сегмент, договор переоформлялся. Правильно спроектированная схема позволяет ответить на вопрос: «Каким было состояние системы в любой прошлый момент времени?»

Зачем хранить историю

-- Сейчас: Алиса — premium-клиент
-- Год назад: Алиса была basic-клиентом
-- Какой тариф был на момент её заказа в марте 2025?

-- Без истории — невозможно ответить
-- С temporal паттерном — тривиальный запрос

Два вида времени (bi-temporal)

1. Valid Time (бизнес-время): когда данные истинны в реальном мире
2. Transaction Time (системное время): когда данные были записаны в БД

CREATE TABLE price_history (
    product_id     INTEGER,
    price          NUMERIC,
    -- Valid time: когда цена действует
    valid_from     DATE NOT NULL,
    valid_to       DATE,  -- NULL = сейчас
    -- Transaction time: когда мы узнали об этом
    recorded_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    recorded_by    TEXT NOT NULL DEFAULT current_user
);

SCD Type 1: перезаписать (нет истории)

-- Простейший вариант: обновить и забыть
UPDATE customers SET email = 'new@example.com' WHERE id = 42;
-- Прошлый email потерян навсегда

Когда: данные технические (cached_data, last_login), история не нужна.

SCD Type 2: полная история

Каждое изменение — новая строка:

CREATE TABLE dim_customers (
    id          BIGSERIAL PRIMARY KEY,
    customer_id INTEGER NOT NULL,  -- бизнес-ключ
    email       TEXT,
    segment     TEXT,
    city        TEXT,
    valid_from  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    valid_to    TIMESTAMPTZ,       -- NULL = текущая запись
    is_current  BOOLEAN DEFAULT true,
    version     INTEGER DEFAULT 1
);

-- Добавить изменение: закрыть старую + создать новую
WITH old_record AS (
    UPDATE dim_customers
    SET valid_to = NOW(), is_current = false
    WHERE customer_id = 42 AND is_current = true
    RETURNING version
)
INSERT INTO dim_customers (customer_id, email, segment, city, version)
SELECT 42, 'new@example.com', 'premium', 'Москва', version + 1
FROM old_record;

-- Актуальное состояние
SELECT * FROM dim_customers WHERE is_current = true;

-- Состояние на конкретный момент
SELECT * FROM dim_customers
WHERE customer_id = 42
  AND valid_from <= '2025-06-15'
  AND (valid_to IS NULL OR valid_to > '2025-06-15');

SCD Type 4: разделить текущее и историческое

-- Основная таблица: только текущие данные (быстрый доступ)
CREATE TABLE customers (
    id      INTEGER PRIMARY KEY,
    email   TEXT,
    segment TEXT,
    city    TEXT
);

-- Таблица истории: все изменения
CREATE TABLE customers_history (
    id          BIGSERIAL PRIMARY KEY,
    customer_id INTEGER,
    email       TEXT,
    segment     TEXT,
    city        TEXT,
    changed_at  TIMESTAMPTZ DEFAULT NOW(),
    change_type TEXT  -- 'insert', 'update', 'delete'
);

-- Триггер для автоматической записи истории
CREATE OR REPLACE FUNCTION track_customer_changes()
RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO customers_history (customer_id, email, segment, city, change_type)
    VALUES (COALESCE(NEW.id, OLD.id), COALESCE(NEW.email, OLD.email),
            COALESCE(NEW.segment, OLD.segment), COALESCE(NEW.city, OLD.city),
            TG_OP);
    RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER track_customers
AFTER INSERT OR UPDATE OR DELETE ON customers
FOR EACH ROW EXECUTE FUNCTION track_customer_changes();

PostgreSQL: Period/Range для temporal данных

-- tstzrange: диапазон временных меток
CREATE TABLE contracts (
    id          SERIAL PRIMARY KEY,
    client_id   INTEGER,
    amount      NUMERIC,
    valid_period TSTZRANGE NOT NULL  -- [от, до)
);

-- Вставка с периодом
INSERT INTO contracts (client_id, amount, valid_period)
VALUES (42, 100000, '[2026-01-01, 2027-01-01)');

-- Найти активные на дату
SELECT * FROM contracts WHERE valid_period @> NOW()::timestamptz;

-- Найти перекрывающиеся периоды (конфликты расписания)
SELECT a.id, b.id
FROM contracts a, contracts b
WHERE a.id < b.id
  AND a.client_id = b.client_id
  AND a.valid_period && b.valid_period;  -- && = пересечение

-- Индекс на диапазон
CREATE INDEX idx_contracts_period ON contracts USING gist(valid_period);

Запрос через time: «как было в прошлом»

-- Выручка по тарифу, который был актуален на момент заказа
SELECT
    o.order_id,
    o.created_at,
    o.amount,
    c.segment AS segment_at_order_time  -- тариф на момент заказа, не сейчас!
FROM orders o
JOIN dim_customers c
    ON c.customer_id = o.customer_id
    AND o.created_at BETWEEN c.valid_from AND COALESCE(c.valid_to, 'infinity')
WHERE o.created_at >= '2025-01-01';

Audit Trail: полная запись всех изменений

CREATE TABLE audit_trail (
    id          BIGSERIAL PRIMARY KEY,
    table_name  TEXT NOT NULL,
    row_id      TEXT NOT NULL,
    operation   TEXT NOT NULL,  -- INSERT / UPDATE / DELETE
    old_values  JSONB,
    new_values  JSONB,
    changed_by  TEXT DEFAULT current_user,
    changed_at  TIMESTAMPTZ DEFAULT NOW(),
    session_id  TEXT DEFAULT pg_backend_pid()::TEXT
);

-- Универсальный триггер
CREATE OR REPLACE FUNCTION universal_audit()
RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO audit_trail (table_name, row_id, operation, old_values, new_values)
    VALUES (
        TG_TABLE_NAME,
        COALESCE(NEW.id, OLD.id)::TEXT,
        TG_OP,
        CASE WHEN TG_OP = 'INSERT' THEN NULL ELSE row_to_json(OLD)::jsonb END,
        CASE WHEN TG_OP = 'DELETE' THEN NULL ELSE row_to_json(NEW)::jsonb END
    );
    RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;

Архивирование старых данных

-- Переместить старые заказы в архив
WITH archived AS (
    DELETE FROM orders
    WHERE created_at < NOW() - INTERVAL '3 years'
    RETURNING *
)
INSERT INTO orders_archive SELECT * FROM archived;

-- Для больших таблиц — батчами
DO $$
DECLARE rows_moved INTEGER := 1;
BEGIN
    WHILE rows_moved > 0 LOOP
        WITH moved AS (
            DELETE FROM orders WHERE id IN (
                SELECT id FROM orders
                WHERE created_at < NOW() - INTERVAL '3 years'
                LIMIT 10000
            )
            RETURNING *
        )
        INSERT INTO orders_archive SELECT * FROM moved;
        GET DIAGNOSTICS rows_moved = ROW_COUNT;
        PERFORM pg_sleep(0.1);
    END LOOP;
END;
$$;

Итог: когда какой паттерн

Храните историю там где она нужна. Не усложняйте схему там где достаточно updated_at.