Ingest

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

This guide shows the ways to load data into VAST DataBase tables: inserting Arrow data with the Python SDK, sizing your batches, importing Parquet files that are already on the cluster, and loading partitioned tables. It also explains what the SDK does when a batch is too big for one request, and ends with loading through Trino, Spark and the VAST Query Engine.

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

Choose a method

Only import_files() keeps the data off your client.

For JSON, CSV and Parquet files, including fixing the types PyArrow guesses and streaming big files, see load files. For NetCDF and GRIB2 arrays, see load scientific data.

Tables can also be populated synchronously from event topics; see blob expansion in the VAST Event Broker docs.

Before you start

You need the connection settings from the quickstart in your environment. To import Parquet files you also need:

• s3:TabularImportData on the database, on top of the actions for opening a table. See actions each SDK operation needs.
• S3 write access to a bucket on the cluster, to put the files there. This guide uses boto3 (pip install boto3) and writes to the database bucket itself.

Connect, and create the schema for this guide:

import io
import os
import tempfile

import boto3
import pyarrow as pa
import pyarrow.parquet as pq
import vastdb
from vastdb import errors

session = vastdb.connect()
BUCKET = os.environ["VASTDB_BUCKET"]
SCHEMA = "guide_ingest"

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

with session.transaction() as tx:
    tx.bucket(BUCKET).create_schema(SCHEMA, fail_if_exists=False)
    print("Schema ready:", SCHEMA)

Schema ready: guide_ingest

Insert Arrow data

insert() takes a pyarrow.Table or pyarrow.RecordBatch whose column names match the table's. It returns the row IDs the database assigned. Every insert runs in a transaction, and nothing is visible to other transactions until it commits.

events_schema = pa.schema([
    ("id", pa.int64()),
    ("device", pa.utf8()),
    ("value", pa.float64()),
])

def make_events(start, count):
    ids = range(start, start + count)
    return pa.table({
        "id": list(ids),
        "device": [f"dev-{i % 10}" for i in ids],
        "value": [i * 0.5 for i in ids],
    }, schema=events_schema)

with session.transaction() as tx:
    events = tx.bucket(BUCKET).schema(SCHEMA).create_table("events", events_schema, fail_if_exists=False)
    row_ids = events.insert(make_events(0, 1000))
    print("Inserted", len(row_ids), "rows")

Inserted 1000 rows

What the SDK does with a big batch

Each request the SDK sends to the cluster is limited to 5 MiB (a client-side constant in SDK 2.1.0). You don't have to keep your batches under that size yourself. For each insert() call, the SDK:

1. Combines a pa.Table into a single record batch. This copies the data in memory.
2. Sends the batch as one request if it fits.
3. If it doesn't fit, inserts as many whole columns as fit, splitting the rows across several requests if it has to. Columns you can't update later (sort keys, partition columns and the vector column) always go into this first step. This step returns the row IDs.
4. Writes the remaining columns as updates against those row IDs, again splitting by rows if needed.

All of these requests belong to your transaction, so they commit or roll back together. See insert in the SDK reference.

So a batch that's too large still works, but it costs more requests than the same data sent as right-sized batches, and step 1 holds a second copy of the data in memory.

Size your batches

Aim for batches that fit in one request. You can measure a batch with pa.ipc.get_record_batch_size(), which is the estimate the SDK itself uses. This example writes 300,000 rows to a local Parquet file, then streams the file into the table in batches of 100,000 rows, all in one transaction:

path = os.path.join(tempfile.mkdtemp(), "events.parquet")
pq.write_table(make_events(1000, 300_000), path)

with session.transaction() as tx:
    events = tx.bucket(BUCKET).schema(SCHEMA).table("events")
    for batch in pq.ParquetFile(path).iter_batches(batch_size=100_000):
        size_mib = pa.ipc.get_record_batch_size(batch) / 2**20
        events.insert(batch)
        print(f"{batch.num_rows} rows, {size_mib:.2f} MiB")

