Substrait — это промежуточный формат (IR) для обмена планами запросов между системами. Он снимает боль диалектов SQL, позволяет делать pushdown в разные бэкенды и избавляет от повторного парсинга/оптимизации федеративных системах и позволяет относительно безболезненно заменять один бэкенд другим. Ниже - зачем он нужен, как устроен и кто поддерживает.

Введение

Часто нужные данные находятся в разных хранилищах: например, продажи лежат в Parquet/S3, а справочники — в Postgres. Бизнес-вопрос требует один запрос поверх всего этого (например, джойн продаж с продуктами). Для этого можно использовать федеративный движок (CedrusData/Trino/Dremio/Presto), который парсит SQL написанный аналитиком, строит единый план исполнения и спускает (push down) части плана к источникам: фильтры и проекции — в ридер Parquet, подзапросы — в Postgres. Для работы с Postgres используется JDBC-коннектор федеративного движка. Это обычно означает что после оптимизации федеративному движку снова приходится генерировать фрагмент SQL в диалекте целевой БД, отправить его туда, забрать промежуточный результат, а остаток плана (джойн между источниками, финальные агрегации) посчитать у себя.

Этот подход имеет ряд определенных проблем:

• Диалекты SQL. Разные системы говорят на разных диалектах, с разным поведением функций и NULL. Продюсер запроса (где мы пишем запрос) вынужден знать каждого потребителя (где запрос или его часть исполняется).

• Повторная тяжёлая работа. Каждый движок заново парсит и оптимизирует SQL, сгенерированный в конекторах федеративного движка. Подсказки и тонкости плана теряются по дороге.

• Несовместимая семантика. Переполнение int32 + int32, деление на 0, сравнение NULL - всё это может вести себя по‑разному в разных движках

• NoSQL и множество DSL. MongoDB, Cassandra и разнообразные API не говорят на SQL. Аналитики же пишут ещё и на Python (Ibis), R (dplyr), … — хочется единый «общий знаменатель» для всех.

Идея Substrait — не отправлять строку SQL, а передавать сериализованный план виде промежуточного представления (intermediate representation — IR) на protobuf: с чётко описанными операторами, типами и функциями.

Это позволяет:

• делать pushdown фрагментов плана в подходящие бэкенды,

• не парсить/оптимизировать одно и то же по нескольку раз,

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

• снижать связность между фронтендами (SQL/Ibis/dplyr/…​) и бэкендами (Velox/DuckDB/DataFusion/…​).

Место Substrait в мире СУБД хорошо демонстрирует следующая иллюстрация:

Когда Substrait НЕ нужен:

• Вы живёте в одной СУБД/DI-стеке, без федерации источников.

• Нет сценариев или необходимости шеринга логики между движками.

• Внутренний IR одной системы полностью вас устраивает и нет планов заменить его на некий универсальный и легкозаменяемый бэкенд.

Арихитектура Substrait

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

Одна из систем, которая генерирует планы исполнения, называется продюсером (Producer), вторая, которая принимает план и исполняет его — консьюмером (Consumer).

Начать общаться эти системы могут с обмена своими возможностями (Capabilities capabilities.proto), чтобы они обе понимали что умеет а что нет их коллега. Проще говоря, они обмениваются текущей поддерживаемой версией Substrait и именами с описаниями тех функций и типов которые они поддерживают. Это описано с помощью так называемых SimpleExtensions, которые содержатся в Capabilities. 

В Substrait принято определять поддерживаемые функции и типы (SimpleExtensions) с помощью yaml. Типичный пример extensions можно найти, например, тут functions_arithmetic.yaml. Здесь описаны поддерживаемые арифметические функции вот в таком виде:

urn: extension:io.substrait:functions_arithmetic
scalar_functions:
  -
    name: "add"
    description: "Add two values."
    impls:
      - args:
          - name: x
            value: i8
          - name: y
            value: i8
        options:
          overflow:
            values: [ SILENT, SATURATE, ERROR ]
        return: i8
      - args:
          - name: x
            value: i16
          - name: y
            value: i16
        options:
          overflow:
            values: [ SILENT, SATURATE, ERROR ]
        return: i16
…

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

В этом и есть мощь Substrait и его отличие от других способов передачи планов, ведь учитываются даже такие тонкие моменты. Кроме того, каждая функция помечена URN (в ранних версиях Substrait — URI) — это сделано для разрешения конфликта имен по аналогии с package в java и namespace в C++, так как возможны ситуации, когда функции с одним и тем же именем в разных системах могут делать разные вещи. Типичный пример: DATEDIFF — одно имя, разные сигнатуры/семантика.

• MySQL: DATEDIFF(date1, date2) — дни между датами (date1 — date2).

• SQL Server: DATEDIFF(day, start, end) — разность в указанных единицах (не только дни); другой порядок аргументов.

Типы данных (кроме встроенных в Substrait) описываются похожим образом в yaml. Вот пример из самого Substrait extension_types.yaml, где описан тип Point:

urn: extension:io.substrait:extension_types
types:
  - name: point
    structure:
      latitude: i32
      longitude: i32
  - name: line
    structure:
      start: point
      end: point

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

