Fluss Catalog

Doris читает таблицы Apache Fluss через Fluss Catalog. Fluss — система потокового хранения данных для аналитики в реальном времени. Её служба переноса данных может непрерывно записывать данные таблиц в озеро данных Paimon. Doris может напрямую читать актуальные данные из Fluss. Кроме того, Doris может объединять историю, уже перенесённую в Paimon, с более новыми строками, которые ещё находятся в журнале Fluss, и выполнять запросы к ним как к одной таблице. Этот второй способ чтения называется Union Read.

примечание

• Эта функция является экспериментальной и доступна начиная с версии 5.0.0.
• Поддерживается только чтение. Запись и операторы управления базами данных или таблицами не поддерживаются.
• Для таблиц с переносом данных в озеро поддерживается только формат Paimon.

Сценарии использования

Сценарий	Описание
Запросы в реальном времени	Прямые запросы к последним данным в таблицах журналов Fluss и таблицах с первичным ключом.
Union Read	Для таблицы с включённым переносом данных в озеро один запрос читает историю, перенесённую в озеро Paimon, вместе со строками журнала Fluss, которые ещё не были перенесены. Это совместное чтение озера данных и журнала.
Интеграция данных	Чтение данных Fluss во внутренние таблицы Doris или их соединение с внутренними таблицами и таблицами из других Catalog в федеративных запросах.
Обратная запись	Не поддерживается.

Как это работает

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

Тип таблицы	Как Doris читает данные
Таблица журналов	Читает журнал Fluss. Каждый бакет образует один диапазон сканирования.
Таблица с первичным ключом	Читает последний KV-снимок каждого бакета, применяет журнал изменений, записанный после снимка, и объединяет данные по первичному ключу, получая последнее состояние.
Таблица с переносом данных в озеро (table.datalake.enabled = true)	Читает перенесённые данные из озера Paimon, применяет журнал Fluss после смещения переноса и объединяет обе части в одном сканировании. Это Union Read.

Данные озера читает встроенный в Doris Paimon Connector; у Fluss Catalog нет собственного средства чтения озера. Fluss сохраняет сведения о доступном для чтения снимке Paimon для каждой таблицы с переносом данных, а Doris читает данные Paimon на момент этого снимка. Поэтому при таком чтении используются нативные средства чтения ORC/Parquet и файловый кеш.

При Union Read таблицы с первичным ключом и переносом данных в озеро обе части объединяются по первичному ключу. Во время планирования FE добавляет диапазон смещений хвоста журнала бакета к каждому сплиту данных озера для этого бакета. Когда BE читает данные озера, он отбрасывает строки, которые с тех пор были обновлены или удалены в хвосте журнала, а затем дополнительно выдаёт последнее состояние самого хвоста журнала в отдельном диапазоне сканирования. Каждая строка выдаётся ровно один раз, и результат совпадает с чтением всей таблицы только из Fluss.

Настройка Catalog

Синтаксис

CREATE CATALOG [IF NOT EXISTS] catalog_name PROPERTIES (
    'type' = 'fluss',
    'fluss.bootstrap.servers' = '<bootstrap_servers>',
    {FlussProperties},
    {LakeProperties},
    {CommonProperties}
);

• <bootstrap_servers>

  Обязательный параметр. Начальные адреса подключения к кластеру Fluss: список host:port через запятую. Обычно это адреса CoordinatorServer, например 'fluss.bootstrap.servers' = '10.0.0.1:9123,10.0.0.2:9123'.

• {FlussProperties}

  Раздел FlussProperties содержит параметры, относящиеся к Fluss Catalog.

  Параметр	Обязательный	По умолчанию	Описание
  fluss.union_read.mode	Нет	auto	Режим Union Read для таблиц с переносом данных в озеро. Возможные значения: auto, required или disabled. См. раздел «Режимы Union Read».
  fluss.union_read.max_tail_rows	Нет	2000000	Максимально допустимое количество строк в хвосте журнала одного бакета при Union Read таблицы с первичным ключом. См. раздел «Режимы Union Read».
  fluss.union_read.max_total_tail_rows	Нет	20000000	Максимально допустимое суммарное количество строк в хвостах журналов всех бакетов за одно сканирование при Union Read таблицы с первичным ключом. Не должно быть меньше fluss.union_read.max_tail_rows.
  enable.mapping.varbinary	Нет	false	Определяет, следует ли сопоставлять типы Fluss BINARY/BYTES с типом Doris varbinary. По умолчанию они сопоставляются с string.
  enable.mapping.timestamp_tz	Нет	false	Определяет, следует ли сопоставлять тип Fluss TIMESTAMP_LTZ с типом Doris timestamptz. По умолчанию он сопоставляется с datetime.

  Помимо перечисленных выше параметров Doris и fluss.lake.paimon.*, все остальные свойства с префиксом fluss. передаются клиенту Fluss без изменений, но с удалённым префиксом. FE и BE используют один и тот же набор параметров. Doris не ограничивает имена параметров, поэтому таким способом можно передать любой параметр конфигурации, который понимает клиент Fluss. Например, если в кластере Fluss включена аутентификация SASL:

  'fluss.client.security.protocol' = 'sasl',
  'fluss.client.security.sasl.mechanism' = 'PLAIN',
  'fluss.client.security.sasl.username' = '<username>',
  'fluss.client.security.sasl.password' = '<password>'

  Полный список см. в справочнике по конфигурации Fluss и руководстве по аутентификации Fluss.

  Настройки клиента Fluss на FE и BE различаются только количеством сетевых потоков. Каждое соединение Fluss на BE обслуживает один диапазон сканирования за раз, поэтому, если в Catalog не задан fluss.netty.client.num-network-threads, BE открывает соединения с одним сетевым потоком, чтобы экономить потоки и файловые дескрипторы на BE, а соединение FE сохраняет значение клиента Fluss по умолчанию. Обычно задавать этот параметр не требуется; если он задан, FE и BE используют указанное значение.

