Динамические таблицы — это распределённая база данных, key‑value‑пары которой объединяются в привычные пользователям реляционных СУБД таблицы. В YTsaurus в них можно хранить огромные массивы данных, при этом их можно быстро читать — поэтому YTsaurus используют почти все сервисы Яндекса: Реклама, Маркет, Такси, даже Поиск при построении поисковой базы, и другие.

Я руковожу службой разработки динамических таблиц в Yandex Infrastructure и раньше уже рассказывал, как мы оптимизировали чтение, улучшали выборку строк в SQL‑запросах и защищались от перегрузок. Сегодня вышла новая версия YTsaurus 24.1.0, в которой динамические таблицы получили ещё несколько долгожданных доработок. В статье расскажу про них подробнее.

Вторичные индексы

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

В новой версии мы реализовали семантические вторичные индексы — они реализованы через отдельную индексную таблицу, запись в которую осуществляется системой. А также добавили простое использование таких индексов через оператор WITH INDEX.

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

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

Любая запись в индексируемую таблицу отразится в индексной таблице. Если у нас идёт перезапись, нам необходимо знать, какое значение перезаписывается, чтобы сделать нужные удаления и вставки в индексной таблице. Поэтому каждая запись в индексируемую таблицу — это также и чтение из неё же. Эти вычисления происходят на прокси прозрачно для пользователя.

Если таблица уже существует, то для неё нужно построить индексную. Это делается с помощью специальной map‑reduce‑операции.

Пока индекс строится с помощью map‑reduce‑операции, в исходную таблицу может продолжаться запись. При этом в индексе будут данные для свежих строк, но их ещё не будет для исторических данных.

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

WebAssembly

В своё время мы поддержали User Defined Functions, в которых давали возможность писать функции на C++ и использовать их в запросах. Но вскоре мы поняли, что для multitenant‑системы это была не самая лучшая идея. Код на С++ небезопасен, а если непроверенная функция работает в рамках того же процесса, что и нода YTsaurus, то проблемы в пользовательской функции потенциально могут затронуть весь процесс ноды. Что ещё хуже, может быть некорректно затронута чужая память, и мы побьём данные. В общем, нужна серьёзная изоляция пользовательского C++ кода.

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

Немного поясним, как работает наша виртуальная машина. На вход приходит программа на WebAssembly: такую можно получить как из C++ с помощью clang, так и из LLVM‑представления, c использованием wasm‑бэкенда для LLVM. Далее виртуальная машина уже сама JIT‑компилирует такую программу в безопасный код под x86.

У каждого инстанса виртуальной машины есть свой выделенный кусок памяти. В начале работы ВМ аллоцирует память при помощи mmap и запрещает запись в последнюю страницу при помощи mprotect. Безопасная работа достигается при помощи нескольких техник:

1. Каждое обращение к памяти проверяется на выход за разрешённые диапазоны. Эта проверка выполняется достаточно быстро: адрес сравнивается с максимально допустимым, после чего делается cmov. Если адрес не лежал в пространстве адресов ВМ, мы обратимся к guard‑страничке и попадём в обработчик SIGSEGV. После этого мы остановим работу нашей виртуальной машины.

2. Два стека. Для стека вызовов используется обычный x86-стек. Стек с данными находится в защищённой области памяти. Таким образом, перетереть адрес возврата из функции не получится.

3. В WebAssembly нет инструкции jump. JIT‑скомпилированный код не может «прыгать» по произвольным адресам (и по невыровненным тоже).

Мы научились выполнять запросы с UDF, написанными на C++ в WebAssembly‑окружении, для чего потребовалось не только запустить виртуальную машину, но и адаптировать libc и libc++ для запуска внутри неё. К сожалению, пока процесс создания, загрузки и подключения новых UDF остаётся сложным. В будущем мы планируем сделать интерфейс более удобным.

Оптимизация и рефакторинг GROUP BY

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

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

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

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

