Flight SQL 驱动程序¶

Flight SQL 驱动程序提供对实现 Arrow Flight SQL 兼容端点的任何数据库的访问。

安装¶

C/C++

从 conda-forge 安装 libadbc-driver-flightsql

mamba install libadbc-driver-flightsql

安装 C/C++ 驱动程序，然后使用 Go 驱动程序管理器。需要 CGO。

go get github.com/apache/arrow-adbc/go/adbc/drivermgr

Python

从 conda-forge 安装 adbc-driver-flightsql

mamba install adbc-driver-flightsql

从 PyPI 安装 adbc-driver-flightsql

pip install adbc-driver-flightsql

从 R-multiverse 安装 adbcflightsql

install.packages("adbcflightsql", repos = "https://community.r-multiverse.org")

此外，该驱动程序可以通过驱动程序管理器从 C/C++、C#、GLib、Go、R、Ruby 和 Rust 使用。

用法¶

要连接到数据库，在构造 AdbcDatabase 时提供“uri”参数。

C++

#include "arrow-adbc/adbc.h"

// Ignoring error handling
struct AdbcDatabase database;
AdbcDatabaseNew(&database, nullptr);
AdbcDatabaseSetOption(&database, "driver", "adbc_driver_flightsql", nullptr);
AdbcDatabaseSetOption(&database, "uri", "grpc://:8080", nullptr);
AdbcDatabaseInit(&database, nullptr);

Python

注意

有关详细示例，请参见 Flight SQL 食谱。

from adbc_driver_flightsql import DatabaseOptions
from adbc_driver_flightsql.dbapi import connect

headers = {"foo": "bar"}

with connect(
    "grpc+tls://:8080",
    db_kwargs={
        DatabaseOptions.AUTHORIZATION_HEADER.value: "Bearer <token>",
        DatabaseOptions.TLS_SKIP_VERIFY.value: "true",
        **{
            f"{DatabaseOptions.RPC_CALL_HEADER_PREFIX.value}{k}": v
            for k, v in headers.items()
        },
    }
) as conn:
    pass

食谱来源：example_usage_test.go

// Tests that use the SQLite server example.

package flightsql_test

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log"

	"github.com/apache/arrow-adbc/go/adbc"
	drv "github.com/apache/arrow-adbc/go/adbc/driver/flightsql"
	"github.com/apache/arrow-go/v18/arrow/array"
	"github.com/apache/arrow-go/v18/arrow/flight"
	"github.com/apache/arrow-go/v18/arrow/flight/flightsql"
	sqlite "github.com/apache/arrow-go/v18/arrow/flight/flightsql/example"
	"github.com/apache/arrow-go/v18/arrow/memory"
	_ "modernc.org/sqlite"
)

var headers = map[string]string{"foo": "bar"}

func FlightSQLExample(uri string) (err error) {
	ctx := context.Background()
	options := map[string]string{
		adbc.OptionKeyURI: uri,
	}

	for k, v := range headers {
		options[drv.OptionRPCCallHeaderPrefix+k] = v
	}

	var alloc memory.Allocator
	drv := drv.NewDriver(alloc)
	db, err := drv.NewDatabase(options)
	if err != nil {
		return fmt.Errorf("failed to open database: %s\n", err.Error())
	}
	defer func() {
		err = errors.Join(err, db.Close())
	}()

	cnxn, err := db.Open(ctx)
	if err != nil {
		return fmt.Errorf("failed to open connection: %s", err.Error())
	}
	defer func() {
		err = errors.Join(err, cnxn.Close())
	}()

	stmt, err := cnxn.NewStatement()
	if err != nil {
		return fmt.Errorf("failed to create statement: %s", err.Error())
	}
	defer func() {
		err = errors.Join(err, stmt.Close())
	}()

	if err = stmt.SetSqlQuery("SELECT 1 AS theresult"); err != nil {
		return fmt.Errorf("failed to set query: %s", err.Error())
	}

	reader, _, err := stmt.ExecuteQuery(ctx)
	if err != nil {
		return fmt.Errorf("failed to execute query: %s", err.Error())
	}
	defer reader.Release()

	for reader.Next() {
		arr, ok := reader.RecordBatch().Column(0).(*array.Int64)
		if !ok {
			return fmt.Errorf("result data was not int64")
		}
		for i := 0; i < arr.Len(); i++ {
			if arr.IsNull(i) {
				fmt.Println("theresult: NULL")
			} else {
				fmt.Printf("theresult: %d\n", arr.Value(i))
			}
		}
	}

	return nil
}

