vgi.worker

Module overview

VGI Worker base class for hosting user-defined functions and catalogs.

A worker is a subprocess that communicates via stdin/stdout using Arrow IPC. Workers are spawned by a client as needed and terminate once they detect their input stream has been closed.

SUPPORTED FUNCTION TYPES

The worker supports three function types, dispatched based on class inheritance:

1. ScalarFunction / ScalarFunctionGenerator: Transforms input batches to single-column output with 1:1 row mapping. Use for per-row computations.

2. TableInOutFunction / TableInOutGenerator: Reads input batches, produces output batches. Use for transforming, filtering, or aggregating input.

3. TableFunctionGenerator: Generates output batches without reading input. Use for data generation functions like sequence(), range(), etc.

QUICK START

Create a worker by subclassing Worker and listing your functions:

from vgi.worker import Worker
from vgi.scalar_function import ScalarFunction
from vgi.table_in_out_function import TableInOutGenerator
from vgi.table_function import TableFunctionGenerator

class DoubleColumn(ScalarFunction):
  # Single-column output with 1:1 row mapping
  ...

class EchoFunction(TableInOutGenerator):
  # Transforms input batches
  ...

class SequenceFunction(TableFunctionGenerator):
  # Generates output without input
  ...

class MyWorker(Worker):
  functions = [DoubleColumn, EchoFunction, SequenceFunction]

if __name__ == "__main__":
  MyWorker().run()

Function names are derived from metadata (Meta.name or class name converted to snake_case). No manual name mapping required.

KEY CLASSES

Worker      - Base class to subclass (set functions attribute)

See Also:

vgi.client.Client : Spawns workers and sends data to them vgi.function.Function : Base class for all functions vgi._test_fixtures.worker : Example worker with built-in functions

function run_table_buffering_finalize_tick

source

run_table_buffering_finalize_tick(
state: Any,
out: Any,
ctx: Any,
) -> None

One tick of cls.finalize(params, fid, state, out).

Lazy-imported by TableBufferingFinalizeState.produce() to break the protocol→worker import cycle. Cold-resolves func_cls + params on every call (no in-process cache — different worker processes may handle different ticks under HTTP).

Applies the pushdown contract symmetric with the streaming TableInOutExchangeState (protocol.py:1106-1186): narrow params.output_schema to the projected slots and, when Meta.auto_apply_filters is True, wrap out in a filtering collector so the user’s finalize() doesn’t need to know.

class Worker

source

Description

Base class for VGI workers that host user-defined functions.

Subclass this and define a functions class attribute listing your function classes. Function names are derived from metadata (Meta.name or snake_case of class name). The worker handles the VGI protocol via vgi_rpc.RpcServer.

Multiple functions can share the same name if they have different argument signatures (function overloading). The worker will select the appropriate function based on the invocation’s arguments.

Catalog interface

If catalog_interface is not set but functions is non-empty, a default read-only catalog interface is created automatically. This exposes the worker’s functions via the catalog protocol, allowing clients to discover available functions.

To customize the catalog, set catalog_interface to a CatalogInterface subclass. To disable the catalog entirely, set catalog_interface = None and catalog_name = None. A catalog-less worker is reachable only from the pure-Python Client — DuckDB reaches VGI functions exclusively through ATTACH, which requires a catalog.

Attributes

attribute functions

Sequence[type[Function]]

Function classes this worker hosts.

attribute protocol_class

type[VgiProtocol]

Protocol class handed to RpcServer; defaults to the real VgiProtocol. Test fixtures override it with a subclass that redeclares protocol_version to exercise version-mismatch enforcement.

attribute catalog_interface

type[CatalogInterface] | None

Custom CatalogInterface subclass, or None to use the auto-generated default (or disable the catalog).

attribute catalog_name

str | None

Name of the default catalog; set to None to disable the default catalog.

attribute catalog

Catalog | None

Optional declarative Catalog describing the worker’s schemas, tables, and views.

Methods

method resolve_token

source

resolve_token(token: str) -> TokenIdentity | None

Resolve an opaque bearer credential to the identity it authenticates as.

Override to enable POST {prefix}/__introspect_token__, which a reverse proxy calls when it terminates the only public listener and must know which principal a credential is before it can authorize anything. Until it is overridden the route does not exist at all — not “exists and refuses”, absent — so no worker grows a credential-to-identity oracle by upgrading a dependency.

