YDB — отказоустойчивая геораспределённая СУБД класса Distributed SQL. Она появилась в недрах Яндекса более десяти лет назад и прошла длительный путь от узкоспециализированного хранилища, применявшегося в поисковом движке, до полновесной СУБД общего назначения. Открытие исходного кода YDB в 2022 году стало одной из наиболее ярких опенсорс-инициатив Яндекса.

В центре внимания разработчиков YDB долгое время была обработка транзакционной нагрузки, однако в последние годы активно развивается и аналитическое направление. Одна из востребованных аналитических функций — возможность исполнения федеративных SQL-запросов, адресованных к внешним источникам данных.

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

О федеративных системах

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

Страницы истории

Несмотря на то, что далеко не каждый рядовой пользователь СУБД сегодня сталкивается с необходимостью писать федеративные запросы, понятие «федеративность» в контексте баз данных имеет богатую историю. Как часто бывает в таких ситуациях, термин применялся в разных контекстах и с годами приобретал новые смыслы.

Ранние годы

Наиболее ранняя публикация с упоминанием федеративных баз данных относится к 1979 году. К этому моменту базы данных уже начали своё шествие по индустрии программного обеспечения. Но в их проектировании превалировал подход с жёсткой интеграцией: данные всех пользователей заносились в одну большую базу с общей схемой. Этой базой управлял отдельный администратор баз данных (DBA), отвечавший за то, чтобы СУБД могла одновременно удовлетворять противоположным интересам разных пользователей. Но обычно это приводило к тому, что СУБД работала одинаково плохо для всех. А поскольку DBA зачастую был сотрудником сторонней организации, возникали вопросы сохранения целостности и изоляции данных. А тесная связанность внутри жёстко интегрированной базы данных исключала оперативную реакцию на изменяющиеся потребности пользователей.

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

В течение 1980-х годов на рынок выходили всё новые и новые СУБД: Informix (1981), IBM Db2 (1983), Teradata (1984), Sybase (1986), Microsoft SQL Server (1989), Postgres (1989). Пользователи зачастую уже не ограничивались единственной базой, из-за чего разработчикам всё чаще приходилось конструировать с помощью подручных средств инструменты для консолидации данных из разнородных источников, а также для миграции данных между ними.

Исследователи Amir Sheth (Bellcore) и James Larson (Intel) обобщили опыт построения федеративных СУБД этого десятилетия в своей обзорной статье. Они определяют федеративную базу данных как совокупность взаимодействующих, но автономных баз данных, тогда как федеративная СУБД (federated database system, FDBS) выступает в качестве логического слоя, благодаря которому пользователь может работать со множеством автономных баз, как если бы они были единым целым.

Федеративными запросами стали именоваться SQL-запросы, адресованные федеративной СУБД к внешним источникам данных. Федеративные запросы делают интеграцию разнотипных источников прозрачной для конечного пользователя: например, в одном запросе можно извлечь, объединить и проанализировать данные, размещённые в нескольких разнородных внешних системах.

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

Федеративные запросы для транзакций и аналитики

С начала 1990-х годов крупные производители реляционных СУБД внедряют в свои решения федеративные возможности. В 1992 году в Oracle 7 появилась функция выполнения распределённых запросов между несколькими базами через Database Links. В 1999 году благодаря технологии Heterogeneous Services в Oracle 8i стала возможна бесшовная работа и с базами других типов — SQL Server, IBM Db2, Sybase. Чуть позже аналогичная функциональность появилась также и у основных конкурентов Oracle — Sybase и Informix. В Microsoft SQL Server федеративные запросы пришли в конце 1998 года, а в IBM Db2 — только в 2002 году.

С отставанием примерно в 10 лет федеративные возможности стали просачиваться и в мир СУБД с открытым исходным кодом: в 2005 году в MySQL 5.7 был анонсирован Federated Storage Engine, а в 2011 году был представлен механизм Foreign Data Wrappers в PostgreSQL 9.1. К 2020 году функциональность федеративных запросов появляется и в аналитических СУБД, таких как Greenplum (Platform Extension Framework) и ClickHouse.

2010-е годы также ознаменовались появлением распределённых движков обработки запросов с федеративными возможностями. Большую роль в развитии этой продуктовой ниши сыграла экосистема Hadoop и облачные DWH.