Конечно же, любое изменение работы операторов должно быть хорошо протестировано. Для GROUP BY мы сделали специальный стресс‑тест: генерируем большое количество случайных запросов с группировками и сортировками. Эти запросы мы выполняем при помощи нашего движка, а также при помощи специально написанной для тестов реализации GROUP BY на Python. После выполнения запросов мы сравниваем их результаты. Для некоторых запросов может быть много правильных результатов, например, когда ключ сортировки не указан целиком в операторе проекции. Результат выполнения таких запросов мы тоже проверяем на корректность.

Когда мы стали внедрять эти изменения на наших кластерах, оказалось, что нашего стресс‑теста недостаточно. Чтобы расширить тестовую базу и разнообразить сценарии, мы собрали большое количество запросов с наших кластеров вместе с семплом данных. Динтаблицами пользуются самые разные команды в Яндексе, и такое расширенное тестирование позволило отладить код и довести его до продакшн‑стадии.

Рефакторинг открыл дорогу к оптимизациям группировок. К релизу YTsaurus 24.1.0 мы завершили оптимизацию, объединяющую операторы группировки и сортировки в один. Выполнение одновременно группировки и сортировки позволит не держать в памяти все группы, и сразу отбрасывать те, которые точно не попадут в ответ. Таким образом можно оптимизировать выполнение запросов, в которых одновременно указаны GROUP BY, ORDER BY и LIMIT.

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

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

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

Multicell и шардированный сбор метрик 

Первая версия балансировщика жила внутри кода мастер‑сервера. Большой плюс этого подхода — все данные есть под рукой, очень легко манипулировать объектами. Но есть и минусы: мастер шардирован, и таблицы могут жить на разных мастерах, на которых независимо работает балансировка. В итоге, по мнению каждого мастер‑сервера, его таблицы могут быть распределены достаточно равномерно, но в совокупности возникает заметный дисбаланс. Кроме этого, у мастеров очень длинный релизный цикл, а новый алгоритм хотелось тестировать и отлаживать побыстрее.

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

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

Уменьшаем дисперсию 

После того как метрика каждого таблета зафиксирована, задача балансировки превращается в вариацию классической NP‑полной задачи Bin packing: есть несколько объектов и контейнеры, в которые необходимо эти объекты упаковать, минимизируя размер максимального контейнера. Какое‑то разбиение при этом уже имеется, и желательно не менять его очень сильно: каждое перемещение таблета — даунтайм таблицы. Несмотря на все усилия, решить NP‑полную задачу мы не смогли и применили жадный алгоритм.

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

Решардирование 

Раньше балансировщик умел решардировать таблеты только по размеру. Это не работает в ситуации, когда нагрузка сильно неравномерно распределена по таблице: маленький таблет может оказаться очень сильно нагружен. Мы поддержали учёт метрик в решардировании: если таблет «маленький» и по нагрузке, и по размеру, то он будет склеен с соседом, а если «большой» хотя бы по одному из параметров, то поделён на несколько.

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

Выбор метрик 

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

Compaction Digests 

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

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

На регулярную компактификацию на наших серверах сейчас приходится порядка 30% всех компактифицируемых данных. Выходных данных при этом всего на 8% меньше, чем входных, и этот показатель не отличается от обычной фоновой компактификации. Это значит, что алгоритм фактически вхолостую перерабатывает одни и те же строки. Зная распределение временных меток строк в чанке, мы могли бы принимать решение о компактификации, исходя из доли устаревших строк. Если бы каждая компактификация, запущенная с целью очистки старых строк, удаляла 50% данных, мы бы сэкономили половину потока в диски, при этом гарантируя, что каждая таблица состоит из мусора не более чем на 50%.

Структура для сжатого хранения распределения называется quantile digest. Есть несколько вариаций, большинство из которых основано на одной идее: изначально будем хранить добавленные элементы абсолютно точно, а с ростом занимаемой памяти сжимать информацию о соседних элементах, запоминая, что в некотором отрезке [L, R] суммарно было V записей.

Например, Q‑digest хранит элементы в полном неявном бинарном дереве с ограничениями на размер узлов. Каждая вершина соответствует некоторому отрезку: лист — отрезку вида [i, i+1), вершины следующего уровня — отрезкам вида [2j, 2j+2), корень — отрезку [0, 2^64). Каждый добавленный элемент учитывается ровно в одной вершине на пути от соответствующего листа до корня.

