Ускоряем федеративные запросы в StarRocks

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

В этой статье постараемся немного расширить кругозор читателей, которым интересна данная тема: рассмотрим федеративные запросы на примере набирающего популярность и активно развивающегося StarRocks. Из статьи вы узнаете: что такое федеративные запросы, как обстоят дела с реализацией гетерогенного федеративного доступа в этой СУБД и какие изменения команда решения Data Ocean Nova реализовала для оптимизации в StarRocks и Impala с целью улучшения функционала доступа к внешним данным.

Что такое федеративные запросы

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

SELECT ...FROM db.log -- журнал действий пользователей...JOIN db.purchase ... -- витрина с покупками пользователейJOIN db.user_profile ... -- профайл пользователей с контактами

Если все эти объекты (log, purchase, user_profile) лежат в одной СУБД, то все относительно просто и привычно: СУБД сама управляет данными, сохраненными в понятном ей формате.

Но что если эти объекты находятся в разных системах? Предположим, что:

• log из-за своих размеров живет в Lakehouse;

• Purchase-витрина все еще собирается в старом добром Data Warehouse на базе Greenplum;

• user_profile находится в CRM на базе PostgreSQL.

SELECT ...FROM lh.log -- журнал действий пользователей в Lakehouse...JOIN dwh.purchase ... -- витрина с покупками пользователей в DWH (Greenplum)JOIN crm.user_profile ... -- профайл пользователей с ключевыми контактами в CRM (PostgreSQL)

Или распишем листинг чуть подробнее, приближая его к реальному кейсу:

SELECT    u.full_name as customer_name,    u.email as customer_email    u.phone as customer_phone,    p.order_date as bed_purchase_date,    p.order_amount as bed_purchase_amount,    p.product_subcategory as bed_category,    intent.last_view_date as mattress_last_view_dateFROM (    SELECT        user_id,        MAX(event_date) as last_view_date    FROM lh.log    WHERE 1=1        AND log.event_date >= CURRENT_DATE() - INTERVAL 14 DAY        AND log.event_type = 'PRODUCT_VIEW'        AND log.product_category = 'MATTRESS'    GROUP BY user_id    HAVING COUNT(1) >= 5  -- считаем что 5 раз - уже не случайность) intentLEFT JOIN dwh.dm.purchase p ON 1=1    AND intent.user_id = p.user_id    AND p.order_date >= CURRENT_DATE() - INTERVAL 60 DAY    AND p.product_category = 'BED'LEFT JOIN crm.public.user_profile u ON 1=1    AND intent.user_id = u.user_id

Это уже федеративный запрос (federated query), который одновременно обращается к нескольким источникам данных и должен объединить все данные «на лету» для формирования результирующего набора. Система, обрабатывающая федеративный запрос, не владеет всеми данными. Для их получения она должна взаимодействовать с другими системами и запрашивать у них недостающую информацию. Решать эту задачу можно разными техниками оптимизации — рассмотрим, на какие моменты стоит обратить внимание, кроме количества поддерживаемых источников.

Исполнение федеративных запросов – что важно

Федеративный запрос — это такой же SQL-запрос. Его необходимо распарсить, представить в виде abstract syntax tree (AST), подвязать (bind) метаданные, провалидировать (включая выражения, типы данных и т.п.), составить logical plan tree, провести оптимизацию, подготовить distributed physical execution plan tree, декомпозировать его на фрагменты — и уже их передать на выполнение.

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

Следующий шаг будет уже более сложным, важным и интересным. Системе необходимо решить, какие части logical plan tree выполнить на своей стороне, а какие – отдать на выполнение во внешнюю систему (это называют push down’ом).

Возможности перекладывания части работы на источник (push down)

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

• Точка «0» — не делаем никаких push down’ов, запрашиваем у источника всю таблицу целиком, а фильтрацию производим на стороне приемника.

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

• Точка «1» — делаем push down предикатов (where, limit и т.п.) и column projection (запрашиваем только нужные колонки).

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

  Большинство инструментов находится именно в окрестностях этой точки: кто-то справляется чуть лучше, кто-то — чуть хуже. Часто это зависит от конкретной реализации коннектора к системе-источнику. Например, похожим функционалом обладают следующие инструменты (есть поддержка по крайней мере для некоторых источников): Presto, Greenplum (PXF), ClickHouse, Impala, StarRocks;