100000 rows, 2.38 MiB
100000 rows, 2.38 MiB
100000 rows, 2.38 MiB

Guidelines:

• Fewer, larger calls beat many small ones. Every insert() call is at least one request. Inserting row by row, or a few rows at a time, spends most of the time on request overhead.
• Stay under about 5 MiB per batch when you can, so that each insert() is one request. Wider rows mean fewer rows per batch.
• Stream large inputs. Read files with iter_batches(), or from a RecordBatchReader, instead of loading a whole dataset and passing one huge pa.Table.
• Decide how much goes in one transaction. One transaction for a whole load gives you all-or-nothing behaviour. One transaction per batch, or per few batches, means a failure only loses the current batch, but a retry has to know where to restart. See transactions in practice.

Rows that are too big for one request

If a single row can't fit in one request even after the SDK splits its columns, insert() raises errors.TooWideRow. A single cell is limited to 5 MB anyway; see limits.

The SDK raises TooWideRow while it's splitting the work, which can be after it has written the row's first columns. If you catch the exception inside the transaction and let the transaction commit, you commit a partial row:

notes_schema = pa.schema([("id", pa.int64()), ("text", pa.utf8())])
too_wide = pa.table({"id": [1], "text": ["x" * (6 * 2**20)]}, schema=notes_schema)

# Don't do this: the exception is caught inside the transaction, which then commits.
with session.transaction() as tx:
    notes = tx.bucket(BUCKET).schema(SCHEMA).create_table("notes", notes_schema, fail_if_exists=False)
    try:
        notes.insert(too_wide)
    except errors.TooWideRow as e:
        print("Rejected:", e)

with session.transaction() as tx:
    notes = tx.bucket(BUCKET).schema(SCHEMA).table("notes")
    print(notes.select().read_all().to_pylist())

Rejected: One of the rows in the given batch is too big for an rpc
[{'id': 1, 'text': None}]

The id was committed without its text. Catch the exception outside the transaction instead, so that the transaction rolls back:

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).schema(SCHEMA)
    schema.table("notes").drop()
    schema.create_table("notes", notes_schema)

try:
    with session.transaction() as tx:
        tx.bucket(BUCKET).schema(SCHEMA).table("notes").insert(too_wide)
except errors.TooWideRow as e:
    print("Rejected and rolled back:", e)

with session.transaction() as tx:
    print("Rows in notes:", tx.bucket(BUCKET).schema(SCHEMA).table("notes").select().read_all().num_rows)

Rejected and rolled back: One of the rows in the given batch is too big for an rpc
Rows in notes: 0

Store large objects as files, and keep their paths in the table.

Import Parquet files

import_files() has the cluster read Parquet files that are already stored in a bucket on the same VAST cluster. The data doesn't pass through your client, so this is the fastest way to load data that's already in files.

Stage the files

Put the files in a bucket with any S3 client. The paths you pass to import_files() have the form /<bucket>/<key>.

Use a key prefix that isn't the name of a schema in the same bucket. In our tests on 5.5.0-SP3, an S3 PUT to a key that started with the name of an existing schema (for example guide_ingest/readings.parquet) was refused with AccessDenied. This guide uses staging/.

readings_schema = pa.schema([
    ("station", pa.utf8()),
    ("day", pa.int32()),
    ("rain_mm", pa.float64()),
])

def stage(key, table):
    buf = io.BytesIO()
    pq.write_table(table, buf)
    s3.put_object(Bucket=BUCKET, Key=key, Body=buf.getvalue())
    return f"/{BUCKET}/{key}"

files = [
    stage("staging/guide_ingest/readings-1.parquet", pa.table({
        "station": ["north", "south"], "day": [1, 1], "rain_mm": [2.5, 0.0]}, schema=readings_schema)),
    stage("staging/guide_ingest/readings-2.parquet", pa.table({
        "station": ["north", "south", "east"], "day": [2, 2, 2], "rain_mm": [7.1, 1.2, 0.4]}, schema=readings_schema)),
]
print(files[0].replace(BUCKET, "<bucket>"))

