vgi.transactor

Module overview

db-transactor — transactional database access for VGI workers.

The transactor is a long-lived subprocess that owns a single DuckDB connection. VGI worker processes communicate with it via vgi_rpc over Unix domain sockets, using the same streaming exchange patterns that DuckDB uses with VGI workers.

Architecture:

VGI Worker(s) ──── vgi_rpc (Unix socket) ──── db-transactor
                                                │
                                            DuckDB file

class TransactorClient

source

Description

Client that connects to (and optionally spawns) a db-transactor.

The transactor process is auto-spawned on first use if not already running. A single transactor serves all databases.

Methods

method get_proxy

source

get_proxy() -> Any

Get the typed RPC proxy, spawning the transactor if needed.

method close

source

close() -> None

Close the connection.

class TransactorProtocol

source

Bases: Protocol

Description

RPC interface for the db-transactor subprocess.

Methods

method register

source

register(
attach_opaque_data: bytes,
catalog_name: str = ‘’,
ddl_statements: list[str] | None = None,
) -> None

Register a new database for this attach_opaque_data and run initial DDL.

method catalog_version

source

catalog_version(attach_opaque_data: bytes) -> int

Return the catalog version for the database (incremented on DDL).

method begin

source

begin(attach_opaque_data: bytes) -> bytes

Begin a transaction. Returns the transactor-generated tx_id.

method commit

source

commit(attach_opaque_data: bytes, tx_id: bytes) -> None

Commit a transaction.

method rollback

source

rollback(attach_opaque_data: bytes, tx_id: bytes) -> None

Rollback a transaction.

method insert

source

insert(
attach_opaque_data: bytes,
tx_id: bytes,
table_name: str,
schema_name: str = ‘’,
returning: bool = False,
) -> Stream[ExchangeState]

Insert rows into a table via lockstep exchange.

method delete

source

delete(
attach_opaque_data: bytes,
tx_id: bytes,
table_name: str,
schema_name: str = ‘’,
returning: bool = False,
) -> Stream[ExchangeState]

Delete rows from a table via lockstep exchange.

method update

source

update(
attach_opaque_data: bytes,
tx_id: bytes,
table_name: str,
schema_name: str = ‘’,
columns: list[str] | None = None,
returning: bool = False,
) -> Stream[ExchangeState]

Update rows in a table via lockstep exchange.

method scan

source

scan(
attach_opaque_data: bytes,
tx_id: bytes,
table_name: str,
columns: list[str],
schema_name: str = ‘’,
pushdown_filters: bytes | None = None,
) -> Stream[ProducerState]

Scan rows from a table with optional predicate pushdown.

method execute_ddl

source

execute_ddl(attach_opaque_data: bytes, sql: str) -> None

Execute a DDL statement on the database (non-transactional).

method execute_ddl_tx

source

execute_ddl_tx(
attach_opaque_data: bytes,
tx_id: bytes,
sql: str,
strip_catalog: str | None = None,
) -> None

Execute DDL within a transaction.

method list_schemas

source

list_schemas(attach_opaque_data: bytes, tx_id: bytes) -> list[str]

List schema names within a transaction.

method list_user_tables

source

list_user_tables(
attach_opaque_data: bytes,
tx_id: bytes,
schema_name: str = ‘main’,
) -> list[str]

List user-created table names in the given schema within a transaction.

method table_schema

source

table_schema(
attach_opaque_data: bytes,
table_name: str,
tx_id: bytes,
) -> bytes

Get Arrow schema for a table as serialized IPC bytes.

method table_comment

source

table_comment(
attach_opaque_data: bytes,
table_name: str,
tx_id: bytes,
) -> str | None

Get the comment on a table, or None if no comment is set.

method list_user_views

source

list_user_views(
attach_opaque_data: bytes,
tx_id: bytes,
schema_name: str = ‘main’,
) -> list[str]

List user-created view names in the given schema within a transaction.

method view_info

source

view_info(
attach_opaque_data: bytes,
view_name: str,
tx_id: bytes,
) -> str

Get view info as JSON (definition, comment).

method ping

source

ping() -> None

Health check.

method shutdown

source

shutdown() -> None

Graceful shutdown.