驱动程序示例¶

方案源码： driver_example.cc

在此，我们将展示如何使用 ADBC 驱动程序框架库在 C++ 中构建 ADBC 驱动程序。该库与 ADBC 用于构建其 SQLite 和 PostgreSQL 驱动程序的库相同。它抽象了 C 可调用对象及目录/元数据函数的实现细节，这些功能实现起来可能比较困难，但对于有效利用 ADBC 生态系统的其余部分至关重要。

概括而言，我们将构建一个驱动程序，其“数据库”是一个目录，数据库中的每个“表”都是一个包含 Arrow IPC 流的文件。可以使用批量导入功能写入表，并使用 SELECT * FROM (文件) 形式的简单查询来读取表。

安装¶

本快速入门实际上是一个可读的 C++ 文件。您可以克隆仓库，构建示例并跟随操作。

我们假设您使用 conda-forge 来管理依赖项。需要 CMake、C++17 编译器和 ADBC 库。安装方式如下：

mamba install cmake compilers libadbc-driver-manager

构建¶

我们将在此处使用 CMake。从 ADBC 仓库的源码检出目录开始

mkdir build
cd build
cmake ../docs/source/cpp/recipe_driver -DADBC_DRIVER_EXAMPLE_BUILD_TESTS=ON
cmake --build .
ctest

使用 C++ 构建 ADBC 驱动程序¶

让我们从一些头文件包含开始。特别需要注意的是，我们将需要驱动程序框架头文件和 nanoarrow，在该示例驱动程序中，我们将使用它来创建和使用 Arrow C 数据接口结构。

#include "driver_example.h"

#include <cstdio>
#include <string>

#include "driver/framework/connection.h"
#include "driver/framework/database.h"
#include "driver/framework/statement.h"

#include "nanoarrow/nanoarrow.hpp"
#include "nanoarrow/nanoarrow_ipc.hpp"

#include "arrow-adbc/adbc.h"

接下来，我们将一些基本的框架类型引入命名空间，以减少实现的冗长性

adbc::driver::Option ：选项可以在 ADBC 数据库、连接和语句上设置。它们可以是字符串、不透明二进制数据、双精度浮点数或整数。Option 类抽象了获取、设置和解析这些值的细节。
adbc::driver::Status：Status 是 ADBC 驱动程序框架的错误处理机制：没有返回值且可能失败的函数会返回 Status。您可以使用 UNWRAP_STATUS(some_call()) 作为 Status status = some_call(); if (!status.ok()) return status; 的简写，以简洁地传播错误。
adbc::driver::Result：Result<T> 用作函数的返回值，该函数在成功时返回类型为 T 的值，在失败时通过 Status 传达其错误。您可以使用 UNWRAP_RESULT(value, some_call()) 作为简写。
```
some_type value;
Result<some_type> maybe_value = some_call();
if (!maybe_value.status().ok()) {
  return maybe_value.status();
} else {
  value = *maybe_value;
}
```

using adbc::driver::Option;
using adbc::driver::Result;
using adbc::driver::Status;

namespace {

接下来，我们将提供数据库实现。驱动程序框架使用了奇异递归模板模式（CRTP）。其细节由框架处理，但从功能上讲，这仍然只是重写处理细节的基类中的方法。

在此，我们的数据库实现将简单地记录用户传入的 uri。我们将其解读为一个 file:// URI，指向一个目录，我们的 IPC 文件应在此处写入和/或从中读取。这就是 ADBC 中数据库的角色：一个共享的数据库句柄，它可能在连接之间缓存一些共享状态，但仍然允许多个连接并发地针对数据库执行操作。

class DriverExampleDatabase : public adbc::driver::Database<DriverExampleDatabase> {
 public:
  [[maybe_unused]] constexpr static std::string_view kErrorPrefix = "[example]";

  Status SetOptionImpl(std::string_view key, Option value) override {
    // Handle and validate options implemented by this driver
    if (key == "uri") {
      std::string_view uri;
      UNWRAP_RESULT(uri, value.AsString());

      if (uri.find("file://") != 0) {
        return adbc::driver::status::InvalidArgument(
            "[example] uri must start with 'file://'");
      }

      uri_ = uri;
      return adbc::driver::status::Ok();
    }

    // Defer to the base implementation to handle state managed by the base
    // class (and error for all other options).
    return Base::SetOptionImpl(key, value);
  }

