VAST DataBase and Vector Search from a function

Applies to VAST Cluster 5.5.

A DataEngine function can read and write VAST DataBase tables with the VAST DataBase Python SDK (vastdb), the same SDK you use from any other Python program. This page shows the pattern: add the SDK to the function's dependencies, pass the S3 keys as a secret, connect once in init, and write a row per event in its own transaction. The example writes one row for every object uploaded under a prefix, and then reads the table back from outside DataEngine.

This isn't a built-in DataEngine integration in 5.5. DataEngine has no table output or database connector of its own: your handler calls the SDK, with credentials you give it. The SDK's API (sessions, transactions, schemas, tables, inserts and queries) is documented in the VAST DataBase section, and isn't repeated here: see the Python SDK and the quickstart. A vector is a column type on VAST DataBase, so writing embeddings from a function is the same pattern with a vector column: see ingest and the Vector Search Python SDK page.

Commands use the environment variables from the DataEngine quickstart. The outputs come from a lab cluster running VAST Cluster 5.5.0-SP6 with vastde v5.5.0-sp2 and vastdb 2.1.0, where the registry host is 127.0.0.1:5000 and the trigger's source bucket is docs-src. Your names, GUIDs, addresses and times will differ.

Before you start

You need everything the DataEngine quickstart needs, plus the VAST DataBase side, which your platform team sets up:

• A database to write to: a bucket on a view with the DATABASE protocol, where your function's user can create schemas and tables, or an existing table it can insert into.
• A user for the function with an S3 access key pair and an identity policy that allows the s3:Tabular* actions it needs. See the DataBase setup prerequisites and S3 keys for functions.
• A database endpoint the compute cluster can reach: the address of a VIP pool that serves the view. The function's pods connect to it from the compute cluster, not from your machine, so ask your platform team to confirm the route.

In the lab, the database is the bucket media-db, and the function creates the schema ingest and the table uploads in it the first time it starts.

Add the SDK to the function

The function's requirements.txt holds one line:

vastdb==2.1.0

Pin the SDK version, so every build of the image gets the SDK you tested with. See versions for which SDK versions work with your cluster.

Write the handler

init runs once per function instance, before the instance takes events. It reads the keys from the secret and the endpoint and table names from environment variables, connects, and creates the schema and table if they don't exist yet. vastdb.connect asks the cluster for its version when it connects, so do it once here and keep the session, not once per event. The handler writes one row per event in a transaction of its own:

# dbwriter/main.py
import os
import time

import pyarrow as pa
import vastdb

SCHEMA = pa.schema([
    ("object_key", pa.utf8()),
    ("size", pa.int64()),
    ("etag", pa.utf8()),
    ("event_id", pa.utf8()),
    ("written_at_us", pa.int64()),
])

session = None

def init(ctx):
    global session
    creds = ctx.secrets["db-creds"]
    endpoint = os.environ["DB_ENDPOINT"]
    session = vastdb.connect(
        endpoint=endpoint,
        access=creds["access-key"],
        secret=creds["secret-key"],
    )
    bucket, schema, table = os.environ["DB_BUCKET"], os.environ["DB_SCHEMA"], os.environ["DB_TABLE"]
    with session.transaction() as tx:
        b = tx.bucket(bucket)
        s = b.schema(schema, fail_if_missing=False) or b.create_schema(schema)
        t = s.table(table, fail_if_missing=False) or s.create_table(table, SCHEMA)
    ctx.logger.info("connected to %s, table %s/%s/%s ready", endpoint, bucket, schema, table)

