ADBC for Vector Search

Applies to VAST Cluster 5.5.

ADBC is the main client for VAST Vector Search. Searches run in the VAST Query Engine, and your application reaches the Query Engine through VAST's ADBC driver, adbc_driver_vastdb. This page covers what's specific to vector search: connecting, the options that matter for searches, bulk ingest of vectors and table names. For the driver in general (installing it, using it through the Python SDK, running queries as an end user), see the VAST DataBase ADBC page.

The driver works on Linux x86_64 from any language that has an ADBC driver manager, for example C++, Rust, Python, Ruby, Go and Java (see VAST Query Engine). These pages use Python.

Run the code blocks in order in one Python session. Below each block is the output you should see.

Before you start

You need a Query Engine endpoint, an access key pair, and an identity or bucket policy with s3:QueryEngine* and s3:Tabular* on the database. Setup has the full list. Admin task: send to your platform team. See Accessing VAST DataBases using the Query Engine.

Install the driver and the driver manager:

pip install adbc-driver-vastdb adbc-driver-manager

Set QE_ENDPOINT, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and VASTDB_BUCKET in your environment, as in setup.

Connect

import os
import adbc_driver_vastdb
from adbc_driver_manager import dbapi

BUCKET = os.environ["VASTDB_BUCKET"]
SCHEMA = f'"{BUCKET}/vs_adbc"'

def connect():
    return dbapi.connect(
        driver=adbc_driver_vastdb.get_driver_path(),
        db_kwargs={
            "vast.db.endpoint": os.environ["QE_ENDPOINT"],
            "vast.db.access_key": os.environ["AWS_ACCESS_KEY_ID"],
            "vast.db.secret_key": os.environ["AWS_SECRET_ACCESS_KEY"],
            # optional: "vast.db.default_bucket": BUCKET,
        },
        autocommit=True,
    )

conn = connect()
with conn.cursor() as cur:
    cur.execute("SELECT 1")
    print(cur.fetchall())
print("autocommit:", conn.adbc_connection.get_option("adbc.connection.autocommit"))

[(1,)]
autocommit: true

The endpoint is a Query Engine endpoint: an HTTP URL on the Query Engine's virtual IP pool, with the cluster's S3 port, such as http://query-engine.example.com:9090. A URL on the wrong port fails with Connection refused (os error 111) (errors).

Use autocommit=True. The driver's default is autocommit on, but dbapi.connect() switches it off unless you pass autocommit=True. With autocommit off, DDL and DML that you don't commit() are lost when the connection closes.

To run searches as an end user, pass vast.db.end_user when you connect, as described in Run queries as an end user. Row and column security then applies to the search as it does to any other query by that user; see security.

Database options

Pass these in db_kwargs:

Connection options

The vector search settings, their formats and what each one controls are listed in search session settings. They apply to every query on the connection.

Session settings and variables can't be set with SQL SET: the Query Engine answers NotImplemented. Use set_options. The driver doesn't check setting names, so a misspelt name is accepted and has no effect.

cur = conn.cursor()
cur.execute(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}")
cur.execute(f"""
CREATE TABLE IF NOT EXISTS {SCHEMA}.docs (
    id INTEGER,
    title VARCHAR,
    embedding VECTOR(FLOAT, 4)
) VECTOR CLUSTER BY embedding (metric = 'l2sq')
""")
cur.execute(f"""
INSERT INTO {SCHEMA}.docs VALUES
    (1, 'a', [0.0, 0.0, 0.0, 0.0]::FLOAT[4]),
    (2, 'b', [1.0, 1.0, 1.0, 1.0]::FLOAT[4]),
    (3, 'c', [2.0, 2.0, 2.0, 2.0]::FLOAT[4])
""")
print(cur.fetchall())