Enabling it also requires an allowlist of principals permitted to ask (--introspect-principals / VGI_INTROSPECT_PRINCIPALS). There is no permissive default: authenticating and introspecting are different capabilities, and a deployment where any valid credential may introspect lets any user resolve any other user’s credential to its owner. Overriding this without setting the allowlist is a startup error rather than a silently-open endpoint.

This is deliberately not “run the credential back through the worker’s own authenticate chain” — see vgi_rpc.http.server._introspect for the four ways that breaks. Write a narrow lookup against whatever store issued the credential.

Parameters

token

The opaque credential. Never a JWS — three-segment credentials are refused before they reach here, because routing one onward would hand a third party a token the asker may itself have rejected.

Returns

A TokenIdentity whose principal is in the exact form this worker would derive itself, or None when the credential does not resolve. Never claims: a pass-through claims field would let a worker choose its caller’s tenant routing and policy branch.

Raises

AuthUnavailableError

When the answer is not knowable — the backing store is down, a timeout, a 5xx from a remote authority. Distinct from None, which means the store answered and the credential is unknown. A caller that negative-caches the second must not cache the first.

method main

source

main() -> None

Run this worker as a CLI application with logging options.

By default, serves over stdin/stdout (pipe transport). Pass --http to serve over HTTP instead.

Supports --quiet, --debug, --log-level, --log-logger, and --log-format for logging control.

HTTP-specific options (only used with --http): --host, --port, --prefix, --cors-origins, --describe/--no-describe.

Requires the http extra for HTTP mode: pip install vgi[http]

method bind

source

bind(request: BindRequest, ctx: CallContext) -> BindResponse

Resolve output schema and validate arguments.

Implements VgiProtocol.bind().

method table_function_cardinality

source

table_function_cardinality(
request: TableFunctionCardinalityRequest,
ctx: CallContext,
) -> TableCardinality

Estimate the cardinality of a table function’s output.

Implements VgiProtocol.table_function_cardinality().

method table_function_statistics

source

table_function_statistics(
request: TableFunctionStatisticsRequest,
ctx: CallContext,
) -> bytes | None

Return per-column statistics for a table function’s output.

Implements VgiProtocol.table_function_statistics(). Returns IPC bytes of the serialized ColumnStatistics batch (same wire shape as catalog_table_column_statistics_get), or None when stats are unknown.

method table_function_dynamic_to_string

source

table_function_dynamic_to_string(
request: TableFunctionDynamicToStringRequest,
ctx: CallContext,
) -> TableFunctionDynamicToStringResponse

Return user diagnostics for EXPLAIN ANALYZE Extra Info.

Implements VgiProtocol.table_function_dynamic_to_string(). Fired once per parallel scan thread post-execution. Best-effort: any exception (including a misbehaving user override) is logged and an empty response is returned so the EA query never aborts.

method aggregate_bind

source

aggregate_bind(
request: AggregateBindRequest,
ctx: CallContext,
) -> AggregateBindResponse

Bind an aggregate function, return output schema and execution_id.

method aggregate_update

source

aggregate_update(
request: AggregateUpdateRequest,
ctx: CallContext,
) -> AggregateUpdateResponse

Accumulate rows from a DataChunk into per-group state.

method aggregate_combine

source

aggregate_combine(
request: AggregateCombineRequest,
ctx: CallContext,
) -> AggregateCombineResponse

Merge source states into target states.

method aggregate_finalize

source

aggregate_finalize(
request: AggregateFinalizeRequest,
ctx: CallContext,
) -> AggregateFinalizeResponse

Produce results for a chunk of group_ids.

method aggregate_destructor

source

aggregate_destructor(
request: AggregateDestructorRequest,
ctx: CallContext,
) -> AggregateDestructorResponse

Best-effort cleanup of aggregate states.

method table_buffering_process

source

table_buffering_process(request: Any, ctx: CallContext) -> Any

Sink one input batch; return worker-chosen state_id (unary).

method table_buffering_combine

source

table_buffering_combine(request: Any, ctx: CallContext) -> Any

End-of-input bridge: hand all state_ids to user combine().

method table_buffering_destructor

source

table_buffering_destructor(request: Any, ctx: CallContext) -> Any

