vgi.table_in_out_function

Module overview

Framework for implementing streaming table-in-table-out functions.

TableInOutGenerator processes input batches via a per-batch callback. Each call to process() emits one output batch via out.emit().

TableInOutFunction provides a simpler callback API (transform/finish) with automatic state serialization for distributed processing.

function `pack_int_cursor`

source

pack_int_cursor(value: int) -> bytes

Encode a signed int64 cursor (e.g., last log_id consumed).

Parameters

value: The signed integer cursor to encode.

Returns

The cursor as 8 little-endian bytes.

class `RowTransformFunction`

source

Bases: TableInOutGenerator[TArgs, None]

Description

Blended (“UNNEST-style”) table-in-out: positional args ARE per-row input columns.

A RowTransformFunction collapses the classic either/or between a standard table function (literal args only) and a table-in-out function (an explicit TABLE subquery arg). Its positional Arg\s declare its per-row input columns — real typed args, NO synthetic TABLE placeholder — so ONE registration serves every call shape:

f(52, 13)                       -- literal   -> one input row
FROM t, f(t.x, t.y)             -- columns    -> streaming input
SELECT ... FROM t, LATERAL f(t.x, t.y)

Contract.

Positional args are the input columns; they are read from batch in process() (by declared name for fixed args, positionally for varargs — use :meth:input_columns). They are NOT surfaced on params.args.
Named (str-position) args stay bind-time scalars on params.args.
Map-shaped, per-row: implement process() to emit output via out.emit(). 1->1, 1->N, 1->0 all work. There is no finalize — a finalize()/finish() override is rejected at resolve_metadata (DuckDB forbids FinalExecute under correlated LATERAL, one of the call shapes blended must serve). Accumulating functions use a classic TableInput table-in-out or a TableBufferingFunction.
A positional const arg is rejected (in the column form DuckDB sweeps a constant into the input subquery; in the literal form it is indistinguishable from an input column). Use a named arg for optional config, or classic TableInput mode for a required constant.

Subclassing RowTransformFunction (not a Meta flag) IS the blended signal — a per-arg or Meta flag could be forgotten on one of N same-named overloads; inheritance cannot. function_type stays TABLE; the resolver sets ResolvedMetadata.input_from_args=True (see metadata._detect_input_from_args).

Methods

method `input_columns`

source

input_columns(batch: pa.RecordBatch) -> list[pa.Array]

This row-batch’s input columns, positionally.

For a varargs blended function the runtime column names are not known at declaration time, so read them positionally with this helper. For fixed-arity blended functions read by declared name (batch.column(name)).

Inherited members (19)

get_metadata method · from MetadataMixin — Get the resolved metadata for this function class.
describe method · from MetadataMixin — Get metadata as a dictionary (for JSON serialization).
logger attribute · from Function
storage attribute · from Function
FunctionArguments attribute · from TableFunctionBase
on_bind method · from TableInOutGenerator — Pass-through default — output schema is the input schema.
bind method · from TableFunctionBase — Bind protocol entry point. Do not override; use on_bind().
on_init method · from TableFunctionBase — One-time setup after bind, before processing batches.
global_init method · from TableFunctionBase — Global init protocol entry point. Do not override; use on_init().
cardinality method · from TableFunctionBase — Return the cardinality for the output.
dynamic_to_string method · from TableFunctionBase — Return diagnostics rendered as Extra Info under EXPLAIN ANALYZE.
statistics method · from TableFunctionBase — Return per-output-column statistics for this invocation.
pushdown_filters method · from TableFunctionBase — Get deserialized pushdown filters, or None if not present.
state_class attribute · from TableInOutGenerator
has_finalize_override method · from TableInOutGenerator — Whether this class’s finalize/finish represents real work.
initial_state method · from TableInOutGenerator — Create initial processing state. Override when TState is used.
process method · from TableInOutGenerator — Process one input batch.
finalize method · from TableInOutGenerator — Finalize processing and produce any remaining output.
on_cancel method · from TableInOutGenerator

class `TableInOutFunction`

source

Bases: TableInOutGenerator[TArgs, TState]

Description

Simplified base class using transform/finish callbacks.

This class provides a simpler API for common use cases where you don’t need to work directly with OutputCollector. Instead of implementing process() directly, you override transform() and optionally finish() as regular methods.

TState is optional. If not provided, state management is disabled and transform() will always receive state=None. When TState is an ArrowSerializableDataclass, state is automatically saved to storage after each transform() call for distributed processing.