• {LakeProperties}

  Раздел LakeProperties содержит настройки подключения к озеру Paimon для таблиц с переносом данных в озеро. Эти настройки имеют префикс fluss.lake.paimon.. Часть после префикса записывается точно так же, как в datalake.paimon.* в файле server.yaml кластера Fluss, поэтому значения можно скопировать оттуда. См. раздел «Настройки подключения к озеру данных».

• {CommonProperties}

  Раздел CommonProperties содержит общие свойства. См. раздел «Общие свойства» в обзоре Catalog.

Примеры

Минимальная конфигурация. Её достаточно, если вы читаете только собственные данные Fluss или если хранилище озера не требует учётных данных:

CREATE CATALOG fluss PROPERTIES (
    'type' = 'fluss',
    'fluss.bootstrap.servers' = '10.0.0.1:9123'
);

Озеро Paimon размещено в объектном хранилище (в этом примере — в S3-совместимом хранилище):

CREATE CATALOG fluss_lake PROPERTIES (
    'type' = 'fluss',
    'fluss.bootstrap.servers' = '10.0.0.1:9123',
    'fluss.lake.paimon.s3.endpoint' = 'http://minio:9000',
    'fluss.lake.paimon.s3.access-key' = '<ak>',
    'fluss.lake.paimon.s3.secret-key' = '<sk>'
);

В кластере Fluss включена аутентификация SASL:

CREATE CATALOG fluss_sasl PROPERTIES (
    'type' = 'fluss',
    'fluss.bootstrap.servers' = '10.0.0.1:9123',
    'fluss.client.security.protocol' = 'sasl',
    'fluss.client.security.sasl.mechanism' = 'PLAIN',
    'fluss.client.security.sasl.username' = '<username>',
    'fluss.client.security.sasl.password' = '<password>'
);

Требования к развёртыванию

• FE использует клиент Fluss для чтения метаданных и планирования запросов, а BE — для чтения данных. Оба должны иметь доступ к CoordinatorServer и TabletServer кластера Fluss.
• BE повторно использует соединения с кластером Fluss. После чтения диапазона сканирования соединение остаётся открытым для следующих диапазонов сканирования и запросов и закрывается только после простоя более 60 секунд. Поэтому на стороне Fluss соединения от BE могут отображаться и после завершения запроса. Соединения разделяются по конфигурации клиента: после изменения свойств Catalog (например, имени пользователя и пароля SASL) BE открывает новые соединения с новыми настройками.
• KV-снимки таблиц с первичным ключом и архивные сегменты журнала находятся в удалённом хранилище, настроенном во Fluss (remote.data.dir). BE читает эти файлы напрямую, поэтому ему также необходим доступ к этому удалённому хранилищу.
• Для чтения таблиц с переносом данных в озеро и FE, и BE должны иметь доступ к хранилищу озера Paimon.
• Данные озера читает плагин Paimon Connector, встроенный в Doris. Плагин поставляется вместе с Doris и не требует отдельной установки.

Поддерживаемые версии Fluss

Версия Doris	Версия клиента Fluss
5.0	1.0

Кластеры Fluss старше версии 1.0 не поддерживаются: клиентские API, от которых зависит Doris, впервые появились во Fluss 1.0.

Сопоставление метаданных

Иерархия метаданных Fluss имеет вид «База данных -> Таблица» и напрямую соответствует иерархии Doris. Дополнительного сопоставления имён нет.

Для таблицы с включённым переносом данных в озеро SHOW TABLES выводит только саму таблицу. tbl$lake и tbl$log — два способа чтения одной таблицы, поэтому они не отображаются как отдельные объекты.

SHOW CREATE TABLE показывает столбцы и комментарии таблицы Fluss, но не её свойства. Чтобы узнать, включён ли для таблицы перенос данных в озеро (table.datalake.enabled) и какой формат озера используется (table.datalake.format), проверьте определение таблицы на стороне Fluss, например во Flink SQL. В Doris только таблицы с включённым переносом данных в озеро имеют tbl$lake и tbl$log.

Сопоставление типов столбцов

Тип Fluss	Тип Doris	Комментарий
BOOLEAN	boolean
TINYINT	tinyint
SMALLINT	smallint
INT	int
BIGINT	bigint
FLOAT	float
DOUBLE	double
DECIMAL(P, S)	decimal(P, S)
CHAR(N)	char(N)	Сопоставляется с string, если N больше 255
STRING	string
BINARY(N)	string / varbinary(N)	Управляется параметром enable.mapping.varbinary. При значении по умолчанию false сопоставляется с string; при true — с varbinary
BYTES	string / varbinary	Аналогично предыдущей строке
DATE	date
TIME	UNSUPPORTED	В Doris нет типа с такой же семантикой. К другим столбцам таблицы по-прежнему можно выполнять запросы; ошибкой завершается только запрос, обращающийся к этому столбцу
TIMESTAMP(P)	datetime(P)	Сопоставляется с datetime(6), если точность больше 6, что может привести к потере точности
TIMESTAMP_LTZ(P)	datetime(P) / timestamptz(P)	Точность больше 6 преобразуется в 6. Управляется параметром enable.mapping.timestamp_tz. При значении по умолчанию false сопоставляется с datetime; при true — с timestamptz
ARRAY	array
MAP	map
ROW	struct

Примечание:

Для столбца TIMESTAMP_LTZ значение WITH_TIMEZONE отображается в столбце Extra результата DESCRIBE. При сопоставлении с datetime Doris перед возвратом значения преобразует его в текущий часовой пояс сеанса (SET time_zone = <tz>).

Для таблицы с включённым переносом данных в озеро tbl и tbl$lake имеют одинаковые типы столбцов. Приведённое выше сопоставление согласовано с преобразованием типов, которое Fluss применяет при записи в Paimon, и с сопоставлением типов в Paimon Catalog.

Секционированные таблицы

Столбцы секционирования в Doris являются обычными столбцами. SHOW PARTITIONS выводит секции в формате k1=v1/k2=v2. Предикаты по столбцам секционирования используются для отсечения секций, что уменьшает количество читаемых бакетов.

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