def handler(ctx, event):
    record = event.get_data()["Records"][0]["s3"]["object"]
    row = pa.table({
        "object_key": [record["key"]],
        "size": [record["size"]],
        "etag": [record["eTag"]],
        "event_id": [event.id],
        "written_at_us": [time.time_ns() // 1000],
    }, schema=SCHEMA)
    with session.transaction() as tx:
        tx.bucket(os.environ["DB_BUCKET"]).schema(os.environ["DB_SCHEMA"]).table(os.environ["DB_TABLE"]).insert(row)
    ctx.logger.info("inserted row for %s", record["key"])

The event is an element event; its Records[0].s3.object holds the object's key, size and ETag. See events.

A few things to keep in mind:

• A retried event writes again. If the handler raises after its insert commits, DataEngine delivers the event again, and the row is inserted twice. The event_id column lets you find duplicates later, because a retried event keeps its ID. If you can't tolerate duplicates, check for the event's ID before you insert. See retries and the dead-letter topic.
• Handlers run concurrently. The runtime runs a sync handler in a pool of threads, so several events can use the session at once. Each event here gets its own transaction. The SDK doesn't document whether one session is safe to share between threads; see threads and processes.
• init must succeed. An instance takes events only after init returns, so a wrong endpoint or wrong keys stop the function from handling anything. Check the pipeline's status and logs after every deploy.

Build the image

Build with --save-build-log, to keep the log of what the builder installed:

vastde functions build "$REGISTRY/docs/dbwriter" --target ./dbwriter --image-tag v1 --push --save-build-log

Detected language: python
Validating Python version 3.12.*...
Python version 3.12.* resolved to 3.12.13
Building 127.0.0.1:5000/docs/dbwriter:v1
App Path: ./dbwriter
Handlers File: main.py
2026/10/08 14:03:07 [Started] Python Builder: 127.0.0.1:5000/docs/dbwriter:v1
2026/10/08 14:04:21 [Completed] Python Builder: 127.0.0.1:5000/docs/dbwriter:v1
Build log saved to: dbwriter/build_dbwriter_v1_2026-10-08-14-03-06.log
Build completed: 127.0.0.1:5000/docs/dbwriter:v1

The builder installs your requirements under constraints taken from the DataEngine runtime. They're in the build log:

sed -n '/Runtime dependency constraints applied/,/VAST Constraints Buildpack completed/p' dbwriter/build_dbwriter_v1_*.log

Runtime dependency constraints applied:
aioboto3>=12.0.0; extra == 'all'
aioboto3>=12.0.0; extra == 's3-async'
aiofiles<25.0.0,>=24.1.0
boto3>=1.28.0; extra == 'all'
boto3>=1.28.0; extra == 's3'
cloudevents<2.0.0,>=1.11.0
fastapi-cloudevents<3.0.0,>=2.0.2
fastapi<1.0.0,>=0.115.11
httpx<1.0.0,>=0.28.1
opentelemetry-api==1.39.0
opentelemetry-exporter-otlp==1.39.0
opentelemetry-processor-baggage==0.60b0
opentelemetry-sdk==1.39.0
pydantic<3.0.0,>=2.1.1
pyyaml<7.0.0,>=6.0.3
uvicorn<1.0.0,>=0.34.0
---> VAST Constraints Buildpack completed

vastdb 2.1.0 fits these constraints. If a package you add needs a different OpenTelemetry, FastAPI or pydantic version than these, pip can't satisfy both and the build fails, so add packages one at a time and keep the build log.

The SDK brings PyArrow, NumPy, Ibis and PyIceberg with it. The build installs your requirements and the runtime's in separate steps, so some packages appear twice in the log:

grep -o 'Successfully installed .*' dbwriter/build_dbwriter_v1_*.log | tr ' ' '\n' | grep -E '^(vastdb|pyarrow|numpy|ibis-framework|pyiceberg|protobuf|pydantic|requests|opentelemetry-sdk|opentelemetry-proto)-[0-9]'

ibis-framework-10.8.0
numpy-2.5.3
protobuf-7.36.2
pyarrow-23.0.1
pydantic-2.13.5
pyiceberg-0.12.0
requests-2.34.2
vastdb-2.1.0
opentelemetry-proto-1.39.0
opentelemetry-sdk-1.39.0
protobuf-6.33.6
pydantic-2.13.5
requests-2.34.2

Two versions of protobuf are installed: 7.36.2 for your requirements (vastdb doesn't limit the protobuf version) and 6.33.6 for the runtime's OpenTelemetry packages. In the running function, Python imported protobuf 7.36.2, and the function's logs still reached DataEngine. If your function's traces or metrics go missing after you add the SDK, check the build log for this.

The dependencies make the image bigger. In the lab it was 505 MiB compressed, against 394 MiB for a function without them, which adds to the time a new instance takes to pull the image the first time.

Create the function from the image, as in the quickstart:

vastde functions create --name dbwriter --description "writes a row per object to VAST DataBase" --artifact-type image --artifact-source docs/dbwriter --image-tag v1 --container-registry "$REGISTRY_NAME" --publish

Function created: dbwriter
Name: dbwriter
Description: writes a row per object to VAST DataBase
Tags: []
GUID: 63cb0e42-c52c-4cc5-b0eb-aed106057853
Owner: [id: 4294967295, id-type: vid]
Created At: 2026-10-08T14:04:56Z
Updated At: 2026-10-08T14:04:56Z
VRN: vast:dataengine:functions:dbwriter
Last Revision: 1

And an element trigger for objects under db/:

vastde triggers create element --name db-uploads --source-bucket "$BUCKET" --event "ObjectCreated:*" --name-prefix db/

Trigger created: db-uploads
Name: db-uploads
Description: N/A
Tags: []
GUID: ab813477-9369-4042-ab52-659afa078f5f
Owner: {4294967295 vid 0xc000146e00 map[]}
CreatedAt: 2026-10-08T14:05:07Z
UpdatedAt: 2026-10-08T14:05:07Z
VRN: vast:dataengine:triggers:db-uploads
Type: Element
Events:
  -: ObjectCreated:*
Topic: vast:dataengine:topics:docs-kafka/default
SourceBucket: docs-src

Pass the credentials

The S3 keys go in a secret; the endpoint and the table's location go in environment variables, which aren't secret.

Keep the secret file out of Git. One way is to keep a template in Git and fill it from your shell or CI secrets store, with permissions only you can read:

# db-creds.yaml.tmpl
name: db-creds
kubernetes_cluster_vrn: vast:dataengine:kubernetes-clusters:test-cluster
namespace: dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91
entries:
  - key: access-key
    value: ${S3_AK}
  - key: secret-key
    value: ${S3_SK}

export S3_AK="<function user's access key>" S3_SK="<function user's secret key>"
(umask 077; envsubst < db-creds.yaml.tmpl > db-creds.yaml)

This prints nothing. db-creds.yaml is readable only by you; delete it when you no longer need it.

Don't name the environment variables VAST_ENDPOINT, VAST_ACCESS_KEY and so on: the VAST_ prefix is reserved, and the cluster rejects the pipeline. A pipeline file that set VAST_ENDPOINT on the function deployment:

# not-run
        environment_variables:
          - name: VAST_ENDPOINT
            value: http://<DB endpoint>

vastde pipelines create --config @reserved.yaml --secret-file db-creds.yaml

Creating secret: db-creds
Secret created: db-creds
pipeline creation failed: The following environment variable names are reserved for internal use and cannot be set by users: VAST_ENDPOINT. Reserved prefixes: VASTPIPELINE_, K_, VAST_. Reserved names: HOSTNAME, PERFORMANCE, PORT, VAST_RUNTIME_HANDLER_THREADS

The CLI creates the secret before it sends the pipeline, so the secret exists even though the pipeline doesn't. The list of reserved names is in write a function. Use a prefix of your own, such as DB_.

Create and deploy the pipeline

Save this as uploads-to-db.yaml, with your compute cluster, namespace and database endpoint:

name: uploads-to-db
description: write a row per uploaded object to a VAST DataBase table
kubernetes_cluster_vrn: vast:dataengine:kubernetes-clusters:test-cluster
namespace: dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91
manifest:
  triggers:
    - name: uploads
      vrn: vast:dataengine:triggers:db-uploads
  function_deployments:
    - name: dbwriter
      function_vrn: vast:dataengine:functions:dbwriter
      config:
        environment_variables:
          - name: DB_ENDPOINT
            value: http://<DB endpoint>
          - name: DB_BUCKET
            value: media-db
          - name: DB_SCHEMA
            value: ingest
          - name: DB_TABLE
            value: uploads
        secrets:
          - db-creds
  links:
    - source: [uploads]
      destination: [dbwriter]

vastde pipelines create --config @uploads-to-db.yaml --secret-file db-creds.yaml --deploy

Creating secret: db-creds
Secret created: db-creds
Pipeline created: uploads-to-db
Name: uploads-to-db
Description: write a row per uploaded object to a VAST DataBase table
Tags: []
GUID: 6428ee26-91ef-4134-9e96-2b74b47eeed3
Owner: [id: 4294967295, id-type: vid]
Created At: 2026-10-08T14:06:24Z
Updated At: 2026-10-08T14:06:24Z
VRN: vast:dataengine:pipelines:uploads-to-db

Deploying pipeline: uploads-to-db
Pipeline deployed successfully

Wait for it to be ready:

until s=$(vastde pipelines get uploads-to-db -o json | jq -r .status); [ "$s" = Ready ] || [ "$s" = Error ]; do sleep 5; done; vastde pipelines get uploads-to-db -o json | jq -c "{status, reason}"

{"status":"Ready","reason":"Pipeline is ready and active."}

The function's start-up logs show init connecting and creating the schema and table. The vastdb.* lines are the SDK's own log messages:

vastde logs get uploads-to-db --since 5m

2026-10-08 14:07:01.57 [dbwriter] [INFO]  [vast-runtime] Print function intercepted and redirected to 'user-print' logger
2026-10-08 14:07:01.95 [dbwriter] [INFO]  [vast-runtime] Runtime is listening on port 8080
2026-10-08 14:07:01.95 [dbwriter] [INFO]  [vast-runtime] Sync handler detected, concurrency limited to 20 via middleware
2026-10-08 14:07:02.17 [dbwriter] [INFO]  [vast-runtime] ConcurrencyLimitMiddleware initialized with max_concurrent=20
2026-10-08 14:07:02.17 [dbwriter] [INFO]  [vast-runtime] Configured private thread pool executor with 20 workers
2026-10-08 14:07:02.94 [dbwriter] [INFO]  [vast-runtime] Initialized HTTP client with HTTP/2 support for CloudEvent sending
2026-10-08 14:07:02.94 [dbwriter] [INFO]  [vast-runtime] Call Init Handler: init
2026-10-08 14:07:02.96 [dbwriter] [INFO]  [user] vastdb.features: VAST version: (5, 5, 0, 1)
2026-10-08 14:07:03.53 [dbwriter] [INFO]  [user] vastdb.schema: Created schema: ingest
2026-10-08 14:07:04.14 [dbwriter] [INFO]  [user] vastdb.schema: Created table: uploads
2026-10-08 14:07:04.51 [dbwriter] [INFO]  [user] connected to http://<DB endpoint>, table media-db/ingest/uploads ready
2026-10-08 14:07:04.51 [dbwriter] [INFO]  [vast-runtime] Init Handler Completed, duration: 0:00:01.564103
2026-10-08 14:07:04.51 [dbwriter] [INFO]  [vast-runtime] Handler is now ready to accept requests

Run it

Upload two objects under db/:

echo "hello database" > db-1.txt; aws --endpoint-url "$S3_ENDPOINT" s3 cp db-1.txt "s3://$BUCKET/db/db-1.txt" >/dev/null
echo "much longer second object for the table" > db-2.txt; aws --endpoint-url "$S3_ENDPOINT" s3 cp db-2.txt "s3://$BUCKET/db/db-2.txt" >/dev/null
sleep 10; vastde logs get uploads-to-db --since 1m --scope user

2026-10-08 14:07:04.14 [dbwriter] [INFO]  [user] vastdb.schema: Created table: uploads
2026-10-08 14:07:04.51 [dbwriter] [INFO]  [user] connected to http://<DB endpoint>, table media-db/ingest/uploads ready
2026-10-08 14:07:51.46 [dbwriter] [INFO]  [user] inserted row for db/db-2.txt
2026-10-08 14:07:52.00 [dbwriter] [INFO]  [user] inserted row for db/db-1.txt

Here the second upload was handled first. See ordering for when the order of events is kept.

Read the table back

The table is an ordinary VAST DataBase table, so anything that reads tables can read it: the SDK, the Query Engine, Trino or Spark. This script reads it with the SDK:

# readback.py   (env DB_ENDPOINT, DB_BUCKET, DB_SCHEMA, DB_TABLE, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
import os
import vastdb

session = vastdb.connect(endpoint=os.environ["DB_ENDPOINT"],
                         access=os.environ["AWS_ACCESS_KEY_ID"], secret=os.environ["AWS_SECRET_ACCESS_KEY"])
with session.transaction() as tx:
    t = tx.bucket(os.environ["DB_BUCKET"]).schema(os.environ["DB_SCHEMA"]).table(os.environ["DB_TABLE"])
    for row in sorted(t.select().read_all().to_pylist(), key=lambda r: r["written_at_us"]):
        print(row)

Run python readback.py wherever vastdb is installed. If your machine doesn't have a Python the SDK supports, the function's image already has one with the SDK in it. Here the script runs in the image, with the image's launcher, which sets up its Python environment. Your machine needs a route to the database endpoint:

export DB_ENDPOINT="http://<DB endpoint>"
docker run --rm --network host -v $PWD/readback.py:/tmp/readback.py:ro \
  -e DB_ENDPOINT="$DB_ENDPOINT" -e DB_BUCKET=media-db -e DB_SCHEMA=ingest -e DB_TABLE=uploads \
  -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY \
  --entrypoint /cnb/lifecycle/launcher "$REGISTRY/docs/dbwriter:v1" python /tmp/readback.py

{'object_key': 'db/db-2.txt', 'size': 40, 'etag': '24f36adb24aca85675b7044de117f9fe', 'event_id': '87982405058577', 'written_at_us': 1791468471044994}
{'object_key': 'db/db-1.txt', 'size': 15, 'etag': '3baf1f01389d8743d04ebdaa871245b0', 'event_id': '88051124535314', 'written_at_us': 1791468471593661}

Write embeddings

To write embeddings instead of (or as well as) metadata, compute the embedding in the handler and insert it into a vector column, in the same kind of transaction. The table, its vector column and its index are designed and created as in any other Vector Search application; create the table up front rather than from init if it needs an index. See:

• Ingest: inserting Arrow data with the SDK, and what the cluster rejects.
• Create an index and the Vector Search Python SDK page.

Clean up

# not-run
vastde pipelines delete uploads-to-db
vastde triggers delete db-uploads
vastde functions delete dbwriter

Drop the schema and table with the SDK (see the DataBase quickstart), and delete the secret's file.

See also

• Python SDK: transactions, create-or-get, retries and errors.
• Write a function: environment variables, secrets and dependencies.
• Security: S3 keys for functions.
• Dead-letter topic: what happens to events whose handler keeps failing.
• Integrations
• VSS blueprint and Research assistant agent: complete applications whose DataEngine functions write embeddings to VAST DataBase (reference applications on GitHub, not verified against these docs).