• Отдаление от точки «1» — здесь добавляются дополнительные возможности push down’а.

  Чаще всего следующим шагом добавляют поддержку push down join’ов или агрегаций — именно ее для ряда источников можно найти в Trino. Это полезно, когда по итогам выполнения join’а или агрегации на порядок и более сокращается размер итоговой выборки.

  Но важно понимать: такие оптимизации не всегда полезны. Не все системы-источники в принципе готовы к аналитическим запросам с join’ами и агрегациями — многие изначально проектировались под OLTP. Далее— вычислительные ресурсы: рассчитан ли сайзинг систем-источников на такую нагрузку? какой объем ресурсов доступен в источнике, а какой — в Lakehouse? Распараллелить выгрузку с join’ами и агрегациями так, чтобы не увеличить нагрузку в N раз, еще сложнее, а в ряде случаев — невозможно.

  Поэтому в каждом конкретном случае нужно взвешивать все за и против и решать: использовать эти оптимизации или нет.

• +INF — недостижимый максимум, когда мы:

  • умеем делать push down не просто любых вычислений, но и данных, полученных в другом месте (локально, из третьей системы);

  • понимаем, когда это нужно делать, а когда нет (включая текущую нагрузку по всем внешним системам);

  • работаем лучше всех внутренних оптимизаторов всех СУБД на любых запросах/сценариях/workload’ах…

  К этой точке стремятся системы для виртуализации данных, все дальше отдаляясь от точки «1».

Подытожим: push down предикатов (where, limit и т.п.) и column projection — золотая середина, необходимая для эффективного выполнения федеративных запросов в большинстве кейсов.

Возможности параллельной выгрузки

Параллелизм вычислений — фундамент современных аналитических платформ. StarRocks, Trino, Impala, Spark и другие стараются распараллелить вычисления, насколько возможно, поэтому и сканирование исходных данных, например, в parquet-файлах, идет в несколько потоков.

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

Например, в Open Source версиях Trino и StarRocks есть поддержка JDBC-подключений к нескольким СУБД, можно работать с PostgreSQL или Oracle. Но в текущей версии они умеют сканировать только в один поток, что на больших объемах может быть мучительно долго.

В случае с MPP СУБД Greenplum для эффективной выгрузки важно не только уметь направить N параллельных запросов, но и обойти bottleneck с Master-узлом (можно поднять Primary + Standby, но обслуживать пользовательские запросы сможет только Primary). Для серьезной выгрузки используют gpfdist, позволяющий забирать данные напрямую с сегментов Greenplum.

С точки зрения физической модели данных важно понимать, какие есть возможности избежать выполнения full table scan N раз — то есть разбить чтение на N непересекающихся частей. Например:

• В MPP СУБД обычно есть способ прочитать данные с одного или нескольких узлов. Например, в Greenplum первым потоком можно прочитать сегменты с 0 по 9, вторым — с 10 по 19 и т.д. В Teradata аналогично можно читать первым потоком AMP с 0 по 9, вторым — с 10 по 19 (о чем я рассказывал в предыдущем материале, где освещал детально, как мы сделали самый эффективный коннектор Trino для Teradata);

• Если таблица «нарезана» на партиции, можно читать каждую партицию в отдельном потоке;

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

Но даже если таких вариантов не нашлось, несколько full table scan все равно могут дать ускорение выгрузки, пусть и за счет дополнительной нагрузки на исходную систему.

Федеративные запросы в StarRocks

Управление внутренними и внешними данными в StarRocks реализуется через объекты под названием каталоги (Catalogs). Для собственных данных в StarRocks существует default_catalog, который может быть только один. Для «подключения» внешних систем нужно создать каталог (CREATE EXTERNAL CATALOG <catalog_name> ...) одного из поддерживаемых типов, например:

• JDBC-каталог — поддерживается подключение к PostgreSQL, Oracle, SQL Server, MySQL, ClickHouse (экспериментально);

• Unified-каталог позволяет работать с Iceberg, Hudi, Delta, Hive (HMS), Kudu и Paimon. Для каждого из них есть каталог своего конкретного типа:

  • Iceberg-каталог позволяет работать с Iceberg-таблицами, поддерживаются HMS/REST/JDBC-каталоги, данные могут располагаться в S3 или HDFS;

  • … Hudi-каталог, Delta-каталог и т.д. — возможности и ограничения для каждого такого каталога будут верными и для Unified-каталога.

• Elasticsearch-каталог и еще один специфичный для Alibaba Cloud — на них останавливаться сейчас не буду.

При работе с данными из CRM (PostgreSQL) создание каталога выглядит примерно так:

CREATE EXTERNAL CATALOG crmPROPERTIES (    "type"         = "jdbc",     "user"         = "***",    "password"     = "***",    "jdbc_uri"     = "jdbc:postgresql://crm-hostname:5432/crm",    "driver_url"   = "file:///opt/usr/jdbc-drivers/postgresql-42.3.3.jar",    "driver_class" = "org.postgresql.Driver");