/<bucket>/staging/guide_ingest/readings-1.parquet

Import into an existing table

The table must have every column that's in the files, with the same types. Table columns that a file doesn't have are left null.

with session.transaction() as tx:
    readings = tx.bucket(BUCKET).schema(SCHEMA).create_table("readings", readings_schema, fail_if_exists=False)
    readings.import_files(files)

with session.transaction() as tx:
    readings = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    print(readings.select().read_all().sort_by([("day", "ascending"), ("station", "ascending")]))

pyarrow.Table
station: string
day: int32
rain_mm: double
----
station: [["north","south","east","north","south"]]
day: [[1,1,2,2,2]]
rain_mm: [[2.5,0,0.4,7.1,1.2]]

If a file doesn't match the table, you get errors.ImportFilesError, and its error_dict["res"] says why, for example TabularMismatchColumnType. A missing file raises errors.NotFound. See importing Parquet files in the error reference.

See which files have been imported

Each table records the files imported into it in its imports table, which you can query like any other table:

with session.transaction() as tx:
    readings = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    imported = readings.imports_table().select(columns=["ObjectName"]).read_all()
    print(sorted(imported["ObjectName"].to_pylist()))

['staging/guide_ingest/readings-1.parquet', 'staging/guide_ingest/readings-2.parquet']

New databases keep track of which files each table has imported (avoiding duplicate imports). If you import a file that's already been imported, the Python SDK skips it: there's no error and no rows are added. If the file was changed since it was imported, the import fails with ImportFilesError. Importing through the VAST Query Engine instead fails with "File was already imported to table". The cluster recognises a file by the stored object itself, not only by its name.

Create a table from files

vastdb.util.create_table_from_files() reads the files' Parquet schemas, creates a table with the merged schema, and imports the files, all in one call:

from vastdb import util

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).schema(SCHEMA)
    from_files = util.create_table_from_files(schema, "readings_copy", files)
    print([(f.name, str(f.type)) for f in from_files.arrow_schema])
    print(from_files.select(columns=["station"]).read_all().num_rows, "rows")

[('station', 'string'), ('day', 'int32'), ('rain_mm', 'double')]
5 rows

By default every file's schema must be a subset of the largest one. Pass schema_merge_func=util.strict_schema_merge to require identical schemas, or util.union_schema_merge to combine them. See Import Parquet files in the SDK reference.

Fill in columns that aren't in the files

When files are laid out by a key (for example station=west/...) and the key's value isn't stored inside each file, use import_partitioned_files(). It maps each file to a one-row record batch that holds the values for the columns the file doesn't have:

station_schema = pa.schema([("station", pa.utf8())])
west_file = stage("staging/guide_ingest/station=west/day3.parquet",
                  pa.table({"day": pa.array([3, 3], pa.int32()), "rain_mm": [0.5, 0.7]}))

with session.transaction() as tx:
    readings = tx.bucket(BUCKET).schema(SCHEMA).table("readings")
    readings.import_partitioned_files({
        west_file: pa.record_batch([pa.array(["west"])], schema=station_schema),
    })
    west = readings.select(predicate=readings["station"] == "west").read_all()
    print(west.sort_by("rain_mm").to_pylist())

[{'station': 'west', 'day': 3, 'rain_mm': 0.5}, {'station': 'west', 'day': 3, 'rain_mm': 0.7}]

Import settings

Pass an ImportConfig to tune an import:

The SDK splits the file list into batches and sends them from several threads in parallel, so pass all your files in one call rather than calling import_files() once per file.

from vastdb.config import ImportConfig

table.import_files(paths, config=ImportConfig(import_concurrency=16))

Add columns to existing rows

With ImportConfig(key_names=[...]), an import doesn't add rows. The cluster matches each row in the file to an existing row by the key columns, and writes the file's other columns into those rows. A column the table doesn't have yet is added. The keys in the file and in the table must be in the same order, and missing or extra keys make the import fail; see importing files into a VAST DataBase.

