Распределенное выполнение запросов в CrateDB: что нужно знать

Эта статья была написана на основе материалов, опубликованных Матиасом Фюссенеггером в серии статей о внутреннем устройстве CrateDB. Дополнительные примеры того, как CrateDB выполняет запросы, смотрите в оригинальных статьях.

Язык запросов в CrateDB

CrateDB полагается на SQL для запросов и манипулирования данными. Использование SQL в качестве языка запросов сокращает время обучения, облегчает перенос и позволяет пользователям сосредоточиться на логике запросов, а не на низкоуровневых деталях распределенной системы. Более того, CrateDB также позволяет пользователям писать UDF (например, определяемые пользователем функции) для манипулирования данными в случае необходимости.

Операторы SQL внутренне преобразуются в ряд операторов, которые должны применяться по порядку, и каждый оператор может обрабатывать только свои входные данные. Например, WHERE является одним из самых внутренних операторов, а SELECT применяется почти в конце. Порядок выполнения операторов зависит от оптимизатора. Как и в других системах баз данных, оптимизатор в CrateDB применяет правила оптимизации к дереву операторов или к подмножеству дерева, чтобы переписать дерево в эквивалентный вариант, который дешевле в исполнении.

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

Компиляция и выполнение запросов

Логический план

Вообще говоря, логический план — это абстракция всех шагов преобразования, необходимых для выполнения запроса. Чтобы понять логический план в CrateDB, выполните SQL-запрос с помощью оператора https://crate.io/docs/crate/reference/en/4.8/sql/statements/explain.html. Например, рассмотрим простой запрос SELECT:

EXPLAIN SELECT name FROM users;
-> Collect[doc.users | [name] | true]

В этом случае логический план состоит из одного оператора Collect. Параметры включают doc.users — имя таблицы, name — собираемый атрибут, и true — выражение запроса. Добавление фильтра в предложение WHERE изменит выражение запроса только следующим образом:

EXPLAIN SELECT name FROM users WHERE name = 'name';
-> Collect[doc.users | [name] | (name = 'name')]

Такое поведение связано с тем, что реализация оператора Collect в CrateDB достаточно надежна, и в настоящее время на уровне логического планировщика нет различия между сканированием таблицы и поиском индекса.

При более сложных запросах генерируются более сложные логические планы. Например, рассмотрим подзапрос с оператором SELECT:

EXPLAIN SELECT name FROM (SELECT name FROM users WHERE name = 'name') AS n 
WHERE n.name = 'foo';

И соответствующий логический план:

Rename[name] AS n
  └ Collect[doc.users | [name] | ((name = 'foo') AND (name = 'name'))]

Логический план приводит к оператору Rename. Интересно то, что внутреннее предложение WHERE и внешнее WHERE объединились в (name = 'foo') AND (name = 'name'), и это выражение является частью оператора Collect. Это произошло из-за оптимизации filter pushdown, которая пытается переместить предикаты как можно дальше вниз по дереву, чтобы уменьшить количество строк, которые необходимо исследовать.

Если оператор Collect содержит выражение, CrateDB будет использовать индекс Lucene, если это возможно. В предыдущем случае CrateDB будет искать термины foo и name в инвертированном индексе. Инвертированный индекс сопоставляет термины с набором идентификаторов документов, как показано в нашей предыдущей статье об индексировании и хранении в CrateDB. Использование индекса Lucene значительно дешевле, чем загрузка строк и вычисление name = 'foo' с конкретным значением name для каждой строки.

Как показано на рисунке, EXPLAIN выводит список логических операторов. Перед выполнением запроса CrateDB преобразует логический план в физический план выполнения.

Физический план выполнения

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

Физический план выполнения определяет, где находятся данные и как они должны попасть на узел, с которым общается клиент. Мы называем этот узел узлом-обработчиком, поскольку он управляет общением с клиентом и инициирует выполнение запроса. Узлы, с которых собираются данные, называются узлами сбора, иногда в процесс вовлекаются узлы слияния, которые объединяют результаты, полученные с узлов сбора.

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

Рассмотрим физический план выполнения для простого запроса SELECT name FROM users;:

RoutedCollectPhase:
  toCollect: [Ref{name}]
  routing:
    node-0:
      users: [0, 1]
    node-1:
      users: [2, 3]


MergePhase:
  executionNodes: [node-0]
  numUpstreams: 2

