Ingest vectors

Applies to VAST Cluster 5.5. Examples use the VAST DataBase Python SDK 2.1.0.

This guide shows the ways to load embeddings into a vector-indexed VAST DataBase table, what the cluster rejects, how to change a row's embedding, and what to expect while the index catches up with new rows.

Create the table, with its index, before you load the data. You can't add a vector index to a table that already exists (5.5.0 release notes). See create a vector index.

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

Choose a method

The DB guide Ingest covers batch sizes, Parquet staging and import settings in detail. Everything there applies to vector-indexed tables too. This page covers what's specific to vectors.

Before you start

You need the environment variables from setup (see create a vector index for the list), and for the Parquet import, S3 write access to a bucket on the cluster and s3:TabularImportData (setup prerequisites). This guide uses boto3 and writes the files to the database bucket.

Connect, and create a cosine-indexed table with a 4-dimension vector:

import io
import os
import re

import adbc_driver_manager.dbapi as dbapi
import adbc_driver_vastdb
import boto3
import pyarrow as pa
import pyarrow.parquet as pq
import vastdb
from vastdb import errors

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

conn = 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"],
    },
    autocommit=True,
)

def sql(query):
    with conn.cursor() as cur:
        cur.execute(query)
        return cur.fetchall()

session = vastdb.connect()
s3 = boto3.client("s3", endpoint_url=os.environ["AWS_S3_ENDPOINT_URL"])

sql(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}")
sql(f"""CREATE TABLE IF NOT EXISTS {SCHEMA}.docs (
          id BIGINT, model VARCHAR, title VARCHAR, embedding VECTOR(FLOAT, 4))
        VECTOR CLUSTER BY embedding (metric = 'cosine')""")
print("ready")

ready

Insert with SQL

Write each vector as a list literal cast to FLOAT[<d>]:

print(sql(f"""INSERT INTO {SCHEMA}.docs VALUES
    (1, 'demo-embed-v1', 'red apple',   [0.9, 0.1, 0.0, 0.0]::FLOAT[4]),
    (2, 'demo-embed-v1', 'green apple', [0.8, 0.3, 0.0, 0.0]::FLOAT[4])"""))

[(2,)]

The result is the number of rows inserted.

A SQL statement can be at most 5 MB (5.5.0 release notes). Embeddings written as literals are long, and a multi-row INSERT ... VALUES of them reaches a different limit first: the query plan's size. Such a statement, well under 5 MB, fails with Logical plan is too big: <size> > 4194304; one over 5 MB fails with RequestDataTooLarge (errors). For large batches, use the SDK or ADBC bulk ingest.

Insert Arrow data with the SDK

The vector column's Arrow type must be a fixed-size list whose items aren't nullable. Build the table with an explicit schema:

DIM = 4
arrow_schema = pa.schema([
    ("id", pa.int64()),
    ("model", pa.utf8()),
    ("title", pa.utf8()),
    ("embedding", pa.list_(pa.field("item", pa.float32(), nullable=False), DIM)),
])
batch = pa.table({
    "id": [3, 4],
    "model": ["demo-embed-v1"] * 2,
    "title": ["banana", "cherry"],
    "embedding": [[0.1, 0.9, 0.1, 0.0], [0.7, 0.0, 0.6, 0.0]],
}, schema=arrow_schema)

with session.transaction() as tx:
    tx.bucket(BUCKET).schema(SCHEMA_NAME).table("docs").insert(batch)
print(sql(f"SELECT count(*) FROM {SCHEMA}.docs"))

[(4,)]

Append with ADBC bulk ingest

Any ADBC client can append Arrow data with the driver's bulk ingest. Pass the table name, and the bucket and schema path as db_schema_name. Use a cursor only for the ingest: a cursor that has run an ingest refuses SQL afterwards, with Cannot configure query options on a statement configured for ingestion.

more = pa.table({
    "id": [5], "model": ["demo-embed-v1"], "title": ["plum"],
    "embedding": [[0.5, 0.0, 0.0, 0.8]],
}, schema=arrow_schema)

with conn.cursor() as cur:
    cur.adbc_ingest("docs", more, mode="append", db_schema_name=f"{BUCKET}/{SCHEMA_NAME}")
