Updates and deletes

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

This guide shows how to change and remove rows with the Python SDK: updating and deleting by row ID, updating only some columns, what you can't update on sorted and partitioned tables, how tables with user-defined row IDs behave, and how to delete whole partitions. Then it does the same with Trino, Spark and the VAST Query Engine.

Run the code blocks in order in one Python session. Below each block is the output it produced on a VAST Cluster 5.5 system.

Set up

Connect the same way as in the quickstart, and create a table with a few rows:

import os
import pyarrow as pa
import vastdb
from vastdb import errors

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

columns = pa.schema([
    ("id", pa.int64()),
    ("city", pa.utf8()),
    ("temperature", pa.float64()),
])
rows = pa.table({
    "id": [1, 2, 3, 4],
    "city": ["Lisbon", "Oslo", "Cairo", "Lima"],
    "temperature": [21.5, 4.0, 30.2, 18.9],
}, schema=columns)

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).create_schema("guide_updates", fail_if_exists=False)
    schema.create_table("readings", columns, fail_if_exists=False).insert(rows)

def readings(tx):
    return tx.bucket(BUCKET).schema("guide_updates").table("readings")

def show(tx):
    print(readings(tx).select().read_all().sort_by("id").to_pydict())

with session.transaction() as tx:
    show(tx)

{'id': [1, 2, 3, 4], 'city': ['Lisbon', 'Oslo', 'Cairo', 'Lima'], 'temperature': [21.5, 4.0, 30.2, 18.9]}

How updates and deletes find rows

update() and delete() don't take a filter. They take rows that carry a $row_id column, which identifies each row in the table. To get row IDs, select the rows you want to change with internal_row_id=True:

Select with row IDs, then update or delete, in one transaction.

with session.transaction() as tx:
    t = readings(tx)
    oslo = t.select(predicate=t["city"] == "Oslo", internal_row_id=True).read_all()
    print(oslo.schema.field("$row_id"))
    print(oslo.num_rows, "row")

pyarrow.Field<$row_id: uint64>
1 row

Do the select and the change in the same transaction. Row IDs from one transaction aren't a safe input to another one: another client may have changed or deleted those rows in the meantime. If you pass rows with no $row_id column, you get MissingRowIdColumn.

The type of $row_id depends on the table. It's uint64 on regular tables and decimal128(38, 0) on sorted, partitioned and vector-indexed tables. If you take $row_id from select(..., internal_row_id=True), it already has the right type.

Update rows

update(rows, columns=None) writes new values into the rows identified by $row_id. Pass columns to say which columns to write. If you leave it out, every column in rows except $row_id is written.

The usual pattern is select, change in Arrow, update:

import pyarrow.compute as pc

with session.transaction() as tx:
    t = readings(tx)
    hot = t.select(columns=["temperature"], predicate=t["temperature"] > 20,
                   internal_row_id=True).read_all()
    fahrenheit = pc.add(pc.multiply(hot["temperature"], 1.8), 32)
    t.update(hot.set_column(0, "temperature", fahrenheit), columns=["temperature"])
    show(tx)

{'id': [1, 2, 3, 4], 'city': ['Lisbon', 'Oslo', 'Cairo', 'Lima'], 'temperature': [70.7, 4.0, 86.36, 18.9]}

Select only the columns you plan to change, plus whatever you need to compute the new values. Columns you don't pass, or don't name in columns, keep their values.

If a $row_id doesn't match any row, for example because the row was deleted, update() and delete() don't raise an error. Nothing happens for that row:

with session.transaction() as tx:
    t = readings(tx)
    ghost = pa.table({"$row_id": pa.array([1000], pa.uint64()), "city": ["Nowhere"]})
    t.update(ghost)
    t.delete(ghost)
    show(tx)

{'id': [1, 2, 3, 4], 'city': ['Lisbon', 'Oslo', 'Cairo', 'Lima'], 'temperature': [70.7, 4.0, 86.36, 18.9]}

So don't use the absence of an error to check that a row existed. If you need to know, select it first in the same transaction.

If rows has a column the table doesn't have, the cluster rejects the update with NotFound (404), code TabularMismatchColumnName. The message mentions a "file to import", even though this is an update.

Delete rows

delete(rows) deletes the rows identified by $row_id. Any other columns in rows are ignored, so you can pass the result of a select directly:

with session.transaction() as tx:
    t = readings(tx)
    cold = t.select(columns=["id"], predicate=t["temperature"] < 10, internal_row_id=True).read_all()
    t.delete(cold)
    show(tx)

{'id': [1, 3, 4], 'city': ['Lisbon', 'Cairo', 'Lima'], 'temperature': [70.7, 86.36, 18.9]}

Updates and deletes are part of your transaction, so they commit or roll back with everything else in it. If another transaction changes the same rows at the same time, one of them gets a conflict error. See transactions in practice.

The SDK sorts the rows by $row_id and sends them in slices of up to 512K rows per request. That's client-side behaviour of SDK 2.1.0; you don't need to split large updates yourself.

