Интерфейс Arrow Flight

​

Обзор

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

​

Включение сервера Arrow Flight

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

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

​

Настройка TLS

<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>

​

Аутентификация

​

Базовая аутентификация

​

Аутентификация с Bearer-токеном

​

Пример на Python

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])

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])

​

Управление сеансами

​

Справочник по конфигурации сервера

​

Поддерживаемые методы RPC

​

GetFlightInfo

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

# Запрос по имени таблицы
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())

​

PollFlightInfo

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

​

GetSchema

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

​

DoGet

• Тикет, возвращённый 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()

​

DoPut

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()

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

​

DoAction

​

CancelFlightInfo

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

​

SetSessionOptions

​

GetSessionOptions

​

CreatePreparedStatement

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

• При определении схемы каждый ? заменяется на NULL, поэтому плейсхолдер, определяющий столбец результата, получает тип от этого NULL, а не от значения, которое вы привяжете позже. SELECT ? AS x определяет столбец типа Nothing, но при привязке 5 отдаётся UInt8. У плейсхолдера, используемого только в предикате, как в SELECT id, name FROM t WHERE id = ?, такой проблемы нет, поскольку типы результата берутся из таблицы.
• Столбец, не имеющий эквивалента в Arrow, получает свой тип Arrow из output_format_arrow_unsupported_types, которое каждый вызов разрешает из сеанса, из которого он выполняется. Поскольку дескриптор принадлежит пользователю, а не одному сеансу, последующий вызов может разрешить его иначе и отдать binary там, где был заявлен utf8, или наоборот. Установка режима внутри самого подготовленного запроса фиксирует его для обоих случаев.

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

​

ClosePreparedStatement

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

​

Команды Flight SQL

​

Поддерживаются через GetFlightInfo / GetSchema

​

Поддерживается через DoPut

​

Не поддерживается в ClickHouse

​

Полный пример

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

​

Формат данных

• Flight SQL фиксирует схему своих ответов с метаданными. Например, CommandGetTables обязан возвращать catalog_name: utf8, db_schema_name: utf8, table_name: utf8 not null, table_type: utf8 not null, table_schema: bytes not null. Если бы настройка сеанса превращала эти столбцы utf8 в binary, ClickHouse перестал бы соответствовать спецификации для любого драйвера Flight SQL, а также изменилась бы схема отдельных таблиц, которую ClickHouse объявляет внутри table_schema.
• Клиент Flight получает схему и данные разными вызовами (GetFlightInfo или GetSchema, затем DoGet). Любая настройка, способная изменить схему, открывает возможность рассогласования между объявленной схемой и переданным потоком, если в промежутке сеанс изменится.

​

Совместимость

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

​

Возможности ArrowFlight на стороне клиента

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

​

См. также

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