واجهة Arrow Flight

يدعم ClickHouse بروتوكول Apache Arrow Flight — وهو إطار RPC عالي الأداء لنقل البيانات العمودية بكفاءة باستخدام تنسيق Arrow IPC عبر gRPC.

يتضمن هذا التنفيذ دعم Arrow Flight SQL، مما يتيح لأدوات BI والتطبيقات التي تدعم بروتوكول Flight SQL الاستعلام من ClickHouse مباشرةً.

القدرات الأساسية:

• تنفيذ استعلامات SQL واسترجاع النتائج بتنسيق Apache Arrow.
• إدراج البيانات في الجداول باستخدام تنسيق Arrow.
• الاستعلام عن البيانات الوصفية (catalogs وschemas وtables وprimary keys) عبر أوامر 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. وعند نجاح المصادقة، يُرجِع الخادم Bearer token في ترويسة الاستجابة.

يمكن للطلبات اللاحقة استخدام رمز Bearer المُعاد من المصادقة الأساسية عبر ترويسة Authorization: Bearer <token>. ويُجدَّد الرمز تلقائيًا عند كل استخدام، وتنتهي صلاحيته وفقًا لإعداد الخادم default_session_timeout (الافتراضي: 60 ثانية).

import pyarrow.flight as flight

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

# Basic auth returns a bearer token for subsequent calls
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 خام، أو أمر Flight SQL protobuf مُسلسل (راجع Flight SQL Commands).

يُنفَّذ الاستعلام بالكامل، وتُخزَّن النتائج في تذاكر على جانب الخادم. وتنتج كل كتلة بيانات نقطة نهاية/تذكرة منفصلة، مما يتيح للعملاء استرجاع البيانات بالتوازي.

# Query by table name
descriptor = flight.FlightDescriptor.for_path("my_table")
info = client.get_flight_info(descriptor, options)

# Query by SQL
descriptor = flight.FlightDescriptor.for_command(
    "SELECT * FROM my_table WHERE id > 100"
)
info = client.get_flight_info(descriptor, options)

# Retrieve results
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.
• سلسلة استعلام Raw SQL كقيمة للتذكرة.

# Using a ticket from GetFlightInfo
reader = client.do_get(endpoint.ticket, options)
table = reader.read_all()

# Using a raw SQL query as 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. كما يُلغي أيضًا أي واصفات استطلاع مرتبطة بالاستعلام.

# Start a long-running query via PollFlightInfo, then cancel it
cancel_request = flight.CancelFlightInfoRequest(info)
result = client.cancel_flight_info(cancel_request, options)
# result.status is CancelStatus.CANCELLED if successful

يضبط إعدادات خادم ClickHouse للجلسة الحالية. ويتطلب ذلك تعيين معرّف جلسة عبر الترويسة x-clickhouse-session-id.

أنواع القيم المدعومة: string وboolean وinteger وdouble وقوائم من string.

إذا كان اسم الإعداد غير معروف، فسيُعاد الخطأ INVALID_NAME. وإذا تعذّر تحليل القيمة، فسيُعاد الخطأ INVALID_VALUE.

يعيد جميع إعدادات ClickHouse الحالية وقيمها الخاصة بالجلسة. ويعيد خريطة تربط أسماء الإعدادات بقيم نصية (ويستعلم داخليًا من system.settings).

ينشئ عبارة مُحضَّرة على جانب الخادم ويُرجع معرّفًا لها. يحتوي الطلب على نص استعلام SQL مع عناصر نائبة ?.

transaction_id غير مدعوم لهذا الإجراء. وإذا تم توفيره، يعيد ClickHouse الخطأ NotImplemented.

بالنسبة إلى تعليمات الاستعلام، قد تتضمن الاستجابة ما يلي:

• dataset_schema: مخطط مجموعة النتائج.
• parameter_schema: مخطط معلمات التعليمة.

إذا فشل استنتاج المخطط لاستعلام صالح (على سبيل المثال، عندما لا يكون استبدال العناصر النائبة بـ NULL صالحًا لهذا الاستعلام)، فسيواصل ClickHouse إنشاء العبارة المُحضَّرة ويُرجع المعرّف من دون dataset_schema.

تكون العبارات المُحضَّرة مملوكة للمستخدم الذي جرت مصادقته، وليس لجلسة واحدة بعينها. وإذا فتحت عدة جلسات بالمستخدم نفسه، يمكنك تنفيذ معرّف التعليمة نفسه، وإعادة الربط به، وإغلاقه من أيٍّ من تلك الجلسات.

لا يمكن للمستخدمين الآخرين تنفيذ معرّف تعليمة لم ينشئوه، أو إجراء bind له، أو إغلاقه.

يتحكم arrowflight.prepared_statements_lifetime_seconds في سلوك انتهاء الصلاحية:

• > 0: استخدم القيمة المُعدّة على أنها مدة بقاء العبارة. ويُجدَّد انتهاء الصلاحية مع كل طلب لكلٍّ من التعليمات المرتبطة بجلسة وغير المرتبطة بجلسة.
• 0: لا تنتهي صلاحية العبارات المُحضَّرة تلقائيًا.
• -1 (default): إذا أُنشئت العبارة داخل جلسة، فإن مدة بقائها تتبع مهلة تلك الجلسة ويُجدَّد مع كل طلب داخلها. وإذا أُنشئت العبارة من دون جلسة، فلا تنتهي صلاحيتها تلقائيًا.

تُزال العبارات منتهية الصلاحية، ولا تعود تُحتسب ضمن arrowflight.max_prepared_statements_per_user.

يغلق عبارة مُحضَّرة ويحرّر موارد جهة الخادم المرتبطة بها عندما يحتوي الطلب على معرّف عبارة غير فارغ.

يدعم ClickHouse أيضًا الإغلاق المجمّع باستخدام ClosePreparedStatement عندما يكون المعرّف فارغًا:

• إذا كان x-clickhouse-session-id موجودًا، فسيُغلق جميع العبارات المُحضَّرة للمستخدم المُصادَق عليه ضمن تلك الجلسة.
• إذا لم يكن هناك معرّف جلسة، فسيُغلق فقط العبارات المُحضَّرة غير المرتبطة بجلسة للمستخدم المُصادَق عليه.

إذا أُنشئت عبارة مُحضَّرة داخل جلسة (عبر x-clickhouse-session-id)، فستُغلق أيضًا تلقائيًا عند إغلاق تلك الجلسة.

عندما يحتوي الواصف CMD على رسالة Protobuf لـ Flight SQL مُسلسلة، يدعم ClickHouse الأوامر التالية:

تشير هذه الأوامر إلى ميزات لا يوفّرها ClickHouse، لذلك فهي غير مدعومة في واجهة Arrow Flight SQL.

import pyarrow as pa
import pyarrow.flight as flight

# Connect and authenticate
client = flight.FlightClient("grpc://localhost:9090")
token = client.authenticate_basic_token("default", "")
options = flight.FlightCallOptions(headers=[token])

# Insert data using DoPut with a PATH descriptor
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()

# Query data using 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 غير المدعومة ستُسلسَل على شكل بيانات ثنائية.

واجهة Arrow Flight متوافقة مع أي عميل أو أداة تدعم بروتوكول Arrow Flight أو Arrow Flight SQL، بما في ذلك:

• بايثون (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