from vastdb.config import ImportConfig

prices_schema = pa.schema([("id", pa.int64()), ("symbol", pa.utf8())])
prices_file = stage("staging/guide_ingest/prices.parquet", pa.table({
    "id": [0, 1, 2], "symbol": ["a", "b", "c"], "price": [10.0, 11.0, 12.0]}))

with session.transaction() as tx:
    prices = tx.bucket(BUCKET).schema(SCHEMA).create_table("prices", prices_schema, fail_if_exists=False)
    prices.insert(pa.table({"id": [0, 1, 2], "symbol": ["a", "b", "c"]}, schema=prices_schema))

with session.transaction() as tx:
    prices = tx.bucket(BUCKET).schema(SCHEMA).table("prices")
    prices.import_files([prices_file], config=ImportConfig(key_names=["id", "symbol"]))

with session.transaction() as tx:
    prices = tx.bucket(BUCKET).schema(SCHEMA).table("prices")
    print(prices.select().read_all().sort_by("id").to_pydict())

{'id': [0, 1, 2], 'symbol': ['a', 'b', 'c'], 'price': [10.0, 11.0, 12.0]}

When incoming data has new columns

Data from files and events often gains fields over time. An insert must use the table's columns. A batch with a column the table doesn't have is rejected, so add the column first. Schema changes are transactional, so you can add the column and insert in one transaction; see schema management.

Start with a table and a first batch:

with session.transaction() as tx:
    feed = tx.bucket(BUCKET).schema(SCHEMA).create_table(
        "feed", pa.schema([("id", pa.int64()), ("city", pa.utf8())]), fail_if_exists=False)
    feed.insert(pa.table({"id": [1], "city": ["Oslo"]}))
print("Created feed")

Created feed

A later batch has a new field, temp. Inserting it as it is fails:

new_batch = pa.table({"id": [2], "city": ["Cairo"], "temp": [30.1]})
try:
    with session.transaction() as tx:
        tx.bucket(BUCKET).schema(SCHEMA).table("feed").insert(new_batch)
except errors.NotFound as e:
    print(e.status, e.code)

404 TabularMismatchColumnName

Compare the batch's fields with the table's columns, add the missing ones, then insert:

def add_missing_columns(table, incoming: pa.Schema) -> list[str]:
    existing = set(table.columns().names)
    missing = [field for field in incoming if field.name not in existing]
    for field in missing:
        table.add_column(pa.schema([field]))
    return [field.name for field in missing]

with session.transaction() as tx:
    feed = tx.bucket(BUCKET).schema(SCHEMA).table("feed")
    added = add_missing_columns(feed, new_batch.schema)
    feed.insert(new_batch)
print("Added columns:", added)

Added columns: ['temp']

Rows inserted before the column existed read null for it:

with session.transaction() as tx:
    rows = tx.bucket(BUCKET).schema(SCHEMA).table("feed").select().read_all()
print(sorted(rows.to_pylist(), key=lambda r: r["id"]))

[{'id': 1, 'city': 'Oslo', 'temp': None}, {'id': 2, 'city': 'Cairo', 'temp': 30.1}]

Things to know:

• A batch can leave columns out. Columns that aren't in the batch read null for the new rows. Only extra columns need adding first.
• Types must match. If a field's type differs from the existing column's, for example id arriving as a string, the insert fails with BadRequest, code TabularMismatchColumnType. Cast the batch to the table's types before you insert. You can't change a column's type.
• Two loaders adding the same column at once conflict. The second gets Conflict (TabularWwc). Retry the transaction; on the retry the column already exists, so nothing is added twice. See conflicts between transactions.
• Vector-indexed tables can't add columns in 5.5. See VAST Vector Search limits.

Load a partitioned table

