---
title: "VAST DataBase and Vector Search from a function"
slug: "dev-dataengine-integrations-vast-database-and-vector-search"
description: "Learn to use the VAST DataBase Python SDK with DataEngine to read/write tables, manage transactions, and handle embeddings efficiently."
tags: ["DataEngine"]
updated: 2026-10-30T18:19:01Z
published: 2026-10-30T18:19:01Z
canonical: "docs.vastdata.com/dev-dataengine-integrations-vast-database-and-vector-search"
---

> ## Documentation Index
> Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt
> Use this file to discover all available pages before exploring further.

# 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](/develop/docs/dev-database-integrations-python-sdk) and the [quickstart](/develop/docs/dev-database-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](/develop/docs/dev-vector-search-guides-ingest#insert-arrow-data-with-the-sdk) and the [Vector Search Python SDK page](/develop/docs/dev-vector-search-integrations-python-sdk).

Commands use the environment variables from the [DataEngine quickstart](/develop/docs/dev-dataengine-quickstart#before-you-start). 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](/develop/docs/dev-dataengine-quickstart#before-you-start) 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](/develop/docs/dev-database-operations-setup-prerequisites#permissions) and [S3 keys for functions](/develop/docs/dev-dataengine-operations-security#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](/develop/docs/dev-database-reference-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](/develop/docs/dev-dataengine-reference-events#element-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](/develop/docs/dev-dataengine-concepts-architecture#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](/develop/docs/dev-database-integrations-python-sdk#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](/develop/docs/dev-dataengine-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](/develop/docs/dev-dataengine-guides-write-a-function#what-the-cluster-rejects) 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](/develop/docs/dev-dataengine-concepts-architecture#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](/develop/docs/dev-vector-search-guides-ingest): inserting Arrow data with the SDK, and what the cluster rejects.
- [Create an index](/develop/docs/dev-vector-search-guides-create-an-index) and the [Vector Search Python SDK page](/develop/docs/dev-vector-search-integrations-python-sdk).

## 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](/develop/docs/dev-database-quickstart#7-clean-up)), and delete the secret's file.

## See also

- [Python SDK](/develop/docs/dev-database-integrations-python-sdk): transactions, create-or-get, retries and errors.
- [Write a function](/develop/docs/dev-dataengine-guides-write-a-function#environment-variables-and-secrets): environment variables, secrets and dependencies.
- [Security](/develop/docs/dev-dataengine-operations-security#s3-keys-for-functions): S3 keys for functions.
- [Dead-letter topic](/develop/docs/dev-dataengine-integrations-dead-letter-topic): what happens to events whose handler keeps failing.
- [Integrations](/develop/docs/dev-dataengine-integrations-index)
- [VSS blueprint](https://github.com/vast-data/vss-blueprint) and [Research assistant agent](https://github.com/vast-data/vast-research-assistant-agent): complete applications whose DataEngine functions write embeddings to VAST DataBase (reference applications on GitHub, not verified against these docs).