Важной вехой стало появление в опенсорсе движка Presto (2013). Его кодовая база легла в основу двух широко известных ныне продуктов: движка обработки запросов Trino, обычно использующегося в режиме on-premise, и сервиса бессерверной аналитики Amazon Athena. Все они внесли значительный вклад в популяризацию концепции Data Lake, согласно которой функции хранения и обработки данных разделяются между двумя независимыми слоями софта: простым, но эффективным хранилищем (обычно S3) и движком обработки SQL-запросов.

Завершая обзор федеративных систем, нельзя не упомянуть близкие к ним продукты, нацеленные на виртуализацию данных, — например, Denodo и TIBCO Data Virtualization. Если в федеративных СУБД и движках источники данных прямо упоминаются в тексте SQL-запросов, то системы виртуализации конструируют поверх источников виртуальные представления (Virtual Views). Так они избавляют пользователя от необходимости указывать в запросах сетевые адреса и пароли от внешних систем и даже писать JOIN данных из источников.

Область применения

Итак, сегодня функциональность федеративных запросов можно обнаружить сразу в нескольких классах ПО: классических транзакционных СУБД, аналитических СУБД, а также в распределённых движках обработки запросов. Последние отличаются от аналитических баз в основном отсутствием встроенного слоя хранения данных.

По субъективным наблюдениям, в классических «реляционках» с присущими им проблемами масштабирования федеративные запросы не могут в полной мере раскрыть свой потенциал. Однако они могут быть полезны и там: например, в сценарии импорта данных, который возникает при замене корпоративной СУБД, когда необходимо перенести данные из старой системы в новую.

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

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

Рассмотрим почти классический пример гетерогенного хранилища, работающего в бэкенде типичного интернет-магазина. Основные данные о товарах и пользователях там обычно хранятся в реляционной PostgreSQL, пользовательская корзина — в документоориентированной MongoDB, а статистика пользовательской активности — в оптимизированной под запись колоночной ClickHouse.

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

SELECT 
    users.user_name,
    products.product_name,
    events.view_count,
    cart.total_value
FROM 
    -- Извлекаем данные о пользователе из таблицы в PostgreSQL
    postgres.users AS users
    
    -- В MongoDB узнаём, какой именно товар лежит у пользователя в корзине
    JOIN (
        SELECT 
            user_id,
            total_value,
            product_id  -- Single product in cart
        FROM mongodb.carts
        WHERE user_id = 12345
    ) AS cart ON cart.user_id = users.user_id
    
    -- Из ещё одной таблицы в PostgreSQL получаем общую информацию о товаре
    JOIN postgres.products AS products 
        ON products.product_id = cart.product_id
    
  -- У ClickHouse запрашиваем информацию о том, сколько раз пользователь просматривал товар перед покупкой
    JOIN (
        SELECT 
            user_id,
            COUNT(*) AS view_count
        FROM clickhouse.events
        WHERE user_id = 12345 
          AND event_type = 'product_view'
          AND product_id = cart.product_id
        GROUP BY user_id
    ) AS events ON events.user_id = users.user_id
WHERE 
    users.user_id = 12345;

Важно отметить, что круг потенциальных источников данных для федеративной системы не ограничивается традиционными СУБД. Полноценным источником может стать практически любой API, раздающий данные по сети. Логика проста: если данные можно привести к табличной форме, значит, их можно обработать и с помощью SQL-запросов. А это открывает безграничные возможности для аналитики.

Сложности разработки

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

Федеративная система — всегда система распределённая. В сегодняшних реалиях едва ли кто-то будет ограничиваться единственным сервером для хостинга всех хранилищ данных, а значит, серверов будет много, и взаимодействовать они будут через потенциально ненадёжную сеть. Если в федеративной системе не будут предусмотрены механизмы отказоустойчивости, по-настоящему масштабные аналитические запросы запускать в ней просто не получится. Кроме того, развёрнутая на большом парке железа федеративная система должна эффективно его использовать с помощью параллельных алгоритмов. Это позволяет рассматривать её ещё и как систему массивно-параллельной обработки данных (MPP).

Источники данных, входящие в состав федерации, — автономные. Параллельно с запросами, поступающими от федеративных СУБД или движков, они продолжают обрабатывать запросы сторонних пользователей. Монопольного доступа к данным внешних источников у федеративной системы, как правило, нет. В этих условиях неизбежно возникают вопросы о консистентности данных, извлечённых снаружи, особенно если чтение выполнялось в несколько потоков, что является естественным для MPP-системы.