A partitioned table accepts ordinary insert() calls. But partitioned tables benefit when inserts are collected in memory per partition and written in batches (database partitioning). session.buffered_inserter() does that for you. You add record batches in any order. It buffers the rows by partition, and flushes each partition's rows through a thread pool that you provide. It needs cluster 5.5 or later.

Create a table partitioned by device, and some batches that mix devices:

import datetime as dt
from concurrent.futures import ThreadPoolExecutor

from pyiceberg.transforms import IdentityTransform
from vastdb import BufferedInsertConfig, PartitionKey, PartitionSpec, TableRef

metrics_schema = pa.schema([
    ("device", pa.utf8()),
    ("ts", pa.timestamp("us")),
    ("value", pa.float64()),
])
spec = PartitionSpec([PartitionKey("device", IdentityTransform())])

with session.transaction() as tx:
    tx.bucket(BUCKET).schema(SCHEMA).create_table(
        "metrics", metrics_schema, partition_spec=spec, fail_if_exists=False)

start = dt.datetime(2026, 1, 1)
batches = [
    pa.record_batch({
        "device": [f"d{(i + j) % 3}" for j in range(1000)],
        "ts": [start + dt.timedelta(seconds=i * 1000 + j) for j in range(1000)],
        "value": [float(j) for j in range(1000)],
    }, schema=metrics_schema)
    for i in range(5)
]
print(len(batches), "batches of", batches[0].num_rows, "rows")

5 batches of 1000 rows

Load them in one transaction by passing it as BufferedInsertConfig(tx=...):

with session.transaction() as tx:
    with ThreadPoolExecutor(max_workers=4) as pool:
        config = BufferedInsertConfig(tx=tx)
        with session.buffered_inserter(TableRef(BUCKET, SCHEMA, "metrics"), pool, config) as inserter:
            for batch in batches:
                inserter.add_batch(batch)
        # Leaving the inserter's block flushes what's left and waits for every write.

with session.transaction() as tx:
    metrics = tx.bucket(BUCKET).schema(SCHEMA).table("metrics")
    print(sorted(metrics.partitions().select(columns=["device"]).read_all()["device"].to_pylist()))
    print(metrics.select(columns=["value"], predicate=metrics["device"] == "d1").read_all().num_rows, "rows for d1")

['d0', 'd1', 'd2']
1667 rows for d1

If you don't pass tx, each flush commits in its own transaction. That suits a long-running stream, but a failure part-way through leaves the earlier flushes committed. The other BufferedInsertConfig fields decide when a partition is flushed:

Each flush is written with the same splitting rules as insert(). For choosing partition keys, see data modeling. For what partitioned tables can't be combined with, see limits.

Load with Trino, Spark or the VAST Query Engine

Each engine can insert rows and copy query results into a table. Trino and Spark can also import Parquet files the way import_files() does: the cluster reads the files from the bucket, and the data doesn't pass through the engine (Importing Data to VAST DataBase). The examples below use the database (bucket) vastdb; use yours instead.

Create a table for these examples, and stage one Parquet file each for Trino and Spark to import:

with session.transaction() as tx:
    tx.bucket(BUCKET).schema(SCHEMA).create_table("rain", readings_schema, fail_if_exists=False)

engine_files = {
    engine: stage(f"staging/guide_ingest/rain-{engine}.parquet", pa.table({
        "station": ["north", "south"], "day": [day, day], "rain_mm": [rain, rain / 2]}, schema=readings_schema))
    for engine, day, rain in [("trino", 3, 4.0), ("spark", 4, 6.0)]
}
print(engine_files["trino"].replace(BUCKET, "<bucket>"))

/<bucket>/staging/guide_ingest/rain-trino.parquet

Trino

To import files, insert their paths into the table's vast.import_data() form. Give one path per row of VALUES, in the form /<bucket>/<key>:

INSERT INTO vast."vastdb/guide_ingest"."rain vast.import_data()" ("$parquet_file_path")
VALUES ('/vastdb/staging/guide_ingest/rain-trino.parquet');