После этого можем запрашивать данные из внешней системы:

SELECT * FROM crm.public.user_profile;

Можем сразу работать как с данными Lakehouse, так и внешней системы:

SELECT log.*FROM lh.log lJOIN crm.public.user_profile u on l.user_id = u.user_id AND u.user_category = 'VIP'WHERE l.errors IS NOT NULL;

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

Остановимся на работе с JDBC чуть детальнее.

Ускоряем чтение по JDBC: параллельное чтение

Как уже говорили выше, в Open Source версии StarRocks выгрузка данных идет в один поток на одном BE/CN-узле:

На небольших объемах это не так страшно, но при значительных объемах это ограничение — очевидное «узкое горлышко» (bottleneck). Дополнительно оно приводит к перекосу (skew) по вычислениям и промежуточным данным на узлах.

Можно использовать workaround с ручным разбиением через UNION ALL:

SELECT ... FROM crm.public.user_profile WHERE user_id % 2 = 0UNION ALLSELECT ... FROM crm.public.user_profile WHERE user_id % 2 = 1

Но это неудобно: при каждом изменении количества потоков необходимо переписывать запрос. И не самый очевидный нюанс — нельзя забыть про обработку NULL-значений, если колонка допускает их (нет ограничения NOT NULL), иначе можно потерять строки.

В Data Ocean Nova StarRocks, начиная с версии 2026.1.0, появилась поддержка параллельного чтения внешних JDBC-таблиц без необходимости в описанных выше магических заклинаниях. Для работы механизма достаточно указать, по какому полю выполнить разбиение (split), всю остальную работу он возьмет на себя:

SELECT /*+ SET_VAR(jdbc_split_column='crm.public.user_profile.user_id') */    ...FROM crm.public.user_profile

В итоге вместо прежней схемы передачи в один поток получаем совсем иную картину:

Теперь все узлы участвуют в передаче данных, работает внутриузловой параллелизм, больше нет bottleneck’а и skew в вычислениях и данных. Совсем другое дело.

Если в запросе несколько таблиц, в jdbc_split_column используем перечисление:

SELECT /*+ SET_VAR(jdbc_split_column='crm.public.user_profile.user_id,dwh.dm.purchase.segment_id') */    ...FROM ...JOIN dwh.dm.purchase ...JOIN crm.public.user_profile ...

StarRocks для перечисленных в jdbc_split_column таблиц выполнит запрос к источнику для получения min() и max() по указанной split-колонке, а полученный интервал «нарежет» на N диапазонов, которые будет запрашивать у источника параллельно. Если колонка может содержать NULL-значения, в одной из N порций будет запрос с NULL.

На текущий момент есть ограничение – поддерживаются только целочисленные (TINYINT, SMALLINT, INTEGER, BIGINT) колонки для split’а.

Ускоряем чтение из Greenplum: добавляем Greenplum-каталог в StarRocks с поддержкой gpfdist

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

В Data Ocean Nova StarRocks, начиная с версии 2026.2.0, добавили Greenplum-каталог с поддержкой работы через gpfdist:

CREATE EXTERNAL CATALOG dwhPROPERTIES (    "type" = "greenplum",    "user" = "***",    "password" = "***",    "jdbc_uri" = "jdbc:postgresql://213.165.209.171:5432/dwh",    "driver_url"="file:///opt/usr/jdbc-drivers/postgresql-42.3.3.jar",    "driver_class" = "org.postgresql.Driver",    "gp_external_table_temp_schema" = "public",    "gp_external_table_format" = "csv",    "gp_external_table_delimiter" = ",");

При создании Greenplum-каталога указываем тип greenplum . Для работы со словарем Greenplum используется JDBC-подключение, но все данные идут через gpfdist. В дополнение к properties, используемым при JDBC-подключении, здесь появляется несколько дополнительных:

• gp_external_table_temp_schema – схема в Greenplum, в которой создаются external writable tables;

• gp_external_table_format – формат таблицы external writable table, создаваемой в Greenplum;

• gp_external_table_delimiter – разделитель столбцов в CSV-файле.

Как можно догадаться из этих параметров, во время выполнения запроса создаются временные external writable tables. Их будет столько, сколько существует потоков выгрузки. По завершении выгрузки они удаляются. Важный момент: для этого сценария рекомендуем размещать временные таблицы в схеме с отключенным журналированием (UNLOGGED), чтобы не создавать дополнительную нагрузку на WAL-журналы СУБД.

Количество потоков (или Scan Instances) определяется как наименьшее из:

• Количества сегментов в Greenplum;