Источники данных почти всегда относятся к системам разных типов. В мире СУБД невозможно найти две полностью эквивалентные и взаимозаменяемые базы данных — все они похожи в общем, но каждую отличает определённое своеобразие. Задача федеративной системы — сгладить различия между разнородными (гетерогенными) источниками, чтобы сделать возможной их интеграцию друг с другом. Это трудно, однако именно широта спектра поддерживаемых источников является характеристикой, отличающей наиболее продвинутые федеративные системы. Например, лидеры рынка — движок Trino и аналитическая СУБД ClickHouse — могут работать с источниками более чем 30 различных типов.

Формы проявления гетерогенности источников

О гетерогенности источников данных стоит поговорить особо, поскольку она проявляется сразу в нескольких формах. Прежде всего — в разнообразии моделей данных. Понятно, что реляционная модель была и остаётся наиболее универсальным способом организации данных: к ней относятся все транзакционные и аналитические РСУБД, а также S3, в котором могут храниться файлы табличных форматов.

Чуть меньшая популярность у систем, относящихся к документной модели. К ней относятся многие NoSQL СУБД и всё тот же S3, в случае если он хранит JSON или другие форматы данных с иерархической структурой.

Также в качестве источников данных для федеративных систем могут использоваться базы с моделью данных типа «ключ — значение», системы мониторинга, базы временных рядов, очереди сообщений, а в качестве экзотики — даже графовые СУБД.

Гетерогенность моделей данных влечёт за собой гетерогенность интерфейсов доступа к ним. С одной стороны, язык SQL — своего рода lingua franca для мира баз данных. С другой стороны, в каждой базе формируется собственный диалект SQL с присущим ему набором ключевых слов и синтаксических конструкций. И это несмотря на долгую историю его стандартизации — с 1986 по 2023 годы выпущено 11 ревизий стандарта SQL.

В то же время существует целый класс NoSQL-решений, в которых, как следует из названия, применяются либо принципиально другие языки запросов (например, MQL в MongoDB или Cypher в Neo4j), либо узкоспециализированные API, сосредоточенные вокруг набора CRUD-операций над сущностями базы данных (Redis, DynamoDB). А для работы с S3, даже если в нём хранятся структурированные данные, приходится использовать низкоуровневый HTTP API.

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

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

В этом отношении меньше всего проблем доставляют строго схематизированные источники (strictly schematized) — к ним относятся все реляционные СУБД. Схема таблицы у этих источников задаётся в момент её создания, и каждый раз вновь записываемые данные валидируются на предмет соответствия схеме. Отсюда и второе название этой группы — Schema-on-Write. В момент исполнения запроса федеративная система легко может получить схему внешней таблицы, обратившись к метаданным источника.

Но существуют и такие источники, схему данных которых можно узнать, только прочитав сами данные (Schema-on-Read), и даже принципиально бессхемные (Schemaless). Работая с ними, федеративная система вынуждена либо использовать специальные эвристики для выведения схемы данных на лету (но без гарантий точности), либо обращаться к внешним хранилищам метаданных — например, к Hive Metastore. Кроме того, пользователь может облегчить задачу федеративной системе, указав схему внешнего источника прямо в теле запроса, — правда, почти всегда в таких случаях запрос становится очень громоздким.

Наконец, гетерогенность выражается в разнообразии систем типов источников данных. Нестыковки в типах данных встречаются порой в самых неожиданных местах. В качестве примера можно привести типы беззнаковых целых чисел, привычно встречающихся во всех языках программирования, восходящих к Си. Но оказывается, что беззнаковые числа есть только в относительно современных РСУБД (MySQL, ClickHouse, YDB). В базах, вышедших в свет на рубеже 1980–1990-х годов (PostgreSQL, SQL Server), до сих предлагаются только знаковые типы. А у «мастодонта» мира баз данных — Oracle (1976) — система типов выглядит совсем уж архаично: на все случае жизни там предлагается использовать единственный числовой тип NUMBER переменной длины (от 1 до 21 байта).

Следовательно, если в федеративном запросе потребуется перекладывать данные из источника с беззнаковыми числами в приёмник, в котором таких чисел нет, сразу же возникнет проблема конвертации типов. И если uint32 ещё можно представить в виде int64, то как представить uint64?