Attributes

attribute `state_class`

type[TState] | None

The TState dataclass type, inferred automatically from the generic type parameters; None disables state management.

Methods

method `transform`

source

transform(
batch: pa.RecordBatch,
params: ProcessParams[TArgs],
state: TState | None,
) -> pa.RecordBatch | list[pa.RecordBatch]

Transform a single input batch.

Override this method to implement your transformation logic. This is called once for each input batch.

Parameters

batch: Input RecordBatch to transform.
params: ProcessParams containing arguments, schemas, and settings.
state: Mutable state that should be updated and will be serialized as needed.

Returns

- A single pa.RecordBatch: The transformed output - A list of pa.RecordBatch: Multiple outputs (will be concatenated)

method `finish`

source

finish(
params: ProcessParams[TArgs],
states: list[TState],
) -> list[pa.RecordBatch]

Return final batches after all input is processed.

Override this method to emit results after all input batches have been processed. This is useful for aggregations, sorting, or any operation that needs to see all data before producing output.

Parameters

params: The process parameters — function args, settings, secrets.
states: The accumulated per-partition states from transform().

Returns

List of pa.RecordBatch to emit as final output. Return an empty list if no finalization output is needed.

method `initial_state`

source

initial_state(params: ProcessParams[TArgs]) -> TState | None

Create the initial state for processing.

Override this method to initialize the state object before processing begins.

Parameters

params: ProcessParams containing arguments, schemas, and settings.

Returns

An instance of TState representing the initial state.

method `process`

source

process(
params: ProcessParams[TArgs],
state: TState,
batch: pa.RecordBatch,
out: OutputCollector,
) -> None

Process input batches by calling transform(). Do not override.

This method implements the exchange protocol by calling your transform() method for each input batch. State is automatically saved to storage after each call for distributed processing.

Parameters

params: Process parameters including arguments and schemas.
state: Mutable state persisted between calls. None if TState unused.
batch: The input RecordBatch to process.
out: OutputCollector for emitting output and logging.

method `finalize`

source

finalize(params: ProcessParams[TArgs]) -> list[pa.RecordBatch]

Emit final batches by calling finish(). Do not override.

This method collects serialized states from all workers, deserializes them, and passes them to your finish() method.

Parameters

params: Process parameters including arguments and schemas.

Returns

List of output RecordBatches produced by finish().

Inherited members (15)

get_metadata method · from MetadataMixin — Get the resolved metadata for this function class.
describe method · from MetadataMixin — Get metadata as a dictionary (for JSON serialization).
logger attribute · from Function
storage attribute · from Function
FunctionArguments attribute · from TableFunctionBase
on_bind method · from TableInOutGenerator — Pass-through default — output schema is the input schema.
bind method · from TableFunctionBase — Bind protocol entry point. Do not override; use on_bind().
on_init method · from TableFunctionBase — One-time setup after bind, before processing batches.
global_init method · from TableFunctionBase — Global init protocol entry point. Do not override; use on_init().
cardinality method · from TableFunctionBase — Return the cardinality for the output.
dynamic_to_string method · from TableFunctionBase — Return diagnostics rendered as Extra Info under EXPLAIN ANALYZE.
statistics method · from TableFunctionBase — Return per-output-column statistics for this invocation.
pushdown_filters method · from TableFunctionBase — Get deserialized pushdown filters, or None if not present.
has_finalize_override method · from TableInOutGenerator — Whether this class’s finalize/finish represents real work.
on_cancel method · from TableInOutGenerator

class `TableInOutFunctionStateNoOp`

source

Bases: ArrowSerializableDataclass

Description

No-op state class for TableInOutFunction when no state is needed.

class `TableInOutGenerator`

source

Bases: TableFunctionBase[TArgs]

Description

Base class for streaming table functions that transform Arrow RecordBatches.

Each call to process() should emit exactly one output batch via out.emit(). Use TState to persist state between process() calls.

For functions that need a finalize phase (e.g., aggregation), override finalize() to return the final output batches.

Attributes

attribute `state_class`

type[ArrowSerializableDataclass] | None

Concrete ArrowSerializableDataclass type subclasses set to opt into framework-managed state; None (the default) means process()/finalize() get state=None and the framework skips its round-trip.

Methods

method `has_finalize_override`

source

has_finalize_override() -> bool