Best-effort end-of-query cleanup.

method aggregate_window_init

source

aggregate_window_init(
request: AggregateWindowInitRequest,
ctx: CallContext,
) -> AggregateWindowInitResponse

Cache a partition on the worker for windowed aggregation.

method aggregate_window

source

aggregate_window(
request: AggregateWindowRequest,
ctx: CallContext,
) -> AggregateWindowResponse

Compute one output row for a windowed aggregate.

method aggregate_window_batch

source

aggregate_window_batch(
request: AggregateWindowBatchRequest,
ctx: CallContext,
) -> AggregateWindowBatchResponse

Compute count window output rows in a single batched RPC.

method aggregate_window_destructor

source

aggregate_window_destructor(
request: AggregateWindowDestructorRequest,
ctx: CallContext,
) -> AggregateWindowDestructorResponse

Evict a cached partition from storage.

method aggregate_streaming_open

source

aggregate_streaming_open(
request: AggregateStreamingOpenRequest,
ctx: CallContext,
) -> AggregateStreamingOpenResponse

Open a streaming-partitioned aggregate session.

method aggregate_streaming_chunk

source

aggregate_streaming_chunk(
request: AggregateStreamingChunkRequest,
ctx: CallContext,
) -> AggregateStreamingChunkResponse

Process one chunk of streaming input.

method aggregate_streaming_close

source

aggregate_streaming_close(
request: AggregateStreamingCloseRequest,
ctx: CallContext,
) -> AggregateStreamingCloseResponse

End a streaming-partitioned aggregate session.

method init

source

init(
request: InitRequest,
ctx: CallContext,
) -> Stream[ProcessState, GlobalInitResponse]

Initialize a function execution and return a processing stream.

Implements VgiProtocol.init(). Creates the appropriate state object based on function type and creates the appropriate state object.

method catalog_catalogs

source

catalog_catalogs() -> CatalogsResponse

List available catalog discovery records.

method catalog_attach

source

catalog_attach(
request: CatalogAttachRequest,
*,
ctx: CallContext | None = None,
) -> CatalogAttachResult

Attach to a catalog with options.

method catalog_detach

source

catalog_detach(attach_opaque_data: bytes) -> None

Detach from a catalog.

method catalog_create

source

catalog_create(request: CatalogCreateRequest) -> None

Create a new catalog.

method catalog_drop

source

catalog_drop(name: str) -> None

Drop a catalog.

method catalog_version

source

catalog_version(
attach_opaque_data: bytes,
transaction_opaque_data: bytes | None = None,
*,
ctx: CallContext | None = None,
) -> CatalogVersionResponse

Get the current catalog version.

method catalog_transaction_begin

source

catalog_transaction_begin(
attach_opaque_data: bytes,
) -> TransactionBeginResponse

Begin a new transaction.

method catalog_transaction_commit

source

catalog_transaction_commit(
attach_opaque_data: bytes,
transaction_opaque_data: bytes,
) -> None

Commit a transaction.

method catalog_transaction_rollback

source

catalog_transaction_rollback(
attach_opaque_data: bytes,
transaction_opaque_data: bytes,
) -> None

Rollback a transaction.

method catalog_schemas

source

catalog_schemas(
attach_opaque_data: bytes,
transaction_opaque_data: bytes | None = None,
) -> SchemasResponse

List schemas in the catalog.

method catalog_schema_get

source

catalog_schema_get(
attach_opaque_data: bytes,
name: str,
transaction_opaque_data: bytes | None = None,
) -> SchemasResponse

Get information about a schema. Returns 0 or 1 items.

method catalog_schema_create

source

catalog_schema_create(
attach_opaque_data: bytes,
name: str,
on_conflict: OnConflict = OnConflict.ERROR,
comment: str | None = None,
tags: dict[str, str] | None = None,
transaction_opaque_data: bytes | None = None,
) -> None

Create a new schema.

method catalog_schema_drop

source