Также нередко встречается ситуация, когда типы со схожими названиями в разных источниках данных имеют совершенно разную семантику. Например, в любой базе есть типы, которые используются для хранения даты и времени, но диапазоны значений, которые могли бы быть сохранены в этих типах, отличаются на порядки: в одних системах это сотни лет, в других — сотни тысяч лет. Например, диапазон возможных значений типа Timestamp в PostgreSQL в принципе не сводится к Timestamp в любой из других популярных СУБД. А если добавить сюда ещё и разрешающую способность типов, то картина станет совсем запутанной и сложной: вспомнить хотя бы тип Datetime в MS SQL Server с нестандартным разрешением в 3,33 мс.

Кроме того, можно выделить две категории типов данных для работы со временем: «наивное» время, не имеющее привязки к часовому поясу, и зональное время, «идущее» в определённой точке земного шара. Поддержка «наивного» времени — отличительная черта классических РСУБД (PostgreSQL, MySQL, MS SQL Server, Oracle), но в современных базах (ClickHouse, YDB) «наивное» время отсутствует. Реализация зонального времени в источниках сильно различается: иногда значение зоны физически хранится непосредственно рядом со значением времени, иногда хранится только «наивное» время, а зона прописывается в метаданных таблицы, либо зависит от настроек сервера базы данных или параметров пользовательской сессии. Это разнообразие добавляет потенциальных проблем при консолидации данных из разных источников.

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

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

Обработка федеративных запросов

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

Этапы обработки

Прежде всего, федеративный запрос — это обычный SQL-запрос. Когда он поступает в федеративную систему, он парсится и трансформируется во внутреннее представление — абстрактное синтаксическое дерево (Abstract Syntax Tree, AST). На основе AST компилятор запросов формирует логический план выполнения и начинает его оптимизировать. Он последовательно приводит его к некоторой форме, которая, как ожидается, обеспечит минимальное время выполнения запроса. На этапе компиляции запроса федеративная система выполняет первое обращение к внешнему источнику данных: ей нужны метаданные источников, а именно схемы внешних таблиц (и в некоторых случаях статистика по ним).

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

После извлечения данные обрабатываются; вычисляется результат запроса. При необходимости он может сохраняться во внутреннем хранилище (если оно имеется у федеративной системы), а затем направляется в сторону клиента.

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

Функции коннекторов

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

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

Если пользователь хочет объединить таблицу из PostgreSQL c табличными файлами, хранящимися в S3, то коннектор к PostgreSQL должен извлечь список колонок требуемой таблицы из другой служебной таблицы, тогда как коннектор к S3 должен выполнить листинг объектов в бакете. При этом если в S3 находятся файлы бессхемного формата CSV, и пользователь не указал их схему в федеративном запросе, то коннектору к S3 придётся вычитать некоторое количество файлов и вывести их схему прямо на месте.

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

Ещё одна функция коннектора — потоковое чтение данных из источника и их приведение к единообразному виду. Из PostgreSQL данные вычитываются построчно по бинарному протоколу Frontend/Backend, тогда как из S3 данные извлекаются по HTTP сразу целыми блоками в виде файлов разнообразных форматов, которые к тому же могут быть сжаты. Коннекторы должны конвертировать эти потоки разнородных данных в блоки унифицированного формата, доступные для обработки движком.

Внутреннее устройство коннекторов

Проектируя федеративную систему, разработчики оказываются перед выбором механизма реализации коннекторов. Можно сформулировать два важных трейдоффа, помогающих принять правильное решение в отношении архитектуры и интерфейса коннекторов. Трейдофф № 1: встроенные или внешние коннекторы. Прежде всего, нужно решить, должен ли коннектор встраиваться непосредственно в федеративную систему и работать с ней в едином адресном пространстве, либо же он должен быть реализован в виде отдельного приложения, например микросервиса, с которым федеративная система будет взаимодействовать по сети. У каждого из этих вариантов есть свои плюсы и минусы. Встроенный коннектор, реализуемый в виде библиотеки, позволяет достичь максимальной производительности. В отличие от него, внешний коннектор добавляет накладные расходы на межпроцессное взаимодействие, а также дополнительный хоп по сети в процесс извлечения данных из внешнего источника. Встроенный коннектор также более удобен с точки зрения организации деплоя, поскольку он разворачивается автоматически вместе с самой системой.

