Интерфейс Arrow Flight

ClickHouse поддерживает протокол Apache Arrow Flight — высокопроизводительный RPC-фреймворк для эффективной передачи столбцовых данных в формате Arrow IPC поверх gRPC.

Реализация также поддерживает Arrow Flight SQL, что позволяет BI-инструментам и приложениям, работающим по протоколу Flight SQL, отправлять запросы к ClickHouse напрямую.

Основные возможности:

• Выполнение SQL-запросов и получение результатов в формате Apache Arrow.
• Вставка данных в таблицы с использованием формата Arrow.
• Запрос метаданных (каталогов, схем, таблиц, первичных ключей) через команды Flight SQL.
• Создание, связывание, выполнение и закрытие серверных подготовленных операторов через Flight SQL.
• Управление сеансами и настройками через действия Flight SQL.
• Шифрование TLS и аутентификация по имени пользователя и паролю.
• Поэтапное получение результатов через PollFlightInfo.
• Отмена запросов через CancelFlightInfo.

Чтобы включить сервер Arrow Flight, добавьте настройку arrowflight_port в конфигурацию сервера ClickHouse:

<clickhouse>
    <arrowflight_port>9090</arrowflight_port>
</clickhouse>

При запуске в журнале появляется сообщение, подтверждающее, что интерфейс активен:

{} <Information> Application: Arrow Flight compatibility protocol: 0.0.0.0:9090

Чтобы включить TLS для интерфейса Arrow Flight, настройте следующие параметры:

<clickhouse>
    <arrowflight_port>9090</arrowflight_port>
    <arrowflight>
        <enable_ssl>true</enable_ssl>
        <ssl_cert_file>/path/to/server-cert.pem</ssl_cert_file>
        <ssl_key_file>/path/to/server-key.pem</ssl_key_file>
    </arrowflight>
</clickhouse>

Когда TLS включен, клиенты должны подключаться по схеме grpc+tls:// вместо grpc://.

Интерфейс Arrow Flight поддерживает два метода аутентификации:

Клиенты проходят аутентификацию по имени пользователя и паролю через стандартный HTTP-заголовок Authorization: Basic. При успешной аутентификации server возвращает Bearer-токен в заголовке ответа.

В последующих запросах можно использовать Bearer-токен, полученный при базовой аутентификации, передавая его в заголовке Authorization: Bearer <token>. Токен автоматически обновляется при каждом использовании и действует до истечения времени, заданного настройкой сервера default_session_timeout (по умолчанию — 60 секунд).

import pyarrow.flight as flight

client = flight.FlightClient("grpc://localhost:9090")

# Базовая аутентификация возвращает bearer-токен для последующих запросов
token_pair = client.authenticate_basic_token("default", "")
options = flight.FlightCallOptions(headers=[token_pair])

С использованием TLS:

import pyarrow.flight as flight

with open("ca-cert.pem", "rb") as f:
    tls_root_certs = f.read()

client = flight.FlightClient(
    "grpc+tls://localhost:9090",
    tls_root_certs=tls_root_certs,
)

token_pair = client.authenticate_basic_token("default", "password")
options = flight.FlightCallOptions(headers=[token_pair])

Интерфейс Arrow Flight поддерживает сеансы ClickHouse с помощью пользовательских заголовков метаданных gRPC:

Сеансы позволяют задавать сохраняемые настройки ClickHouse через действие SetSessionOptions (см. DoAction).

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

Принимает FlightDescriptor, который может быть одним из следующих:

• дескриптор PATH: Путь с одним компонентом, интерпретируемый как имя таблицы. Генерирует SELECT * FROM <table>.
• дескриптор CMD: Либо исходную строку SQL-запроса, либо сериализованную protobuf-команду Flight SQL (см. Команды Flight SQL).

Запрос выполняется полностью, а результаты сохраняются в тикетах на стороне сервера. Каждый block данных создает отдельную конечную точку/тикет, что позволяет клиентам получать данные параллельно.

# Запрос по имени таблицы
descriptor = flight.FlightDescriptor.for_path("my_table")
info = client.get_flight_info(descriptor, options)

# Запрос по SQL
descriptor = flight.FlightDescriptor.for_command(
    "SELECT * FROM my_table WHERE id > 100"
)
info = client.get_flight_info(descriptor, options)