  Result<Option> GetOption(std::string_view key) override {
    // Return the value of options implemented by this driver
    if (key == "uri") {
      return Option(uri_);
    }

    // Defer to the base implementation to handle state managed by the base
    // class (and error for all other options).
    return Base::GetOption(key);
  }

  // This is called after zero or more calls to SetOption() on
  Status InitImpl() override {
    if (uri_.empty()) {
      return adbc::driver::status::InvalidArgument(
          "[example] Must set uri to a non-empty value");
    }

    return Base::InitImpl();
  }

  // Getters for members needed by the connection and/or statement:
  const std::string& uri() { return uri_; }

 private:
  std::string uri_;
};

接下来，我们实现连接。虽然数据库的角色通常是存储或缓存信息，但连接的角色是提供获取成本可能较高的资源句柄（例如，连接数据库时协商身份验证）。因为我们的示例“数据库”只是一个目录，所以除了为子语句提供访问数据库 URI 的方法外，我们无需在连接方面进行太多的资源管理。

连接的另一个角色是提供有关表、列、统计信息以及调用者在发出查询前可能想了解的其他目录类信息的元数据。驱动程序框架基类提供了帮助程序来实现这些功能，这样您基本上可以使用 C++17 标准库来实现它们（而不是自己构建 C 级别的数组）。

class DriverExampleConnection : public adbc::driver::Connection<DriverExampleConnection> {
 public:
  [[maybe_unused]] constexpr static std::string_view kErrorPrefix = "[example]";

  // Get information from the database and/or store a reference if needed.
  Status InitImpl(void* parent) {
    auto& database = *reinterpret_cast<DriverExampleDatabase*>(parent);
    uri_ = database.uri();
    return Base::InitImpl(parent);
  }

  // Getters for members needed by the statement:
  const std::string& uri() { return uri_; }

 private:
  std::string uri_;
};

接下来，我们提供语句实现。语句是管理查询执行的地方。由于我们的数据源实际上就是 Arrow 数据，因此我们无需提供管理类型或值转换的层。SQLite 和 PostgreSQL 驱动程序都花费了大量代码行来高效地实现和测试这些转换。nanoarrow 库可用于实现双向转换，这是另一篇文章的主题。

class DriverExampleStatement : public adbc::driver::Statement<DriverExampleStatement> {
 public:
  [[maybe_unused]] constexpr static std::string_view kErrorPrefix = "[example]";

  // Get information from the connection and/or store a reference if needed.
  Status InitImpl(void* parent) {
    auto& connection = *reinterpret_cast<DriverExampleConnection*>(parent);
    uri_ = connection.uri();
    return Base::InitImpl(parent);
  }

  // Our implementation of a bulk ingestion is to write an Arrow IPC stream as a file
  // using the target table as the filename.
  Result<int64_t> ExecuteIngestImpl(IngestState& state) {
    std::string directory = uri_.substr(strlen("file://"));
    std::string filename = directory + "/" + *state.target_table;

    nanoarrow::ipc::UniqueOutputStream output_stream;
    FILE* c_file = std::fopen(filename.c_str(), "wb");
    UNWRAP_ERRNO(Internal, ArrowIpcOutputStreamInitFile(output_stream.get(), c_file,
                                                        /*close_on_release*/ true));

    nanoarrow::ipc::UniqueWriter writer;
    UNWRAP_ERRNO(Internal, ArrowIpcWriterInit(writer.get(), output_stream.get()));

    ArrowError nanoarrow_error;
    ArrowErrorInit(&nanoarrow_error);
    UNWRAP_NANOARROW(nanoarrow_error, Internal,
                     ArrowIpcWriterWriteArrayStream(writer.get(), &bind_parameters_,
                                                    &nanoarrow_error));

    return -1;
  }

