Flight SQL 食譜

其中一些食譜是針對由 SQLite 支援的示範 Flight SQL 服務編寫的。您可以自行執行，如下所示

$ go install github.com/apache/arrow/go/v${ARROW_MAJOR_VERSION}/arrow/flight/flightsql/example/cmd/sqlite_flightsql_server@latest
$ sqlite_flightsql_server -host 0.0.0.0 -port 8080

其他食譜使用 OSS 版本的 Dremio

$ docker run -p 9047:9047 -p 31010:31010 -p 45678:45678 dremio/dremio-oss

如果您已檢出 ADBC 儲存庫並安裝了 Docker Compose，則可以使用我們的設定來執行這兩項服務

$ docker compose up --detach --wait dremio dremio-init flightsql-sqlite-test

連線到未受保護的 Flight SQL 服務

食譜來源：flightsql_sqlite_connect.py

要連線到未受保護的 Flight SQL 服務，只需提供 URI 即可。

import os

import adbc_driver_flightsql.dbapi

uri = os.environ["ADBC_SQLITE_FLIGHTSQL_URI"]
conn = adbc_driver_flightsql.dbapi.connect(uri)

然後我們可以執行一個簡單的查詢。

with conn.cursor() as cur:
    cur.execute("SELECT 1")

    assert cur.fetchone() == (1,)

conn.close()

連線到具有使用者名稱和密碼的 Flight SQL 服務

食譜來源：flightsql_dremio_connect.py

Dremio 需要使用者名稱和密碼。要連線到具有身份驗證的 Flight SQL 服務，請在連線時提供選項。

import os

import adbc_driver_flightsql.dbapi
import adbc_driver_manager

uri = os.environ["ADBC_DREMIO_FLIGHTSQL_URI"]
username = os.environ["ADBC_DREMIO_FLIGHTSQL_USER"]
password = os.environ["ADBC_DREMIO_FLIGHTSQL_PASS"]
conn = adbc_driver_flightsql.dbapi.connect(
    uri,
    db_kwargs={
        adbc_driver_manager.DatabaseOptions.USERNAME.value: username,
        adbc_driver_manager.DatabaseOptions.PASSWORD.value: password,
    },
)

然後我們可以執行一個簡單的查詢。

with conn.cursor() as cur:
    cur.execute("SELECT 1")

    assert cur.fetchone() == (1,)

conn.close()

設定逾時和其他選項

食譜來源：flightsql_sqlite_options.py

Flight SQL 驅動程式支援各種選項。

import os

import adbc_driver_flightsql.dbapi
from adbc_driver_flightsql import ConnectionOptions, DatabaseOptions

uri = os.environ["ADBC_SQLITE_FLIGHTSQL_URI"]

我們可以啟用 Cookie 支援，某些伺服器實作需要此支援。

conn = adbc_driver_flightsql.dbapi.connect(
    uri,
    db_kwargs={DatabaseOptions.WITH_COOKIE_MIDDLEWARE.value: "true"},
)

其他選項在連線或語句上設定。

例如，我們可以將自訂標頭新增至所有傳出的請求。

custom_header = f"{ConnectionOptions.RPC_CALL_HEADER_PREFIX.value}x-custom-header"
conn.adbc_connection.set_options(**{custom_header: "value"})

我們也可以設定逾時。這些以浮點秒為單位。

conn.adbc_connection.set_options(
    **{
        ConnectionOptions.TIMEOUT_FETCH.value: 30.0,
        ConnectionOptions.TIMEOUT_QUERY.value: 30.0,
        ConnectionOptions.TIMEOUT_UPDATE.value: 30.0,
    }
)

這些選項將適用於我們建立的所有游標。

with conn.cursor() as cur:
    cur.execute("SELECT 1")

    assert cur.fetchone() == (1,)

conn.close()

設定最大 gRPC 訊息大小

食譜來源：flightsql_sqlite_max_msg_size.py

預設情況下，Flight SQL 驅動程式會限制傳入/傳出訊息的大小。如果超出這些限制，您可能會看到類似這樣的錯誤

INTERNAL: [FlightSQL] grpc: received message larger than max

可以調整這些限制以避免這種情況。

import os

import adbc_driver_flightsql.dbapi
from adbc_driver_flightsql import DatabaseOptions

uri = os.environ["ADBC_SQLITE_FLIGHTSQL_URI"]

此查詢每個批次產生約 16 MiB，這將觸發預設限制。

query = """
WITH RECURSIVE generate_series(value) AS (
  SELECT 1
  UNION ALL
  SELECT value + 1 FROM generate_series
   WHERE value + 1 <= 2048
)
SELECT printf('%.*c', 16384, 'x') FROM generate_series
"""

當我們執行查詢時，會收到錯誤。

conn = adbc_driver_flightsql.dbapi.connect(uri)
with conn.cursor() as cur:
    cur.execute(query)

    try:
        cur.fetchallarrow()
    except adbc_driver_flightsql.dbapi.InternalError:
        # This exception is expected.
        pass
    else:
        assert False, "Did not raise expected exception"

conn.close()

我們可以改為在連線時變更限制。

conn = adbc_driver_flightsql.dbapi.connect(
    uri,
    db_kwargs={DatabaseOptions.WITH_MAX_MSG_SIZE.value: "2147483647"},
)
with conn.cursor() as cur:
    cur.execute(query)

    assert len(cur.fetchallarrow()) == 2048

conn.close()