Распределённый SQL Picodata: планы, motion и цена ключа шардирования

Запрос к Picodata выглядит как запрос к PostgreSQL, но исполняется иначе: данные лежат на разных узлах, и прежде чем что-то посчитать, движок решает, где именно это считать и что придётся перевезти. Решение видно в плане, а последствия — в лимитах, в которые запрос упирается или не упирается.

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

Третья статья серии; предыдущая — про границы совместимости с PostgreSQL. Планы ниже сняты стендом databases/picodata, сценарий sbroad, на таблице из 200 000 товаров, распределённой по 3000 бакетам между двумя репликасетами.

В статье

• Два слоя исполнения
• Адресация: сколько бакетов трогает запрос
• Агрегация: две фазы и motion между ними
• Коллокация: когда соединение не требует переноса
• Цена неверного ключа шардирования
• Два лимита, о которых стоит знать заранее
• Чем платим
• Что дальше

Два слоя исполнения

Распределённый SQL в Picodata — собственный движок на Rust. Запрос проходит разбор, превращается в дерево, в него вставляются узлы переноса данных, и получившийся план распадается на части: одни исполняются на узлах-хранилищах, другие — на узле, принявшем запрос.

Но исполняет локальную часть не этот движок, а машина ядра — того самого Tarantool, на форке которого построена Picodata. Двухслойность не абстракция: именно из неё следует поведение, которое иначе выглядит необъяснимым. Плагин на Rust, работающий внутри кластера, при выполнении запроса упирается ровно в тот же лимит и получает ровно то же сообщение, что и внешний клиент. Если бы лимит принадлежал внешнему SQL-интерфейсу, код внутри кластера его бы не заметил — а он замечает.

Адресация: сколько бакетов трогает запрос

План завершается строкой, которую стоит читать первой. Запрос по ключу шардирования:

projection ("products"."id"::int -> "id", "products"."title"::string -> "title")
    selection "products"."id"::int = 42::int
        scan "products"
execution options:
    sql_vdbe_opcode_max = 5000000
    sql_motion_row_max = 5000
buckets = [2426]

buckets = [2426] — один бакет. Движок вычислил хеш от ключа и знает, на каком узле лежит строка; остальной кластер запрос не побеспокоит.

Тот же запрос по обычной колонке:

selection "products"."category"::string = 'tools'::string
        scan "products"
buckets = [1-3000]

buckets = [1-3000] — все три тысячи. Условие не позволяет вычислить местоположение, и запрос идёт на все узлы. Это не ошибка и не патология: полный проход по кластеру — нормальный режим для аналитики. Важно понимать, когда вы в нём находитесь, потому что стоимость отличается принципиально.

Агрегация: две фазы и motion между ними

Дальше начинается самое интересное. Простой на вид GROUP BY разворачивается в двухфазную конструкцию:

projection (…sum(("count_1"::int))::int -> "col_1")
    group by ("gr_expr_1"::string) …
        motion [policy: full, program: ReshardIfNeeded]
            projection ("products"."category"::string -> "gr_expr_1",
                        count((*::int))::int -> "count_1")
                group by ("products"."category"::string) …
                    scan "products"

Читать снизу вверх. Внизу — локальная фаза: каждое хранилище само сканирует свою часть данных, группирует и считает частичный count. Посередине — motion, узел переноса: частичные результаты собираются. Наверху — финальная фаза: группировка повторяется и частичные счётчики складываются через sum.

Ключевое в этой картине: по сети едут не строки таблицы, а частичные агрегаты. Категорий шесть, значит с каждого хранилища приезжает шесть пар «категория — счётчик», а не 200 000 строк. Это тот же принцип, что в map-reduce, и он объясняет, почему агрегация по всему кластеру остаётся дешёвой.

Коллокация: когда соединение не требует переноса

Теперь два соединения одной и той же таблицы с собой — отличаются только колонкой, по которой идёт сопоставление.

По ключу шардирования:

join on "a"."id"::int = "b"."id"::int
            scan "products" -> "a"
            scan "products" -> "b"

Никакого motion. Строки с одинаковым id по построению лежат в одном бакете, то есть на одном узле, — соединять их можно локально. Это и называется коллокацией: данные, которые соединяют, заранее размещены вместе.

По обычной колонке:

join on "a"."category"::string = "b"."category"::string
            scan "products" -> "a"
            motion [policy: full, program: ReshardIfNeeded]
                    scan "products" -> "b"

Появился motion [policy: full] над правой стороной. Строки с одинаковой категорией разбросаны по всем узлам, поэтому правую таблицу нужно перевезти целиком, чтобы соединение стало возможным.

Разница между этими двумя планами — это разница между «посчитали на месте» и «сначала перевезли данные по сети».

Цена неверного ключа шардирования

EXPLAIN выше показывает план, но не доказывает, что запрос выполнится. Выполненный, даже узкий вариант соединения по category падает:

sbroad: unexpected number of values:
Exceeded maximum number of rows (5000) in virtual table: 99776