Columns you can't update

Some columns can't be changed once a row is inserted:

The SDK doesn't check this before it sends the update. The cluster rejects it with BadRequest (400), code InvalidArgument and message Invalid Argument, which doesn't say which column is the problem. Here's a sorted table:

with session.transaction() as tx:
    sorted_t = tx.bucket(BUCKET).schema("guide_updates").create_table(
        "sorted_readings", columns, sorting_key=["id"], fail_if_exists=False)
    sorted_t.insert(rows)

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema("guide_updates").table("sorted_readings")
    lima = t.select(columns=["id", "temperature"], predicate=t["city"] == "Lima",
                    internal_row_id=True).read_all()
    print("$row_id type:", lima.schema.field("$row_id").type)

    t.update(lima, columns=["temperature"])  # not a sort column: allowed
    print("Updated temperature")

    try:
        new_id = lima.set_column(0, "id", pa.array([40], pa.int64()))
        t.update(new_id, columns=["id"])     # new value in a sort column: rejected
    except errors.BadRequest as e:
        print(e.status, e.code, e.message)

$row_id type: decimal128(38, 0)
Updated temperature
400 InvalidArgument Invalid Argument

The rejected update didn't stop the transaction from committing the update before it. Writing a sort column's existing value back to it is accepted; only a new value is rejected. Check the table you're updating against the table above before you write the update.

To change a value in a column you can't update, delete the row and insert it again with the new value, in one transaction.

Partitioned tables

On a partitioned table, you can update a partition column only if the row stays in the same partition. Here the table is partitioned on city:

from pyiceberg.transforms import IdentityTransform

by_city = vastdb.PartitionSpec([vastdb.PartitionKey("city", IdentityTransform())])

with session.transaction() as tx:
    part_t = tx.bucket(BUCKET).schema("guide_updates").create_table(
        "readings_by_city", columns, partition_spec=by_city, fail_if_exists=False)
    part_t.insert(rows)

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema("guide_updates").table("readings_by_city")
    oslo = t.select(columns=["city"], predicate=t["id"] == 2, internal_row_id=True).read_all()

    t.update(oslo)  # same value, same partition: allowed
    print("Same partition: updated")

    moved = oslo.set_column(0, "city", pa.array(["Bergen"]))
    try:
        t.update(moved)  # different partition: rejected
    except errors.BadRequest as e:
        print(e.status, e.code, e.message)

Same partition: updated
400 InvalidArgument Invalid Argument

Delete whole partitions

To remove every row in a partition, use delete_partitions() instead of selecting and deleting rows. Deleting whole partitions is significantly faster than deleting rows one by one; see Database partitioning.

delete_partitions() isn't transactional. Changing partitions while other writes target the table can leave the data in an indeterminate state. The SDK refuses to run it unless you pass allow_non_acid=True, and it sends the request outside your transaction. A rollback doesn't bring the rows back:

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema("guide_updates").table("readings_by_city")
    try:
        t.delete_partitions(t.partitions().select().read_all().to_batches()[0])
    except AssertionError as e:
        print(e)

try:
    with session.transaction() as tx:
        t = tx.bucket(BUCKET).schema("guide_updates").table("readings_by_city")
        p = t.partitions()
        cairo = p.select(predicate=p["city"] == "Cairo").read_all()
        t.delete_partitions(cairo.combine_chunks().to_batches()[0], allow_non_acid=True)
        raise RuntimeError("roll back")
except RuntimeError:
    pass

with session.transaction() as tx:
    t = tx.bucket(BUCKET).schema("guide_updates").table("readings_by_city")
    print(sorted(t.select(columns=["city"]).read_all()["city"].to_pylist()))

Beware, deleting partitions is a non acid operation! In order to proceed please use allow_non_acid=True
['Lima', 'Lisbon', 'Oslo']

The transaction rolled back, but the Cairo partition is gone. Each row you pass to delete_partitions() identifies one partition, using the columns that partitions() returns. Before you use it:

• Make sure no other client is writing to the table at the same time.
• Don't put it in a transaction you might retry, and don't rely on a rollback to undo it.
• Your identity policy needs s3:TabularDeletePartitions.

Trino and Spark delete partitions with a vast.allow_non_acid flag on the statement; see Database partitioning.

Tables with user-defined row IDs

On a table created with use_external_row_ids_allocation=True, you choose each row's ID yourself, in an int64 column named vastdb_rowid (User-defined row IDs). Values can't be larger than 2^47. In this capture, $row_id had the same value as vastdb_rowid, so you can update rows by the IDs you assigned without selecting them first:

with session.transaction() as tx:
    ext = tx.bucket(BUCKET).schema("guide_updates").create_table(
        "stations",
        pa.schema([("vastdb_rowid", pa.int64()), ("name", pa.utf8())]),
        use_external_row_ids_allocation=True, fail_if_exists=False)
    returned = ext.insert(pa.table({"vastdb_rowid": pa.array([100, 200], pa.int64()),
                                    "name": ["north", "south"]}))
    print("Row IDs returned by insert:", returned.to_pylist())

