API ADBC

Arrow Database Connectivity (ADBC), аналогично ODBC и JDBC, представляет собой API в стиле C, что обеспечивает переносимость кода между различными системами баз данных. Это позволяет разработчикам без труда создавать приложения, взаимодействующие с системами баз данных, не используя код, специфичный для конкретной системы. Основное различие между ADBC и ODBC/JDBC заключается в том, что ADBC использует Arrow для передачи данных между системой базы данных и приложением. DuckDB имеет драйвер ADBC, который использует преимущества безопасной интеграции DuckDB и Arrow для эффективной передачи данных.

Драйвер ADBC DuckDB в настоящее время поддерживает версию 0.7 ADBC.

Для более подробного обсуждения ADBC и подробного объяснения API, обратитесь к странице документации ADBC.

Реализованные Функциональности

Драйвер DuckDB-ADBC реализует полную спецификацию ADBC, за исключением функций ConnectionReadPartition и StatementExecutePartitions. Обе эти функции предназначены для поддержки систем, которые внутренне разбивают результаты запроса, что не применяется к DuckDB. В этом разделе мы опишем основные функции, которые существуют в ADBC, вместе с аргументами, которые они принимают, и предоставим примеры для каждой функции.

База данных

Набор функций, которые работают с базой данных.

Подключение

Набор функций, которые создают и уничтожают подключение для взаимодействия с базой данных.

Набор функций, которые извлекают метаданные о базе данных. Как правило, эти функции возвращают объекты Arrow, а именно ArrowArrayStream.

Набор функций с семантикой транзакций для подключения. По умолчанию все подключения начинаются в режиме автоматического подтверждения, но это можно отключить с помощью функции ConnectionSetOption.

Запрос

Запросы хранят состояние, связанное с выполнением запроса. Они представляют как одноразовые запросы, так и подготовленные запросы. Они могут быть повторно использованы; однако, это сделает устаревшими предыдущие наборы результатов от данного запроса.

Функции, используемые для создания, уничтожения и настройки параметров запроса:

Функции, связанные с выполнением запроса:

Функции, связанные с привязкой, используемые для массового вставки или в подготовленных запросах.

Примеры

Независимо от используемого языка программирования, для использования ADBC с DuckDB потребуются два параметра базы данных. Первый — driver, который принимает путь к библиотеке DuckDB. Второй параметр — entrypoint, который представляет собой экспортированную функцию из драйвера DuckDB-ADBC, инициализирующую все функции ADBC. После настройки этих двух параметров можно (необязательно) установить параметр path, указав путь на диске для хранения базы данных DuckDB. Если не указано, создается база данных в оперативной памяти. После настройки всех необходимых параметров можно приступить к инициализации базы данных. Ниже приведен способ сделать это в различных средах программирования.

C++

Начнём наш пример на C++ объявлением необходимых переменных для запроса данных через ADBC. Эти переменные включают Error, Database, Connection, обработку Statement и поток Arrow для передачи данных между DuckDB и приложением.

AdbcError adbc_error;
AdbcDatabase adbc_database;
AdbcConnection adbc_connection;
AdbcStatement adbc_statement;
ArrowArrayStream arrow_stream;

Затем мы можем инициализировать нашу переменную базы данных. Перед инициализацией базы данных необходимо установить параметры driver и entrypoint, как указано выше. Затем мы устанавливаем параметр path и инициализируем базу данных. В примере ниже строка "path/to/libduckdb.dylib" должна быть путем к динамической библиотеке DuckDB. Она будет .dylib на macOS и .so на Linux.

AdbcDatabaseNew(&adbc_database, &adbc_error);
AdbcDatabaseSetOption(&adbc_database, "driver", "path/to/libduckdb.dylib", &adbc_error);
AdbcDatabaseSetOption(&adbc_database, "entrypoint", "duckdb_adbc_init", &adbc_error);
// By default, we start an in-memory database, but you can optionally define a path to store it on disk.
AdbcDatabaseSetOption(&adbc_database, "path", "test.db", &adbc_error);
AdbcDatabaseInit(&adbc_database, &adbc_error);

После инициализации базы данных необходимо создать и инициализировать подключение к ней.

AdbcConnectionNew(&adbc_connection, &adbc_error);
AdbcConnectionInit(&adbc_connection, &adbc_database, &adbc_error);

Теперь мы можем инициализировать наше выражение и выполнить запросы через наше соединение. После AdbcStatementExecuteQuery поле arrow_stream заполняется результатом.