print(sql(f"SELECT count(*) FROM {SCHEMA}.docs"))

[(5,)]

Prepared statements with bound parameters (cursor.executemany()) aren't available: the driver returns NOT_IMPLEMENTED: AdbcStatementPrepare.

Import Parquet files

import_files() works on a vector-indexed table. In the Parquet file, the vector column has the same Arrow type as for insert():

buf = io.BytesIO()
pq.write_table(pa.table({
    "id": [6, 7], "model": ["demo-embed-v1"] * 2, "title": ["lemon", "lime"],
    "embedding": [[0.2, 0.2, 0.9, 0.1], [0.1, 0.4, 0.8, 0.1]],
}, schema=arrow_schema), buf)
s3.put_object(Bucket=BUCKET, Key="staging/vs_guide_ingest/part-1.parquet", Body=buf.getvalue())

with session.transaction() as tx:
    tx.bucket(BUCKET).schema(SCHEMA_NAME).table("docs").import_files(
        [f"/{BUCKET}/staging/vs_guide_ingest/part-1.parquet"])
print(sql(f"SELECT count(*) FROM {SCHEMA}.docs"))

[(7,)]

For large imports, pass all the files in one call, and tune ImportConfig. import_concurrency sets the number of client threads. By default, import_files() spreads the files across the endpoints the cluster reports for the table (SDK 2.1.0): the virtual IPs of the pool that the session's own endpoint belongs to. So a session opened on the Query Engine endpoint imports through the Query Engine pool. To keep loading off the pool that serves your searches, open the import session on the database endpoint, or pass data_endpoints with protocol endpoints from your platform team.

import_files() of a file that's already been imported into the table can add no rows and raise no error. Don't rely on that either way: track which files you've loaded.

from vastdb.config import ImportConfig

table.import_files(paths, config=ImportConfig(
    import_concurrency=16,
    data_endpoints=["http://<protocol-vip-1>:9090", "http://<protocol-vip-2>:9090"],
))

See import settings in the DB guide.

What the cluster rejects

The cluster checks the vector column on every insert. These are the messages for the common mistakes:

def sdk_insert(rows, schema=arrow_schema):
    try:
        with session.transaction() as tx:
            tx.bucket(BUCKET).schema(SCHEMA_NAME).table("docs").insert(pa.table(rows, schema=schema))
        return "OK"
    except errors.BadRequest as e:
        return f"{type(e).__name__}: {e.code}: {re.sub(r'^.*?]: ', '', e.message)}"
    except errors.NotSupportedSchema as e:
        return f"{type(e).__name__}: {e.cause.code}"

row = {"id": [90], "model": ["demo-embed-v1"], "title": ["bad"]}
nullable_items = pa.schema(list(arrow_schema)[:3] + [("embedding", pa.list_(pa.float32(), 4))])
five_dims = pa.schema(list(arrow_schema)[:3] + [("embedding", pa.list_(pa.field("item", pa.float32(), nullable=False), 5))])

print("nullable item type:", sdk_insert({**row, "embedding": [[1, 0, 0, 0]]}, nullable_items))
print("wrong dimension:   ", sdk_insert({**row, "embedding": [[1, 0, 0, 0, 0]]}, five_dims))
print("null vector:       ", sdk_insert({**row, "embedding": [None]}))

nullable item type: NotSupportedSchema: TabularUnsupportedColumnType
wrong dimension:    BadRequest: TabularMismatchColumnType: Mismatched column type between file to import and existing table.
null vector:        BadRequest: InvalidArgument: Vector index column contains null values

With SQL, a vector of the wrong length fails when it's cast to the column type, before it reaches the table. But a vector with a null element is accepted:

def try_sql(query):
    try:
        return sql(query)
    except Exception as e:
        msg = str(e).replace('\\"', '"')
        m = re.search(r'failed with: (.*?)", request_id', msg, re.S) or re.search(r"internal error: ([^)]*)\)", msg)
        return m.group(1) if m else msg

print("wrong dimension:", try_sql(f"INSERT INTO {SCHEMA}.docs VALUES (91, 'demo-embed-v1', 'bad', [1, 0, 0]::FLOAT[3])"))
print("null element:   ", try_sql(f"INSERT INTO {SCHEMA}.docs VALUES (92, 'demo-embed-v1', 'bad', [1, 0, 0, NULL]::FLOAT[4])"))
sql(f"DELETE FROM {SCHEMA}.docs WHERE id = 92")

