ADBC 클라이언트

DuckDB의 ADBC(Arrow Database Connectivity) 클라이언트를 직접 써보고 싶다면, libduckdb 아카이브를 플랫폼에 맞게 다운로드하고 아래 설치 안내를 따라가면 돼요. ADBC는 ODBC·JDBC처럼 서로 다른 데이터베이스 사이에서 코드를 그대로 옮겨 쓸 수 있게 해주는 C 스타일 API라서, DBMS에 종속되지 않는 범용 애플리케이션을 만들 때 특히 유용하죠. DuckDB는 Arrow와의 zero-copy 연동 덕분에 데이터를 아주 효율적으로 주고받을 수 있어요.

출처: 문서

본문

Arrow Database Connectivity (ADBC)는 ODBC, JDBC와 마찬가지로 서로 다른 데이터베이스 시스템 사이에서 코드 이식성을 보장하는 C 스타일 API예요. 이를 통해 개발자는 특정 데이터베이스에 종속된 코드를 쓰지 않고도 다양한 데이터베이스 시스템과 통신하는 애플리케이션을 쉽게 만들 수 있어요. ADBC가 ODBC/JDBC와 다른 핵심 차이는 데이터를 주고받을 때 Arrow를 사용한다는 점이에요. DuckDB에는 ADBC 드라이버가 있으며, DuckDB와 Arrow 사이의 zero-copy 통합을 활용해 데이터를 효율적으로 전송해요.

ADBC에 대한 더 자세한 논의와 상세한 API 설명은 ADBC 문서 페이지를 참고해요.

구현된 기능

DuckDB-ADBC 드라이버는 ConnectionReadPartition과 StatementExecutePartitions 함수를 제외한 전체 ADBC 스펙을 구현해요. 이 두 함수는 쿼리 결과를 내부적으로 파티셔닝하는 시스템을 지원하기 위한 것인데, DuckDB에는 해당되지 않아요. 이 섹션에서는 ADBC에 존재하는 주요 함수와 각 함수가 받는 인자, 그리고 예제를 살펴볼게요.

Database

데이터베이스에 대해 동작하는 함수 모음이에요.

함수명	설명	인자	예제
`DatabaseNew`	새(초기화되지 않은) 데이터베이스를 할당	`(AdbcDatabase database, AdbcError error)`	`AdbcDatabaseNew(&adbc_database, &adbc_error)`
`DatabaseSetOption`	char* 옵션 설정	`(AdbcDatabase database, const char key, const char value, AdbcError error)`	`AdbcDatabaseSetOption(&adbc_database, "path", "test.db", &adbc_error)`
`DatabaseInit`	옵션 설정을 마무리하고 데이터베이스 초기화	`(AdbcDatabase database, AdbcError error)`	`AdbcDatabaseInit(&adbc_database, &adbc_error)`
`DatabaseRelease`	데이터베이스 파괴	`(AdbcDatabase database, AdbcError error)`	`AdbcDatabaseRelease(&adbc_database, &adbc_error)`

Database 옵션

옵션	설명
`driver`	DuckDB 공유 라이브러리 경로 (`libduckdb.so`, `libduckdb.dylib`, 또는 `duckdb.dll`).
`entrypoint`	엔트리 포인트 함수명. `duckdb_adbc_init`이어야 함.
`path`	DuckDB 데이터베이스 파일 경로. 설정하지 않으면 메모리 내 데이터베이스가 생성됨.
`uri`	`path`의 대안. 일반 경로 또는 `file:` URI(예: `file:test.db`, `file:///absolute/path.db`) 허용. 둘 다 설정하면 `path`보다 우선함.

Connection

데이터베이스와 상호작용하기 위한 커넥션을 생성·파괴하는 함수 모음이에요.

함수명	설명	인자	예제
`ConnectionNew`	새(초기화되지 않은) 커넥션 할당	`(AdbcConnection, AdbcError)`	`AdbcConnectionNew(&adbc_connection, &adbc_error)`
`ConnectionSetOption`	ConnectionInit 전에 옵션 설정 가능	`(AdbcConnection, const char, const char, AdbcError)`	`AdbcConnectionSetOption(&adbc_connection, ADBC_CONNECTION_OPTION_AUTOCOMMIT, ADBC_OPTION_VALUE_DISABLED, &adbc_error)`
`ConnectionInit`	옵션 설정을 마무리하고 커넥션 초기화	`(AdbcConnection, AdbcDatabase, AdbcError*)`	`AdbcConnectionInit(&adbc_connection, &adbc_database, &adbc_error)`
`ConnectionRelease`	커넥션 파괴	`(AdbcConnection, AdbcError)`	`AdbcConnectionRelease(&adbc_connection, &adbc_error)`