with session.transaction() as tx:
    ext = tx.bucket(BUCKET).schema("guide_updates").table("stations")
    ext.update(pa.table({"$row_id": pa.array([200], pa.uint64()), "name": ["south-2"]}))
    print(ext.select(internal_row_id=True).read_all().sort_by("vastdb_rowid").to_pydict())

Row IDs returned by insert: [100, 200]
{'vastdb_rowid': [100, 200], 'name': ['north', 'south-2'], '$row_id': [100, 200]}

insert() returns the row IDs, which are your vastdb_rowid values. Once a table uses user-defined row IDs, every insert has to supply vastdb_rowid. An insert without it is rejected:

with session.transaction() as tx:
    ext = tx.bucket(BUCKET).schema("guide_updates").table("stations")
    try:
        ext.insert(pa.table({"name": ["east"]}))
    except errors.BadRequest as e:
        print(e.status, e.code, e.message)

400 TabularRowAllocMethodConflict There is a conflict in tabular row allocation method

You can't update vastdb_rowid itself (BadRequest, InvalidArgument, captured). User-defined row IDs can't be combined with sorting or partitioning; see limits.

Update and delete with Trino, Spark or the VAST Query Engine

In SQL you don't fetch row IDs first. UPDATE and DELETE take a WHERE clause, and the engine finds the matching rows for you. The rules above still apply, because the cluster enforces them: a statement that changes a sort column is rejected in all three engines, and Trino also rejects an update that moves a row to another partition. The examples below use the database (bucket) vastdb; use yours instead.

Deleting partitions isn't transactional in any engine, as with delete_partitions() above. Make a fresh copy of the example rows to change:

with session.transaction() as tx:
    tx.bucket(BUCKET).schema("guide_updates").create_table(
        "cities", columns, fail_if_exists=False).insert(rows)
print("cities ready")

cities ready

Trino

UPDATE vast."vastdb/guide_updates".cities SET temperature = 5.0 WHERE city = 'Oslo';
DELETE FROM vast."vastdb/guide_updates".cities WHERE id = 4;
SELECT * FROM vast."vastdb/guide_updates".cities ORDER BY id;

 id |  city  | temperature 
----+--------+-------------
  1 | Lisbon |        21.5 
  2 | Oslo   |         5.0 
  3 | Cairo  |        30.2 
(3 rows)

MERGE INTO ... USING ... WHEN MATCHED THEN UPDATE works too. Updating id in sorted_readings, the sorted table above, fails with Failed updating rows, the same rejection the SDK gets.

Spark

The WHERE clause can use anything Spark can evaluate, including LIKE:

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
CITIES = f"ndb.`{BUCKET}`.guide_updates.cities"

spark.sql(f"UPDATE {CITIES} SET temperature = temperature + 1 WHERE city LIKE 'L%'").collect()
spark.sql(f"DELETE FROM {CITIES} WHERE id = 3").collect()
spark.sql(f"SELECT * FROM {CITIES} ORDER BY id").show()

+---+------+-----------+
| id|  city|temperature|
+---+------+-----------+
|  1|Lisbon|       22.5|
|  2|  Oslo|        5.0|
+---+------+-----------+

Writes need spark.task.maxFailures=1 and spark.speculation=false; see Spark.

VAST Query Engine

The Query Engine runs the statement inside the cluster. 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,
)
CITIES_SQL = f'"{BUCKET}/guide_updates".cities'

with conn.cursor() as cur:
    cur.execute(f"UPDATE {CITIES_SQL} SET city = 'Lisboa' WHERE id = 1")
    cur.execute(f"DELETE FROM {CITIES_SQL} WHERE temperature < 10")
    cur.execute(f"SELECT id, city, temperature FROM {CITIES_SQL} ORDER BY id LIMIT 10")
    print(cur.fetchall())
conn.close()

[(1, 'Lisboa', 22.5)]

On a sorted table the Query Engine updates other columns, but rejects a new value in a sort column with Invalid Argument.

Performance

What's known so far:

• Batch your changes. Each update() or delete() call sends at least one request, so changing many rows in one call costs fewer round trips than one call per row.
• Deleting a whole partition is faster than deleting its rows (Database partitioning). If you regularly delete data by time or category, consider partitioning on that column; see data modeling.
• Sorted tables: after rows are added or changed, the table is re-sorted in the background, and its sorting grade drops until that finishes (Sorted tables).

What happens after you commit

See write path.

Clean up

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).schema("guide_updates")
    for table in schema.tables():
        table.drop()
    schema.drop()
    print("Dropped guide_updates")

Dropped guide_updates

See also

• Event sourcing and CQRS: append events instead of updating rows that many writers share, and fold them into a current-state table.
• Write path: what updates and deletes do on the cluster.