• Поддерживаются: CHAR, STRING, BOOLEAN, TINYINT, SMALLINT, INT, BIGINT, DATE, а также BINARY/BYTES, если enable.mapping.varbinary выключен (значение секции представляет собой шестнадцатеричную текстовую запись байтов).
• Не поддерживаются: FLOAT, DOUBLE, TIME, TIMESTAMP, TIMESTAMP_LTZ. Описание таких таблиц по-прежнему можно получить с помощью DESCRIBE, но запросы и SHOW PARTITIONS завершаются ошибкой. В сообщении об ошибке указывается столбец секционирования, вызвавший её.

Выполнение запросов

Базовый запрос

-- Switch to the catalog, then query
SWITCH fluss;
USE db;
SELECT * FROM tbl LIMIT 10;

-- Use the fully qualified name
SELECT * FROM fluss.db.tbl WHERE dt = '20260101';

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

Таблицы с переносом данных в озеро и Union Read

Для таблицы с table.datalake.enabled = true Doris предлагает три способа чтения. По умолчанию данные озера и журнала читаются совместно с помощью Union Read:

SELECT * FROM fluss.db.tbl;         -- Union Read: lake + log merged into the full data set (default)
SELECT * FROM fluss.db.`tbl$lake`;  -- Only the tiered data in the Paimon lake
SELECT * FROM fluss.db.`tbl$log`;   -- Only the data in the Fluss log that has not been tiered yet

tbl$lake и tbl$log разделяют данные по месту хранения. Они не пересекаются и вместе составляют все данные tbl. Их имена описывают расположение данных, а не их актуальность: если служба переноса данных останавливается, tbl$log по-прежнему содержит всё, что осталось в журнале.

tbl$lake

• Paimon Connector напрямую читает таблицу Paimon, записанную службой переноса данных. Для таблицы, данные которой перенесены средствами Fluss 1.0, tbl$lake имеет ровно те же столбцы, что и tbl. Системных столбцов __bucket, __offset и __timestamp, которые более ранние версии Fluss добавляли в таблицы озера, больше нет.
• Чтение выполняется на момент доступного для чтения снимка озера, сведения о котором хранит Fluss, либо из последнего состояния озера, если такого снимка ещё нет. Fluss создаёт таблицу Paimon одновременно с таблицей Fluss, поэтому до первой фиксации данных службой переноса tbl$lake возвращает пустой результат, а не ошибку.
• Если Doris не может найти таблицу озера с текущими настройками подключения, запрос завершается ошибкой nothing has been tiered. Обычно это означает, что настройки, например fluss.lake.paimon.warehouse, не указывают на хранилище, в которое записывает данные кластер Fluss. См. раздел «Настройки подключения к озеру данных».
• Запросы к историческим состояниям и инкрементальные запросы Paimon Catalog не поддерживаются.

tbl$log

• Начинает чтение со смещения журнала, соответствующего снимку озера, и возвращает только данные, записанные после этого смещения.
• Поддерживаются только таблицы журналов. Для таблицы с первичным ключом журнал после снимка озера представляет собой поток изменений с обновлениями и удалениями ключей, уже существующих в озере, поэтому его нельзя вернуть как набор независимых строк. Запрос к $log для таблицы с первичным ключом завершается ошибкой.
• У таблицы уже должен быть доступный для чтения снимок озера, иначе запрос завершается ошибкой.
• У таблиц без переноса данных в озеро нет $lake и $log.

Режимы Union Read

Будет ли запрос к самой таблице с переносом данных в озеро (SELECT * FROM tbl) использовать Union Read, определяется свойством Catalog fluss.union_read.mode:

Значение	Поведение
auto (по умолчанию)	Выполняет Union Read, если существует доступный для чтения снимок озера. Если условия не выполнены, переключается на чтение данных из Fluss вместо завершения с ошибкой. См. случаи ниже.
required	Union Read обязателен. Если условия не выполнены, запрос завершается ошибкой без переключения на другой способ чтения.
disabled	Union Read не используется. Читаются только данные, которые сейчас находятся во Fluss; озеро игнорируется, включая секции, существующие только в нём.

Переменная сеанса fluss_union_read_mode переопределяет настройку Catalog для текущего сеанса:

SET fluss_union_read_mode = 'required';  -- No fallback to a Fluss-only read; fail instead
SET fluss_union_read_mode = 'disabled';  -- Read from Fluss only
SET fluss_union_read_mode = '';          -- Default: follow the catalog setting

Чтобы изменить режим только для одного оператора, используйте подсказку SET_VAR:

SELECT /*+ SET_VAR(fluss_union_read_mode='required') */ * FROM fluss.db.tbl;

Режим Union Read определяет путь чтения данных. Пока Fluss хранит все данные таблицы, то есть ни одна запись журнала не удалена по TTL и ни одна секция, которая ещё существует в озере, не удалена из Fluss, все три режима возвращают одинаковые строки. required полезен для регрессионных тестов и диагностики, поскольку при несовпадении пути чтения с ожидаемым запрос завершается ошибкой, а не незаметно использует другой путь.

Случаи переключения на другой способ чтения в режиме auto

Случай	Описание
Нет доступного для чтения снимка озера	Служба переноса данных ещё ничего не зафиксировала, поэтому все данные остаются во Fluss. Таблица читается только из Fluss.
key-type	Столбец первичного ключа в таблице с первичным ключом, не являющийся столбцом секционирования, имеет тип, значения которого невозможно точно сравнить между озером и журналом: FLOAT, DOUBLE, TIME, TIMESTAMP с точностью больше 6 и TIMESTAMP_LTZ. Когда enable.mapping.timestamp_tz имеет значение false (по умолчанию), столбец TIMESTAMP_LTZ любой точности попадает в этот случай, поскольку разные моменты времени могут иметь одинаковое местное время при переводе часов назад. При значении true сюда попадают только столбцы с точностью больше 6.
partition-type	Столбец секционирования таблицы с первичным ключом имеет тип, отличный от STRING. Сплиты озера сопоставляются с секциями Fluss по текстовой форме значения секции, и только STRING гарантирует одинаковую запись с обеих сторон.
tail-truncated	Часть журнала после снимка озера для таблицы с первичным ключом уже удалена по TTL Fluss, поэтому хвост журнала невозможно прочитать полностью.
tail-too-large	Хвост журнала таблицы с первичным ключом содержит больше строк, чем допускают fluss.union_read.max_tail_rows или fluss.union_read.max_total_tail_rows.