План состоит из CollectPhase и MergePhase. В этом примере оператор Collect создает CollectPhase. CollectPhase является источником, что означает, что он должен выводить строки, считывая их с диска или из памяти. Оператор Collect создает этот CollectPhase, комбинируя информацию, которую он хранит сам, с информацией из состояния кластера. Состояние кластера — это снимок, который представляет текущее состояние кластера. Он включает в себя такую информацию, как какие таблицы существуют, какие столбцы они имеют, и на каком узле расположены различные осколки таблиц.

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

Свойство toCollect сообщает нам, какие атрибуты должны быть собраны, а свойство routing говорит нам, откуда. Маршрутизация включает в себя идентификаторы шардов таблиц или разделов, которые должны быть запрошены, и на каких узлах они находятся. В данном случае исполнитель должен получить данные с node-0 и node-1, поскольку оба узла содержат две копии таблицы.

Фаза MergePhase используется для указания того, что узел должен объединить данные со всех узлов, участвующих в фазе CollectPhase. Обычно эта фаза слияния назначается узлу-обработчику. В данном случае это node-0, и он ожидает результатов от двух других узлов. В этом сценарии узел node-0 является одновременно узлом-обработчиком и узлом-сборщиком. Он ожидает результатов как от себя, так и от одного другого узла.

Как только планировщик/оптимизатор завершает создание физического плана выполнения, он выполняет его.

Выполнение

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

В случае CollectPhase это включает создание запроса Lucene из выражения фильтра (например, name = 'name'), получение читателя Lucene для каждого шарда и итерации по совпадающим документам с применением всех преобразований, которые являются частью фазы, перед отправкой результата обратно в узел обработчика.

Наконец, результаты фаз выполнения могут быть направлены на узлы слияния (например, node-0 в предыдущем примере), прежде чем окончательный результат будет отправлен на узел обработчика, а затем обратно клиенту.

Пример: Выполнение запроса, затем выборка

Давайте расширим простой запрос SELECT с помощью LIMIT, а чтобы сделать пример более реалистичным, добавим предложение WHERE и выберем дополнительный столбец. Теперь запрос выглядит следующим образом:

SELECT name, age FROM users WHERE name LIKE 'A%' LIMIT 10`

Давайте изучим логический план с помощью оператора EXPLAIN:

Fetch[name, age]
  └ Limit[50::bigint;0]
    └ Collect[doc.users | [_fetchid] | (name LIKE 'A%')]

Начиная сверху вниз, план содержит следующие операторы:

• Оператор Fetch: Fetch принимает на вход отношение и ожидает, что оно будет содержать один или несколько столбцов _fetchid. Он использует эти значения _fetchid для получения значений других атрибутов, в данном примере он получает name и age.

• Оператор Limit. Limit берет отношение и ограничивает его не более чем 50 записями.

• Оператор Collect. Оператор Collect в этом примере указывает, что атрибут под названием _fetchid должен быть извлечен из таблицы doc.users. Оператор также включает выражение запроса name LIKE 'A%', указывая, что только записи, соответствующие этому выражению, должны быть включены. _fetchid — это системный столбец, который может использоваться оператором Fetch.

Поскольку данные в CrateDB распределены по нескольким шардам и узлам, CrateDB не может точно знать заранее, сколько записей хранится на каждом шарде и сколько из этих записей будут соответствовать выражению name LIKE 'A%'. Поэтому в данном примере CrateDB должен получить не более 50 записей из каждого узла и затем объединить их вместе, остановившись, как только будет достигнут предел.

Операция Fetch использует readerId, который закодирован в _fetchid, чтобы определить, к какому узлу ей нужно обратиться, чтобы получить фактическое значение. Вся операция Fetch работает пакетно, поэтому несколько записей извлекаются одновременно. Это немного похоже на асинхронную плоскую карту. Основное преимущество такого подхода заключается в том, что теперь каждый узел должен загружать только те значения name и age, которые необходимы. Например, если данные хранятся на трех узлах, первый узел может загрузить 16 записей, второй — 17, третий — оставшиеся 17, итого 50.

Какой вывод мы можем сделать?

Эта статья дает первое представление о том, как CrateDB выполняет распределенные запросы. Мы начали с обзора логического плана, физического плана выполнения и уровня выполнения на примере простого оператора SELECT. Затем мы проиллюстрировали некоторые оптимизации и стратегии выполнения на более сложном примере с использованием оператора LIMIT.

Узнайте больше о CrateDB здесь и ознакомьтесь с нашей официальной документацией. Получите информацию от нашего сообщества.

Чтобы получить больше информации о том, как CrateDB выполняет другие типы запросов, обратитесь к оригинальным статьям Матиаса Фюссенеггера.