Однако внешний коннектор легче масштабировать. В отличие от тяжеловесной федеративной системы, он является относительно лёгким stateless-сервисом, который без проблем можно развернуть в том же K8s. Кроме того, внешний коннектор можно реализовать на любом языке программирования и благодаря этому повысить скорость разработки федеративной системы, особенно если использовать средства кодогенерации наподобие gRPC. Есть ещё несколько нюансов, актуальных для федеративных систем, написанных на С или C++. Дело в том, что встроенный коннектор, будучи влинкованным в исполняемый файл федеративной системы, привносит большое количество внешних зависимостей — зачастую неизвестного качества. В этой ситуации любой segfault или неотловленное исключение в области кода коннектора приведут к падению системного процесса, что недопустимо, особенно в условиях мультитенантной среды. По этой причине изоляция всех неконтролируемых библиотек в отдельном внешнем коннекторе повышает надёжность системы в целом, а также повышает скорость её сборки и снижает итоговый размер исполняемого файла.

Встроенные коннекторы есть практически у всех популярных федеративных систем, а вот внешние коннекторы в обязательном порядке появляются у систем, написанных на С и C++. К их группе также примыкает написанный на Java облачный сервис AWS Athena (фактически мультитенантная SaaS-версия Trino): в нём внешние коннекторы реализованы в качестве serverless-приложений (AWS Lambda).

Трейдофф № 2: гибкий или ограниченный интерфейс коннектора. Для максимальной эффективности федеративных запросов очень важно, чтобы коннекторы умели выполнять пушдаун (push down), то есть передавать некоторые логические конструкции запроса для исполнения на внешний источник данных. Например, федеративный запрос SELECT * FROM external_source.table WHERE id = 123 будет выполняться гораздо быстрее, если фильтрация данных произойдёт во внешнем источнике: SELECT * FROM table WHERE id = 123. Ну а если колонка id окажется ещё и первичным ключом внешней таблицы, то такой запрос будет выполнен почти мгновенно.

Если же коннектор по каким-то причинам не умеет выполнять пушдаун фильтров, то он запросит у внешнего источника полное сканирование таблицы: SELECT * FROM table. Фильтрация полностью выгруженной из источника таблицы выполнится уже на стороне федеративной системы, при этом будет затрачено гораздо больше вычислительных и сетевых ресурсов, а время обработки такого запроса будет линейно зависеть от размера таблицы.

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

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

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

Именно в этом выборе между максимальной гибкостью интерфейса коннектора и простотой разработки и заключается второй трейдофф при проектировании коннекторов. Гибкий интерфейс потенциально позволяет сделать пушдаун любого выражения, но делает разработку коннекторов весьма дорогой и долгой. Однако её можно ускорить, если сфокусировать API коннектора только на самых важных вариантах пушдауна и отказаться от сложных и редко используемых конструкций. Как видно из таблицы, этот подход вполне жизнеспособен и часто встречается в федеративных системах.

Федеративные возможности YDB

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

Архитектура движка обработки федеративных запросов

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

Ранее упоминалось, что в архитектуре федеративных систем можно выделить два относительно независимых слоя: слой движка обработки запросов и слой коннекторов. Для федеративных запросов в YDB отдельный движок не разрабатывался. Вместо этого мы переиспользовали уже давно развивавшийся движок обработки запросов YQL, который применяется при обработке запросов к таблицам внутри YDB. В него в качестве расширений были подключены коннекторы к внешним источникам данных. Более подробно с устройством движка можно ознакомиться, посмотрев доклад Алексея Озерицкого.

Изначально коннекторы в YDB работали с логическим планом выполнения запросов напрямую и были реализованы в виде библиотек на С++ — в терминологии YDB такие библиотеки называются провайдерами. В виде провайдеров реализованы коннекторы к S3, шине данных Data Streams, облачному мониторингу и некоторым другим источникам. Эти коннекторы хорошо оптимизированы, используются в проде (например, в облачном сервисе Yandex Query) и продолжают своё развитие в рамках избранной парадигмы.

Однако для организации взаимодействия YDB с реляционными СУБД выбрали всё же другую схему — внешние коннекторы с ограниченным, строго типизированным gRPC-интерфейсом. Это решение было мотивировано несколькими факторами.

Прежде всего, нам нужно было увеличить темпы добавления новых источников. Для этого мы перевели разработку коннекторов на более выразительный язык программирования и в целом минимизировали вмешательство в С++ кода YDB при добавлении очередного источника. В YDB был реализован обобщённый Generic-провайдер, взаимодействующий с внешними коннекторами по сети. Связка из Generic-провайдера и коннектора, с одной стороны, полностью абстрагирует YDB от специфики внешнего источника данных (например, от особенностей синтаксиса языка запросов и сетевого протокола), с другой — почти избавляет разработчиков коннекторов от необходимости погружаться в логику движка обработки запросов.