Последние четыре случая применимы только к таблицам с первичным ключом; EXPLAIN сообщает о них в поле degraded. При переключении таблицы с первичным ключом на другой способ чтения секции, которые ещё существуют во Fluss, полностью читаются из Fluss. Данные при этом не теряются, поскольку Fluss хранит полное состояние этих секций; запрос лишь не может использовать для них столбцовые файлы озера и выполняется медленнее. Секции, удалённые из Fluss, но ещё существующие в озере, например из-за истечения срока хранения секции, по-прежнему читаются из озера. Несекционированная таблица читается только из Fluss.

В случае partition-type Doris не может надёжно определить, какой секции Fluss принадлежит сплит озера. Если в озере есть секция, которой уже нет во Fluss, или значение секции, записанное с обеих сторон по-разному, запрос завершается ошибкой cannot be matched safely to a live fluss partition вместо переключения на другой способ чтения. Установите режим Union Read в disabled, чтобы читать только данные Fluss.

Для таблиц журналов поведение отличается. Чтение только из Fluss возвращает лишь то, что ещё хранится в журнале Fluss. После удаления ранних записей журнала по TTL результат содержит меньше строк, чем при Union Read. Для таблиц журналов с переносом данных в озеро режим disabled не рекомендуется. Кроме того, таблица журналов не переключается на другой способ чтения, если её хвост журнала неполон: если срок хранения части журнала после снимка озера уже истёк, Union Read завершается ошибкой Part of the log tail has expired и в режиме auto, и в режиме required; запрос к tbl$log также завершается ошибкой в любом режиме. Дождитесь, пока Fluss опубликует более новый доступный для чтения снимок озера.

Ограничения количества строк в хвосте журнала

При Union Read таблицы с первичным ключом BE держит хвост журнала каждого бакета в памяти: одна копия представляет собой набор первичных ключей для фильтрации данных озера, а другая — строки, которые должен выдать сам хвост. Чтобы ограничить использование памяти, fluss.union_read.max_tail_rows задаёт предел количества строк хвоста одного бакета (по умолчанию 2 миллиона строк), а fluss.union_read.max_total_tail_rows — суммарного количества строк хвостов всех бакетов за одно сканирование (по умолчанию 20 миллионов строк). Оба ограничения проверяются при планировании на основе смещений, сообщаемых Fluss. Обычно хвост журнала содержит только данные, записанные за один период table.datalake.freshness, что значительно ниже пределов по умолчанию. Достижение предела обычно означает, что служба переноса данных остановилась или сильно отстаёт. Строки, выдаваемые хвостом, хранятся в куче JVM на BE, поэтому перед увеличением любого из ограничений убедитесь, что в куче JVM достаточно места. См. раздел «Память кучи JVM на BE».

Проверка пути чтения с помощью EXPLAIN

Строка flussScan в выводе EXPLAIN показывает, как запрос фактически читает таблицу:

flussScan: readMode=default, unionRead=yes, lakeSplits=3, suppressedLakeSplits=1, logRanges=0, pkRanges=0, pkTailRanges=1, mode=auto

Поле	Значение
readMode	default читает всю таблицу; log читает tbl$log.
unionRead	Выполнялось ли Union Read. Значение yes также используется для tbl$log, начальные смещения которого берутся из снимка озера; lakeSplits в этом случае равен 0.
lakeSplits	Количество сплитов озера.
suppressedLakeSplits	Количество сплитов озера, которые необходимо фильтровать по хвосту журнала. Только для таблиц с первичным ключом.
logRanges	Количество диапазонов сканирования журнала для таблицы журналов.
pkRanges	Количество бакетов таблицы с первичным ключом, полностью читаемых из Fluss (KV-снимок + журнал изменений).
pkTailRanges	Количество диапазонов сканирования хвоста журнала при Union Read таблицы с первичным ключом.
mode	Действующий режим Union Read. Содержит суффикс (session), если режим задан переменной сеанса.
degraded	Появляется только при переключении таблицы с первичным ключом на другой способ чтения в режиме auto. Значение — одно из key-type, partition-type, tail-truncated и tail-too-large из таблицы выше. Отсутствие доступного для чтения снимка озера здесь не указывается; оно отражается только как unionRead=no. Если секции, существующие только в озере, по-прежнему читаются из озера, unionRead=yes и degraded отображаются вместе.

Настройки подключения к озеру данных

Кластер Fluss записывает настройки datalake.paimon.* из своего файла server.yaml в свойства каждой таблицы с переносом данных в озеро, а Doris на их основе настраивает подключение к озеру Paimon. В большинстве случаев Catalog не нужны собственные настройки подключения к озеру.

Однако перед возвратом свойств таблицы клиенту Fluss удаляет все параметры конфигурации, имена которых содержат key, secret или password. Если озеро Paimon размещено в объектном хранилище, требующем учётных данных, Doris не может получить их от Fluss, поэтому их необходимо указать в Catalog с префиксом fluss.lake.paimon.:

CREATE CATALOG fluss PROPERTIES (
    'type' = 'fluss',
    'fluss.bootstrap.servers' = '10.0.0.1:9123',
    'fluss.lake.paimon.s3.endpoint' = 'http://minio:9000',
    'fluss.lake.paimon.s3.access-key' = '<ak>',
    'fluss.lake.paimon.s3.secret-key' = '<sk>'
);

После префикса используйте собственные имена параметров Paimon — те же, что следуют за datalake.paimon. в файле server.yaml Fluss. Копируйте их без изменений:

server.yaml Fluss	Свойство Catalog в Doris
datalake.paimon.warehouse: s3://bucket/lake	'fluss.lake.paimon.warehouse' = 's3://bucket/lake'
datalake.paimon.metastore: filesystem	'fluss.lake.paimon.metastore' = 'filesystem'
datalake.paimon.s3.endpoint: http://minio:9000	'fluss.lake.paimon.s3.endpoint' = 'http://minio:9000'

Переопределение применяется к каждому ключу отдельно, а не ко всей конфигурации. Ключ, заданный в Catalog, переопределяет такой же ключ, сообщаемый кластером Fluss, а незаданные ключи по-прежнему берутся из кластера. Если указать только учётные данные, warehouse, metastore и остальные настройки по-прежнему будут поступать из кластера.

Параметры хранилища (учётные данные, токены сеанса, конечная точка, регион, стиль адресации и настройки Hadoop) не передаются в Paimon Catalog. Они преобразуются в собственные имена параметров хранилища Doris и применяются на уровне хранения Doris. Причина в том, что эти настройки используются и FE (при чтении манифестов), и BE (при чтении файлов данных), и только уровень хранения обеспечивает обеим сторонам одинаковый набор. Сопоставление выглядит следующим образом:

Параметр Paimon	Параметр хранилища Doris
s3.access-key / s3.access.key / fs.s3a.access.key	s3.access_key
s3.secret-key / s3.secret.key / fs.s3a.secret.key	s3.secret_key
s3.session-token / s3.session.token / fs.s3a.session.token	s3.session_token
s3.endpoint / fs.s3a.endpoint	s3.endpoint
s3.region / fs.s3a.endpoint.region	s3.region
s3.path-style-access / s3.path.style.access / fs.s3a.path.style.access	use_path_style
fs.oss.accessKeyId	oss.access_key
fs.oss.accessKeySecret	oss.secret_key
fs.oss.endpoint	oss.endpoint
fs.obs.access.key / fs.obs.access-key	obs.access_key
fs.obs.secret.key / fs.obs.secret-key	obs.secret_key
fs.obs.endpoint	obs.endpoint

Другие параметры, начинающиеся с fs., dfs. или hadoop., например настройки HDFS HA и Kerberos, также передаются уровню хранения Doris под исходными именами.

fluss.lake.paimon.metastore принимает значения filesystem (по умолчанию), hive и rest. Другие параметры, такие как адрес Hive Metastore или настройки аутентификации REST, записываются так же, как в Paimon Catalog.

ALTER CATALOG может только задавать свойства, но не удалять их. Чтобы отменить переопределение fluss.lake.paimon.*, задайте пустую строку, и Doris вернётся к конфигурации, сообщаемой кластером Fluss:

ALTER CATALOG fluss SET PROPERTIES ('fluss.lake.paimon.warehouse' = '');

Один Fluss Catalog читает таблицы озера с единым набором настроек подключения. После изменения datalake.paimon.* в кластере Fluss удалите и заново создайте Catalog, чтобы Doris загрузил новые настройки. REFRESH CATALOG только очищает кеш метаданных и не перезагружает настройки озера.

Профиль запроса

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

TableReader
├── FlussLogReadTime            8s25ms      Total time on the Fluss side (JNI)
│   ├── FlussPkTailRangeNum          2      Only the range types actually read in this scan are registered
│   ├── FlussPkTailRangeReadTime  6s2ms
│   └── FlussLogRowsReturned         2
├── FlussLakeReadTime           8s25ms      Total time on the Paimon side, including tail reads and filtering
│   ├── FlussLakeRangeNum            1
│   ├── FlussLakeSuppressRangeNum    2
│   └── FlussLakeRowsReturned        7      Rows after filtering; add FlussUnionSuppressedRows to get the count before filtering
├── FlussUnionTailReadTime      5s20ms      One round trip to Fluss per bucket
└── FlussUnionSuppressTime      16.1us      Filtering per block, grows with the number of lake rows

Если вычесть время каждого типа диапазона из общего времени соответствующей стороны, получится стоимость инициализации этого средства чтения (загрузка классов JNI или настройка стека чтения Paimon). У запроса, который читает только таблицу журналов, в профиле нет счётчиков озера или первичного ключа.

Сканирование, читающее сплиты озера, также содержит следующие счётчики в узле TableReader. Они заполняются при Union Read таблицы с первичным ключом:

Счётчик	Значение
FlussUnionSuppressedRows	Строки озера, отброшенные из-за их обновления или удаления в хвосте журнала.
FlussUnionTailKeysRead	Записи журнала изменений, прочитанные из хвостов журналов.
FlussUnionTailKeysRetained	Уникальные первичные ключи, хранящиеся в памяти для фильтрации строк озера.
FlussUnionTailCacheHitCount	Сплиты озера, повторно использовавшие уже прочитанный хвост журнала. На одном BE сплиты озера, относящиеся к одному бакету, совместно используют хвост журнала после его чтения.

Часть данных из Fluss проходит через средство чтения JNI, счётчики которого находятся в узле fluss под TableReader и отображаются только в части DetailProfile профиля. Следующие счётчики относятся к соединениям и куче JVM:

Счётчик	Значение
FlussJniConnectionsOpened	Количество соединений Fluss, открытых для чтения. BE повторно использует простаивающие соединения, поэтому запрос, повторённый сразу после предыдущего, обычно показывает 0.
JvmHeapDeclaredBytes	При включённом контроле допуска JNI к куче — объём кучи JVM, который средства чтения заявили как удерживаемый. Заявки подают только полное чтение таблиц с первичным ключом и чтение хвостов журналов при Union Read таблиц с первичным ключом; для остальных видов чтения отображается 0. См. раздел «Память кучи JVM на BE».
JvmHeapWaitTime	При включённом контроле допуска JNI к куче — время, которое средства чтения провели в очереди в ожидании памяти кучи JVM. См. раздел «Память кучи JVM на BE».

При чтении сплитов озера через Paimon JNI одноимённые счётчики кучи JVM находятся в узле paimon.

Производительность запросов

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