SELECT count(*) AS imported_rows FROM vast."vastdb/guide_ingest".rain;

 imported_rows 
---------------
             2 
(1 row)

Insert rows directly, and copy query results into a new table:

INSERT INTO vast."vastdb/guide_ingest".rain VALUES ('east', 3, 1.5);

CREATE TABLE vast."vastdb/guide_ingest".rain_by_station AS
SELECT station, sum(rain_mm) AS rain_mm
FROM vast."vastdb/guide_ingest".rain
GROUP BY station;

SELECT * FROM vast."vastdb/guide_ingest".rain_by_station ORDER BY station;

 station | rain_mm 
---------+---------
 east    |     1.5 
 north   |     4.0 
 south   |     2.0 
(3 rows)

You can also write the import form as "<table> vast.import_data", without the parentheses, and with partition columns listed in the parentheses. Both forms work.

Spark

The same import in Spark SQL, with the table name and its vast.import_data() suffix quoted in backticks. Where your platform team has set up the connector, getOrCreate() picks up its settings; to try it locally, see Try it from a local PySpark session:

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
RAIN = f"ndb.`{BUCKET}`.{SCHEMA}.rain"

spark.sql(f"INSERT INTO ndb.`{BUCKET}`.{SCHEMA}.`rain vast.import_data()` (`$parquet_file_path`) "
          f"VALUES ('{engine_files['spark']}')").collect()

more = spark.createDataFrame([("east", 4, 2.5)], "station string, day int, rain_mm double")
more.writeTo(RAIN).append()

spark.sql(f"SELECT day, count(*) AS readings FROM {RAIN} GROUP BY day ORDER BY day").show()

+---+--------+
|day|readings|
+---+--------+
|  3|       3|
|  4|       3|
+---+--------+

Spark refuses to write unless spark.task.maxFailures is 1 and spark.speculation is off, because a retried task would insert its rows twice. See Spark.

VAST Query Engine

The Query Engine runs INSERT inside the cluster, so INSERT INTO ... SELECT copies rows without sending them to your client. Connect with the ADBC driver; QE_ENDPOINT is the Query Engine's endpoint from your platform team:

import adbc_driver_manager.dbapi as dbapi
import adbc_driver_vastdb

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,
)
RAIN_SQL = f'"{BUCKET}/{SCHEMA}".rain'

with conn.cursor() as cur:
    cur.execute(f"INSERT INTO {RAIN_SQL} VALUES ('west', 5, 0.5)")
    cur.execute(f"INSERT INTO {RAIN_SQL} SELECT station, 6, rain_mm * 2 FROM {RAIN_SQL} WHERE day = 5")
    cur.execute(f"SELECT count(*) FROM {RAIN_SQL}")
    print("rows in rain:", cur.fetchone()[0])
conn.close()

rows in rain: 8

The VAST Query Engine can import Parquet files too. The table name gets the suffix vast.import_data inside its quotes, and each file is given as <bucket>/<key>, without a leading /: INSERT INTO "<bucket>/<schema>"."<table> vast.import_data" ("$parquet_file_path") VALUES ('<bucket>/<key>'). The form with the plain table name fails with Mismatched column name between file to import and existing table. See the SQL reference.

Clean up

Drop the tables and the schema, and delete the staged files:

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).schema(SCHEMA)
    for name in sorted(schema.tablenames()):
        schema.table(name).drop()
    schema.drop()

staged = s3.list_objects_v2(Bucket=BUCKET, Prefix="staging/guide_ingest/").get("Contents", [])
for obj in staged:
    s3.delete_object(Bucket=BUCKET, Key=obj["Key"])
print(f"Dropped {SCHEMA} and deleted {len(staged)} staged files")

Dropped guide_ingest and deleted 6 staged files

Next steps

• Querying: read the data back efficiently.
• Arrow, pandas and Polars: insert from DataFrames.
• Python SDK reference: every parameter of insert(), import_files() and buffered_inserter().