  // Our implementation of query execution is to accept a simple query in the form
  // SELECT * FROM (the filename).
  Result<int64_t> ExecuteQueryImpl(QueryState& state, ArrowArrayStream* stream) {
    std::string prefix("SELECT * FROM ");
    if (state.query.find(prefix) != 0) {
      return adbc::driver::status::InvalidArgument(
          "[example] Query must be in the form 'SELECT * FROM filename'");
    }

    std::string directory = uri_.substr(strlen("file://"));
    std::string filename = directory + "/" + state.query.substr(prefix.size());

    nanoarrow::ipc::UniqueInputStream input_stream;
    FILE* c_file = std::fopen(filename.c_str(), "rb");
    UNWRAP_ERRNO(Internal, ArrowIpcInputStreamInitFile(input_stream.get(), c_file,
                                                       /*close_on_release*/ true));

    UNWRAP_ERRNO(Internal,
                 ArrowIpcArrayStreamReaderInit(stream, input_stream.get(), nullptr));
    return -1;
  }

  // This path is taken when the user calls Prepare() first.
  Result<int64_t> ExecuteQueryImpl(PreparedState& state, ArrowArrayStream* stream) {
    QueryState query_state{state.query};
    return ExecuteQueryImpl(query_state, stream);
  }

 private:
  std::string uri_;
};

}  // namespace

最后，我们创建驱动程序初始化函数，这是驱动程序管理器提供 Adbc**() 函数实现所必需的，这些函数构成了 ADBC C API。此函数的名称非常重要：该文件将被编译为名为 libdriver_example.(so|dll|dylib) 的共享库，因此当被要求加载驱动程序 "driver_example" 时，驱动程序管理器将查找符号 AdbcDriverExampleInit() 作为默认入口点。

extern "C" AdbcStatusCode AdbcDriverExampleInit(int version, void* raw_driver,
                                                AdbcError* error) {
  using ExampleDriver =
      adbc::driver::Driver<DriverExampleDatabase, DriverExampleConnection,
                           DriverExampleStatement>;
  return ExampleDriver::Init(version, raw_driver, error);
}

底层测试¶

方案源码： driver_example_test.cc

在我们编写了驱动程序草图之后，下一步是确保它能被驱动程序管理器加载，并且数据库、连接和语句实例可以被初始化和释放。

首先，我们将包含驱动程序管理器和 googletest。

#include "driver_example.h"

#include "arrow-adbc/adbc_driver_manager.h"
#include "gtest/gtest.h"

接下来，我们将为基本生命周期声明一个测试用例。

TEST(DriverExample, TestLifecycle) {
  struct AdbcError error = ADBC_ERROR_INIT;

  struct AdbcDatabase database;
  ASSERT_EQ(AdbcDatabaseNew(&database, &error), ADBC_STATUS_OK);
  AdbcDriverManagerDatabaseSetInitFunc(&database, &AdbcDriverExampleInit, &error);
  ASSERT_EQ(AdbcDatabaseSetOption(&database, "uri", "file://foofy", &error),
            ADBC_STATUS_OK);
  ASSERT_EQ(AdbcDatabaseInit(&database, &error), ADBC_STATUS_OK);

  struct AdbcConnection connection;
  ASSERT_EQ(AdbcConnectionNew(&connection, &error), ADBC_STATUS_OK);
  ASSERT_EQ(AdbcConnectionInit(&connection, &database, &error), ADBC_STATUS_OK);

  struct AdbcStatement statement;
  ASSERT_EQ(AdbcStatementNew(&connection, &statement, &error), ADBC_STATUS_OK);

  ASSERT_EQ(AdbcStatementRelease(&statement, &error), ADBC_STATUS_OK);
  ASSERT_EQ(AdbcConnectionRelease(&connection, &error), ADBC_STATUS_OK);
  ASSERT_EQ(AdbcDatabaseRelease(&database, &error), ADBC_STATUS_OK);

  if (error.release) {
    error.release(&error);
  }
}

存在于 apache/arrow-adbc 仓库中的驱动程序可以使用内置的验证库，该库针对全功能 SQL 数据库实现了一个通用测试套件，并提供了测试一系列输入和输出的实用工具。

高层测试¶

方案源码： driver_example.py

在验证了基本驱动程序功能后，我们可以使用 adbc_driver_manager Python 包的内置 dbapi 实现来公开一个现成的 Python 风格数据库 API。这对于高层测试也非常有用！