• Нативное средство чтения C++. BE напрямую читает файлы ORC/Parquet в озере Paimon, используя те же средства чтения, что и Paimon Catalog. JVM при этом не используется, а файловый кеш Doris доступен.
• Средство чтения JNI. BE вызывает Fluss Java SDK через JNI для чтения собственных данных Fluss. Журналы, KV-снимки и журналы изменений можно читать только этим способом. Каждый пакет необходимо преобразовать из объектов Java в столбцовые блоки Doris, что значительно медленнее нативного чтения. Некоторые из этих операций чтения также удерживают большой объём кучи JVM; см. раздел «Память кучи JVM на BE».

Используемый путь зависит от типа диапазона сканирования, которому соответствуют счётчики в строке flussScan вывода EXPLAIN:

Читаемые данные	Счётчик EXPLAIN	Путь чтения
Журнал таблицы журналов	logRanges	JNI
Полное чтение таблицы с первичным ключом (KV-снимок + журнал изменений)	pkRanges	JNI
Хвост журнала таблицы с первичным ключом при Union Read	pkTailRanges	JNI
Сплиты озера для таблицы журналов при Union Read	lakeSplits	Нативное средство чтения C++
Сплиты озера для таблицы с первичным ключом при Union Read	lakeSplits, suppressedLakeSplits	Определяется правилами Paimon Catalog, см. ниже. Фильтрация строк озера по первичным ключам из хвоста журнала выполняется на C++ на BE

tbl$lake планируется напрямую Paimon Connector, поэтому в его EXPLAIN нет строки flussScan. Вместо неё отображается paimonNativeReadSplits=N/M: это означает, что N из M сплитов используют нативное средство чтения. В профиле такого запроса нет счётчиков Fluss.

Сплиты озера планирует Paimon Connector, а использование нативного средства чтения определяется теми же правилами, что и для обычной таблицы Paimon:

• Данные таблицы журналов переносятся в таблицу Paimon с добавлением записей. Каждый сплит читает файлы напрямую, поэтому все сплиты используют нативное средство чтения.
• Данные таблицы с первичным ключом переносятся в таблицу Paimon с первичным ключом. Сплиты, для которых уже выполнена компактация и не требуется слияние с другими файлами, используют нативное средство чтения. Для недавно записанных файлов, ещё не прошедших компактацию, и сплитов, файлы которых имеют пересекающиеся диапазоны ключей и требуют слияния при чтении, используется средство чтения Paimon JNI.
• Переменная сеанса force_jni_scanner = true направляет каждый сплит озера через JNI. Используйте её только для диагностики.

Для таблицы с переносом данных в озеро Union Read позволяет читать часть данных в озере (обычно подавляющее большинство) нативным средством чтения, а через JNI — только небольшой объём журнала после смещения переноса. Чем своевременнее перенос данных (чем меньше table.datalake.freshness), тем меньше данных проходит через JNI. Поэтому такие таблицы по умолчанию используют Union Read. Несколько рекомендаций:

• Если вы анализируете историю и актуальность данных не особенно важна, запрашивайте tbl$lake напрямую. Журнал Fluss при этом полностью пропускается: таблица журналов читается целиком нативным средством чтения, а для таблицы с первичным ключом действуют приведённые выше правила Paimon.
• fluss.union_read.mode = disabled направляет чтение всей таблицы через JNI. Использовать этот режим рекомендуется только для диагностики.
• Когда режим auto переключается на другой способ чтения (в degraded= указывается причина в выводе EXPLAIN), все данные из Fluss также читаются через JNI; из озера по-прежнему читаются только секции, существующие исключительно в нём. Обращайте внимание на случаи key-type и partition-type: они определяются схемой таблицы и устраняются только изменением её определения.
• При Union Read таблицы с первичным ключом первичные ключи хвоста журнала дополнительно читаются для каждого бакета на каждом BE (FlussUnionTailReadTime в профиле), а остальные сплиты озера этого бакета на том же BE повторно используют их. Чем длиннее хвост, тем выше затраты.

Чтобы проверить, какой путь использовал запрос, сравните lakeSplits с logRanges, pkRanges и pkTailRanges в EXPLAIN, а также FlussLakeReadTime с FlussLogReadTime в профиле.

Память кучи JVM на BE

BE читает собственные данные Fluss через JNI; сплиты озера Paimon, требующие слияния, также проходят через JNI (см. раздел «Производительность запросов»). Все эти средства чтения выполняются во встроенной в BE JVM и используют общую кучу с чтением других Catalog через JNI и с Java UDF. Размер кучи задаётся параметром -Xmx в JAVA_OPTS_FOR_JDK_17 файла be.conf и по умолчанию составляет 2 GB (-Xmx2048m).

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

Читаемые данные	Что удерживается в куче
Полное чтение таблицы с первичным ключом из Fluss (pkRanges в EXPLAIN)	Журнал изменений после последнего KV-снимка бакета, загруженный в память и объединённый по первичному ключу. Чем больше изменений после снимка, тем больше требуется памяти кучи.
Хвост журнала при Union Read таблицы с первичным ключом (pkTailRanges)	Последняя строка каждого первичного ключа в хвосте. Чем длиннее хвост, тем больше требуется памяти кучи.
Сплиты озера Paimon, требующие слияния (таблица Paimon с первичным ключом, ещё не прошедшая компактацию)	Одна группа строк из каждого объединяемого файла. Один сплит может занимать сотни MB.

Запрос читает несколько диапазонов сканирования параллельно: каждый экземпляр сканирования одновременно запускает до max_file_scanners_concurrency (по умолчанию 16) средств чтения, а на одном BE могут одновременно читать данные несколько экземпляров и несколько запросов. Если несколько таких ресурсоёмких средств чтения открываются одновременно, они могут заполнить кучу, и запрос завершится ошибкой. Сообщение об ошибке указывает размер кучи и возможные решения:

errCode = 2, detailMessage = (10.0.0.3)[JNI_ERROR]OutOfMemoryError: Java heap space. BE's JVM is out of heap. Its maximum is 2048 MB, the -Xmx in JAVA_OPTS_FOR_JDK_17 of be.conf: raise it and restart the BE, or have the statement hold less of the heap at once - read with fewer scanners by setting max_file_scanners_concurrency below its default of 16, or set enable_jni_heap_admission = true to have JNI readers wait for room by the heap they declare. A statement that ran out of heap frees it only once its scanners have exited

Что делать при нехватке памяти кучи

Есть несколько способов решить эту проблему:

1. Выделите JVM больше памяти кучи. Измените -Xmx в JAVA_OPTS_FOR_JDK_17 файла be.conf; изменение вступит в силу после перезапуска BE. Для кластеров, в которых часто выполняются запросы к таблицам с первичным ключом, в первую очередь рассмотрите этот вариант. Куча JVM также занимает память машины, на которой работает BE, поэтому перед увеличением убедитесь, что на машине достаточно памяти.

2. Открывайте меньше средств чтения одновременно. Уменьшите max_file_scanners_concurrency: при меньшем числе одновременно открытых средств чтения удерживается меньше памяти кучи, но чтение замедляется:

   SET max_file_scanners_concurrency = 4;
   -- For a single statement only
   SELECT /*+ SET_VAR(max_file_scanners_concurrency=4) */ * FROM fluss.db.tbl;

3. Включите контроль допуска JNI к куче, чтобы эти средства чтения ожидали в очереди с учётом объёма памяти, который они будут удерживать. См. ниже.

4. Уменьшите объём данных, которые необходимо загружать в кучу. Чем чаще Fluss создаёт KV-снимки (kv.snapshot.interval кластера Fluss), тем короче журнал изменений после снимка. Чем своевременнее перенос данных (table.datalake.freshness), тем короче хвост журнала. После компактации таблицы озера Paimon с первичным ключом её сплиты не требуют слияния, используют нативное средство чтения и не занимают память кучи JVM.

Контроль допуска JNI к куче

Когда включена переменная сеанса enable_jni_heap_admission (по умолчанию false), FE во время планирования оценивает, сколько памяти кучи будет удерживать каждое из трёх перечисленных выше средств чтения, и передаёт оценку BE вместе с диапазоном сканирования. BE отслеживает суммарный объём, заявленный уже открытыми средствами чтения. Новое средство чтения открывается только в том случае, если эта сумма вместе с его заявкой не превышает долю jni_scanner_heap_budget_ratio от максимального размера кучи JVM (то есть -Xmx). В противном случае средство чтения ожидает в очереди, пока ранее открытые средства чтения не завершат работу, не закроются и не освободят свою долю.

SET enable_jni_heap_admission = true;
-- For a single statement only
SELECT /*+ SET_VAR(enable_jni_heap_admission=true) */ * FROM fluss.db.tbl;

При включённой переменной:

• Только три перечисленных выше вида чтения заявляют объём и ожидают. Чтение журнала таблицы журналов и сплитов озера нативным средством чтения ничего не заявляет и выполняется с одинаковой скоростью независимо от состояния переменной.
• Средства чтения могут лишь ожидать. Результаты запросов не меняются.
• Средства чтения ожидают в порядке поступления, поэтому средство чтения с большой заявкой не будет бесконечно уступать место средствам чтения с меньшими заявками, поступившим позже. Если ни одно средство чтения не удерживает долю бюджета, следующее всегда допускается к работе. Поэтому даже средство чтения, заявившее больше всего бюджета, может открыться; оно просто будет работать в одиночку.
• Средство чтения ожидает не более jni_scanner_heap_max_wait_ms (по умолчанию 60 секунд). После этого оно прекращает ожидание и открывается.
• Бюджет задаётся для каждого BE и совместно используется всеми запросами на этом BE, у которых включена переменная. Запросы без неё не заявляют объём, не ожидают и не учитываются в бюджете. Если несколько запросов одновременно читают таблицы с первичным ключом, используйте SET GLOBAL enable_jni_heap_admission = true, чтобы включить её для всех сеансов.
• Переменная также применяется к сплитам, читаемым через JNI в Paimon Catalog.

Оба связанных параметра конфигурации BE можно изменять во время работы. Порядок изменения см. в разделе «Конфигурация BE»:

Параметр конфигурации	По умолчанию	Описание
jni_scanner_heap_budget_ratio	0.5	Доля максимального размера кучи JVM, которую открытые средства чтения могут заявить суммарно. Остальная часть кучи остаётся для средств чтения, которые ничего не заявляют, для других функций BE, использующих JVM, например Java UDF, и для кратковременных пиков во время чтения. Поэтому не задавайте слишком большое значение.
jni_scanner_heap_max_wait_ms	60000	Максимальное время ожидания средства чтения в очереди, в миллисекундах. После этого оно открывается в любом случае.

JvmHeapDeclaredBytes и JvmHeapWaitTime в профиле (см. раздел «Профиль запроса») показывают, какой объём кучи заявили средства чтения и как долго они ожидали. Если JvmHeapWaitTime часто велик, бюджет слишком мал; рассмотрите увеличение -Xmx.

Заявленные объёмы являются оценками, поэтому переменная по умолчанию выключена:

• Размер значений строковых, двоичных и вложенных типов всегда оценивается в 64 байта. Для таблицы с более длинными значениями, например JSON в строковом столбце или длинными строковыми первичными ключами, средства чтения удерживают больше, чем заявляют. Контроль допуска не может предотвратить это, и запрос по-прежнему может завершиться ошибкой OutOfMemoryError.
• Чтение Paimon со слиянием сопровождается кратковременными пиками потребления памяти, которые могут в несколько раз превышать заявленный объём. Эти пики должны помещаться в часть кучи вне бюджета.
• Если один и тот же первичный ключ многократно обновляется в журнале, заявка превышает объём, фактически удерживаемый средством чтения. Из-за этого оно может ожидать дольше необходимого, замедляя запрос.

После исчерпания памяти кучи