func Example() {
	// For this example we will spawn the Flight SQL server ourselves.

	// Create a new database that isn't tied to any other databases that
	// may be in process.
	db, err := sql.Open("sqlite", "file:example_in_memory?mode=memory")
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		err := db.Close()
		if err != nil {
			log.Fatal(err)
		}
	}()

	srv, err := sqlite.NewSQLiteFlightSQLServer(db)
	if err != nil {
		log.Fatal(err)
	}

	server := flight.NewServerWithMiddleware(nil)
	server.RegisterFlightService(flightsql.NewFlightServer(srv))
	err = server.Init("localhost:8080")
	if err != nil {
		log.Fatal(err)
	}

	go func() {
		if err := server.Serve(); err != nil {
			log.Fatal(err)
		}
	}()

	uri := fmt.Sprintf("grpc://%s", server.Addr().String())
	if err := FlightSQLExample(uri); err != nil {
		log.Printf("Error: %s\n", err.Error())
	}

	server.Shutdown()

	// Output:
	// theresult: 1
}

标准输出¶

theresult: 1

支持的功能¶

Flight SQL 驱动程序通常支持 ADBC API 规范 1.0.0 中定义的功能，以及一些额外的自定义选项。

警告

Java 驱动程序不支持此处的所有选项。请参阅问题 #745。

认证¶

驱动程序默认不进行认证。驱动程序实现了几种可选的认证方案

相互 TLS (mTLS)：请参阅下面的“客户端选项”。
一种模仿 Arrow Flight SQL JDBC 驱动程序的 HTTP 风格方案。

在 AdbcDatabase 上设置选项 username 和 password。或者，设置选项 adbc.flight.sql.authorization_header 以获得完全控制。

客户端通过从客户端向服务器发送 authorization 来提供凭据。服务器随后会在第一个请求上响应一个 authorization 头部。此头部的值随后将作为所有未来请求上的 authorization 头部发送回去。
OAuth 2.0 认证流程。

客户端提供配置以允许客户端应用程序从授权服务器获取访问令牌。获取到的令牌随后将用于所有未来请求的 authorization 头部。

批量摄取¶

Flight SQL 没有专门用于将 Arrow 数据批量摄取到给定表的 API。因此，驱动程序目前不实现批量摄取。

客户端选项¶

用于创建 Flight RPC 客户端的选项可以自定义。

注意

其中许多选项只是封装了 gRPC 选项。有关这些选项功能的更多详细信息，请查阅 gRPC 文档。

adbc.flight.sql.client_option.authority

覆盖 gRPC 的 :authority 伪头部。

Python：adbc_driver_flightsql.DatabaseOptions.AUTHORITY

adbc.flight.sql.client_option.mtls_cert_chain

用于 mTLS 的证书链。

Python：adbc_driver_flightsql.DatabaseOptions.MTLS_CERT_CHAIN

adbc.flight.sql.client_option.mtls_private_key

用于 mTLS 的私钥。

Python：adbc_driver_flightsql.DatabaseOptions.MTLS_PRIVATE_KEY

adbc.flight.sql.client_option.tls_override_hostname

覆盖用于验证服务器 TLS 证书的主机名。

Python：adbc_driver_flightsql.DatabaseOptions.TLS_OVERRIDE_HOSTNAME

adbc.flight.sql.client_option.tls_root_certs

覆盖用于验证服务器 TLS 证书的根证书。

Python：adbc_driver_flightsql.DatabaseOptions.TLS_ROOT_CERTS

adbc.flight.sql.client_option.tls_skip_verify

禁用对服务器 TLS 证书的验证。值应为 true 或 false。

Python：adbc_driver_flightsql.DatabaseOptions.TLS_SKIP_VERIFY

adbc.flight.sql.client_option.with_block

警告

此选项已弃用，因为 gRPC 本身已弃用底层选项。

此选项无效，并将在未来版本中删除。值应为 true 或 false。

adbc.flight.sql.client_option.with_max_msg_size

接受来自服务器的最大消息大小。由于 Flight 服务倾向于返回较大的响应负载，驱动程序默认为 16 MiB。应为正整数字节数。

Python：adbc_driver_flightsql.DatabaseOptions.WITH_MAX_MSG_SIZE

adbc.flight.sql.authorization_header

直接指定要在所有请求上发送的 authorization 头部的值。

Python：adbc_driver_flightsql.DatabaseOptions.AUTHORIZATION_HEADER

adbc.flight.sql.rpc.with_cookie_middleware

启用或禁用处理和处理服务器返回的“set-cookie”元数据头部以及从客户端发送“Cookie”头部中间件。值应为 true 或 false。默认值为 false。

Python：adbc_driver_flightsql.DatabaseOptions.WITH_COOKIE_MIDDLEWARE

自定义调用头部¶

