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

Всё это звучит здорово, но стоять на месте никак нельзя. Поэтому мы постоянно работаем над всякого рода улучшениями и оптимизациями. Зачастую новые фичи хранят под капотом не самую тривиальную идею. И сегодня я хочу рассказать о нескольких таких улучшениях, которые мы затащили в виде новых фич в последнем релизе.

В этой статье разберёмся, как работает xor‑фильтр, в чём особенность чанкового хеш‑индекса и как overload controller повышает стабильность работы. Все примеры разберём на примере YTsaurus, но они будут полезны любому разработчику СУБД.

Зачем нужны динамические таблицы

Для начала немного слов о самих динамических таблицах в YTsaurus. Динамические таблицы (или коротко — динтаблицы) — это распределённая база данных, key‑value‑пары которой объединяются в привычные пользователям реляционных СУБД таблицы. Так же, как и в реляционных СУБД, в динтаблицах может быть много колонок, часть из которых объединена в первичный ключ. Такие таблицы можно пошардировать на много машин, при этом каждый шард будет храниться надёжно и отказоустойчиво. Внутренняя репликация обеспечит отсутствие потерь данных даже в случае, если какой‑нибудь из серверов внезапно придёт в полную негодность (подробнее про устройство распределённого журналирования можно посмотреть в докладе на Hydra'20).

В динтаблицах поддерживаются распределённые транзакции: в одной транзакции можно писать в много шардов и даже в много таблиц. Транзакции такие же надёжные, как и сами данные: вы можете писать и не задумываться над тем, что произойдёт, если вдруг сломается какой‑нибудь из тысяч серверов. Для работы в нескольких дата‑центрах мы поддерживаем репликацию между кластерами YTsaurus. На Highload++'23 мы рассказали о деталях такой репликации и о том, какие получаются гарантии доступности.

Мы уделяем большое внимание тому, с какой скоростью пользователи читают данные. Для максимально быстрого чтения есть возможность положить все данные таблицы в оперативную память (при этом данные будут лежать и на дисках для надёжности), а также разные другие варианты оптимизации. Например, мы сделали отдельный слой хранения пользовательских данных среднего размера, которые читаются эффективно с помощью io_uring с NVMe SSD (подробнее об этом можно посмотреть в докладе на SmartData'23).

Xor-фильтр

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

Практически любой теоретический разговор про LSM (Log‑Structured Merge Tree) упоминает Bloom Filter в качестве структуры данных, которую полезно использовать для уменьшения количества хождений в диски. Bloom Filter позволяет узнать, принадлежит ли объект множеству, и даёт ответ либо «точно нет», либо «возможно, да». Тем самым с некоторой вероятностью он уменьшает число лишних чтений с диска.

Впервые использовать Bloom Filter мы попытались ещё несколько лет назад, но тогда стало понятно, что сами фильтры получаются довольно объёмными: для хорошей вероятности ответа размер памяти под Bloom Filter можно посчитать как 7–8 бит на ключ (или по‑простому — размер таблицы в строках будет равен объёму Bloom Filter в байтах). Даже если включать фильтр ограниченно на части таблиц, получится весьма существенная доля памяти. А если мы посмотрим на наши таблицы сейчас, окажется, что оперативной памяти просто не хватит на то, чтобы хранить фильтры.

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

Есть несколько алгоритмов, решающих ту же задачу, что и Bloom Filter. В 2015 году мы сравнивали Bloom с Cuckoo Filter, а с тех пор появились ещё Xor Filter и Ribbon Filter, используемый в RocksDB. При выборе фильтра для динтаблиц мы решили взять xor-фильтр: он достаточно эффективный как по размеру, так и по скорости, и к тому же легко пишется.

Рассмотрим, как работает xor‑фильтр, на простом примере. Предположим, что у нас есть три массива интов — B1, B2, B3. Хеш‑функции h1, h2, h3 отображают объект в элементы B1, B2, B3, и ещё одна служит для вычисления fingerprint. Функция проверки объекта в фильтре выглядит как fingerprint(x) == B1[h1(x)] ^ B2[h2(x)] ^ B3[h3(x)].

Как же построить такие B1, B2, B3? Построение итеративное: если для выбранных h1, h2 и h3 их построить не удаётся, то предпринимается следующая попытка с множествами большего размера. Каждая попытка выглядит так: нарисуем гиперграф с вершинами, которые соответствуют элементам B1, B2, B3, и гиперрёбрами, которые, в свою очередь, соответствуют нашим объектам x: {h1(x), h2(x), h3(x)}. Если в графе есть вершина v степени 1, это означает, что для отображающегося в ней объекта х (пусть v = h1(x)) мы можем решить задачу: неважно, какие были B2[h2(x)] и B3[h3(x)], ведь мы можем подобрать такое значение B1[h1(x)], чтобы удовлетворить условие фильтрации.

Удалим гиперребро x из нашего графа и повторим процедуру. Если мы удалим все гиперрёбра, то у нас получится последовательность элементов наших массивов в обратном порядке относительно порядка удаления. Их нужно заполнить, чтобы получить корректный фильтр. Если дойти до конца не удалось — повторим попытку.

На картинке ниже показан пример построения xor‑фильтра. Если мы удалим гиперрёбра в последовательности x1 (синий), x3 (жёлтый) и x2 (зелёный) и запомним вершины 1, 2, 3 (обведены красным), то, чтобы построить фильтр, достаточно будет выбрать произвольные значения для всех остальных вершин, а после вычислить значения для вершин 3, 2, 1 (именно в таком, обратном, порядке). Например, для вершины 3 это будет B3[h3(x2)] = fingerprint(x2) ^ B1[h1(x2)] ^ B2[h2(x2)].

Ниже показан результат включения xor‑фильтра на одной из таблиц: на графике показан объём читаемых с диска данных за единицу времени. Хорошо видно, что он сократился втрое: примерно с 1,5 Гб/с до 0,5 Гб/с.

Чанковый хеш-индекс

Идея чанкового хеш‑индекса тоже витала в воздухе долгое время. Когда мы впервые задумались о такой возможности, характеристики NVMe SSD были не до конца понятны, да и вообще такие железки были только в самых свежих поставках — и мы ещё не понимали, насколько же они классные. Чтобы обосновать работоспособность хеш‑индекса, мы даже провели детальный анализ производительности NVMe SSD.

В основе лежит простая мысль: если мы хотим читать из чанка всего одну строчку, но не хотим держать в памяти офсеты для всех строк, давайте положим в чанк хеш‑таблицу, которая хранит этот самый офсет. Это даст несколько мелких чтений с NVMe для самого чтения строки: один блок для индекса и один блок, содержащий строку. Даже несколько мелких чтений с NVMe оказываются быстрее и эффективнее, чем традиционные крупные динтабличные блоки, причём с точки зрения как пропускной способности, так и времени доступа.

В качестве хеш‑таблицы хорошо подходит linear probing: её очень легко реализовать, и она идеально подходит для блоковых чтений: за одно обращение к NVMe мы получим сразу целый диапазон значений, которые будет пробегать linear probing. К тому же чанки иммутабельны, и при записи можно сделать несколько попыток создать хорошую хеш‑таблицу. Если добавить кеширование блоков индекса, то можно рассчитывать на ещё более эффективные чтения.

Чтобы повысить плотность индекса, мы храним в нём не сами ключи, а 64-битный хеш от ключа. При чтении может оказаться, что ключ отличается и нам потребуется продолжить поиск в хеш‑таблице, но вероятность коллизии очень мала.

Но оказалось, что при построении чанков возникает интересный эффект: если писать строки последовательно, то они не выравниваются по блокам, поэтому чтение будет менее эффективным. Например, если размеры строк меньше 4 килобайт (размер блока NVMe), но мы их запишем невыровненно, то на каждую строку придётся читать два блока — это уменьшит пропускную способность системы вдвое. Если каждую строчку начинать с нового блока, то сильно увеличится объём.

Как же стоит выравнивать строки? Если допустить переупорядочение хотя бы в ограниченных пределах (нам достаточно переупорядочивать строки в пределах большого динтабличного блока), то внезапно вся проблема сводится к известной в Computer Science задаче Bin Packing. Её суть в том, что есть корзины фиксированного размера и некоторое количество предметов, надо раскидать предметы по корзинам так, чтобы использовать как можно меньше корзин.

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

Представьте, что у нас есть какое‑то распределение таких обрезанных строк по корзинам. Если мы возьмём и растянем одну из строк до её исходного размера, то получится, что вся картина растянулась на целое число блоков (на рисунке ниже это представлено как переход от верхней части схемы к нижней). При этом расположение границ строк относительно границ соседних блоков осталось прежним. Получается, что никакая другая строка не стала накрывать большее число блоков и мы получили решение задачи.

Ниже показан результат применения хеш‑индекса. Как и на предыдущем графике, мы видим падение объёма данных, читаемых с дисков. Только в этот раз падение заметнее: с 800 Мб/c до 50 Мб/с.

Overload controller

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

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

Здесь на помощь приходит Feedback Control: можно построить систему из двух компонентов, один из которых собирает информацию о системе (feedback), а второй принимает решение о переключении состояния системы (control). Канонический пример такой системы — климат‑контроль: если температура ниже ожидаемой, включается подогрев, а если выше — он выключается. В компьютерных системах feedback control может встретиться, например, на уровне лимитера запросов в микросервисы.

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

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

На картинке показано, как это происходит внутри машины. Слева нарисована компонента RPC (Remote Procedure Call), которая отвечает за приём и передачу сообщений от пользователя. Справа изображены тредпулы и их очереди. Снизу — overload controller. Этот блок собирает feedback в виде размеров очередей тредпулов и отправляет сигнал контроля для RPC: принимать или отклонять запросы.

Чтобы продемонстрировать результат работы overload controller на практике, покажу момент срабатывания. На этом графике — потребление CPU в тредпуле Bus (отвечает за передачу сообщений по сети).

А на этом — общее количество реквестов. Сплошным синим цветом отмечено число реквестов, отменённых из‑за срабатывания контроллера. Можно заметить, что тут перегрузка длилась почти 20 минут. Без overload controller это почти наверняка привело бы к заметной недоступности шардов таблиц, попавших на эту машину. При использовании overload controller мы отказались от выполнения только части пользовательских запросов, а остальные (в том числе внутрисистемные, отвечающие за связность кластера) отработали успешно.

Scalable fair-share thread pool

Fair‑share‑планировщики применяются там, где нужно обеспечить равномерное распределение ресурса между задачами. В динтаблицах мы используем fair‑share в пуле потоков, обрабатывающем запросы select rows, а также при хождении в диски.

Select‑запросы выполняются распределённо и состоят из множества более простых действий. В точке получения запроса в системе (на прокси) запрос аннотируется уникальным тегом (trace id), который в дальнейшем используется для планирования. На каждой машине выполнения CPU распределяется так, чтобы каждому запросу досталось одинаковое количество памяти в единицу времени. Единицы выполнения — простые действия, ограниченные по времени. В связи с тем, что при реализации планировщика внутри пользовательского процесса используется кооперативная многозадачность, в коде в различных местах расставлены yield. Они приостанавливают выполнение действия, если оно выполняется слишком долго, чтобы дать возможность поработать другим действиям.

Если быть честными, в select‑запросах используется двухуровневое планирование по пользователю (или указываемому пулу) и по тегу запроса. Но для понимания концепции достаточно считать, что планирование делается только по тегу.

Fair‑share‑планировщик поддерживает набор корзин, соответствующих тегу планирования и содержащих FIFO‑очереди действий. Корзины организованы в виде очереди с приоритетом — это время, потраченное на выполнение действий из данной корзины. На каждой итерации планирования выбирается корзина с минимальным использованным временем.

Долгое время мы использовали простую реализацию алгоритма планирования. В качестве приоритетной очереди использовалась куча, а конкурентный доступ был под спинлоком. Эта конструкция ожидаемо была не очень эффективной. При этом она позволяла значительно улучшить время выполнения select‑запросов. На графике можно видеть сравнение перцентилей времени ответов на select‑запросы при использовании fair‑share‑пула потоков и без него.

Недостатком такой схемы было то, что при увеличении количества действий до сотен тысяч в секунду производительность начинала значительно деградировать. Если для select‑запросов в большинстве случаев этого было достаточно, то использовать такой пул потоков для некоторых других видов нагрузки (например, lookup rows) не получалось из‑за ограниченной масштабируемости.

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

Можно было бы пробовать шардировать структуру данных и использовать fine‑grained‑блокировки, но это ухудшило бы гарантии планирования. Cтруктуры данных lock‑free даже для FIFO‑очереди оказываются довольно медленными. Либо же возникают проблемы с нарушением свойств FIFO (moody camel).

В итоге мы использовали подход для оптимизации многопоточной синхронизации — Flat Combining. Идея довольно простая: вместо того, чтобы брать блокировку на каждый доступ к структуре данных из каждого потока, множество доступов (чтений или модификаций) к структуре данных оформляется в виде запросов. После получения одним из потоков эксклюзивного доступа к структуре данных происходит выполнение всех накопившихся запросов от других потоков под одной блокировкой. Таким образом, каждый тред публикует запрос, а дальше либо получает доступ и выполняет все запросы, либо ждёт выполнения своего запроса.

Что ещё нам удалось улучшить:

• Добавление действия в тредпул стало lock free. Действия можно добавлять в multiple producer single consumer lock free stack, а потом под блокировкой перекладывать их в корзины в приоритетную очередь.

• Завершение выполнения действия и получение нового объединены в один запрос.

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

В результате пропускная способность fair‑share тредпула увеличилась в несколько раз. Но главное — даже не throughput, а улучшенное latency. Следующий график показывает время ожидания в очереди при выполнении набора задач, равных количеству тредов: задачи не ждут в очереди, для каждой всегда есть свободный поток.

Новый вывод диапазонов в select-запросах

Для гранулярного чтения данных из таблицы при выполнении select‑запросов из условия WHERE выделяются ограничения на ключевые колонки таблицы и строится набор диапазонов чтения. Сложность вывода диапазонов заключается в том, что в схеме таблиц могут быть вычисляемые колонки. Значение в вычисляемой колонке формируется по формуле, которая указана в схеме таблицы. Например, если в таблице есть колонки Hash и UserId, причём для колонки Hash указано выражение hash(UserId), то пользователю достаточно указать только UserId — значение для колонки Hash будет вычислено автоматически.

При выводе диапазонов приходится учитывать специальные случаи вычисляемых колонок.

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

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

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

Рассмотрим сам процесс вывода диапазонов. По предикату из условия WHERE выделяются ограничения на ключевые колонки. Можно считать, что предикат преобразуется в дизъюнктивную нормальную форму, где булевыми переменными будет считаться принадлежность ключевой колонки к определённому диапазону значений. Ограничения представлены в виде дерева — в каждом узле записано ограничение на некоторую колонку. Под ограничениями я имею в виду диапазон допустимых значений.

В качестве примера рассмотрим выражение (k between 1 and 5 or k in (10, 13)) and l between 4 and 7 or k = 7 and l between 5 and 9, где k и l — ключевые колонки. В результате получается такое дерево ограничений:

[1 .. 5]:
. [4 .. 7]: <universe>
7:
. [5 .. 9]: <universe>
10:
. [4 .. 7]: <universe>
13:
. [4 .. 7]: <universe>

Далее из этих ограничений строятся диапазоны чтения. Ограничение k∈[1..5] ∧ l∈[4..7] нельзя представить в виде диапазона чтения, поэтому получается только диапазон [1]..[5, <Max>]. Полный набор выводимых диапазонов чтения для данного выражения получается следующий:

[{[1], [5, <Max>]}, {[7, 5], [7, 9, <Max>]}, {[10, 4], [10, 7, <Max>]}, {[13, 4], [13, 7, <Max>]}]

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

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

• Аргументы не фиксированы, но их диапазоны допустимых значений имеют ограниченный размер. В таком случае предпринимается попытка перебрать значения из диапазона.

Выражение для вычисляемой колонки рассматривается в виде f(k_{i}/N_{i},...) % M. Для каждого аргумента в выражении определяются все используемые с ним делители (если нет деления, то считается, что делитель равен 1), а также учитывается модуль.

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

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

Рассмотрим пример вычисляемой колонки h = hash(k / 3 + l, k / 5, l) % 6 и выражения k between 6 and 14 and l = 1. В этом случае можно перебрать значения k, которые при делении на 3 и на 5 дают разные значения. Получатся следующие диапазоны:

[{[0u, 6], [0u, 14, <Max>]}, {[1u, 6], [1u, 14, <Max>]}, {[3u, 6], [3u, 14, <Max>]}, {[4u, 6], [4u, 14, <Max>]}]

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

[{[<Null>, 6], [<Null>, 15, <Max>]}, {[0u, 6], [0u, 15, <Max>]}, {[1u, 6], [1u, 15, <Max>]}, {[2u, 6], [2u, 15, <Max>]}, {[3u, 6], [3u, 15, <Max>]}, {[4u, 6], [4u, 15, <Max>]}, {[5u, 6], [5u, 15, <Max>]}]

Один из частых сценариев запросов к динтаблице — получение постраничного результата (pagination). В этом случае данные читаются начиная с некоторого ключа. Чтение ограничивается лимитом строк. Также на ключ могут быть наложены дополнительные ограничения.

Рассмотрим пример. Пусть у таблицы есть ключевые колонки h=k, k, l. Нужно вывести диапазоны чтения для предиката (h, k, l) > (1, 1, 2) and k = 1. Из выражения получаются следующие ограничения на ключевые колонки:

1u:
. 1:
. . (2 .. <Max>): <universe>
(1u .. <Max>):
. 1: <universe>

Далее рассматриваются наборы ограничений.

1. Набор h:[1..1], k:[1..1], l:(2..Max). Первую колонку h можно вычислить, так как у колонки k фиксированное значение. Значение h получится равным 1, и это не противоречит ограничению на h (от 1 до 1 включительно). Получим диапазон чтения от [1, 1, 2, <Max>] до [1, 1, <Max>]. Для верхней границы Max в конце означает то, что граница включена, для нижней — что исключена.

2. Набор h:(1..Max), k:[1..1], l:(2..Max). При вычислении значения h получим 1. Оно не пересекается диапазоном (1..Max). Из этого набора не получаем диапазонов чтения.

Раньше при условии такого вида вычисление колонки не происходило при явно заданных на неё ограничениях. Из‑за этого выводился большой дополнительный диапазон — чтение становилось неэффективным. Мы это исправили.

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

• C помощью xor‑фильтра можно сильно снизить количество походов в диски в случае, когда доминируют чтения по несуществующим ключам.

• С помощью чанкового индекса можно сильно уменьшить объём читаемого с дисков, когда размер записи сравним с размером физического блока на NVMe SSD.

• С помощью overload controller мы добиваемся повышения стабильности работы в случае перегрузок пользовательскими запросами.

• С помощью нового fair‑share‑тредпула мы можем заметно повысить число запросов, обрабатываемых системой, не потеряв при этом по времени доступа.

• С помощью нового вывода диапазонов мы эффективнее обрабатываем некоторые классы запросов при использовании вычисляемых колонок в таблицах.

Все эти улучшения доступны в open‑source‑версии YTsaurus. А если у вас возникнут вопросы, то смело задавайте их в нашем community‑чате.