# Получение результатов
for endpoint in info.endpoints:
    reader = client.do_get(endpoint.ticket, options)
    table = reader.read_all()
    print(table.to_pandas())

Позволяет получать результаты длительно выполняющихся запросов по мере готовности. Вместо того чтобы ждать завершения всего запроса (как в случае с GetFlightInfo), PollFlightInfo возвращает результаты блоками.

При первом вызове запрос начинает выполняться. Ответ включает:

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

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

Возвращает схему Arrow для результата запроса без выполнения всего запроса. Поддерживает те же типы дескрипторов, что и GetFlightInfo.

descriptor = flight.FlightDescriptor.for_command(
    "SELECT 1 AS x, 'hello' AS y"
)
schema_result = client.get_schema(descriptor, options)
schema = schema_result.schema
print(schema)  # x: int32, y: string

Извлекает данные по указанному тикету. Принимает одно из следующего:

• Тикет, возвращённый GetFlightInfo или PollFlightInfo.
• Необработанную строку SQL-запроса в качестве значения тикета.

# Использование ticket из GetFlightInfo
reader = client.do_get(endpoint.ticket, options)
table = reader.read_all()

# Использование SQL-запроса в качестве ticket
ticket = flight.Ticket("SELECT number FROM system.numbers LIMIT 10")
reader = client.do_get(ticket, options)
table = reader.read_all()

Отправляет данные в ClickHouse. Принимает FlightDescriptor и поток батчей записей Arrow.

Вставка по имени таблицы (дескриптор PATH):

schema = pa.schema([("id", pa.int64()), ("name", pa.string())])
batch = pa.record_batch(
    [pa.array([1, 2, 3]), pa.array(["Alice", "Bob", "Charlie"])],
    schema=schema,
)

descriptor = flight.FlightDescriptor.for_path("my_table")
writer, _ = client.do_put(descriptor, schema, options)
writer.write_batch(batch)
writer.close()

Вставка с помощью SQL (дескриптор CMD):

descriptor = flight.FlightDescriptor.for_command(
    "INSERT INTO my_table FORMAT Arrow"
)
writer, _ = client.do_put(descriptor, schema, options)
writer.write_batch(batch)
writer.close()

Выполнение DDL/DML через Flight SQL CommandStatementUpdate:

Клиенты Flight SQL используют CommandStatementUpdate для выполнения DDL/DML-команд (CREATE, INSERT, ALTER и т. д.). В ответе возвращается количество затронутых строк.

Пакетный приём через Flight SQL CommandStatementIngest:

Поддерживается только добавление данных в существующие таблицы (TABLE_NOT_EXIST_OPTION_FAIL + TABLE_EXISTS_OPTION_APPEND). Каталоги и временные таблицы для этой команды не поддерживаются.

transaction_id не поддерживается для CommandStatementUpdate и CommandStatementIngest. Если он указан, ClickHouse возвращает ошибку NotImplemented.

Выполняет именованные действия. Поддерживаются следующие из них:

Отменяет выполняемый запрос, связанный с FlightInfo. Идентификатор запроса извлекается из поля app_metadata объекта FlightInfo. Также отменяет все дескрипторы опроса, связанные с этим запросом.

# Запустить длительный запрос через PollFlightInfo, затем отменить его
cancel_request = flight.CancelFlightInfoRequest(info)
result = client.cancel_flight_info(cancel_request, options)
# result.status равен CancelStatus.CANCELLED в случае успеха

Устанавливает настройки сервера ClickHouse для текущего сеанса. Для этого требуется, чтобы идентификатор сеанса был задан через заголовок x-clickhouse-session-id.

Поддерживаемые типы значений: string, boolean, integer, double и списки строк.

Если имя настройки неизвестно, возвращается ошибка INVALID_NAME. Если значение не удаётся разобрать, возвращается ошибка INVALID_VALUE.

Возвращает все текущие настройки ClickHouse и их значения для данного сеанса. Возвращает сопоставление имён настроек со строковыми значениями (внутри выполняет запрос к system.settings).

Создаёт подготовленный оператор на стороне сервера и возвращает дескриптор оператора. Запрос содержит текст SQL-запроса с плейсхолдерами ?.

transaction_id для этого действия не поддерживается. Если он указан, ClickHouse возвращает ошибку NotImplemented.

Для операторов запроса ответ может включать:

• dataset_schema: схема результирующего набора.
• parameter_schema: схема параметров оператора.