tuned = connect()
tuned.adbc_connection.set_options(**{
    "vast.db.setting.vector_search_skip_recent_non_indexed": "0",
    "vast.db.variable.query_vec": "[0.9, 0.9, 0.9, 0.9]",
})
with tuned.cursor() as tcur:
    tcur.execute(f"""
    SELECT id, array_distance(embedding::FLOAT[4],
                              getvariable('query_vec')::FLOAT[4]) AS dist
    FROM {SCHEMA}.docs
    ORDER BY dist
    LIMIT 2
    """)
    for id_, dist in tcur.fetchall():
        print(id_, round(dist, 4))
tuned.close()

[(3,)]
2 0.2
1 1.8

The query vector here comes from a connection variable, which getvariable() reads in the SQL. A table this small has no indexed rows yet, so the block shows how to set a setting, not its effect.

Statement options

with conn.cursor() as cur:
    cur.adbc_statement.set_options(**{"vast.db.query_timeout_secs": 30})
    cur.execute(f"SELECT count(*) FROM {SCHEMA}.docs")
    print(cur.fetchall())
    query_id = cur.adbc_statement.get_option("vast.db.last_query_id")
    print("got a query ID:", bool(query_id))

[(3,)]
got a query ID: True

Bulk ingest

ADBC bulk ingest appends Arrow data to a table without writing INSERT statements. Use a new cursor for it, set the mode to append, the catalog to vastdb, and the schema to <bucket>/<schema> (without quotes):

import pyarrow as pa

vec4 = pa.list_(pa.field("item", pa.float32(), nullable=False), 4)
cols = pa.schema([("id", pa.int32()), ("title", pa.utf8()), ("embedding", vec4)])
batches = [
    pa.record_batch([pa.array([4], pa.int32()), ["d"], pa.array([[3.0] * 4], vec4)], schema=cols),
    pa.record_batch([pa.array([5], pa.int32()), ["e"], pa.array([[4.0] * 4], vec4)], schema=cols),
]

with conn.cursor() as icur:
    n = icur.adbc_ingest(
        "docs",
        pa.Table.from_batches(batches),
        mode="append",
        catalog_name="vastdb",
        db_schema_name=f"{BUCKET}/vs_adbc",
    )
print("ingested", n)

with conn.cursor() as cur:
    cur.execute(f"SELECT count(*) FROM {SCHEMA}.docs")
    print(cur.fetchall())

ingested 2
[(5,)]

The vector column's Arrow type must be a fixed-size list whose items aren't nullable. With nullable items, the cluster rejects the batch with TabularUnsupportedColumnType ("Unsupported column type on file to import"). A cursor that has already run a query can't be reused for ingest: the driver manager raises Cannot configure ingestion options on a statement configured for query.

A SQL statement can be at most 5 MB (limits). Embeddings written as SQL literals are long, and a multi-row INSERT ... VALUES of them hits a different limit first: the size of the query plan. With 1,024-dimension vectors, statements of well under 5 MB already fail (abridged):

InternalError: INTERNAL: ... Logical plan is too big: 5682939 > 4194304

A statement over 5 MB fails with RequestDataTooLarge ... is larger than max available buffer cmd=START_QUERY. Bulk ingest has no statement to grow, so use it for embeddings. For loading larger volumes, see ingest.

Table names

SELECT ... FROM "bucket/schema".table_name

Quote the bucket and schema path together, then add the table name. A nested schema's full path goes inside the quotes. vast.db.default_bucket doesn't let you leave the bucket out of a query.

Queries per second

To run searches in parallel, open a connection for each concurrent search, rather than sharing one connection between threads. A connection, and every query on it, stays on the virtual IP address you gave it; the driver doesn't spread queries across the pool. To use the whole Query Engine pool, give your connections different addresses from the pool, or use a DNS name that resolves to them (spread the load).

Clean up

with conn.cursor() as cur:
    cur.execute(f"DROP TABLE IF EXISTS {SCHEMA}.docs")
    cur.execute(f"DROP SCHEMA IF EXISTS {SCHEMA}")
conn.close()
print("Dropped vs_adbc")

Dropped vs_adbc

See also

• ADBC in the VAST DataBase docs: installing the driver, transactions with the SDK, end users.
• Search session settings.
• Query design: search shapes that use the index.
• Python SDK: vector_search(), which runs its search through this driver.
• SQL reference.