AdbcStatementNew(&adbc_connection, &adbc_statement, &adbc_error);
AdbcStatementSetSqlQuery(&adbc_statement, "SELECT 42", &adbc_error);
int64_t rows_affected;
AdbcStatementExecuteQuery(&adbc_statement, &arrow_stream, &rows_affected, &adbc_error);
arrow_stream.release(arrow_stream)

Помимо выполнения запросов, мы также можем вводить данные через arrow_streams. Для этого необходимо установить параметр с именем таблицы, в которую нужно вставить данные, связать поток и затем выполнить запрос.

StatementSetOption(&adbc_statement, ADBC_INGEST_OPTION_TARGET_TABLE, "AnswerToEverything", &adbc_error);
StatementBindStream(&adbc_statement, &arrow_stream, &adbc_error);
StatementExecuteQuery(&adbc_statement, nullptr, nullptr, &adbc_error);

Python

В первую очередь необходимо использовать pip и установить менеджер драйверов ADBC. Вам также потребуется установить pyarrow для прямого доступа к наборам результатов в формате Apache Arrow (например, используя fetch_arrow_table).

pip install adbc_driver_manager pyarrow

Дополнительную информацию о пакете adbc_driver_manager см. в adbc_driver_manager документации по пакету.

Как и в C++, нам необходимо предоставить параметры инициализации, содержащие расположение разделяемой библиотеки libduckdb и функцию входа. Обратите внимание, что аргумент path для DuckDB передаётся через словарь db_kwargs.

import adbc_driver_duckdb.dbapi

with adbc_driver_duckdb.dbapi.connect("test.db") as conn, conn.cursor() as cur:
    cur.execute("SELECT 42")
    # fetch a pyarrow table
    tbl = cur.fetch_arrow_table()
    print(tbl)

Наряду с fetch_arrow_table, на курсоре также реализованы другие методы из DBApi, такие как fetchone и fetchall. Данные также можно вводить через arrow_streams. Нам просто нужно установить параметры в выражении, чтобы связать поток данных и выполнить запрос.

import adbc_driver_duckdb.dbapi
import pyarrow

data = pyarrow.record_batch(
    [[1, 2, 3, 4], ["a", "b", "c", "d"]],
    names = ["ints", "strs"],
)

with adbc_driver_duckdb.dbapi.connect("test.db") as conn, conn.cursor() as cur:
    cur.adbc_ingest("AnswerToEverything", data)

Go

Сначала убедитесь, что загружена библиотека libduckdb (например, .so в Linux, .dylib в Mac или .dll в Windows) со страницы релизов и поместите её в свою LD_LIBRARY_PATH перед запуском кода (в противном случае ошибка объяснит вам варианты относительно расположения этого файла).

Следующий пример использует базу данных DuckDB в оперативной памяти для изменения наборов записей Arrow в оперативной памяти с помощью SQL-запросов:

package main

import (
    "bytes"
    "context"
    "fmt"
    "io"

    "github.com/apache/arrow-adbc/go/adbc"
    "github.com/apache/arrow-adbc/go/adbc/drivermgr"
    "github.com/apache/arrow/go/v17/arrow"
    "github.com/apache/arrow/go/v17/arrow/array"
    "github.com/apache/arrow/go/v17/arrow/ipc"
    "github.com/apache/arrow/go/v17/arrow/memory"
)

func _makeSampleArrowRecord() arrow.Record {
    b := array.NewFloat64Builder(memory.DefaultAllocator)
    b.AppendValues([]float64{1, 2, 3}, nil)
    col := b.NewArray()

    defer col.Release()
    defer b.Release()

    schema := arrow.NewSchema([]arrow.Field{{Name: "column1", Type: arrow.PrimitiveTypes.Float64}}, nil)
    return array.NewRecord(schema, []arrow.Array{col}, int64(col.Len()))
}

type DuckDBSQLRunner struct {
    ctx  context.Context
    conn adbc.Connection
    db   adbc.Database
}

func NewDuckDBSQLRunner(ctx context.Context) (*DuckDBSQLRunner, error) {
    var drv drivermgr.Driver
    db, err := drv.NewDatabase(map[string]string{
        "driver":     "duckdb",
        "entrypoint": "duckdb_adbc_init",
        "path":       ":memory:",
    })
    if err != nil {
        return nil, fmt.Errorf("failed to create new in-memory DuckDB database: %w", err)
    }
    conn, err := db.Open(ctx)
    if err != nil {
        return nil, fmt.Errorf("failed to open connection to new in-memory DuckDB database: %w", err)
    }
    return &DuckDBSQLRunner{ctx: ctx, conn: conn, db: db}, nil
}

