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

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

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

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_FLOW、adbc_driver_flightsql.OAuthFlowType

adbc.flight.sql.oauth.client_id

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

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_CLIENT_ID

adbc.flight.sql.oauth.client_secret

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

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_CLIENT_SECRET

adbc.flight.sql.oauth.token_uri

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

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_TOKEN_URI

adbc.flight.sql.oauth.scope

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

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_SCOPE

adbc.flight.sql.oauth.exchange.subject_token

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

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_EXCHANGE_SUBJECT_TOKEN

adbc.flight.sql.oauth.exchange.subject_token_type

主体令牌类型的标识符。查看下面的列表以获取支持的令牌类型。

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_EXCHANGE_SUBJECT_TOKEN_TYPE、adbc_driver_flightsql.OAuthTokenType

adbc.flight.sql.oauth.exchange.actor_token

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

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_EXCHANGE_ACTOR_TOKEN

adbc.flight.sql.oauth.exchange.actor_token_type

行动方令牌类型的标识符。查看下面的列表以获取支持的令牌类型。

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_EXCHANGE_ACTOR_TOKEN_TYPE、adbc_driver_flightsql.OAuthTokenType

adbc.flight.sql.oauth.exchange.aud

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

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_EXCHANGE_AUD

adbc.flight.sql.oauth.exchange.resource

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

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_EXCHANGE_RESOURCE

adbc.flight.sql.oauth.exchange.scope

为新令牌请求的特定权限

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_EXCHANGE_SCOPE

adbc.flight.sql.oauth.exchange.requested_token_type

客户端希望交换接收的令牌类型。查看下面的列表以获取支持的令牌类型。

Python： adbc_driver_flightsql.DatabaseOptions.OAUTH_EXCHANGE_REQUESTED_TOKEN_TYPE、adbc_driver_flightsql.OAuthTokenType

支持的令牌类型

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

Python： adbc_driver_flightsql.OAuthTokenType

分布式结果集¶

驱动程序将以未指定的顺序获取服务器返回的所有分区（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，其中包含原始响应中的一个 FlightEndpoints。希望自省分区的客户端可以通过反序列化 ADBC 分区中包含的 FlightInfo 来实现。（例如，希望在多个工作程序或机器上分发工作的客户端可能希望利用 ADBC 没有的本地性信息。）