데이터베이스에 대한 메타데이터를 가져오는 함수 모음이에요. 일반적으로 이 함수들은 Arrow 객체, 구체적으로는 ArrowArrayStream을 반환해요.

함수명	설명	인자	예제
`ConnectionGetInfo`	드라이버와 데이터베이스에 대한 메타데이터 가져오기	`(AdbcConnection, const uint32_t, size_t, ArrowArrayStream, AdbcError)`	`AdbcConnectionGetInfo(&adbc_connection, NULL, 0, &arrow_stream, &adbc_error)`
`ConnectionGetObjects`	모든 카탈로그, 데이터베이스 스키마, 테이블, 컬럼의 계층적 뷰 가져오기	`(AdbcConnection, int, const char, const char, const char, const char*, const char, ArrowArrayStream, AdbcError)`	`AdbcDatabaseInit(&adbc_database, &adbc_error)`
`ConnectionGetTableSchema`	테이블의 Arrow 스키마 가져오기	`(AdbcConnection, const char, const char, const char, ArrowSchema, AdbcError)`	`AdbcDatabaseRelease(&adbc_database, &adbc_error)`
`ConnectionGetTableTypes`	데이터베이스의 테이블 타입 목록 가져오기	`(AdbcConnection, ArrowArrayStream, AdbcError*)`	`AdbcDatabaseNew(&adbc_database, &adbc_error)`

ConnectionGetInfo 함수는 다음 정보 코드를 지원해요.

정보 코드	상수	설명
0	`ADBC_INFO_VENDOR_NAME`	데이터베이스 벤더명 (`duckdb`).
1	`ADBC_INFO_VENDOR_VERSION`	데이터베이스 버전.
100	`ADBC_INFO_DRIVER_NAME`	드라이버명.
101	`ADBC_INFO_DRIVER_VERSION`	드라이버 버전.
102	`ADBC_INFO_DRIVER_ARROW_VERSION`	Arrow 라이브러리 버전.
103	`ADBC_INFO_DRIVER_ADBC_VERSION`	드라이버가 지원하는 ADBC 스펙 버전 (정수로, 예: 1.1.0은 `1001000`).

커넥션에 대한 트랜잭션 시맨틱을 지닌 함수 모음이에요. 기본적으로 모든 커넥션은 auto-commit 모드가 켜진 상태로 시작하지만, ConnectionSetOption 함수로 끌 수 있어요.

함수명	설명	인자	예제
`ConnectionCommit`	대기 중인 트랜잭션 커밋	`(AdbcConnection, AdbcError)`	`AdbcConnectionCommit(&adbc_connection, &adbc_error)`
`ConnectionRollback`	대기 중인 트랜잭션 롤백	`(AdbcConnection, AdbcError)`	`AdbcConnectionRollback(&adbc_connection, &adbc_error)`

Statement

Statement는 쿼리 실행과 관련된 상태를 담아요. 일회성 쿼리와 준비된(prepared) 문장을 모두 나타내며, 재사용할 수 있어요. 다만 재사용하면 해당 statement의 이전 결과 집합은 무효화돼요.

statement를 생성·파괴·옵션 설정하는 함수는 다음과 같아요.

함수명	설명	인자	예제
`StatementNew`	주어진 커넥션에 대한 새 statement 생성	`(AdbcConnection, AdbcStatement, AdbcError*)`	`AdbcStatementNew(&adbc_connection, &adbc_statement, &adbc_error)`
`StatementRelease`	statement 파괴	`(AdbcStatement, AdbcError)`	`AdbcStatementRelease(&adbc_statement, &adbc_error)`
`StatementSetOption`	statement에 문자열 옵션 설정	`(AdbcStatement, const char, const char, AdbcError)`	`StatementSetOption(&adbc_statement, ADBC_INGEST_OPTION_TARGET_TABLE, "TABLE_NAME", &adbc_error)`

쿼리 실행과 관련된 함수는 다음과 같아요.