自定义 HTTP 头部可以通过适用于 AdbcDatabase、AdbcConnection 和 AdbcStatement 的选项附加到请求中。

adbc.flight.sql.rpc.call_header.<HEADER NAME>

将头部 <HEADER NAME> 以给定值添加到传出请求中。

Python：adbc_driver_flightsql.ConnectionOptions.RPC_CALL_HEADER_PREFIX

警告

头部名称必须全部小写。

OAuth 2.0 选项¶

支持使用 OAuth 2.0 认证流程获取令牌的配置。

adbc.flight.sql.oauth.flow

指定要使用的 OAuth 2.0 流程类型。可能的值：client_credentials、token_exchange

adbc.flight.sql.oauth.client_id

授权服务器颁发给客户端应用程序的唯一标识符

adbc.flight.sql.oauth.client_secret

与 client_id 关联的密钥。用于向授权服务器验证客户端应用程序

adbc.flight.sql.oauth.token_uri

客户端应用程序从授权服务器请求令牌的端点 URL

adbc.flight.sql.oauth.scope

客户端请求访问的权限列表，以空格分隔（例如 "read.all offline_access"）

adbc.flight.sql.oauth.exchange.subject_token

客户端应用程序希望交换的安全令牌

adbc.flight.sql.oauth.exchange.subject_token_type

主题令牌类型的标识符。请查看下面的列表以了解支持的令牌类型。

adbc.flight.sql.oauth.exchange.actor_token

表示执行方身份的安全令牌

adbc.flight.sql.oauth.exchange.actor_token_type

参与者令牌类型的标识符。请查看下面的列表以了解支持的令牌类型。

adbc.flight.sql.oauth.exchange.aud

请求的安全令牌的预期受众

adbc.flight.sql.oauth.exchange.resource

客户端打算使用请求的安全令牌的资源服务器

adbc.flight.sql.oauth.exchange.scope

为新令牌请求的特定权限

adbc.flight.sql.oauth.exchange.requested_token_type

客户端希望接收的令牌类型。请查看下面的列表以了解支持的令牌类型。

支持的令牌类型

urn:ietf:params:oauth:token-type:access_token
urn:ietf:params:oauth:token-type:refresh_token
urn:ietf:params:oauth:token-type:id_token
urn:ietf:params:oauth:token-type:saml1
urn:ietf:params:oauth:token-type:saml2
urn:ietf:params:oauth:token-type:jwt

分布式结果集¶

驱动程序将以未指定顺序获取服务器返回的所有分区（FlightEndpoints）（请注意，Flight SQL 本身并未定义这些分区的顺序）。如果一个端点没有位置，数据将使用原始服务器连接获取。否则，驱动程序将按顺序尝试给定的每个位置，直到请求成功。如果连接或请求失败，它将尝试下一个位置。

驱动程序目前不缓存或池化这些辅助连接。它也不会重试连接或请求。

所有分区都并行获取。每个分区都有有限数量的批次排队。数据按分区顺序返回给客户端。

一些行为可以在 AdbcStatement 上配置

adbc.rpc.result_queue_size

每个分区排队的批次数量。默认为 5。

Python：adbc_driver_flightsql.StatementOptions.QUEUE_SIZE

增量执行¶

通过设置 ADBC_STATEMENT_OPTION_INCREMENTAL，您可以使用此驱动程序进行非阻塞执行。这仅更改 AdbcStatementExecutePartitions() 的行为。启用后，ExecutePartitions 将在每次服务器有新分区（在 Flight SQL 术语中，当有新的 FlightEndpoints）时返回，而不是阻塞直到查询完成。

一些行为可以在 AdbcStatement 上配置

adbc.flight.sql.statement.exec.last_flight_info

获取服务返回的最新 FlightInfo 的序列化字节。这是一个用于高级用途的低级选项。它在启用增量执行时最有用，用于检查最新的服务器响应，而无需等待 AdbcStatementExecutePartitions() 返回。

Python：adbc_driver_flightsql.StatementOptions.LAST_FLIGHT_INFO

元数据¶

驱动程序目前不会在 AdbcConnectionGetObjects() 中填充列约束信息（外键、主键等）。此外，目录过滤器被评估为简单的字符串匹配，而不是 LIKE 样式模式。

分区结果集¶

Flight SQL 驱动程序支持 ADBC 的分区结果集。当请求时，结果集的每个分区都包含一个序列化的 FlightInfo，其中包含原始响应的一个 FlightEndpoint。希望检查分区的客户端可以通过反序列化 ADBC 分区中包含的 FlightInfo 来实现。（例如，希望在多个工作程序或机器之间分配工作的客户端可能希望利用 ADBC 没有的局部性信息。）