首先，我们将导入 pathlib 进行路径计算，并导入 adbc_driver_manager 的 dbapi 模块。

from pathlib import Path

from adbc_driver_manager import dbapi

接下来，我们将定义一个 connect() 函数，该函数使用我们在前一节中使用 cmake 构建的共享库位置来包装 dbapi.connect()。对于本教程，它将位于 CMake build/ 目录中。

def connect(uri: str):
    build_dir = Path(__file__).parent / "build"
    for lib in [
        "libdriver_example.dylib",
        "libdriver_example.so",
        "driver_example.dll",
    ]:
        driver_lib = build_dir / lib
        if driver_lib.exists():
            return dbapi.connect(
                driver=str(driver_lib.resolve()), db_kwargs={"uri": uri}
            )

    # Try to find it on the dynamic loader path
    return dbapi.connect(driver="driver_example", db_kwargs={"uri": uri})

接下来，我们可以尝试运行我们的驱动程序了！我们在驱动程序中实现的两个部分是“批量导入”功能和“全选”功能，让我们看看它是否有效！

if __name__ == "__main__":
    import os

    import pyarrow

    with connect(uri=Path(__file__).parent.as_uri()) as con:
        data = pyarrow.table({"col": [1, 2, 3]})
        with con.cursor() as cur:
            cur.adbc_ingest("example.arrows", data, mode="create")

        with con.cursor() as cur:
            cur.execute("SELECT * FROM example.arrows")
            print(cur.fetchall())
            # Output: [(1,), (2,), (3,)]

        os.unlink(Path(__file__).parent / "example.arrows")

标准输出¶

[(1,), (2,), (3,)]

高层测试也可以使用 adbcdrivermanager 包在 R 中编写。

library(adbcdrivermanager)

drv <- adbc_driver("build/libdriver_example.dylib")
db <- adbc_database_init(drv, uri = paste0("file://", getwd()))
con <- adbc_connection_init(db)

data.frame(col = 1:3) |> write_adbc(con, "example.arrows")
con |> read_adbc("SELECT * FROM example.arrows") |> as.data.frame()
unlink("example.arrows")

驱动程序清单¶

方案源码： driver_example_manifest.py

首先，我们将导入 pathlib 进行路径计算，并导入 adbc_driver_manager 的 dbapi 模块。

from pathlib import Path

from adbc_driver_manager import dbapi

接下来，我们将定义一个 connect() 函数，该函数使用指向我们在前一节中使用 cmake 构建的共享库的 .toml 清单文件位置来包装 dbapi.connect()。在本教程中，它将位于当前目录中。

def connect(uri: str):
    # we can point to the manifest file directly
    manifest_file = Path(".") / "driver_example.toml"
    if manifest_file.exists():
        return dbapi.connect(
            driver=str(manifest_file.resolve()), db_kwargs={"uri": uri}
        )

    # alternatively, it can look for the manifest file in the user's config
    # directory ($HOME/.config/adbc/drivers/driver_example.toml) or the system's
    # config directory (/etc/adbc/drivers/driver_example.toml)
    return dbapi.connect(driver="driver_example", db_kwargs={"uri": uri})

接下来，我们可以尝试运行我们的驱动程序了！我们在驱动程序中实现的两个部分是“批量导入”功能和“全选”功能，让我们看看它是否有效！

if __name__ == "__main__":
    import pyarrow

    with connect(uri=Path(__file__).parent.as_uri()) as con:
        data = pyarrow.table({"col": [1, 2, 3]})
        with con.cursor() as cur:
            cur.adbc_ingest("example.arrows", data, mode="create")

        with con.cursor() as cur:
            cur.execute("SELECT * FROM example.arrows")
            print(cur.fetchall())
            # Output: [(1,), (2,), (3,)]

        (Path(__file__).parent / "example.arrows").unlink()

stdout¶

[(1,), (2,), (3,)]

驱动程序清单可以通过 TOML 文件提供一种安装和管理 ADBC 驱动程序的更简便方法，这些文件描述了一些元数据以及驱动程序共享库的路径。驱动程序管理器可以读取这些清单以定位并动态加载驱动程序。