vgi.client
Module overview
VGI client package for communicating with VGI workers.
This package provides:
- Client: A class for programmatic interaction with VGI workers, including both function invocation and catalog operations
- ClientError: Exception raised by Client function operations
- CatalogClientMixin: Mixin class providing catalog operations
- AggregateClientMixin: Mixin class providing aggregate invocation
- AggregateSession / AggregateStreamingSession: Handles on a bound aggregate
- OutputWriter: Helper for writing output in various formats
- main: CLI entry point
Usage (API):
from vgi.client import Client, ClientErrorfrom vgi.arguments import Arguments
with Client("./my_worker.py") as client: for batch in client.table_in_out_function( function_name="echo", arguments=Arguments(positional=[], named={}), input=input_batches, ): process(batch)Usage (Catalog API):
from vgi.client import Client
client = Client("./my_worker")result = client.catalog_attach( name="my_catalog", options={}, data_version_spec=None, implementation_version=None)Usage (CLI):
vgi-client --input data.parquet --function echovgi-client --input data.parquet --function sum_all_columnsclass AggregateClientMixin
Section titled “class AggregateClientMixin”Description
Mixin adding aggregate-function invocation to the VGI Client.
Every aggregate RPC is unary and runs on the client’s primary worker
connection, so start() (or the context-manager protocol) must have run
first — same requirement as scalar_function / table_function.
Methods
method aggregate_session
Section titled “method aggregate_session”aggregate_session(
*,
function_name: str,
schema_name: str,
input_schema: pa.Schema | None = None,
arguments: Arguments | None = None,
settings: dict[str, Any] | None = None,
secrets: dict[str, Any] | None = None,
) -> Iterator[AggregateSession]Bind an aggregate and yield a session over its raw RPCs.
The session is destroyed on exit, including on exception, so worker-side
group state never outlives the with block.
method aggregate_bind
Section titled “method aggregate_bind”aggregate_bind(
*,
function_name: str,
schema_name: str,
input_schema: pa.Schema | None = None,
arguments: Arguments | None = None,
settings: dict[str, Any] | None = None,
secrets: dict[str, Any] | None = None,
) -> AggregateSessionBind an aggregate function without taking responsibility for teardown.
Prefer :meth:aggregate_session, which destroys the session for you.
This entry point exists for callers that must own the lifetime
explicitly (e.g. a proxy that hands the execution_id to another
process); they must call :meth:AggregateSession.destroy themselves.
method aggregate_function
Section titled “method aggregate_function”aggregate_function(
*,
function_name: str,
schema_name: str,
input: Iterable[pa.RecordBatch] = (),
group_by: Sequence[str] = (),
arguments: Arguments | None = None,
settings: dict[str, Any] | None = None,
secrets: dict[str, Any] | None = None,
input_schema: pa.Schema | None = None,
finalize_chunk_size: int = DEFAULT_FINALIZE_CHUNK_SIZE,
) -> pa.RecordBatchRun an aggregate over input and return one row per group.
This is the SELECT <group_by>, agg(...) FROM t GROUP BY <group_by>
shape, driven client-side: group ids are allocated per distinct
group_by key in first-seen order (DuckDB’s hash-aggregate order),
every batch is pumped through aggregate_update, and the groups are
finalized in chunks of finalize_chunk_size.
Every input column that is not named in group_by is passed to the
aggregate as a value column, in the order it appears in the batch — so
order your input columns to match the aggregate’s declared parameters.
method aggregate_streaming
Section titled “method aggregate_streaming”aggregate_streaming(
*,
function_name: str,
schema_name: str,
input_schema: pa.Schema,
partition_key_count: int,
order_key_count: int = 0,
arguments: Arguments | None = None,
settings: dict[str, Any] | None = None,
secrets: dict[str, Any] | None = None,
output_schema: pa.Schema | None = None,
) -> Iterator[AggregateStreamingSession]Open a streaming-partitioned aggregate session and yield it.
The session is closed on exit, including on exception.
class AggregateSession
Section titled “class AggregateSession”Description
A bound aggregate execution — the raw RPC surface, one method per call.
Obtained from :meth:AggregateClientMixin.aggregate_session. Group ids are
caller-allocated int64s; the worker keys its per-group state on them and
never invents one. Reuse the same id across update calls to accumulate
into one group, exactly as DuckDB reuses the id stamped on a state pointer.
Attributes
attribute execution_id
Section titled “attribute execution_id”bytes
Worker-minted identifier for this aggregate execution. Scopes every piece of worker-side state, including window partitions.
attribute output_schema
Section titled “attribute output_schema”Schema the aggregate’s finalize produces (typically a
single result column).
Methods
method update
Section titled “method update”update(
*,
group_ids: Sequence[int] | pa.Array[Any],
batch: pa.RecordBatch | None = None,
) -> NoneAccumulate one chunk of rows into per-group state.
method combine
Section titled “method combine”combine(
*,
source_group_ids: Sequence[int] | pa.Array[Any],
target_group_ids: Sequence[int] | pa.Array[Any],
) -> NoneMerge each source group’s state into the paired target group.
Mirrors DuckDB’s combine step, where thread-local hash tables are merged into the global one.
method finalize
Section titled “method finalize”finalize(group_ids: Sequence[int] | pa.Array[Any]) -> pa.RecordBatchProduce the result row for each of group_ids, in that order.
A group id that was never updated finalizes to whatever the function
returns for an absent state — NULL for SUM/AVG, 0 for COUNT.
method destroy
Section titled “method destroy”destroy() -> NoneRelease every piece of worker state for this execution (best-effort).
Called for you when :meth:AggregateClientMixin.aggregate_session exits.
Like the C++ destructor it never raises: a failure here means leaked
worker rows, not a wrong answer, and the worker reclaims them on
execution timeout anyway.
method window_init
Section titled “method window_init”window_init(
*,
partition_id: int,
partition: pa.RecordBatch,
filter_mask: Sequence[bool] | pa.BooleanArray | None = None,
frame_stats: tuple[tuple[int, int], tuple[int, int]] | None = None,
all_valid: Sequence[bool] | None = None,
) -> NoneShip one window partition to the worker so it can be queried by frame.
method window
Section titled “method window”window(
*,
partition_id: int,
rid: int,
frames: Frames,
) -> pa.RecordBatchCompute the aggregate for one output row of a window partition.
method window_batch
Section titled “method window_batch”window_batch(
*,
partition_id: int,
row_idx: int,
frames: Sequence[Frames],
) -> pa.RecordBatchCompute len(frames) consecutive window output rows in one RPC.
method window_destroy
Section titled “method window_destroy”window_destroy(partition_id: int) -> NoneEvict one window partition from worker storage (best-effort).
class AggregateStreamingSession
Section titled “class AggregateStreamingSession”Description
An open streaming-partitioned aggregate session.
Obtained from :meth:AggregateClientMixin.aggregate_streaming. Each
:meth:chunk returns one output row per input row — the aggregate’s value
at that row’s position within its partition.
Attributes
attribute execution_id
Section titled “attribute execution_id”bytes
Worker-minted identifier for this streaming session.
attribute output_schema
Section titled “attribute output_schema”Schema of every batch :meth:chunk returns.
Methods
method chunk
Section titled “method chunk”chunk(batch: pa.RecordBatch) -> pa.RecordBatchProcess one input chunk and return its per-row output.
method close
Section titled “method close”close() -> NoneEnd the session and free its worker-side state (best-effort).
class CatalogClientMixin
Section titled “class CatalogClientMixin”Description
Mixin that adds catalog operations to a VGI Client.
Catalog methods spawn ephemeral connections under the hood — for
subprocess transport a pooled subprocess worker; for HTTP transport a
short-lived http_connect session reusing the Client’s shared
httpx2.Client (bearer token, headers); for TCP transport a short-lived
tcp_connect session. Browsing catalogs over HTTP is the canonical
non-DuckDB use case this mixin supports.
Other attributes expected from Client: _transport (subprocess,
http, tcp, or launch), _base_url (HTTP base URL), _tcp_host /
_tcp_port (TCP endpoint), _launch_argv / _launch_idle_timeout /
_launch_state_dir / _launch_socket_path (launcher worker
identity + lifecycle), and _get_or_create_httpx_client() (shared
HTTP client factory).
Attributes
attribute server_path
Section titled “attribute server_path”str | Sequence[str]
Worker command used for subprocess transport — a string to be split, or an argv sequence taken as-is.
Methods
method catalogs
Section titled “method catalogs”catalogs() -> list[CatalogInfo]Get list of catalog discovery records from the worker.
method catalog_attach
Section titled “method catalog_attach”catalog_attach(
*,
name: str,
options: dict[str, Any] | None = None,
data_version_spec: str | None,
implementation_version: str | None,
) -> CatalogAttachResultAttach to a catalog.
method catalog_detach
Section titled “method catalog_detach”catalog_detach(*, attach_opaque_data: AttachOpaqueData) -> NoneDetach from a catalog.
method catalog_create
Section titled “method catalog_create”catalog_create(
*,
name: str,
on_conflict: OnConflict = OnConflict.ERROR,
options: dict[str, Any] | None = None,
) -> NoneCreate a new catalog.
method catalog_drop
Section titled “method catalog_drop”catalog_drop(*, name: str) -> NoneDrop a catalog.
method catalog_version
Section titled “method catalog_version”catalog_version(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
) -> intGet the current catalog version.
method catalog_transaction_begin
Section titled “method catalog_transaction_begin”catalog_transaction_begin(
*,
attach_opaque_data: AttachOpaqueData,
) -> TransactionOpaqueData | NoneBegin a new transaction.
method catalog_transaction_commit
Section titled “method catalog_transaction_commit”catalog_transaction_commit(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData,
) -> NoneCommit a transaction.
method catalog_transaction_rollback
Section titled “method catalog_transaction_rollback”catalog_transaction_rollback(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData,
) -> NoneRollback a transaction.
method schemas
Section titled “method schemas”schemas(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
) -> list[SchemaInfo]List schemas in the catalog.
method schema_get
Section titled “method schema_get”schema_get(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
name: str,
) -> SchemaInfo | NoneGet information about a schema.
method schema_create
Section titled “method schema_create”schema_create(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
name: str,
comment: str | None = None,
tags: dict[str, str] | None = None,
) -> NoneCreate a new schema.
method schema_drop
Section titled “method schema_drop”schema_drop(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
name: str,
ignore_not_found: bool = False,
cascade: bool = False,
) -> NoneDrop a schema.
method schema_contents
Section titled “method schema_contents”schema_contents(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
name: str,
type: SchemaObjectType,
) -> Sequence[TableInfo | ViewInfo | FunctionInfo | MacroInfo | IndexInfo]List contents of a schema (tables, views, functions, macros, indexes).
method copy_formats
Section titled “method copy_formats”copy_formats(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
) -> Sequence[CopyFromFormatInfo]List the custom COPY formats this catalog advertises.
Catalog-level rather than schema-scoped, matching the C++ extension,
which registers one DuckDB CopyFunction per entry at ATTACH. Each
entry’s handler is the worker function name to pass to
Client.copy_from / Client.copy_to, and direction says which
of the two it serves ("from", "to", or "both").
method table_get
Section titled “method table_get”table_get(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
at_unit: str | None = None,
at_value: str | None = None,
) -> TableInfo | NoneGet information about a table.
method table_column_statistics
Section titled “method table_column_statistics”table_column_statistics(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
) -> list[ColumnStatistics]Fetch a table’s column statistics, decoded.
Workers may inline statistics on TableInfo (see
TableInfo.column_statistics) or serve them lazily through this per-table
call, which is what TableInfo.supports_column_statistics advertises. Prefer
the inlined copy when present and fall back to this.
method table_create
Section titled “method table_create”table_create(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
columns: SerializedSchema,
on_conflict: OnConflict = OnConflict.ERROR,
not_null_constraints: list[int] | None = None,
unique_constraints: list[list[int]] | None = None,
check_constraints: list[str] | None = None,
) -> NoneCreate a new table.
method table_drop
Section titled “method table_drop”table_drop(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
ignore_not_found: bool = False,
cascade: bool = False,
) -> NoneDrop a table.
method table_scan_function_get
Section titled “method table_scan_function_get”table_scan_function_get(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
at_unit: str | None = None,
at_value: str | None = None,
) -> ScanFunctionResultGet the scan function for a table.
Returns a ScanFunctionResult that tells the VGI DuckDB extension which
DuckDB function to call to obtain the table data.
method table_scan_branches_get
Section titled “method table_scan_branches_get”table_scan_branches_get(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
at_unit: str | None = None,
at_value: str | None = None,
) -> ScanBranchesResultGet the list of scan branches for a (possibly multi-source) table.
Multi-branch tables compose a logical scan from N physical sources
(canonical case: Kafka hot tier + Iceberg cold tier). The VGI DuckDB
extension’s optimizer rewrites the placeholder scan into
LogicalSetOperation(UNION_ALL, ...), one arm per branch — a
caller of this method is expected to do the equivalent (e.g.
pl.concat over one scan per branch in vgi-polars).
Falls back to :meth:table_scan_function_get, wrapped as a
single-branch ScanBranchesResult, when the worker raises
MethodNotImplementedError for the additive catalog_table_scan_ branches_get RPC — mirroring exactly how the C++ extension reads
that same fallback, so a worker written before this RPC existed
still works, just as a one-branch table.
method table_comment_set
Section titled “method table_comment_set”table_comment_set(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
comment: str | None,
ignore_not_found: bool = False,
) -> NoneSet or clear the comment on a table.
method table_rename
Section titled “method table_rename”table_rename(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
new_name: str,
ignore_not_found: bool = False,
) -> NoneRename a table.
method table_column_add
Section titled “method table_column_add”table_column_add(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
column_definition: SerializedSchema,
ignore_not_found: bool = False,
if_column_not_exists: bool = False,
) -> NoneAdd a new column to a table.
method table_column_drop
Section titled “method table_column_drop”table_column_drop(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
column_name: str,
ignore_not_found: bool = False,
if_column_exists: bool = False,
cascade: bool = False,
) -> NoneDrop a column from a table.
method table_column_rename
Section titled “method table_column_rename”table_column_rename(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
column_name: str,
new_column_name: str,
ignore_not_found: bool = False,
) -> NoneRename a column.
method table_column_default_set
Section titled “method table_column_default_set”table_column_default_set(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
column_name: str,
expression: SqlExpression,
ignore_not_found: bool = False,
) -> NoneSet the default value expression for a column.
method table_column_default_drop
Section titled “method table_column_default_drop”table_column_default_drop(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
column_name: str,
ignore_not_found: bool = False,
) -> NoneRemove the default value from a column.
method table_column_type_change
Section titled “method table_column_type_change”table_column_type_change(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
column_definition: SerializedSchema,
expression: SqlExpression | None = None,
ignore_not_found: bool = False,
) -> NoneChange the type of a column.
method table_not_null_drop
Section titled “method table_not_null_drop”table_not_null_drop(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
column_name: str,
ignore_not_found: bool = False,
) -> NoneRemove NOT NULL constraint from a column.
method table_not_null_set
Section titled “method table_not_null_set”table_not_null_set(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
column_name: str,
ignore_not_found: bool = False,
) -> NoneAdd NOT NULL constraint to a column.
method view_get
Section titled “method view_get”view_get(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
) -> ViewInfo | NoneGet information about a view.
method view_create
Section titled “method view_create”view_create(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
definition: str,
on_conflict: OnConflict = OnConflict.ERROR,
) -> NoneCreate a new view.
method view_drop
Section titled “method view_drop”view_drop(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
ignore_not_found: bool = False,
cascade: bool = False,
) -> NoneDrop a view.
method view_rename
Section titled “method view_rename”view_rename(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
new_name: str,
ignore_not_found: bool = False,
) -> NoneRename a view.
method view_comment_set
Section titled “method view_comment_set”view_comment_set(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
comment: str | None,
ignore_not_found: bool = False,
) -> NoneSet or clear the comment on a view.
method macro_get
Section titled “method macro_get”macro_get(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
) -> MacroInfo | NoneGet information about a macro.
method macro_create
Section titled “method macro_create”macro_create(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
macro_type: MacroType,
parameters: list[str],
definition: str,
on_conflict: OnConflict = OnConflict.ERROR,
parameter_default_values: pa.RecordBatch | None = None,
arguments_schema: pa.Schema | None = None,
) -> NoneCreate a new macro.
method macro_drop
Section titled “method macro_drop”macro_drop(
*,
attach_opaque_data: AttachOpaqueData,
transaction_opaque_data: TransactionOpaqueData | None = None,
schema_name: str,
name: str,
ignore_not_found: bool = False,
) -> NoneDrop a macro.
class Client
Section titled “class Client”Bases: CatalogClientMixin, AggregateClientMixin
Description
Canonical VGI client — HTTP is the path other-language ports mirror.
Two transports:
- HTTP (
Client.from_http(base_url, bearer_token=...)). The canonical non-DuckDB path. Usesvgi_rpc.http.http_connectunder the hood; transparently resolves pointer batches returned by workers that externalize large outputs (demo storage, S3). Transparently externalizes large input batches when the server advertises upload-URL support. - Subprocess (
Client(server_path)). Python-only convenience for local workers. Uses shell subprocesses + aWorkerPoolfor reuse. Ports don’t need to mirror this.
Catalog operations (catalogs(), schema_contents(), etc.) are
provided by CatalogClientMixin and don’t require start(). They
open a short-lived connection per call (HTTP) or borrow a pooled
subprocess worker.
Aggregate invocation (aggregate_function, aggregate_session,
aggregate_streaming) comes from AggregateClientMixin. Custom
COPY formats reuse the table-function and buffered drivers with a COPY
context attached — see copy_from / copy_to, and copy_formats()
for discovery.
Function invocation (scalar_function, table_function,
table_in_out_function, aggregate_function) requires start() —
typically via the context-manager protocol:
with Client.from_http("http://host:port", bearer_token="...") as c: for batch in c.table_function(function_name="sequence", ...): ...Attributes
attribute THREAD_JOIN_TIMEOUT
Section titled “attribute THREAD_JOIN_TIMEOUT”float
Seconds to wait for a worker thread to join during shutdown.
attribute PROCESS_WAIT_TIMEOUT
Section titled “attribute PROCESS_WAIT_TIMEOUT”float
Seconds to wait for a worker process to exit during
shutdown before killing it. A hang guard — teardown never blocks past
this, and never raises TimeoutExpired at the caller.
attribute STDERR_DRAIN_TIMEOUT
Section titled “attribute STDERR_DRAIN_TIMEOUT”float
Seconds to wait for a worker’s stderr to finish draining before an error message is built from it.
attribute server_path
Section titled “attribute server_path”str | Sequence[str]
Subprocess-only. The VGI worker command. A string is
split with shlex.split; pass a sequence to give
the argv exactly, which is what you want for arguments carrying
spaces or quotes ([sys.executable, "-c", script]). No shell
is involved either way, so shell syntax — pipes, redirection,
VAR=value prefixes, ~ expansion — is not interpreted.
attribute passthrough_stderr
Section titled “attribute passthrough_stderr”Subprocess-only. If True, worker stderr is passed through to the parent process’s stderr in real-time.
attribute supports_resumable_scan
Section titled “attribute supports_resumable_scan”bool
Whether this client’s transport can drive :meth:table_scan_resumable.
True only for HTTP, whose producer streams round-trip state in continuation tokens. The pipe/subprocess transport holds a live stream with no serializable resume point.
Methods
method from_http
Section titled “method from_http”from_http(
base_url: str,
*,
bearer_token: str | None = None,
oauth: bool = False,
oauth_refresh_token: str | None = None,
oauth_flow: Literal[‘auto’, ‘device_code’, ‘pkce’] = ‘auto’,
oauth_timeout_seconds: float = 120.0,
oauth_prompt: Literal[‘none’, ‘login’, ‘select_account’, ‘consent’] = ‘none’,
httpx_client: Any | None = None,
external_location: Any | None = None,
worker_limit: int | None = None,
attach_opaque_data: bytes | None = None,
) -> ClientCreate a Client bound to a remote HTTP VGI worker.
Canonical entry point for non-DuckDB callers (e.g. a TypeScript port
browsing catalog contents). Subprocess-specific kwargs are not
accepted; pool/stderr semantics do not apply. See Client.__init__
for what oauth/oauth_refresh_token/oauth_flow do.
method from_tcp
Section titled “method from_tcp”from_tcp(
host: str,
port: int,
*,
external_location: Any | None = None,
worker_limit: int | None = None,
attach_opaque_data: bytes | None = None,
) -> ClientCreate a Client bound to a running TCP VGI worker.
Connects via vgi_rpc.rpc.tcp_connect (raw Arrow-IPC framing). The
framing carries no authentication or encryption — only connect to
trusted endpoints on loopback or a trusted network; use
Client.from_http(...) for untrusted networks. Spin up a matching
worker with vgi-fixture-worker --tcp [HOST:]PORT.
method from_launch
Section titled “method from_launch”from_launch(
worker_argv: Sequence[str],
*,
idle_timeout: float = 300.0,
state_dir: str | None = None,
socket_path: str | None = None,
external_location: Any | None = None,
worker_limit: int | None = None,
attach_opaque_data: bytes | None = None,
) -> ClientCreate a Client bound to a launcher-managed warm worker.
Spawns (or reuses) a worker process serving over an AF_UNIX
socket via vgi_rpc.launcher — every client across the machine
pointing at the same worker_argv shares one warm worker,
coordinated by a per-command-hash flock, and the worker
self-terminates after idle_timeout idle seconds with zero
connected clients. This is the Python client-side counterpart to the
VGI DuckDB extension’s launch:<argv> LOCATION scheme. Requires
the vgi-python[launch] extra.
method get_worker_stderr
Section titled “method get_worker_stderr”get_worker_stderr() -> strReturn all captured stderr from the worker processes.
Returns stderr output from the primary worker and all additional workers spawned for parallel processing. The output is accumulated in a shared buffer throughout the client’s lifetime.
This method is thread-safe and can be called while processing is ongoing, though the buffer may not yet contain all output until the workers have completed.
Note
This method only returns data when passthrough_stderr=False was set in the constructor. When passthrough_stderr=True, stderr goes directly to the parent process’s stderr and this method returns an empty string.
method oauth_identity
Section titled “method oauth_identity”oauth_identity() -> Any | NoneReturn the signed-in OAuth identity’s parsed id_token claims.
Returns None when this client isn’t using OAuth, or hasn’t
completed a login yet (the identity is only known once a real
exchange has happened — calling this before the first request
never triggers one). See vgi_rpc.http.OAuthIdentity for the
fields (sub/email/name/issuer/claims).
method start
Section titled “method start”start() -> NoneStart the primary worker subprocess.
Spawns the worker process using the server_path configured in init,
sets up RPC transport, and creates a typed VgiProtocol proxy for
method calls.
After this method returns, the client is ready to invoke functions via table_in_out_function(), table_function(), or scalar_function(). When using the context manager protocol (with statement), this method is called automatically.
The stderr buffer is cleared when start() is called, so any stderr from previous runs is discarded.
method stop
Section titled “method stop”stop(*, force: bool = False) -> intStop all worker subprocesses and clean up resources.
Terminates all workers in the following order:
- Stops all additional workers (spawned for parallel processing)
- Stops the primary worker
- Waits for all stderr drain threads to complete (with timeout)
- Resets all internal state
After this method returns, the client can be started again with start(). When using the context manager protocol (with statement), this method is called automatically on exit.
Cancelling an in-flight call
A graceful stop waits for a worker that is blocked inside a handler, so it
cannot be used to abandon a scan that has overrun its budget. Pass
force=True to SIGKILL direct subprocess workers first, which unblocks any
thread waiting on their output immediately. Note this only applies to direct
subprocess workers: a pooled worker is returned to its pool rather than owned
by this client, so construct the client with pool=None when you need to be
able to cancel it. HTTP and TCP workers are already prompt to close.
method server_capabilities
Section titled “method server_capabilities”server_capabilities() -> AnyReturn the HTTP server’s advertised capabilities.
Only valid for HTTP-mode clients. The returned
HttpServerCapabilities carries max_request_bytes,
upload_url_support, and max_upload_bytes — the fields the
client consults before deciding to externalize large input batches
via upload URLs (see Phase 4 of the whimsical-mccarthy plan).
method table_in_out_function
Section titled “method table_in_out_function”table_in_out_function(
*,
function_name: str,
schema_name: str,
input: Iterator[pa.RecordBatch],
arguments: Arguments | None = None,
bind_result_callback: Callable[[BindResponse], None] | None = None,
projection_ids: list[int] | None = None,
pushdown_filters: bytes | None = None,
join_keys: list[pa.RecordBatch] | None = None,
settings: dict[str, Any] | None = None,
transaction_opaque_data: bytes | None = None,
parent_row_callback: Callable[[list[int]], None] | None = None,
has_finalize: bool = True,
) -> Generator[pa.RecordBatch]Invoke a table-in-out function on the worker and stream results.
For parallel processing (max_workers > 1), input batches are distributed round-robin across workers using dedicated threads. Output order may not match input order in parallel mode. Only the primary worker receives the FINALIZE phase and produces final aggregated output.
method table_buffering_function
Section titled “method table_buffering_function”table_buffering_function(
*,
function_name: str,
schema_name: str,
input: Iterator[pa.RecordBatch],
arguments: Arguments | None = None,
bind_result_callback: Callable[[BindResponse], None] | None = None,
projection_ids: list[int] | None = None,
pushdown_filters: bytes | None = None,
settings: dict[str, Any] | None = None,
transaction_opaque_data: bytes | None = None,
copy_to: CopyToContext | None = None,
input_schema: pa.Schema | None = None,
) -> Generator[pa.RecordBatch]Invoke a TableBufferingFunction (Sink+Source) and stream results.
This mirrors the C++ PhysicalVgiTableBufferingFunction operator
rather than the streaming INPUT/FINALIZE path used by
:meth:table_in_out_function. The sequence is:
bind→init(phase=TABLE_BUFFERING)on the primary worker. The sink init persists init metadata to cold storage so any pool worker can serve subsequent process/combine RPCs; its stream carries no data, so it is closed immediately after the header.table_buffering_process(unary) per input batch — the worker sinks the batch and returns an opaquestate_id.table_buffering_combine(unary) once at end-of-input — the worker hands allstate_ids to usercombine()and returns opaquefinalize_state_ids (the source-side partition keys).init(phase=TABLE_BUFFERING_FINALIZE, finalize_state_id=...)per finalize key — a producer stream driving userfinalize()per tick. Output batches are yielded in finalize-key order.table_buffering_destructor(unary, best-effort) for cleanup.
Unlike :meth:table_in_out_function this driver runs entirely on the
primary worker connection (process/combine are unary RPCs); the
worker buffers all input regardless, so the aggregate result is
identical to the distributed C++ path.
method bind
Section titled “method bind”bind(
*,
function_name: str,
schema_name: str,
arguments: Arguments | None = None,
function_type: FunctionType = FunctionType.TABLE,
settings: dict[str, Any] | None = None,
transaction_opaque_data: bytes | None = None,
) -> BindResponseResolve a function’s bind response without running init()/process().
Runs only the bind() RPC — no init(), no worker execution, no
data produced. This is the schema-discovery primitive table_function()/
table_in_out_function()/scalar_function() lack a standalone version
of: their own bind_result_callback only fires as a side effect of a
generator’s first next(), which has already started init() and
real execution by the time it runs. Prefer the catalog RPCs
(Client.table_get, Client.schema_contents(type=TABLE_FUNCTION))
when a catalog attach is available — those are equally zero-execution
and additionally expose pushdown-capability flags this method does
not. Use this method for a bare (non-catalog) function name, where no
attach exists to ask instead.
method table_function_plan
Section titled “method table_function_plan”table_function_plan(
*,
function_name: str,
schema_name: str,
arguments: Arguments | None = None,
projection_ids: list[int] | None = None,
pushdown_filters: bytes | None = None,
join_keys: list[pa.RecordBatch] | None = None,
settings: dict[str, Any] | None = None,
transaction_opaque_data: bytes | None = None,
target_split_bytes: int | None = None,
min_splits: int | None = None,
max_splits_per_response: int | None = None,
cursor: bytes | None = None,
) -> PlanResponsePlan a table-function scan into named, independently redeemable splits.
Runs bind() then on_plan() and returns the resulting
PlanResponse, whose splits are each individually redeemable by
:meth:table_function via its split_tokens argument — from this
process or, since a split names work rather than describing it (“these
three files at version 47”, not “rows 0-999 of whatever this returns
now”), any other. Workers that don’t opt in via supports_splits
(FunctionInfo.supports_splits) inherit a framework default (commonly
one split for the whole scan) — check that flag first if the caller
cares whether real parallelism/checkpointing is available versus a
single degenerate split.
A response’s next_cursors is normally empty or one entry; more than
one means the plan is paginated across parallel, disjoint enumeration
branches — the caller is responsible for that disjointness (VGI
itself does not verify it; see the “Split disjointness is a worker
contract” note in the VGI extension’s own docs). For a single
sequential caller, following next_cursors one at a time (via
cursor=) and concatenating each response’s splits is always
correct, whether the plan doled out one cursor or several.
method table_function
Section titled “method table_function”table_function(
*,
function_name: str,
schema_name: str,
arguments: Arguments | None = None,
bind_result_callback: Callable[[BindResponse], None] | None = None,
projection_ids: list[int] | None = None,
pushdown_filters: bytes | None = None,
join_keys: list[pa.RecordBatch] | None = None,
settings: dict[str, Any] | None = None,
transaction_opaque_data: bytes | None = None,
copy_from: CopyFromContext | None = None,
split_tokens: list[bytes] | None = None,
split_execution_id: bytes | None = None,
split_init_opaque_data: bytes | None = None,
batch_metadata_callback: Callable[[pa.KeyValueMetadata | None], None] | None = None,
at_unit: str | None = None,
at_value: str | None = None,
) -> Generator[pa.RecordBatch]Invoke a table function (source function) and stream output batches.
Table functions generate output batches without receiving input data. They are useful for data sources, generators, or functions that produce results based solely on their arguments.
For parallel processing (max_workers > 1), output is read from all
workers concurrently using threads. Output order is non-deterministic.
This is unrelated to (and mutually exclusive in effect with) redeeming
splits — see split_tokens below.
method table_scan_resumable
Section titled “method table_scan_resumable”table_scan_resumable(
*,
function_name: str,
schema_name: str,
arguments: Arguments | None = None,
projection_ids: list[int] | None = None,
pushdown_filters: bytes | None = None,
settings: dict[str, Any] | None = None,
transaction_opaque_data: bytes | None = None,
resume_token: bytes | None = None,
) -> ResumableTableScanOpen (or resume) a resumable table-function scan.
Resumable variant of :meth:table_function: the returned
:class:ResumableTableScan yields (batch, token) one batch at a
time, surfacing the worker’s continuation token so a stateless caller
can persist it and resume on another process/node.
When resume_token is given, the scan continues from that token
(the bind/init is still issued — the upstream’s first turn is produced
and discarded — so the same function_name/projection/filters must
be supplied). When None, a fresh scan starts.
method table_scan_continue
Section titled “method table_scan_continue”table_scan_continue(
*,
resume_token: bytes,
output_schema: pa.Schema | None = None,
) -> ResumableTableScanResume a producer table scan from a continuation token WITHOUT re-binding.
The cheap counterpart to table_scan_resumable(resume_token=...): a continuation
token is a signed, self-describing snapshot of the worker’s producer state, so the
server recovers state + schemas + function identity from the token alone. This skips
the bind/init round-trip (and the discarded first turn) that
table_scan_resumable pays — the right primitive for a stateless relay that holds
a per-batch token and resumes on any node every batch.
The client must be started and connected to a worker that honours the token (the
token is verified against the caller’s auth identity, and routed by the same
init stream method that minted it). HTTP transport only.
method copy_from
Section titled “method copy_from”copy_from(
*,
function_name: str,
schema_name: str,
format: str,
file_path: str,
expected_schema: pa.Schema,
arguments: Arguments | None = None,
bind_result_callback: Callable[[BindResponse], None] | None = None,
projection_ids: list[int] | None = None,
pushdown_filters: bytes | None = None,
settings: dict[str, Any] | None = None,
transaction_opaque_data: bytes | None = None,
) -> Generator[pa.RecordBatch]Read a custom COPY ... FROM format and stream the parsed rows.
A CopyFromFunction is an ordinary producer-mode table function that
additionally receives a CopyFromContext, so this is
:meth:table_function with that context attached — the same shape the
C++ extension’s copy_from_bind produces.
Discover the (format, handler) pairs a catalog advertises with
client.copy_formats(attach_opaque_data=...); handler is the
function_name to pass here.
method copy_to
Section titled “method copy_to”copy_to(
*,
function_name: str,
schema_name: str,
format: str,
file_path: str,
input: Iterator[pa.RecordBatch],
input_schema: pa.Schema | None = None,
arguments: Arguments | None = None,
bind_result_callback: Callable[[BindResponse], None] | None = None,
settings: dict[str, Any] | None = None,
transaction_opaque_data: bytes | None = None,
) -> NoneWrite input to a custom COPY ... TO format and close it.
A CopyToFunction is a buffered Sink+Combine function with no Source
phase, so this is :meth:table_buffering_function with a
CopyToContext attached: every batch is sunk via
table_buffering_process and the terminal write happens once inside
table_buffering_combine. Returns when the destination is closed.
Unlike the C++ path this drives a single worker connection, so ordered
writers (Meta.sink_order_dependent) see source order for free.
method scalar_function
Section titled “method scalar_function”scalar_function(
*,
function_name: str,
schema_name: str,
input: Iterator[pa.RecordBatch],
arguments: Arguments | None = None,
bind_result_callback: Callable[[BindResponse], None] | None = None,
settings: dict[str, Any] | None = None,
secrets: dict[str, Any] | None = None,
transaction_opaque_data: bytes | None = None,
) -> Generator[pa.RecordBatch]Invoke a scalar function on the worker and stream results.
Scalar functions transform input batches to single-column output with 1:1 row mapping. Processing ends when input is exhausted.
For parallel processing (max_workers > 1), input batches are distributed round-robin across workers using dedicated threads. Output order may not match input order in parallel mode.
Inherited members (43)
aggregate_sessionmethod · from AggregateClientMixin — Bind an aggregate and yield a session over its raw RPCs.aggregate_bindmethod · from AggregateClientMixin — Bind an aggregate function without taking responsibility for teardown.aggregate_functionmethod · from AggregateClientMixin — Run an aggregate overinputand return one row per group.aggregate_streamingmethod · from AggregateClientMixin — Open a streaming-partitioned aggregate session and yield it.catalogsmethod · from CatalogClientMixin — Get list of catalog discovery records from the worker.catalog_attachmethod · from CatalogClientMixin — Attach to a catalog.catalog_detachmethod · from CatalogClientMixin — Detach from a catalog.catalog_createmethod · from CatalogClientMixin — Create a new catalog.catalog_dropmethod · from CatalogClientMixin — Drop a catalog.catalog_versionmethod · from CatalogClientMixin — Get the current catalog version.catalog_transaction_beginmethod · from CatalogClientMixin — Begin a new transaction.catalog_transaction_commitmethod · from CatalogClientMixin — Commit a transaction.catalog_transaction_rollbackmethod · from CatalogClientMixin — Rollback a transaction.schemasmethod · from CatalogClientMixin — List schemas in the catalog.schema_getmethod · from CatalogClientMixin — Get information about a schema.schema_createmethod · from CatalogClientMixin — Create a new schema.schema_dropmethod · from CatalogClientMixin — Drop a schema.schema_contentsmethod · from CatalogClientMixin — List contents of a schema (tables, views, functions, macros, indexes).copy_formatsmethod · from CatalogClientMixin — List the customCOPYformats this catalog advertises.table_getmethod · from CatalogClientMixin — Get information about a table.table_column_statisticsmethod · from CatalogClientMixin — Fetch a table’s column statistics, decoded.table_createmethod · from CatalogClientMixin — Create a new table.table_dropmethod · from CatalogClientMixin — Drop a table.table_scan_function_getmethod · from CatalogClientMixin — Get the scan function for a table.table_scan_branches_getmethod · from CatalogClientMixin — Get the list of scan branches for a (possibly multi-source) table.table_comment_setmethod · from CatalogClientMixin — Set or clear the comment on a table.table_renamemethod · from CatalogClientMixin — Rename a table.table_column_addmethod · from CatalogClientMixin — Add a new column to a table.table_column_dropmethod · from CatalogClientMixin — Drop a column from a table.table_column_renamemethod · from CatalogClientMixin — Rename a column.table_column_default_setmethod · from CatalogClientMixin — Set the default value expression for a column.table_column_default_dropmethod · from CatalogClientMixin — Remove the default value from a column.table_column_type_changemethod · from CatalogClientMixin — Change the type of a column.table_not_null_dropmethod · from CatalogClientMixin — Remove NOT NULL constraint from a column.table_not_null_setmethod · from CatalogClientMixin — Add NOT NULL constraint to a column.view_getmethod · from CatalogClientMixin — Get information about a view.view_createmethod · from CatalogClientMixin — Create a new view.view_dropmethod · from CatalogClientMixin — Drop a view.view_renamemethod · from CatalogClientMixin — Rename a view.view_comment_setmethod · from CatalogClientMixin — Set or clear the comment on a view.macro_getmethod · from CatalogClientMixin — Get information about a macro.macro_createmethod · from CatalogClientMixin — Create a new macro.macro_dropmethod · from CatalogClientMixin — Drop a macro.
class ClientError
Section titled “class ClientError”Bases: Exception
Description
Error raised by Client operations.
The first line of str(ClientError) is the remote exception as the
worker raised it ({error_type}: {error_message}), so that whatever
a user typed into their raise ValueError(...) shows up at the top of
their traceback instead of being buried under VGI framing. Remote
traceback and worker-stderr excerpts, when present, follow after an
empty line.
Methods
method from_rpc_error
Section titled “method from_rpc_error”from_rpc_error(e: RpcError) -> ClientErrorCreate a ClientError from an RpcError, including remote traceback.
Lead with the user’s exception (error_type: error_message) so
the most actionable line is first. The Remote traceback section
trails and is only included when the worker produced one.
function main
Section titled “function main”main() -> NoneCLI entry point for vgi-client.
class OutputWriter
Section titled “class OutputWriter”Description
Handles writing output batches in various formats.
Supported formats
- json: JSON Lines format (one JSON object per row)
- csv: CSV with header
- parquet: Apache Parquet columnar format
- arrow-ipc: Apache Arrow IPC streaming format (useful for debugging)
The arrow-ipc format writes batches in the standard Arrow IPC streaming format, which can be read by any Arrow implementation. This is useful for:
- Debugging VGI protocol issues- Inspecting raw output data with tools like pyarrow or arrow CLI- Piping data to other Arrow-aware toolsAttributes
attribute output_file
Section titled “attribute output_file”Path to output file, “-” for stdout, or None for logging.
attribute format
Section titled “attribute format”Output format (“parquet”, “csv”, “json”, or “arrow-ipc”).
Methods
method write_batch
Section titled “method write_batch”write_batch(batch: pa.RecordBatch) -> NoneWrite a batch to the output destination in the configured format.
method close
Section titled “method close”close() -> NoneClose the underlying writer if one exists.
class ResumableTableScan
Section titled “class ResumableTableScan”Description
A resumable, one-batch-at-a-time handle on an upstream table-function scan.
Unlike :meth:Client.table_function (a live generator that hides the
server’s continuation token), each :meth:next returns (batch, token)
where token is the worker’s serialized producer state AFTER batch.
A stateless client (e.g. a load-balanced proxy) can persist token, drop
the connection, and resume on another node via
Client.table_scan_resumable(resume_token=token, ...).
Single-worker: reads the primary stream only (parallel max_workers>1
reads are unordered and not resumable from a single token).
Methods
method next
Section titled “method next”next() -> tuple[pa.RecordBatch | None, bytes | None]Return (batch, resume_token); (None, None) at end-of-stream.
resume_token resumes the scan AFTER batch on any node.
method close
Section titled “method close”close() -> NoneRelease the underlying stream (no-op over HTTP — stateless).
class ResumeUnsupported
Section titled “class ResumeUnsupported”Bases: ClientError
Description
Raised when a resumable scan is requested on a non-resumable transport.
Only the HTTP transport round-trips producer state in continuation tokens,
so only HTTP clients can drive :meth:Client.table_scan_resumable. On the
pipe/subprocess transport the stream is a live connection with no
serializable resume point; the caller must keep the live stream in-process
instead.
Inherited members (1)
from_rpc_errormethod · from ClientError — Create aClientErrorfrom anRpcError, including remote traceback.