func serializeRecord(record arrow.Record) (io.Reader, error) {
    buf := new(bytes.Buffer)
    wr := ipc.NewWriter(buf, ipc.WithSchema(record.Schema()))
    if err := wr.Write(record); err != nil {
        return nil, fmt.Errorf("failed to write record: %w", err)
    }
    if err := wr.Close(); err != nil {
        return nil, fmt.Errorf("failed to close writer: %w", err)
    }
    return buf, nil
}

func (r *DuckDBSQLRunner) importRecord(sr io.Reader) error {
    rdr, err := ipc.NewReader(sr)
    if err != nil {
        return fmt.Errorf("failed to create IPC reader: %w", err)
    }
    defer rdr.Release()
    stmt, err := r.conn.NewStatement()
    if err != nil {
        return fmt.Errorf("failed to create new statement: %w", err)
    }
    if err := stmt.SetOption(adbc.OptionKeyIngestMode, adbc.OptionValueIngestModeCreate); err != nil {
        return fmt.Errorf("failed to set ingest mode: %w", err)
    }
    if err := stmt.SetOption(adbc.OptionKeyIngestTargetTable, "temp_table"); err != nil {
        return fmt.Errorf("failed to set ingest target table: %w", err)
    }
    if err := stmt.BindStream(r.ctx, rdr); err != nil {
        return fmt.Errorf("failed to bind stream: %w", err)
    }
    if _, err := stmt.ExecuteUpdate(r.ctx); err != nil {
        return fmt.Errorf("failed to execute update: %w", err)
    }
    return stmt.Close()
}

func (r *DuckDBSQLRunner) runSQL(sql string) ([]arrow.Record, error) {
    stmt, err := r.conn.NewStatement()
    if err != nil {
        return nil, fmt.Errorf("failed to create new statement: %w", err)
    }
    defer stmt.Close()

    if err := stmt.SetSqlQuery(sql); err != nil {
        return nil, fmt.Errorf("failed to set SQL query: %w", err)
    }
    out, n, err := stmt.ExecuteQuery(r.ctx)
    if err != nil {
        return nil, fmt.Errorf("failed to execute query: %w", err)
    }
    defer out.Release()

    result := make([]arrow.Record, 0, n)
    for out.Next() {
        rec := out.Record()
        rec.Retain() // .Next() will release the record, so we need to retain it
        result = append(result, rec)
    }
    if out.Err() != nil {
        return nil, out.Err()
    }
    return result, nil
}

func (r *DuckDBSQLRunner) RunSQLOnRecord(record arrow.Record, sql string) ([]arrow.Record, error) {
    serializedRecord, err := serializeRecord(record)
    if err != nil {
        return nil, fmt.Errorf("failed to serialize record: %w", err)
    }
    if err := r.importRecord(serializedRecord); err != nil {
        return nil, fmt.Errorf("failed to import record: %w", err)
    }
    result, err := r.runSQL(sql)
    if err != nil {
        return nil, fmt.Errorf("failed to run SQL: %w", err)
    }

    if _, err := r.runSQL("DROP TABLE temp_table"); err != nil {
        return nil, fmt.Errorf("failed to drop temp table after running query: %w", err)
    }
    return result, nil
}

func (r *DuckDBSQLRunner) Close() {
    r.conn.Close()
    r.db.Close()
}

func main() {
    rec := _makeSampleArrowRecord()
    fmt.Println(rec)

    runner, err := NewDuckDBSQLRunner(context.Background())
    if err != nil {
        panic(err)
    }
    defer runner.Close()

    resultRecords, err := runner.RunSQLOnRecord(rec, "SELECT column1+1 FROM temp_table")
    if err != nil {
        panic(err)
    }

    for _, resultRecord := range resultRecords {
        fmt.Println(resultRecord)
        resultRecord.Release()
    }
}

Запуск этого примера даёт следующий результат:

record:
  schema:
  fields: 1
    - column1: type=float64
  rows: 3
  col[0][column1]: [1 2 3]

record:
  schema:
  fields: 1
    - (column1 + 1): type=float64, nullable
  rows: 3
  col[0][(column1 + 1)]: [2 3 4]