catalog_schema_drop(
attach_opaque_data: bytes,
name: str,
ignore_not_found: bool = False,
cascade: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Drop a schema.

method catalog_schema_contents_tables

source

catalog_schema_contents_tables(
attach_opaque_data: bytes,
name: str,
transaction_opaque_data: bytes | None = None,
) -> TablesResponse

List tables in a schema.

method catalog_schema_contents_views

source

catalog_schema_contents_views(
attach_opaque_data: bytes,
name: str,
transaction_opaque_data: bytes | None = None,
) -> ViewsResponse

List views in a schema.

method catalog_schema_contents_functions

source

catalog_schema_contents_functions(
attach_opaque_data: bytes,
name: str,
type: SchemaObjectType,
transaction_opaque_data: bytes | None = None,
) -> FunctionsResponse

List functions in a schema (scalar or table).

method catalog_copy_from_formats

source

catalog_copy_from_formats(
attach_opaque_data: bytes,
transaction_opaque_data: bytes | None = None,
) -> CopyFromFormatsResponse

List custom COPY ... FROM formats advertised by this catalog.

method catalog_table_get

source

catalog_table_get(
attach_opaque_data: bytes,
schema_name: str,
name: str,
at_unit: str | None = None,
at_value: str | None = None,
transaction_opaque_data: bytes | None = None,
) -> TablesResponse

Get information about a table. Returns 0 or 1 items.

method catalog_table_create

source

catalog_table_create(request: TableCreateRequest) -> None

Create a new table.

method catalog_table_drop

source

catalog_table_drop(
attach_opaque_data: bytes,
schema_name: str,
name: str,
ignore_not_found: bool = False,
cascade: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Drop a table.

method catalog_table_scan_function_get

source

catalog_table_scan_function_get(
attach_opaque_data: bytes,
schema_name: str,
name: str,
at_unit: str | None = None,
at_value: str | None = None,
transaction_opaque_data: bytes | None = None,
) -> bytes

Get the scan function for a table. Returns ScanFunctionResult as IPC bytes.

method catalog_table_scan_branches_get

source

catalog_table_scan_branches_get(
attach_opaque_data: bytes,
schema_name: str,
name: str,
at_unit: str | None = None,
at_value: str | None = None,
transaction_opaque_data: bytes | None = None,
) -> bytes

Get the list of scan branches for a multi-branch table.

Returns ScanBranchesResult as IPC bytes. The CatalogInterface base provides a default-impl shim that wraps the legacy table_scan_function_get as a one-branch result, so every existing single-source worker automatically responds correctly here without further code changes.

method catalog_table_column_statistics_get

source

catalog_table_column_statistics_get(
attach_opaque_data: bytes,
schema_name: str,
name: str,
transaction_opaque_data: bytes | None = None,
) -> bytes | None

Get column statistics for a table. Returns IPC bytes or None.

method catalog_table_insert_function_get

source

catalog_table_insert_function_get(
attach_opaque_data: bytes,
schema_name: str,
name: str,
transaction_opaque_data: bytes | None = None,
writable_branch_function_name: str | None = None,
) -> bytes

Get the insert function for a table. Returns WriteFunctionResult as IPC bytes.

method catalog_table_update_function_get

source

catalog_table_update_function_get(
attach_opaque_data: bytes,
schema_name: str,
name: str,
transaction_opaque_data: bytes | None = None,
) -> bytes

Get the update function for a table. Returns WriteFunctionResult as IPC bytes.

method catalog_table_delete_function_get

source

catalog_table_delete_function_get(
attach_opaque_data: bytes,
schema_name: str,
name: str,
transaction_opaque_data: bytes | None = None,
) -> bytes

Get the delete function for a table. Returns WriteFunctionResult as IPC bytes.

method catalog_table_comment_set

source

catalog_table_comment_set(
attach_opaque_data: bytes,
schema_name: str,
name: str,
comment: str | None = None,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Set or clear the comment on a table.

method catalog_table_column_comment_set

source

catalog_table_column_comment_set(
attach_opaque_data: bytes,
schema_name: str,
name: str,
column_name: str,
comment: str | None = None,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Set or clear the comment on a table column.

method catalog_table_rename

source

catalog_table_rename(
attach_opaque_data: bytes,
schema_name: str,
name: str,
new_name: str,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Rename a table.

method catalog_table_column_add

source

catalog_table_column_add(
attach_opaque_data: bytes,
schema_name: str,
name: str,
column_definition: bytes,
ignore_not_found: bool = False,
if_column_not_exists: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Add a new column to a table.

method catalog_table_column_drop

source

catalog_table_column_drop(
attach_opaque_data: bytes,
schema_name: str,
name: str,
column_name: str,
ignore_not_found: bool = False,
if_column_exists: bool = False,
cascade: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Drop a column from a table.

method catalog_table_column_rename

source

catalog_table_column_rename(
attach_opaque_data: bytes,
schema_name: str,
name: str,
column_name: str,
new_column_name: str,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Rename a column.

method catalog_table_column_default_set

source

catalog_table_column_default_set(
attach_opaque_data: bytes,
schema_name: str,
name: str,
column_name: str,
expression: str,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Set the default value expression for a column.

method catalog_table_column_default_drop

source

catalog_table_column_default_drop(
attach_opaque_data: bytes,
schema_name: str,
name: str,
column_name: str,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Remove the default value from a column.

method catalog_table_column_type_change

source

catalog_table_column_type_change(
attach_opaque_data: bytes,
schema_name: str,
name: str,
column_definition: bytes,
expression: str | None = None,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Change the type of a column.

method catalog_table_not_null_drop

source

catalog_table_not_null_drop(
attach_opaque_data: bytes,
schema_name: str,
name: str,
column_name: str,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Remove NOT NULL constraint from a column.

method catalog_table_not_null_set

source

catalog_table_not_null_set(
attach_opaque_data: bytes,
schema_name: str,
name: str,
column_name: str,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Add NOT NULL constraint to a column.

method catalog_view_get

source

catalog_view_get(
attach_opaque_data: bytes,
schema_name: str,
name: str,
transaction_opaque_data: bytes | None = None,
) -> ViewsResponse

Get information about a view. Returns 0 or 1 items.

method catalog_view_create

source

catalog_view_create(
attach_opaque_data: bytes,
schema_name: str,
name: str,
definition: str,
on_conflict: OnConflict,
transaction_opaque_data: bytes | None = None,
) -> None

Create a new view.

method catalog_view_drop

source

catalog_view_drop(
attach_opaque_data: bytes,
schema_name: str,
name: str,
ignore_not_found: bool = False,
cascade: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Drop a view.

method catalog_view_rename

source

catalog_view_rename(
attach_opaque_data: bytes,
schema_name: str,
name: str,
new_name: str,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Rename a view.

method catalog_view_comment_set

source

catalog_view_comment_set(
attach_opaque_data: bytes,
schema_name: str,
name: str,
comment: str | None = None,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Set or clear the comment on a view.

method catalog_macro_get

source

catalog_macro_get(
attach_opaque_data: bytes,
schema_name: str,
name: str,
transaction_opaque_data: bytes | None = None,
) -> MacrosResponse

Get information about a macro. Returns 0 or 1 items.

method catalog_macro_create

source

catalog_macro_create(request: MacroCreateRequest) -> None

Create a new macro.

method catalog_macro_drop

source

catalog_macro_drop(
attach_opaque_data: bytes,
schema_name: str,
name: str,
ignore_not_found: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Drop a macro.

method catalog_schema_contents_macros

source

catalog_schema_contents_macros(
attach_opaque_data: bytes,
name: str,
type: SchemaObjectType,
transaction_opaque_data: bytes | None = None,
) -> MacrosResponse

List macros in a schema (scalar or table).

method catalog_index_get

source

catalog_index_get(
attach_opaque_data: bytes,
schema_name: str,
name: str,
transaction_opaque_data: bytes | None = None,
) -> IndexesResponse

Get information about an index. Returns 0 or 1 items.

method catalog_index_create

source

catalog_index_create(request: IndexCreateRequest) -> None

Create a new index.

method catalog_index_drop

source

catalog_index_drop(
attach_opaque_data: bytes,
schema_name: str,
name: str,
ignore_not_found: bool = False,
cascade: bool = False,
transaction_opaque_data: bytes | None = None,
) -> None

Drop an index.

method catalog_schema_contents_indexes

source

catalog_schema_contents_indexes(
attach_opaque_data: bytes,
name: str,
transaction_opaque_data: bytes | None = None,
) -> IndexesResponse

List indexes in a schema.

method run

source

run(otel_config: Any = None) -> None

Run the worker, reading from stdin and writing to stdout.

Parameters

otel_config

Optional OtelConfig for OpenTelemetry instrumentation. When provided, instruments the RPC server and creates a VGI tracer.