Журнал изменений (Changelog) / Table Stream

Кратко Apache Doris Table Stream станет публичным API для чтения изменений на уровне строк (вставок, обновлений и удалений) из таблицы Doris, аналогично чтению MySQL binlog. Необходимая инфраструктура уже есть в кодовой базе: отдельный changelog для каждой таблицы, объект STREAM, который фиксирует смещения для каждой партиции, и метаданные потребления. Функция пока не достигла статуса GA, а значение changelog.ttl_seconds по умолчанию составляет 86,400 секунд (одни сутки). Для CDC из Doris в общем случае используйте Flink Doris Connector. Статус: скоро появится (предварительная версия).

Статус: скоро появится

Объект STREAM, свойство таблицы changelog.enable и описанные здесь представления метаданных потребления уже есть в исходном коде Doris, но пока не выпущены как стабильный публичный API. Синтаксис и семантика могут измениться до выхода GA. В промышленных конвейерах CDC сейчас следует использовать интеграцию с Kafka и CDC.

Зачем использовать changelog и Table Stream в Apache Doris?

Changelog и Table Stream в Apache Doris предоставляют потребителям данных поток изменений на уровне строк, позволяя заданию Flink, журналу аудита или поисковому индексу реагировать на каждую вставку, обновление и удаление без повторного чтения таблицы. Apache Doris хорошо справляется с приёмом данных об изменениях. Получать данные об изменениях обратно из Doris сложнее. Если системе-потребителю (заданию Flink, журналу аудита, другому кластеру Doris или поисковому индексу) нужно реагировать на каждую вставку, обновление и удаление, доступные сегодня варианты неудобны:

• Периодически перечитывать таблицу и вычислять различия. Это просто реализовать, но дорого выполнять, и такой подход никогда не обеспечивает работу в реальном времени.
• Настроить у производителей данных двойную запись в Kafka и Doris. Два приёмника, два сценария отказа.

Changelog и Table Stream решают эту проблему. Changelog записывает каждое зафиксированное изменение на уровне строк. Table Stream служит именованным курсором для этого журнала: потребитель подключается к потоку, читает изменения с последнего смещения и подтверждает их обработку. Идея та же, что у MySQL binlog, Snowflake Streams или слота логической репликации Postgres, и реализована как полноценный объект Doris.

Что такое changelog и Table Stream в Apache Doris?

Changelog в Apache Doris представляет собой журнал изменений на уровне строк, который Doris ведёт для каждой таблицы, а Table Stream в Apache Doris предоставляет потребителям доступ к нему в виде CDC на уровне строк. Changelog записывает каждое зафиксированное изменение строки с временной меткой фиксации и, при необходимости, значениями до изменения. Этот журнал отделён от binlog, используемого для межкластерной репликации (CCR): у changelog есть собственные свойства changelog.*, и для одной таблицы нельзя включить оба механизма одновременно.

Table Stream представляет собой объект для потребителей данных, построенный поверх changelog. Поток создаётся командой CREATE STREAM <name> ON TABLE <base_table>, хранит собственные смещения для каждой партиции и предоставляет SQL-интерфейс: SHOW CREATE STREAM и представление в information_schema для отслеживания прогресса потребления.

Ключевые термины

• changelog.enable: свойство таблицы. Значение true включает changelog для этой таблицы. Его можно задать только при создании таблицы.
• changelog.need_historical_value: свойство таблицы. При значении true также записываются значения до каждого обновления и удаления; поддерживается только для таблиц Unique Key Merge-on-Write.
• changelog.ttl_seconds: настройка срока хранения. TTL по умолчанию составляет одни сутки; его можно изменить после создания таблицы.
• STREAM: объект Catalog в Doris, который создаётся поверх базовой таблицы и отслеживает смещения для каждой партиции для одного потребителя.
• Режим потребления потока: detail, append_only или min_delta (по умолчанию). Определяет, выдаёт ли поток каждое изменение строки, только вставки или минимальный набор изменений.
• Отставание потока: разница между видимой версией партиции и смещением, сохранённым потоком. Отображается в столбце LAG.

Как работают changelog и Table Stream в Apache Doris?

Changelog в Apache Doris записывает изменения на уровне строк на BE, а Table Stream дополняет этот журнал смещениями для каждой партиции, чтобы потребитель мог читать изменения инкрементально и подтверждать их обработку.

