Transactions in practice

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

Every Python SDK call runs inside a transaction. This guide shows how to use transactions in your code: what commits and what rolls back, what the SDK retries for you, which errors mean "another transaction got there first", and how to retry a whole transaction yourself. It ends with transactions in 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, with the connection settings in environment variables, and create a small table for this guide:

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()), ("status", pa.utf8())])

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).create_schema("guide_tx", fail_if_exists=False)
    schema.create_table("orders", columns, fail_if_exists=False)

def orders(tx):
    return tx.bucket(BUCKET).schema("guide_tx").table("orders")

print("Ready")

Ready

orders(tx) is a small helper that returns the table handle inside a given transaction. The rest of this guide uses it.

Commit and rollback

with session.transaction() as tx: starts a transaction on the cluster when the block starts. When the block ends normally, the SDK commits it. If an exception leaves the block, the SDK rolls it back and the exception carries on up to your code. There's no separate commit() or rollback() method: the with block is how you control a transaction.

Inside the transaction you see your own changes (except through a bucket snapshot, which shows only what was committed when the snapshot was taken). After a rollback, none of them are kept:

try:
    with session.transaction() as tx:
        t = orders(tx)
        t.insert(pa.table({"id": [1, 2], "status": ["new", "new"]}, schema=columns))
        print("Rows inside the transaction:", t.select().read_all().num_rows)
        raise RuntimeError("something went wrong")
except RuntimeError as e:
    print("Caught:", e)

with session.transaction() as tx:
    print("Rows after the rollback:", orders(tx).select().read_all().num_rows)

Rows inside the transaction: 2
Caught: something went wrong
Rows after the rollback: 0

When it rolls back, the SDK also logs the exception at ERROR level on the vastdb.transaction logger, with the message rolling back txid=.... You'll see that in your logs even when your code handles the exception.

Make each transaction one unit of work: the changes that must all happen or not happen together. One transaction can create tables, and insert, update and delete rows, across several tables.

One exception: deleting whole partitions with delete_partitions() isn't transactional, and a rollback doesn't undo it. See updates and deletes.

Objects belong to their transaction

A table handle, and the reader that select() returns, only work while their transaction is open. Once the with block has ended, using them raises an error:

with session.transaction() as tx:
    t = orders(tx)

try:
    t.select().read_all()
except Exception as e:
    print(type(e).__module__ + "." + type(e).__name__)

vastdb.transaction.TransactionNotActiveError

Read query results fully, for example with read_all(), before the block ends. If a reader is still streaming when its transaction closes, you get MissingTransaction instead. To avoid loading the same table metadata in every transaction, see reusing metadata across transactions.

What other transactions see

Here a reader transaction is opened first. A second transaction inserts a row and commits while the reader is still open:

with session.transaction() as reader:
    with session.transaction() as writer:
        orders(writer).insert(pa.table({"id": [10], "status": ["new"]}, schema=columns))
    # the writer has committed here
    print("Open reader sees:", orders(reader).select().read_all().num_rows, "rows")

with session.transaction() as tx:
    print("New transaction sees:", orders(tx).select().read_all().num_rows, "rows")

Open reader sees: 0 rows
New transaction sees: 1 rows

In this run, the transaction that was already open didn't see a row committed after it started, and a transaction started afterwards did. To read the latest committed data, start a new transaction.

Conflicts between transactions

When two open transactions change the same thing, the cluster rejects one of them. You see this as one of two errors.

Conflict (409, TabularWwc) on the call

If another open transaction has already changed the same row, your update() or delete() call fails straight away with vastdb.errors.Conflict:

def set_status(tx, order_id, status):
    t = orders(tx)
    rows = t.select(columns=["id"], predicate=t["id"] == order_id, internal_row_id=True).read_all()
    t.update(rows.append_column("status", pa.array([status] * rows.num_rows)), columns=["status"])

with session.transaction() as first:
    set_status(first, 10, "paid")
    try:
        with session.transaction() as second:
            set_status(second, 10, "cancelled")
    except errors.Conflict as e:
        print(e.status, e.code, e.message)

with session.transaction() as tx:
    print(orders(tx).select().read_all().to_pydict())

409 TabularWwc Write write conflict
{'id': [10], 'status': ['paid']}

The second transaction's with block rolled it back, and the first one committed. These cases also returned Conflict:

BadRequest (400, TabularTxAlreadyRollbacked) at commit

A conflict doesn't always show up on the call. In this example, the transaction opened second changes the row first. Then the transaction opened first changes the same row: that call succeeds, and the cluster rolls back the other transaction. The other transaction only finds out when it tries to commit:

with session.transaction() as older:
    try:
        with session.transaction() as younger:
            set_status(younger, 10, "shipped")
            set_status(older, 10, "refunded")
    except errors.BadRequest as e:
        print(e.status, e.code, e.message)

with session.transaction() as tx:
    print(orders(tx).select().read_all().to_pydict())

400 TabularTxAlreadyRollbacked Tabular transaction already rollbacked
{'id': [10], 'status': ['refunded']}

You get the same error at commit if you catch a Conflict inside your with block and carry on: after a Conflict, the cluster has already rolled your transaction back, so nothing more in it can commit. Don't catch Conflict inside the transaction it happened in. Let it leave the block, and retry the whole transaction.

Retries the SDK does for you

Which failures the SDK retries, which you retry, and which you fix.

The SDK retries each request according to the session's BackoffConfig, but only for these errors. This is client-side behaviour of SDK 2.1.0:

You can change the retry budget when you connect. The defaults are 10 tries and 60 seconds, with exponential backoff:

from vastdb.config import BackoffConfig