Когда поступает запрос от пользователя в каком-либо виде (SQL, python, dplyr,...), продюсер конвертирует его в protobuf-сообщение в соответствии со спецификацией Substrait. 

Чтобы было нагляднее возьмем типичный use-case пушдауна подзапроса в другую систему. Например, мы должны выполнить этот запрос:

-- Сумма продаж по строкам, где price > 100
SELECT SUM(amount) AS total_amount
FROM sales
WHERE price > 100;

Где таблица sales лежит, допустим, в паркете, и наш Parquet-reader умеет выполнять фильтрацию с простыми предикатами, как в примере WHERE price > 100, но агрегацию (SUM) придется продюсеру  делать самому. Для этого продюсер генерирует Substrait план и отправляет его консьюмеру плана:

{
 // Перечисление использующихся extensions (функций и типов)
 "extensions": [
    // namespase для extensions имеет anchor=1, это ссылка по которой функция привязывается к простанству имен
    { "extensionUri": { "uriAnchor": 1, "uri": "urn:substrait:functions_comparison" } },
    // Определение используемой функции “gt”, сокращенно “greater than”. Имеет anchor по которому на нее ссылаются ниже
    { "extensionFunction": { "functionAnchor": 1, "name": "gt", "uriReference": 1 } }
  ],

  "relations": [
    {
      // Дерево реляционных выражений
      "root": {
        "input": {
          "project": {  // Оператор проеции SELECT(amount) который откидывает лишние колонки
            "input": {
              "filter": {  // Оператор фильтра WHERE price > 100
                "input": {
                  "read": {  // Оператор чтения из таблицы и схема таблицы         
                    "namedTable": { "names": ["sales"] },
                    "baseSchema": {
                      "names": ["id", "price", "amount"],
                      "struct": {
                        "types": [
                          { "i64": {} },
                          { "i32": {} },
                          { "i32": {} }
                        ]
                      }
                    }
                  }
                },
                "condition": {  // Аргумент оператора filter - функция price > 100
                  "scalarFunction": {
                    "functionReference": 1,          // gt(x, y) - операция “Больше” с anchor=1 (см. в начале определние extensionFunction)
                    "arguments": [
                      {
                        "value": {
                          "selection": {
                            "directReference": { "structField": { "field": 1 } },  // поле price - поле #1
                            "rootReference": {}
                          }
                        }
                      },
                      { "value": { "literal": { "i32": 100 } } } // Литерал “100” типа int32
                    ]
                  }
                }
              }
            },
            "expressions": [ // Аргументоператора project - выбираем только колонку amount
              {                                          
                "selection": {
                  "directReference": { "structField": { "field": 2 } }, // amount - поле #2
                  "rootReference": {}
                }
              }
            ]
          }
        },
        "names": ["amount"] // финальная проекция узла root
      }
    }
  ]
}

Консьюмер плана, получив его, начинает исполнение и отдает обратно данные, а именно колонку amount, которая уже отфильтрована в соответствии с условием WHERE price > 100. Остается данные сагрегировать в SELECT SUM(amount) и вернуть пользователю итоговый результат. 

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

Способы использования Substrait

В примере выше показан лишь один из сценариев, где Substrait был бы полезен. А именно, пушдаун подзапроса в другую систему. Но есть и другие кейсы использования Substrait.

Его мы только что видели выше, где одна система отправляет план подазпроса в другую (или в несколько других) систем. При этом подходе Database 1 должна хорошо понимать возможности других систем, чтобы случайно не отправить то, что они не могут выполнить. Например, в Parquet reader нельзя отправлять джойны или агрегаты, а проекции и простые фильтры можно.

В этом случае Database 1 не обязана знать возможности других систем. Но взаимодействие с Database 2 будет немного усложнено. Database 1 может отправить хоть целиком план запроса запроса в Database 2, и она уже сама решит: что возможно исполнить в ее рамках — она исполнит, а что нет, вернет в качестве остаточной (residual) части запроса, так же в формате Substrait, и остаточную часть уже будет исполнять Database 1.

Substrait может также использоваться и в рамках одной системы, заменяя собой внутренний IR для общения между планировщиком запросов и движком исполнения. Это может быть удобным для быстрой замены одного движка на другой при необходимости. Из существующих систем сейчас так работает Spark с движком нативного исполнения Gluten.

Кто поддерживает Substrait

Список активно пополняется. Проверяйте релизы на момент чтения.

Заключение

Substrait приближает нас к миру, где фронтенды (SQL/Ibis/dplyr/…​) и бэкенды (Velox/DuckDB/DataFusion/…​) можно свободно комбинировать. Это не серебряная пуля: придется договориться о семантике и профиле, наладить валидацию и наблюдаемость. Но выигрыш — в скорости интеграций и переносимости логики —  почти всегда стоит усилий.

Полезные ссылки

[1] Substrait site https://substrait.io/

[2] Substrait tutorial https://substrait.io/tutorial/sql_to_substrait/

[3] Substrait repo https://github.com/substrait-io/substrait/blob/main/extensions/extension_types.yaml

[4] Useful exmamples https://medium.com/@omri-levy/one-query-plan-three-different-engines-e5dc74aeb52f

Роман Кондаков

Старший разработчик CedrusData