wrong dimension: Conversion Error: Cannot cast array of size 3 to array of size 4
null element:    [(1,)]

The null element is stored as 0.0, so the row ranks as if that element were zero.

A vector containing NaN or infinity is rejected. Through SQL the message is only internal error: Invalid Argument; the SDK's insert() says vector index column contains non-finite values (has_nan=1 has_inf=0) (errors).

Check embeddings in your pipeline before you load them: right dimension, no nulls, no NaNs.

Keep the embedding contract

Every vector in an indexed column has to come from the same embedding setup, and the query vectors too:

• Use the same model, model version, dimension and normalization for ingest and for queries.
• Record the model name and version, in a metadata column (like model above) or in your deployment configuration.
• Don't mix models in one indexed column. Distances between vectors from different models mean nothing.
• For cosine and ip, many pipelines L2-normalize the vectors. With normalized vectors, the inner product equals the cosine similarity. The cluster doesn't normalize for you, and the two metrics differ on vectors that aren't normalized: cosine treats [1, 0] and [10, 0] as the same direction, while ip ranks [10, 0] above [1, 0].

Load data that clusters well

The index groups similar vectors into clusters, so how well it works depends on your data. Uniform random floats have no structure to find, so they make a poor benchmark for approximate nearest-neighbour search. Test with real embeddings, or synthetic data with clear clusters.

After a load

New rows are searchable straight away, with nearest-neighbour search and with ordinary filters. Updating the index with them can take some time; see the 5.5.0 release notes.

q = "[0.85, 0.2, 0.0, 0.0]::FLOAT[4]"
for row in sql(f"""SELECT id, title, round(array_cosine_distance(embedding::FLOAT[4], {q})::DOUBLE, 4) AS distance
                   FROM {SCHEMA}.docs ORDER BY distance LIMIT 3"""):
    print(row)

(1, 'red apple', 0.0072)
(2, 'green apple', 0.0081)
(4, 'cherry', 0.2609)

Rows that the index hasn't absorbed yet are still searched: the Query Engine compares the query vector with each of them, as well as searching the index. So a large backlog of new rows makes searches slower until the cluster has indexed them. After a large load, expect queries to get faster as the index catches up. The vector index explains this lifecycle, and performance shows how to check how far the index has got.

Change a row's embedding

You can't update the vector column in place. Updating other columns, and deleting rows, works:

print("update title:    ", try_sql(f"UPDATE {SCHEMA}.docs SET title = 'red apple (fresh)' WHERE id = 1"))
print("update embedding:", try_sql(f"UPDATE {SCHEMA}.docs SET embedding = [0.0, 0.0, 1.0, 0.0]::FLOAT[4] WHERE id = 2"))

update title:     [(1,)]
update embedding: Invalid Argument

To change an embedding, delete the row and insert it again with the new vector. The new row is then placed in the index like any other new row:

sql(f"DELETE FROM {SCHEMA}.docs WHERE id = 2")
sql(f"INSERT INTO {SCHEMA}.docs VALUES (2, 'demo-embed-v1', 'green apple', [0.0, 0.0, 1.0, 0.0]::FLOAT[4])")
print(sql(f"SELECT id, title FROM {SCHEMA}.docs WHERE id IN (1, 2) ORDER BY id LIMIT 10"))

[(1, 'red apple (fresh)'), (2, 'green apple')]

Do the delete and the insert in one transaction if readers mustn't see the row missing. With the SDK, that's one session.transaction() block; see updates and deletes.

Clean up

s3.delete_object(Bucket=BUCKET, Key="staging/vs_guide_ingest/part-1.parquet")
sql(f"DROP TABLE IF EXISTS {SCHEMA}.docs")
sql(f"DROP SCHEMA IF EXISTS {SCHEMA}")
conn.close()
print("cleaned up")

cleaned up

Next steps

• Query design: write searches that use the index.
• Measure recall and latency: time searches and measure recall.
• RAG retrieval: chunks, embeddings and permissions for retrieval-augmented generation.