함수명	설명	인자	예제
`StatementSetSqlQuery`	실행할 SQL 쿼리 설정. 이후 `StatementExecuteQuery`로 실행 가능	`(AdbcStatement, const char, AdbcError*)`	`AdbcStatementSetSqlQuery(&adbc_statement, "SELECT * FROM TABLE", &adbc_error)`
`StatementSetSubstraitPlan`	실행할 Substrait 플랜 설정. 이후 `StatementExecuteQuery`로 실행 가능	`(AdbcStatement, const uint8_t, size_t, AdbcError*)`	`AdbcStatementSetSubstraitPlan(&adbc_statement, substrait_plan, length, &adbc_error)`
`StatementExecuteQuery`	statement 실행하고 결과 가져오기	`(AdbcStatement, ArrowArrayStream, int64_t, AdbcError)`	`AdbcStatementExecuteQuery(&adbc_statement, &arrow_stream, &rows_affected, &adbc_error)`
`StatementPrepare`	statement를 여러 번 실행할 준비된(prepared) statement로 변환	`(AdbcStatement, AdbcError)`	`AdbcStatementPrepare(&adbc_statement, &adbc_error)`

바인딩과 관련된 함수로, 벌크 삽입이나 준비된 statement에 사용돼요.

함수명	설명	인자	예제
`StatementBindStream`	Arrow 스트림 바인딩. 벌크 삽입이나 준비된 statement에 사용 가능	`(AdbcStatement, ArrowArrayStream, AdbcError*)`	`StatementBindStream(&adbc_statement, &input_data, &adbc_error)`

수집(Ingestion) 모드

StatementBindStream으로 데이터를 수집할 때, 수집 모드는 ADBC_INGEST_OPTION_MODE로 설정해요.

모드	상수	설명
Create	`ADBC_INGEST_OPTION_MODE_CREATE`	새 테이블 생성 (이미 존재하면 오류). 기본값.
Append	`ADBC_INGEST_OPTION_MODE_APPEND`	기존 테이블에 추가 (테이블이 없으면 오류).
Replace	`ADBC_INGEST_OPTION_MODE_REPLACE`	기존 테이블을 드롭하고 새로 생성.
Create or Append	`ADBC_INGEST_OPTION_MODE_CREATE_APPEND`	테이블이 없으면 새로 만들고, 있으면 추가.

StatementExecuteQuery는 rows_affected 출력 파라미터를 통해 영향받은 행 수를 반환해요.

수집(Ingestion) 옵션

옵션	설명
`adbc.ingest.target_table`	수집 대상 테이블명.
`adbc.ingest.target_catalog`	수집 대상 카탈로그(부착된 데이터베이스). `adbc.ingest.target_db_schema` 없이 설정하면 스키마는 `main`으로 기본 설정됨.
`adbc.ingest.target_db_schema`	수집 대상 스키마.
`adbc.ingest.temporary`	`enabled`로 설정하면 임시 테이블 생성. `target_catalog` 및 `target_db_schema`와 호환되지 않음.

DuckDB ADBC 드라이버 설정

DuckDB를 ADBC 드라이버로 사용하기 전에 libduckdb 공유 라이브러리를 시스템에 설치하고 애플리케이션에서 사용할 수 있도록 해야 해요. 이 라이브러리에는 ADBC 드라이버가 상호작용하는 DuckDB 핵심 엔진이 들어 있어요.

libduckdb 다운로드

DuckDB 릴리즈 페이지에서 플랫폼에 맞는 libduckdb 라이브러리를 다운로드해요.

Linux: libduckdb-linux-amd64.zip (libduckdb.so 포함)
macOS: libduckdb-osx-universal.zip (libduckdb.dylib 포함)
Windows: libduckdb-windows-amd64.zip (duckdb.dll 포함)

아카이브를 압축 해제해 공유 라이브러리 파일을 얻어요.

라이브러리 설치

Linux

다운로드한 아카이브에서 libduckdb.so 파일 압축 해제
코드에서 라이브러리를 사용할 수 있게 준비. 다음 중 하나를 선택해요.
- 시스템 라이브러리 디렉토리에 복사 (루트 권한 필요):
```
sudo cp libduckdb.so /usr/local/lib/
sudo ldconfig
```
- 또는 사용자 정의 디렉토리에 넣고 LD_LIBRARY_PATH에 추가:
```
mkdir -p ~/lib
cp libduckdb.so ~/lib/
export LD_LIBRARY_PATH=~/lib:$LD_LIBRARY_PATH
```

macOS