1. Включите changelog при создании таблицы. CREATE TABLE t (...) PROPERTIES ("changelog.enable" = "true"). Включить changelog для существующей таблицы нельзя. BE хранит изменения на уровне строк вместе с данными tablet; срок хранения определяется changelog.ttl_seconds.
2. Создайте поток для таблицы. CREATE STREAM s ON TABLE t PROPERTIES ("type" = "min_delta"). FE создаёт объект STREAM, фиксирует текущую видимую версию каждой партиции в качестве начального смещения и регистрирует объект в базе данных. Параметр show_initial_rows = true меняет это поведение: при первом чтении возвращается историческое состояние.
3. Читайте изменения. Потребитель данных (сейчас это внешний CDC-коннектор, в будущем появятся встроенные интерфейсы потребления) читает данные из потока. Каждая порция данных продвигает смещения для соответствующих партиций и сохраняет временную метку потребления.
4. Удаляйте устаревшие изменения. Срок хранения изменений старше changelog.ttl_seconds истекает независимо от того, успели ли все потоки их прочитать; затем фоновая операция compaction освобождает занятое ими место. Поток, смещение которого вышло за пределы этого окна, необходимо сбросить, начав заново со свежего снимка.
5. Наблюдайте за состоянием. information_schema.table_streams выводит список всех потоков и их базовых таблиц. table_stream_consumption показывает UNIT (партицию), CONSUMPTION_STATUS, LAG и LAST_CONSUMPTION_TIME для каждой партиции. Эти данные служат панелью мониторинга отставания.

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

-- 1. Enable the changelog when the source table is created
CREATE TABLE orders (
  order_id   BIGINT,
  order_date DATE,
  status     VARCHAR(16),
  amount     DECIMAL(10, 2)
)
UNIQUE KEY(order_id, order_date)
PARTITION BY RANGE(order_date) (
  PARTITION p_2026_05 VALUES [('2026-05-01'), ('2026-06-01'))
)
DISTRIBUTED BY HASH(order_id) BUCKETS 8
PROPERTIES (
  "enable_unique_key_merge_on_write" = "true",
  "changelog.enable" = "true",
  "changelog.need_historical_value" = "true",
  "changelog.ttl_seconds" = "86400"
);

-- 2. Create a Table Stream over it (planned syntax, subject to change)
CREATE STREAM orders_cdc ON TABLE orders
PROPERTIES ("type" = "min_delta", "show_initial_rows" = "false");

-- 3. Inspect the stream and its lag
SHOW CREATE STREAM orders_cdc;
SELECT * FROM information_schema.table_stream_consumption
WHERE STREAM_NAME = 'orders_cdc';

Ожидаемый результат

+---------+------------+-----------+--------------------+-----+----------------------+
| DB_NAME | STREAM_NAME| STREAM_ID | UNIT (partition)   | LAG | LAST_CONSUMPTION_TIME|
+---------+------------+-----------+--------------------+-----+----------------------+
| sales   | orders_cdc | 10042     | p_2026_05          | 0   | 2026-05-09 10:14:33  |
+---------+------------+-----------+--------------------+-----+----------------------+

LAG = 0 означает, что потребитель дошёл до видимой версии партиции. По мере фиксации операций записи в источнике LAG растёт; по мере чтения данных потребителем он уменьшается. Если потребитель отстал больше, чем на changelog.ttl_seconds, ему требуется новый снимок; STALE_REASON не сообщает об этом случае, поэтому сравнивайте задержку потребления со сроком хранения.

Когда стоит использовать changelog и Table Stream в Apache Doris?

После выхода GA используйте changelog и Table Stream в Apache Doris, если системе-потребителю нужны события изменений на уровне строк из таблицы Doris без перечитывания всей таблицы.

Подходящие сценарии (после выхода GA)

• Обработка событий изменений из Doris в заданиях Flink или Spark без повторного чтения таблицы.
• Поддержание внешнего поискового индекса, кэша или журнала аудита, которым нужны все события вставки и удаления.
• Репликация из Doris в Doris, для которой необходимо явно отслеживать смещения потребителя.
• Запуск пересчёта внешнего материализованного представления при изменении LAG партиции.

Неподходящие сценарии (особенно сейчас)

• Промышленные конвейеры CDC на текущий момент. Объект STREAM уже есть в исходном коде, но его статус GA ещё не объявлен. Используйте интеграцию с Kafka и CDC.
• Длительные паузы в работе потребителей при коротком сроке хранения changelog. Если потребитель отстаёт больше, чем на changelog.ttl_seconds, срок хранения нужных ему изменений истекает и потребителю приходится заново загружать начальное состояние.
• Кластеры с разделением хранения и вычислений. Модель поддержки Table Stream для них ещё разрабатывается.
• Чтение таблиц из внешних Catalog (Hive, Iceberg). Table Stream работает только с внутренними таблицами OLAP.

Дополнительные материалы

• Интеграция с Kafka и CDC: поддерживаемый сегодня способ передачи данных об изменениях в Doris и связанным с ним системам.
• Обновление и удаление данных: как таблицы Unique Key Merge-on-Write представляют удаления и обновления, поступающие не по порядку, которые будет передавать поток изменений строк.
• Iceberg: внешнее хранилище для реплицируемых таблиц, в которое можно записывать данные CDC через SQL, не выходя из Doris.
• Unique Key: модель исходной таблицы, которая хранит состояние по принципу «одна строка на ключ» в удобном для потребителей changelog виде.