Фильтр слева перенос не уменьшает: motion [policy: full] перевозит всю правую сторону — здесь 99 776 строк, долю одного репликасета (конкретное число зависит от того, как в кластере разложились данные). При лимите sql_motion_row_max в 5000 строк это отказ. Это стоит проговорить прямо: неверный ключ шардирования превращает замедление в отказ. В классической СУБД такой запрос отработал бы, просто дольше; здесь он завершается ошибкой.

Но в этом запросе смешаны три вещи: перенос, алгоритм соединения и число получающихся пар — у category шесть значений, и каждая строка слева совпадает с десятками тысяч справа. Чтобы их развести, стенд выполняет три соединения на одном и том же фильтре a.id < 10 и сверяет результат с PostgreSQL:

по id (коллокация)            лимиты по умолчанию          -> ok:9       44 мс  (PostgreSQL: 9)
по sku (без коллокации, 1:1)  лимиты по умолчанию          -> motion     56 мс  (PostgreSQL: 9)
по sku (без коллокации, 1:1)  SQL_MOTION_ROW_MAX = 200000  -> ok:9      259 мс  (PostgreSQL: 9)
по category (без коллокации)  лимиты по умолчанию          -> motion     66 мс  (PostgreSQL: 301147)
по category (без коллокации)  SQL_MOTION_ROW_MAX = 200000  -> vdbe      247 мс  (PostgreSQL: 301147)

sku уникален, как и id: каждая строка совпадает ровно с одной, пар столько же. Разница только в коллокации — и её цена видна в чистом виде: соединение по sku упирается в лимит переноса, а с поднятым для одного запроса лимитом (OPTION (SQL_MOTION_ROW_MAX = …)) выполняется и даёт тот же ответ, что PostgreSQL, — 259 мс против 44 мс у коллоцированного соединения по id в том же прогоне (время от клиента, с docker exec и psql; годится для сравнения вариантов между собой, а не как абсолютная латентность; в другом прогоне было 368 против 43). Значит, отказ по лимиту переноса — это цена отсутствия коллокации, а не невозможность соединения.

Соединение по category с тем же поднятым лимитом упирается уже в другой предел — sql_vdbe_opcode_max, Reached a limit on max executed vdbe opcodes. Поднятие одного лимита снимает ограничение на число переносимых строк, но успешный запрос не гарантирует: следующим пределом может стать другой лимит или ресурсы узлов. Поднимать опкоды дальше здесь бессмысленно: отдельной пробой sbroad/vdbe-probe.sh (в plans.sh она не входит из-за длительности, вывод — в sbroad/vdbe-probe.txt) запрос не уложился и в 2 000 000 000 опкодов, проработав 16,8 секунды (в более раннем прогоне — 33). Почему соединение по category настолько дороже соединения по sku при одинаковом переносе, стенд не установил. Число пар — 301 147 против 9 — первое подозрение, но 2 млрд опкодов на 300 тыс. пар — слишком много, чтобы объяснить всё одним им; вывода о том, что любое соединение без коллокации растёт квадратично, эти данные не дают — пример sku показывает обратное. Для исходного варианта с a.id < 2000 PostgreSQL даёт 66,6 млн пар; в Picodata он ведёт себя так же, как узкий: отказ по переносу, а с поднятым лимитом — по опкодам. Всё это с проверкой причины каждого отказа — шаги 6–8 сценария sbroad/plans.sh в стенде, вывод — в sbroad/plans.txt.

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

Два лимита, о которых стоит знать заранее

В каждом плане печатаются два параметра, и оба — реальные ограничители, а не декорация.

sql_motion_row_max (по умолчанию 5000) — сколько строк помещается в промежуточную таблицу переноса. В него упираются соединения без коллокации, как выше, и он же неожиданно всплывает в других местах. Например, оконная функция row_number() спокойно работает на подзапросе из сотни строк и падает на полной таблице — с тем же самым сообщением про 5000. Легко принять это за отсутствие поддержки оконных функций; на самом деле это лимит материализации.

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

Оба параметра меняются на уровне кластера:

ALTER SYSTEM SET sql_vdbe_opcode_max = 5000000 FOR ALL TIERS;

Важно, что это именно предохранители, а не настройки производительности. Они гасят запросы, которые иначе съели бы ресурсы кластера. Поднимать их до потолка «чтобы не мешали» — значит менять внятный отказ на непредсказуемую нагрузку.

Чем платим

• Схема становится решением о запросах. Ключ шардирования определяет, какие соединения останутся локальными, а какие потребуют переноса. Это проектное решение, а не деталь DDL.
• Отказ вместо замедления. Запрос, не уложившийся в лимит переноса, не выполнится. Это честнее незаметной деградации, но требует, чтобы ограничения были известны заранее.
• Планы надо читать. EXPLAIN показывает и адресацию к бакетам, и места переноса — без него природа отказа неочевидна, а сообщения об ошибках говорят про строки в промежуточной таблице, а не про ключ шардирования, который к этому привёл.
• Лимиты общие для внешнего и внутреннего кода. Плагин внутри кластера подчиняется тем же ограничениям — обойти их, «спустившись ниже», не получится.

Что дальше

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

Шардирование и его последствия шире одной СУБД — общая картина в статье про шардирование и консистентное хешированиеСкоро.