다운로드한 아카이브에서 libduckdb.dylib 파일 압축 해제
코드에서 라이브러리를 사용할 수 있게 준비. 다음 중 하나를 선택해요.
- 시스템 라이브러리 디렉토리에 복사:
```
sudo cp libduckdb.dylib /usr/local/lib/
```
- 또는 사용자 정의 디렉토리에 넣고 DYLD_LIBRARY_PATH에 추가:
```
mkdir -p ~/lib
cp libduckdb.dylib ~/lib/
export DYLD_LIBRARY_PATH=~/lib:$DYLD_LIBRARY_PATH
```

Windows

다운로드한 아카이브에서 duckdb.dll 파일 압축 해제
다음 위치 중 하나에 배치:
- 애플리케이션 실행 파일과 같은 디렉토리
- PATH 환경 변수에 있는 디렉토리
- Windows 시스템 디렉토리 (예: C:\Windows\System32)

라이브러리 경로 이해하기

LD_LIBRARY_PATH(Linux)와 DYLD_LIBRARY_PATH(macOS)는 런타임에 공유 라이브러리를 어디서 찾을지 시스템에 알려주는 환경 변수예요. 애플리케이션이 libduckdb를 로드하려고 할 때 시스템은 이 경로들을 검색해 라이브러리 파일을 찾아요.

설치 확인

라이브러리가 제대로 설치되고 접근 가능한지 다음과 같이 확인할 수 있어요.

Linux/macOS:

ldd path/to/your/application  # Linux
otool -L path/to/your/application  # macOS

예제

사용하는 프로그래밍 언어와 무관하게, DuckDB에서 ADBC를 활용하려면 두 가지 데이터베이스 옵션이 필요해요. 첫 번째는 DuckDB 라이브러리 경로를 받는 driver이고, 두 번째는 ADBC 함수를 모두 초기화하는 DuckDB-ADBC 드라이버의 내보낸 함수인 entrypoint예요. 이 두 옵션을 설정한 후 선택적으로 path 옵션을 설정해 DuckDB 데이터베이스를 디스크에 저장할 수 있어요. 설정하지 않으면 메모리 내 데이터베이스가 생성돼요. 필요한 옵션을 모두 설정한 뒤 데이터베이스를 초기화하면 돼요. 다양한 언어 환경에서 이렇게 사용할 수 있어요.

C++

C++ 예제는 ADBC로 데이터를 쿼리하기 위한 필수 변수 선언으로 시작해요. 여기에는 Error, Database, Connection, Statement 처리와 DuckDB와 애플리케이션 사이에 데이터를 전송하는 Arrow Stream이 포함돼요.

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

그런 다음 데이터베이스 변수를 초기화해요. 데이터베이스를 초기화하기 전에 위에서 언급한 driver와 entrypoint 옵션을 설정해야 해요. 그 다음 path 옵션을 설정하고 데이터베이스를 초기화해요. driver 옵션은 설치한 libduckdb 라이브러리를 가리켜야 해요.

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);

이제 statement를 초기화하고 커넥션을 통해 쿼리를 실행할 수 있어요. 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 드라이버 매니저를 설치해요. Apache Arrow 형식의 결과 집합(예: fetch_arrow_table 사용)에 직접 접근하려면 pyarrow도 설치해야 해요.

pip install adbc_driver_manager pyarrow

adbc_driver_manager 패키지에 대한 자세한 내용은 adbc_driver_manager 패키지 문서를 참고해요.

C++와 마찬가지로 libduckdb 공유 객체의 위치와 entrypoint 함수로 구성된 초기화 옵션을 제공해야 해요. DuckDB의 path 인자는 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 외에도 fetchone, fetchall 같은 DBApi의 다른 메서드도 커서에서 구현돼 있어요. arrow_streams를 통해 데이터를 수집할 수도 있어요. statement에 옵션을 설정해 데이터 스트림을 바인딩하고 쿼리를 실행하면 돼요.

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 라이브러리를 설치해야 해요. 다음 예제는 메모리 내 DuckDB 데이터베이스를 사용해 메모리 내 Arrow RecordBatch를 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/v18/arrow"
    "github.com/apache/arrow-go/v18/arrow/array"
    "github.com/apache/arrow-go/v18/arrow/ipc"
    "github.com/apache/arrow-go/v18/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 nil, fmt.Errorf("failed to create IPC reader: %w", err)
    }
    defer rdr.Release()

    _, err = adbc.IngestStream(r.ctx, r.conn, rdr, "temp_table", adbc.OptionValueIngestModeCreate, adbc.IngestStreamOptions{})

    return err
}

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]

더 알아보기 (Learn more)

ADBC 공식 문서에서 ADBC 스펙과 API를 더 자세히 살펴볼 수 있어요.