Whether this class’s finalize/finish represents real work.

Returns True iff either:

The class’s Meta declares has_finalize as True or False (explicit override — the declared value wins, even if it disagrees with the auto-detection).
Auto-detection finds a user subclass (one that is itself a TableInOutGenerator subclass) strictly above the VGI bases in the MRO defining a callable finish or finalize attribute.

The framework uses this to decide whether to advertise a finalize callback to DuckDB; DuckDB rejects LATERAL with correlated input on table functions that register in_out_function_final.

Returns

True if a real finalize/finish override is present.

method `on_bind`

source

on_bind(params: BindParams[TArgs]) -> BindResponse

Pass-through default — output schema is the input schema.

Override to compute a dynamic output type or validate arguments. See TableFunctionBase.on_bind for the broader contract.

Parameters

params: Bind parameters including arguments and the bind request.

Returns

A BindResponse whose output schema equals the input schema.

method `initial_state`

source

initial_state(params: ProcessParams[TArgs]) -> TState | None

Create initial processing state. Override when TState is used.

Called once during init to create the state object that will be passed to process() on each input batch.

Parameters

params: Process parameters including arguments and schemas.

Returns

Initial state, or None if no state is needed.

method `process`

source

process(
params: ProcessParams[TArgs],
state: TState,
batch: pa.RecordBatch,
out: OutputCollector,
) -> None

Process one input batch.

Called once per input batch during the INPUT phase. Must call out.emit(batch) exactly once to produce output.

Use out.client_log(level, message) for in-band logging.

Parameters

params: Process parameters including arguments and schemas.
state: Mutable state persisted between calls. None if TState not used.
batch: The input RecordBatch to process.
out: OutputCollector for emitting output and logging.

method `finalize`

source

finalize(params: ProcessParams[TArgs]) -> list[pa.RecordBatch]

Finalize processing and produce any remaining output.

Called after all input batches have been processed during the FINALIZE phase. Override to emit buffered or aggregated results.

Parameters

params: Process parameters including arguments and schemas.

Returns

List of output RecordBatches, or empty list if no finalization needed.

method `on_cancel`

source

on_cancel(params: ProcessParams[TArgs], state: TState | None) -> None

Inherited members (12)

get_metadata method · from MetadataMixin — Get the resolved metadata for this function class.
describe method · from MetadataMixin — Get metadata as a dictionary (for JSON serialization).
logger attribute · from Function
storage attribute · from Function
FunctionArguments attribute · from TableFunctionBase
bind method · from TableFunctionBase — Bind protocol entry point. Do not override; use on_bind().
on_init method · from TableFunctionBase — One-time setup after bind, before processing batches.
global_init method · from TableFunctionBase — Global init protocol entry point. Do not override; use on_init().
cardinality method · from TableFunctionBase — Return the cardinality for the output.
dynamic_to_string method · from TableFunctionBase — Return diagnostics rendered as Extra Info under EXPLAIN ANALYZE.
statistics method · from TableFunctionBase — Return per-output-column statistics for this invocation.
pushdown_filters method · from TableFunctionBase — Get deserialized pushdown filters, or None if not present.

function `unpack_int_cursor`

source

unpack_int_cursor(cursor: bytes, default: int = -1) -> int

Decode a packed int64 cursor; b"" returns default.

Use default=-1 (before-first sentinel) to start at the beginning of a state_log when no prior cursor exists.

Parameters

cursor: The packed int64 cursor bytes (b““ for none).
default: Value returned when cursor is empty.

Returns

The decoded integer cursor, or default when empty.

vgi.table_in_out_function

function pack_int_cursor

class RowTransformFunction

method input_columns

class TableInOutFunction

attribute state_class

method transform

method finish

method initial_state

method process

method finalize

class TableInOutFunctionStateNoOp

class TableInOutGenerator

attribute state_class

method has_finalize_override

method on_bind

method initial_state

method process

method finalize

method on_cancel

function unpack_int_cursor

function `pack_int_cursor`

class `RowTransformFunction`

method `input_columns`

class `TableInOutFunction`

attribute `state_class`

method `transform`

method `finish`

method `initial_state`

method `process`

method `finalize`

class `TableInOutFunctionStateNoOp`

class `TableInOutGenerator`

attribute `state_class`

method `has_finalize_override`

method `on_bind`

method `initial_state`

method `process`

method `finalize`

method `on_cancel`

function `unpack_int_cursor`