• Количества BE/CN-узлов StarRocks (степень внутреннего параллелизма, Degree Of Parallelism, DoP).

Сегменты распределяются по Scan Instances простым round-robin.

Рассмотрим пример. Допустим, кластер Greenplum состоит из 48 Primary Segments, на стороне StarRocks – 3 BE/CN, а на запросе DoP выставлен равным 2 (pipeline_dop=2). Тогда мы получаем  min(48, 3 * 2) = 6 Scan Instances, а сегменты между ними распределяются так:

• Scan Instance #1 → Segments [1, 7, 13, 19, 25, 31, 37, 43]

• Scan Instance #2 → Segments [2, 8, 14, 20, 26, 32, 38, 44]

• Scan Instance #3 → Segments [3, 9, 15, 21, 27, 33, 39, 45]

• Scan Instance #4 → Segments [4, 10, 16, 22, 28, 34, 40, 46]

• Scan Instance #5 → Segments [5, 11, 17, 23, 29, 35, 41, 47]

• Scan Instance #6 → Segments [6, 12, 18, 24, 30, 36, 42, 48]

На схеме потоки данных будут выглядеть так:

Обратите внимание, что данные идут напрямую между worker-узлами Greenplum и StarRocks’а – на этапе передачи данных мы не загружаем Master лишней работой.

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

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

SET pipeline_dop = 4;SELECT /*+ SET_VAR(jdbc_split_column='crm.public.user_profile.user_id') */    u.full_name as customer_name,    u.email as customer_email    u.phone as customer_phone,    p.order_date as bed_purchase_date,    p.order_amount as bed_purchase_amount,    p.product_subcategory as bed_category    intent.last_view_date as mattress_last_view_dateFROM (    SELECT        user_id,        MAX(event_date) as last_view_date    FROM lh.log    WHERE 1=1        AND log.event_date >= CURRENT_DATE() - INTERVAL 14 DAY        AND log.event_type = 'PRODUCT_VIEW'        AND log.product_category = 'MATTRESS'    GROUP BY user_id    HAVING COUNT(1) >= 5  -- считаем что 5 раз - уже не случайность) intentLEFT JOIN dwh.dm.purchase p ON 1=1    AND intent.user_id = p.user_id    AND p.order_date >= CURRENT_DATE() - INTERVAL 60 DAY    AND p.product_category = 'BED'LEFT JOIN crm.public.user_profile u ON 1=1    AND intent.user_id = u.user_id;

А что там в Impala?

В этой статье мы сфокусировались на StarRocks, но в самом начале упоминали оптимизации в Impala в составе Data Ocean Nova. Кратко отвечу тем, у кого возникли вопросы: на текущий момент реализована поддержка многопоточного чтения и записи данных через JDBC, внутренняя механика практически 1-в-1 с описанным выше. Чтение и запись данных из внешних и во внешние JDBC-таблицы у Impala идет распределенно через узлы-исполнители, а не через один узел координатора, как в open source версии движка. Кстати, в open source версии реализовано только чтение во внешние СУБД, без возможности записи. Запись позволяет доставлять деривативные производные данные (например, витрины) из платформы Data Ocean Nova в системы-потребители, не прибегая к сторонним инструментам или другим движкам типа Spark. Поддержку работы с Greenplum через gpfdist добавим в одном из будущих релизов.

Заключение

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

Trino явно уходит вперед с точки зрения функциональных возможностей в федеративных вычислениях: в арсенале большое количество коннекторов для разных систем, есть поддержка push down и join’ов с агрегацией.

StarRocks пока не может этим похвастаться, но у него свои сильные стороны: он значительно опережает Trino по производительности, стабильнее справляется с большим количеством конкурентных запросов и заметно лучше работает с кешированием данных — это не только ускоряет запросы в самом StarRocks, но и снижает нагрузку на источник.

Раньше в федеративных вычислениях StarRocks был явный bottleneck в виде однопоточного чтения. Его устранение позволяет значительно ускорить запросы во множестве кейсов. Но пока остаются ситуации, когда Trino может оказаться быстрее на федеративных запросах, где дополнительные оптимизации push down’ов приходятся к месту и системы-источники к ним готовы. 

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

Эффективные механизмы пакетного чтения внешних СУБД движками StarRocks и Impala в платформе данных открывают Data Sapience возможность использовать их в инструменте пакетной репликации Data Ocean Flex Loader. В ближайших релизах при выгрузке данных из реляционных СУБД появится возможность вместо Spark выбрать StarRocks или Impala как средство извлечения данных.

Спасибо, что дочитали до конца! Подписывайтесь на блог Data Sapience на Хабре и наш Telegram-канал.