• Запрос, столкнувшийся с OutOfMemoryError, завершается ошибкой, а процесс BE продолжает работать.
• Память кучи, удерживаемая этим запросом, освобождается только после завершения всех его средств чтения. Запросы, запущенные до этого момента, могут завершаться ошибкой Failed to attach the current thread to the JVM, code=-1, за которой следуют те же рекомендации, что и выше. Немного подождите и повторите попытку.
• При исчерпании памяти кучи некоторые фоновые потоки клиента Fluss могут также завершиться и не перезапуститься, из-за чего отдельные операции чтения Fluss могут зависнуть навсегда. Эти потоки и удерживаемая ими память освобождаются только при перезапуске BE. После исчерпания памяти кучи JVM перезапустите BE в удобное время.
• Если вы предпочитаете, чтобы BE завершался сразу после исчерпания памяти кучи JVM, а внешний диспетчер процессов перезапускал его, добавьте -XX:+ExitOnOutOfMemoryError в JAVA_OPTS_FOR_JDK_17. При этом все запросы и операции загрузки данных, выполняющиеся на этом BE, завершатся ошибкой.

Ограничения

• Только чтение. INSERT, UPDATE и DELETE не поддерживаются, как и операторы управления базами данных и таблицами, например CREATE TABLE и DROP TABLE.
• Paimon — единственный поддерживаемый формат озера. Для таблицы с другим значением table.datalake.format запрос к tbl$lake завершается ошибкой. Как только у такой таблицы появляется доступный для чтения снимок озера, запрос к самой таблице также завершается ошибкой в режимах auto и required; установите режим Union Read в disabled, чтобы читать данные Fluss.
• Тип Fluss TIME сопоставляется с UNSUPPORTED, и запросы к этому столбцу выполнять нельзя.
• Предикаты не передаются на выполнение во Fluss: за исключением отсечения секций, Doris фильтрует прочитанные из Fluss данные после чтения. Для озера фильтр передаётся в Paimon при планировании сплитов, и Paimon использует его, чтобы пропускать секции и файлы данных. Поддерживается отсечение столбцов и вложенных столбцов для вложенных типов.
• Запросы к историческим состояниям и инкрементальные запросы не поддерживаются, в том числе для tbl$lake.
• Fluss предоставляет только количество строк на уровне таблицы, без статистики по столбцам.

Часто задаваемые вопросы

1. Запрос к таблице с переносом данных в озеро завершается ошибкой the paimon connector plugin is not available

   Данные озера читает плагин Paimon Connector. Проверьте, что каталог plugins/connector/paimon в каталоге развёртывания FE содержит все необходимые файлы.

2. Запрос к tbl$lake или выполнение Union Read завершается ошибкой доступа к объектному хранилищу

   Fluss не передаёт учётные данные хранилища. Настройте их в Catalog через fluss.lake.paimon.*. См. раздел «Настройки подключения к озеру данных».

3. Ошибка has no readable lake snapshot yet

   Служба переноса данных ещё не зафиксировала данные в озере. Запрос к самой таблице сообщает об этом только в режиме required: дождитесь фиксации данных службой переноса или переключитесь в auto. Запрос к tbl$log сообщает об этом в любом режиме, поскольку tbl$log начинает чтение от снимка озера; вместо этого выполните запрос к самой таблице.

4. Запрос к tbl$lake завершается ошибкой nothing has been tiered

   Doris не может найти таблицу озера с текущими настройками подключения. Fluss 1.0 создаёт таблицу озера одновременно с таблицей Fluss, поэтому обычно это означает, что настройки, например fluss.lake.paimon.warehouse, не указывают на хранилище, в которое записывает данные кластер Fluss. См. раздел «Настройки подключения к озеру данных».

5. Ошибка is already serving lake tables with a different paimon configuration

   Конфигурация datalake.paimon.* кластера Fluss изменилась после того, как Catalog начал читать таблицы озера. Удалите и заново создайте Catalog. REFRESH CATALOG не перезагружает настройки озера.

6. Ошибка Part of the log tail has expired

   Union Read таблицы журналов или запрос к tbl$log требует журнала после снимка озера, а часть этого журнала уже удалена по TTL Fluss. Дождитесь, пока Fluss опубликует более новый доступный для чтения снимок озера. См. раздел «Режимы Union Read».

7. Ошибка cannot be read: its partition column 'xx' has fluss type TIMESTAMP(3)

   Doris не может читать столбцы секционирования этого типа. См. раздел «Секционированные таблицы».

8. Ошибка cannot be matched safely to a live fluss partition

   Таблица с первичным ключом секционирована по столбцу, тип которого отличается от STRING, а в озере есть секция, которую Doris не может сопоставить с секцией Fluss, например секция, уже удалённая из Fluss. См. случай partition-type в разделе «Режимы Union Read». Чтобы прочитать данные, ещё находящиеся во Fluss, установите режим Union Read в disabled.

9. Ошибка OutOfMemoryError: Java heap space с текстом BE's JVM is out of heap в сообщении

   Куча JVM на BE заполнена. Чаще всего это происходит при чтении таблиц с первичным ключом, выполнении Union Read таблицы с первичным ключом или чтении большого числа сплитов озера Paimon, требующих слияния. Увеличьте -Xmx, уменьшите max_file_scanners_concurrency или включите enable_jni_heap_admission. См. раздел «Память кучи JVM на BE». После исчерпания памяти кучи перезапустите BE в удобное время.

10. Ошибка Failed to attach the current thread to the JVM, code=-1

    Обычно она возникает сразу после исчерпания памяти кучи JVM: ещё не все средства чтения запроса, завершившегося с OutOfMemoryError, закончили работу, поэтому занятая им память кучи не освобождена. Немного подождите и повторите попытку. См. раздел «Память кучи JVM на BE».

Отладка

Каталог docker/thirdparties/docker-compose/fluss в исходном коде Doris содержит среду Fluss, используемую регрессионными тестами: Fluss, Flink, службу переноса данных Paimon и MinIO. Следуйте инструкциям в README этого каталога, чтобы развернуть среду и проверить функцию.

Источники

• Документация Apache Fluss
• Хранилище Lakehouse во Fluss
• Paimon Catalog