Если не удаётся определить схему для корректного запроса (например, когда замена плейсхолдеров на NULL недопустима для этого запроса), ClickHouse всё равно создаёт подготовленный оператор и возвращает дескриптор без dataset_schema.

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

Другие пользователи не могут выполнять, привязывать или закрывать дескриптор оператора, который они не создавали.

arrowflight.prepared_statements_lifetime_seconds управляет поведением при истечении срока действия:

• > 0: использовать настроенное значение как время жизни оператора. Срок действия обновляется при каждом запросе как для операторов, привязанных к сеансу, так и для операторов без сеанса.
• 0: подготовленные операторы не истекают автоматически.
• -1 (по умолчанию): если оператор создан в сеансе, его время жизни соответствует тайм-ауту этого сеанса и обновляется при каждом запросе в этом сеансе. Если оператор создан без сеанса, он не истекает автоматически.

Операторы с истекшим сроком действия удаляются и больше не учитываются в arrowflight.max_prepared_statements_per_user.

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

ClickHouse также поддерживает массовое закрытие с помощью ClosePreparedStatement, когда дескриптор пуст:

• Если указан x-clickhouse-session-id, закрываются все подготовленные операторы аутентифицированного пользователя в этом сеансе.
• Если идентификатор сеанса не указан, закрываются только подготовленные операторы без сеанса для аутентифицированного пользователя.

Если подготовленный оператор создан в сеансе (через x-clickhouse-session-id), он также автоматически закрывается при закрытии этого сеанса.

Если дескриптор CMD содержит сериализованное сообщение Flight SQL protobuf, ClickHouse обрабатывает следующие команды:

Эти команды относятся к возможностям, которых в ClickHouse нет, поэтому они не поддерживаются интерфейсом Arrow Flight SQL.

import pyarrow as pa
import pyarrow.flight as flight

# Подключение и аутентификация
client = flight.FlightClient("grpc://localhost:9090")
token = client.authenticate_basic_token("default", "")
options = flight.FlightCallOptions(headers=[token])

# Вставка данных с помощью DoPut с дескриптором PATH
schema = pa.schema([("id", pa.uint32()), ("value", pa.string())])
batch = pa.record_batch(
    [pa.array([1, 2, 3], type=pa.uint32()), pa.array(["a", "b", "c"])],
    schema=schema,
)
descriptor = flight.FlightDescriptor.for_path("test")
writer, _ = client.do_put(descriptor, schema, options)
writer.write_batch(batch)
writer.close()

# Запрос данных с помощью GetFlightInfo + DoGet
descriptor = flight.FlightDescriptor.for_command(
    "SELECT * FROM test ORDER BY id"
)
info = client.get_flight_info(descriptor, options)
for endpoint in info.endpoints:
    reader = client.do_get(endpoint.ticket, options)
    table = reader.read_all()
    print(table.to_pandas())

   id value
0   1     a
1   2     b
2   3     c

Все данные передаются в формате Apache Arrow IPC. Поддерживается только формат Arrow — указание других форматов ClickHouse (например, FORMAT JSON, FORMAT CSV) приводит к ошибке.

При сериализации типы данных ClickHouse сопоставляются с типами Arrow. Параметр output_format_arrow_unsupported_types_as_binary определяет, будут ли неподдерживаемые типы ClickHouse сериализоваться как бинарные BLOB-объекты.

Интерфейс Arrow Flight совместим с любым клиентом или инструментом, поддерживающим протокол Arrow Flight или Arrow Flight SQL, в том числе:

• Python (pyarrow)
• Java (org.apache.arrow.flight)
• C++ (arrow::flight)
• Go (apache/arrow/go)
• драйверы ADBC (Arrow Database Connectivity)
• DBeaver и другие инструменты с поддержкой Flight SQL

Если для вашего инструмента доступен собственный коннектор ClickHouse (например, JDBC, ODBC или собственный протокол), лучше использовать его, если только Arrow Flight не нужен именно для повышения производительности или совместимости форматов.

ClickHouse также может выступать клиентом Flight и читать данные с внешних серверов Arrow Flight. См.:

• Движок таблицы ArrowFlight
• Табличная функция arrowFlight

• Спецификация Apache Arrow Flight
• Спецификация Apache Arrow Flight SQL
• Формат Arrow в ClickHouse