patient_session = vastdb.connect(backoff_config=BackoffConfig(max_tries=20, max_time=300.0))
print("Connected with a longer retry budget")

Connected with a longer retry budget

The SDK logs retries at DEBUG level by default. Set backoff_log_level to log them at a higher level.

A connection failure during commit is ambiguous. The SDK doesn't retry the commit request, and the error doesn't tell you whether the cluster committed the transaction. If that matters to you, design the transaction so you can check afterwards whether it took effect, for example by reading back a row it wrote.

Retrying a whole transaction

For the conflict errors above, the fix is to run the whole transaction again, in a new session.transaction(). A rolled-back transaction leaves nothing behind (see above), so a retry starts from a clean state. Put the transaction's work in a function, and retry the function:

import time

def is_conflict(e):
    return isinstance(e, errors.Conflict) or (
        isinstance(e, errors.BadRequest) and e.code == "TabularTxAlreadyRollbacked")

def run_transaction(session, work, attempts=5, wait=0.5):
    for attempt in range(1, attempts + 1):
        try:
            with session.transaction() as tx:
                return work(tx)
        except Exception as e:
            if not is_conflict(e) or attempt == attempts:
                raise
            time.sleep(wait * attempt)

def mark_delivered(tx):
    set_status(tx, 10, "delivered")
    return "done"

print(run_transaction(session, mark_delivered))

done

Keep these rules in mind:

• Re-read inside the retry. work(tx) must read everything it depends on in the new transaction. In the example, set_status() selects the row and its $row_id again each time. Don't reuse row IDs, handles or results from a failed attempt.
• Keep side effects out of work. Anything outside the database, such as sending a message or writing a file, happens again on every attempt.
• Don't retry delete_partitions(). A rollback doesn't undo it, so part of it may already have happened.
• Retry only conflicts. Forbidden, NotFound and BadRequest with other codes won't go away on their own. Connection errors on writes are ambiguous; see above.

To make conflicts rare, avoid having several clients update the same rows at the same time, and don't run the same DDL from several clients. Concurrent inserts into the same table, and updates of different rows of the same table, don't conflict.

Other failure modes

For every error message, see the error reference.

Transactions in Trino, Spark and the VAST Query Engine

SQL engines commit each statement on its own unless you group statements yourself. The examples below use the database (bucket) vastdb; use yours instead.

Trino

START TRANSACTION;
INSERT INTO vast."vastdb/guide_tx".orders VALUES (100, 'never stored');
ROLLBACK;

START TRANSACTION;
INSERT INTO vast."vastdb/guide_tx".orders VALUES (101, 'stored');
COMMIT;

SELECT id, status FROM vast."vastdb/guide_tx".orders WHERE id IN (100, 101);

 id  | status 
-----+--------
 101 | stored 
(1 row)

Spark

The VAST connector adds three functions to the ndb catalog. ndb.create_tx() starts a transaction that the following statements in the Spark session join, and ndb.commit_tx() or ndb.rollback_tx() ends it. 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()
ORDERS = f"ndb.`{BUCKET}`.guide_tx.orders"

spark.sql("SELECT ndb.create_tx()").collect()
spark.sql(f"INSERT INTO {ORDERS} VALUES (102, 'never stored')").collect()
spark.sql("SELECT ndb.rollback_tx()").collect()

spark.sql("SELECT ndb.create_tx()").collect()
spark.sql(f"INSERT INTO {ORDERS} VALUES (103, 'stored')").collect()
spark.sql("SELECT ndb.commit_tx()").collect()

spark.sql(f"SELECT id, status FROM {ORDERS} WHERE id IN (102, 103)").show()

+---+------+
| id|status|
+---+------+
|103|stored|
+---+------+

Only one transaction can be open per Spark session: ndb.create_tx() fails with "Active transaction was found" if one is already open.

VAST Query Engine, in an SDK transaction

When the ADBC driver is installed, the SDK opens an ADBC connection for each transaction, with your transaction's ID, so SQL on tx.adbc_conn.cursor runs in the same transaction as your SDK calls. It sees your uncommitted changes, and its writes commit or roll back with the rest of the transaction:

try:
    with session.transaction() as tx:
        orders(tx).insert(pa.table({"id": [104], "status": ["from the SDK"]}, schema=columns))
        cur = tx.adbc_conn.cursor
        cur.execute(f'SELECT count(*) FROM "{BUCKET}/guide_tx".orders WHERE id = 104')
        print("SQL sees the uncommitted SDK row:", cur.fetchone()[0])
        cur.execute(f"""INSERT INTO "{BUCKET}/guide_tx".orders VALUES (105, 'from SQL')""")
        raise RuntimeError("roll back")
except RuntimeError:
    pass

with session.transaction() as tx:
    tx.adbc_conn.cursor.execute(f"""INSERT INTO "{BUCKET}/guide_tx".orders VALUES (106, 'from SQL, committed')""")

with session.transaction() as tx:
    t = orders(tx)
    kept = t.select(predicate=t["id"].isin([104, 105, 106])).read_all()
print(kept.to_pydict())

SQL sees the uncommitted SDK row: 1
{'id': [106], 'status': ['from SQL, committed']}

The rollback removed both the SDK insert and the SQL insert. A standalone ADBC connection opened with autocommit=True, as in VAST Query Engine (ADBC), commits each statement on its own instead.

Clean up

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).schema("guide_tx")
    schema.table("orders").drop()
    schema.drop()
    print("Dropped guide_tx")

Dropped guide_tx

See also

• Event sourcing and CQRS: append-only writers and one materializer, a pattern that avoids conflicts between many writers of the same rows.
• Transactions: isolation and conflict rules.