В дереве поддерживается два инварианта в зависимости от параметра k — ограничения на размер: во‑первых, значение в вершине не может быть больше n/k; во‑вторых, для каждой непустой вершины сумма значений в ней, брате и предке не может быть меньше n/k. Эти инварианты несложно поддерживаются при вставке. Их оказывается достаточно для обеспечения погрешности порядка 64/k и линейного относительно k числа ненулевых вершин.

Более продвинутая структура t‑digest явно применяет k‑means кластеризацию для получения лучшего отношения размера к погрешности. Её реализация уже успешно применялась в Яндексе.

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

Запись без блокировок 

Если паттерн записи в таблицу такой, что много транзакций обновляет какое‑то значение, но при этом его не читает, то у таких транзакций можно исключить конфликт записи. Напрашивающийся пример: запись инкрементов в агрегирующую колонку из разных транзакций. Такой режим записи у нас получил название shared write lock. В других системах он также известен как blind write.

Обычная транзакция предполагает что у неё есть время старта и время коммита и в момент коммита проверяется, что все значения, которые транзакция обновляет, не менялись с момента начала транзакции. Такая проверка отражает ожидаемое поведения для read‑modify‑write транзакций, когда транзакция читает какое‑то значение и записывает его модификацию. Несложно проверить, что транзакции получаются сериализуемыми, причём порядок сериализации задаётся временем коммита.

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

Как же это работает? В динтаблицах есть структура данных, куда складываются свежие данные при записи — dynamic store. Она также служит для разрешения конфликтов между транзакциями. Если сильно упростить, то для каждого значения в dynamic store хранится последовательность версий, причём только одно, самое последнее значение, может быть незакоммиченным. По наличию незакоммиченного значения и обнаруживаются конфликты, когда две транзакции хотят обновить одно и то же значение. В случае с shared write lock мы расширили слот незакоммиченных значений с единственного до множества. По мере того как транзакции коммитятся и сериализуются, значения перемещаются из этого множества в строгую последовательность закоммиченных значений.

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

Словарное сжатие строковых значений 

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

Решать эту проблему можно, сжимая каждое значение в отдельности. Но на маленьких объёмах в единицы килобайт кодек общего назначения плохо справляется. Чтобы сжимать эффективно малые объёмы данных нужно пользоваться словарями. Мы воспользовались возможностями библиотеки zstd по построению и использованию словарей для сжатия и разжатия данных.

Проблема подготовки словаря в том, что для эффективного сжатия при построении словаря нужно использовать данные, максимально близкие к реальным. Но данные в таблице могут меняться со временем, и единожды построенный словарь может стать не очень хорошим в будущем. Получается, что нужно периодически создавать новый словарь, при этом оставляя для старых данных старые словари. К счастью, организация данных в виде LSM‑дерева (Log‑Strucutred Mergre Tree) даёт естественную разбивку данных на чанки, и для каждого чанка можно указывать свой словарь.

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

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

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

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

Балансировка запросов к Rpc-Proxy 

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

К сожалению, простой случайный выбор — не самый лучший способ балансировать нагрузку. Даже при абсолютной идентичности прокси по производительности и равенстве нагрузки от запроса теория вероятности утверждает, что будет заметная дисперсия. Если добавить прелести реальной жизни в виде некоторого фона тяжёлых запросов и случайных аномалий в производительности CPU на разных машинах — проблем не избежать.

Мы и правда наблюдали подобные проблемы и в новом релизе переделали алгоритм балансировки. Есть известный алгоритм, называемый power of two choices: нужно выбрать не одну случайную прокси, а две и отправить запрос в ту, которая наименее загружена. У такого алгоритма более качественная теоретическая оценка равномерности распределения и он заметно лучше работает на практике. В частности, используется в Google Maglev и ScyllaDB.

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

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

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

Все новые возможности уже доступны в опенсорс‑версии YTsaurus. Смотрите, пробуйте, делитесь впечатлениями. Будем рады вашим комментариям и пул‑реквестам. Если у вас есть вопросы, будем рады вас видеть в нашем community‑чате.