В то же время нам хотелось избежать раздувания кодовой базы YDB в процессе добавления новых источников, ведь уже сейчас статически слинкованный дистрибутив YDB в релизной сборке весит более 500 МБ. Для сравнения: исполняемый файл PostgreSQL вместе со всеми своими *.so-зависимостями не достигает и 100 МБ.

На данный момент основная реализация внешнего коннектора YDB — это проект fq-connector-go. Он поддерживает наиболее популярные реляционные источники (PostgreSQL, Greenplum, ClickHouse, MySQL, Oracle, Microsoft SQL Server, YDB), а усилиями студентов в него была добавлена поддержка ещё и нескольких NoSQL баз данных: MongoDB, Redis и OpenSearch.

Протокол работы YDB с внешним коннектором

В ходе обработки федеративного запроса Generic-провайдер YDB несколько раз обращается к внешнему коннектору по gRPC.

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

Здесь же, на этапе компиляции, однократно вызывается метод ListSplits. Сплит (split) — это некоторый фрагмент (партиция, шард) большой внешней таблицы и в то же время это элементарная единица работы для системы массивно-параллельной обработки данных, то есть тот объём работы, который будет выполняться последовательно одним из её воркеров. Термин «сплит» часто употребляется в различных аналитических системах; о его значении в контексте распределённого движка обработки запросов Trino, например, подробно рассказывает Владимир Озеров.

При выполнении запросов, подразумевающих full scan внешних таблиц (а это возникает, например, при JOIN таблиц из разных внешних источников), крайне важно уметь выкачивать крупные внешние таблицы по частям и в несколько потоков, тем самым максимизируя пропускную способность системы. При вызове ListSplits коннектор, ориентируясь на метаданные внешней таблицы, либо возвращает список её сплитов, либо сигнализирует о том, что YDB должен прочесть таблицу целиком в один поток.

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

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

Потоковое чтение данных

Важной особенностью протокола работы YDB с внешним коннектором является обмен данных в формате Apache Arrow IPC Streaming. Этот формат сериализации выбирают многие современные системы потоковой обработки — Apache Spark, Dremio, Apache DataFusion Ballista. Он позволяет эффективно работать с данными, которые не помещаются в оперативную память: батчи Arrow-записей обрабатываются инкрементально, без необходимости аллокации памяти сразу под весь объём данных. Для многих операций над данными к тому же реализована семантика zero-copy. Кроме того, благодаря размещению данных в виде колонок формат Apache Arrow особенно удачно подходит для векторизированных вычислений, которые сейчас активно разрабатываются в движке обработки запросов YDB.

Однако коннектору зачастую приходится работать с источниками, хранение данных в которых организовано в виде строк. Это справедливо практически для всех популярных реляционных СУБД, за исключением ClickHouse. В этой ситуации он должен на лету выполнить транспонирование вычитанных строк в колонки Apache Arrow. Это вычислительно сложная операция, асимптотически линейно зависящая от ширины внешней таблицы.

При этом как чтение строк из внешнего источника, так и отправка блоков колонок в сторону YDB должны происходить потоково, без извлечения всего объёма данных из внешнего источника в память коннектора. В связи с этим методы ListSplits и ReadSplits коннектора реализованы в виде gRPC Server-Side Streaming. Вычитанные коннектором из внешнего реляционного источника данные трансформируются, буферизуются и по мере накопления отправляются в сторону YDB. Именно этот процесс и составляет основную часть бизнес-логики коннектора fq-connector-go.

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

Заключение

Федеративные базы данных и движки обработки запросов — особый класс ПО, выполняющий консолидацию данных из различных источников и миграцию данных между ними. Потребность в федеративных системах была осознана уже на ранних этапах развития рынка СУБД, и можно не сомневаться, что федеративные системы будут прочно занимать свою нишу на протяжении всего периода, пока в IT-индустрии будет сохраняться практика построения гетерогенных хранилищ данных.

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

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

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

В настоящее время YDB активно развивается не только как транзакционная, но и как аналитическая СУБД с федеративными возможностями. Архитектура взаимодействия YDB и внешнего коннектора fq-connector-go позволяет организовать потоковое извлечение и массивно-параллельную обработку данных из различных внешних источников — реляционных СУБД и некоторых NoSQL баз данных.