Skip to content
Query.Farm
Talk with Us

vgi.table_buffering_function

Module overview

Framework for implementing table sink+source functions.

TableBufferingFunction is the worker-side base for functions that must see every input row before producing any output (e.g. buffer-then-emit, global aggregations, sort-then-emit). Routed through the C++ PhysicalVgiTableBuffering Sink+Source operator.

Three callbacks, mirroring the operator’s three phases:

  • process(batch, params) -> bytes — ingest one batch, return an opaque state_id naming where the worker stored it.
  • combine(state_ids, params) -> list[bytes] — once per query, on the coordinator worker; group/merge/sort the per-batch state_ids and return finalize_state_ids for the Source phase.
  • finalize(params, finalize_state_id, state, out) — producer-mode streaming RPC mirroring TableFunctionGenerator.process: one tick per call, emit one batch via out.emit(batch) (or out.finish() for EOS), state persists between ticks via wire-serialization.

State_ids are opaque bytes. The worker picks the granularity (per-batch, per-thread, custom partitioning); the framework just round-trips them.

INVARIANT: any state the worker stores in process() that finalize() will need MUST live in cross-process storage scoped by params.execution_id (BoundStorage is the canonical choice). The Source phase may route a given finalize_state_id to a worker process that did NOT run the corresponding process() calls.

class TableBufferingFunction

Section titled “class TableBufferingFunction”
source

Bases: TableFunctionBase[TArgs]

Description

Base class for table sink+source functions.

Subclass to declare a function that must see every input row before producing output. The C++ PhysicalVgiTableBuffering operator routes calls through three phases:

  1. Sink — process(batch, params) -> state_id is called per input batch (parallel across DuckDB threads unless Meta.sink_order_dependent is set).
  2. Combine — combine(state_ids, params) -> finalize_state_ids is called once on the coordinator worker after every process() completes.
  3. Source — finalize(params, fid, state, out) is called per tick by the framework, emitting one batch per call (parallel across finalize_state_ids unless Meta.source_order_dependent).

Cross-process invariant: any state the worker writes during process() that finalize() will read MUST live in cross-process storage scoped by params.execution_id — BoundStorage is the canonical choice. The Source phase routes a given finalize_state_id to whatever worker process the C++ scheduler picks; it is NOT guaranteed to be the same process that ran process().

Methods

method on_bind

Section titled “method on_bind”
source
on_bind(params: BindParams[TArgs]) -> BindResponse

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

Override to validate arguments, compute a dynamic output type, or request secrets via SecretsAccessor. See TableFunctionBase.on_bind for the broader contract.

method process

Section titled “method process”
source
process(
batch: pa.RecordBatch,
params: TableBufferingParams[TArgs],
) -> bytes

Ingest one input batch and return an opaque state_id.

The worker chooses both where to store the batch (BoundStorage, external files, in-memory cross-process structures, etc.) and the granularity of state_ids (per-batch, per-thread, custom partitioning). The framework collects all returned state_ids and passes them to combine() on the coordinator worker.

Common pattern for “one bucket per execution” is to return params.execution_id; combine() then collapses the list of identical state_ids to a single finalize stream.

Cross-process invariant: any state the worker stores here that finalize() will need MUST live in cross-process storage scoped by params.execution_id. The Source phase may route the corresponding finalize_state_id to a different worker process.

Parameters

batch
One input batch from DuckDB. Schema matches the function’s declared input_schema.
params
Process-time params, including identity fields (execution_id, attach_id, transaction_id, function_name) and params.batch_index when Meta.requires_input_batch_index=True.

Returns

Opaque state_id naming where the batch was stored.

method combine

Section titled “method combine”
source
combine(
state_ids: list[bytes],
params: TableBufferingParams[TArgs],
) -> list[bytes]

Group / merge / sort state_ids; return finalize_state_ids.

Called once on the coordinator worker after every process() completes. State_ids are opaque bytes — the framework does not inspect, dedup, or transform them. combine returns the exact list of finalize_state_ids the Source phase will iterate; one finalize stream per returned id.

Typical patterns:

  • Single-bucket execution — process() returns params.execution_id for every call; combine() returns [params.execution_id] so one finalize stream drains the single accumulator.
  • Per-shard fan-out — process() returns a per-shard identifier; combine() returns the list of unique shard ids for parallel finalize.
  • Global sort under Meta.sink_order_dependent — process() returns per-batch ids; combine() reads each, sorts globally, returns [sentinel] so a single ordered finalize stream emits the merged result.

Parameters

state_ids
Every state_id returned from every process() call across every DuckDB thread, in arbitrary order. Duplicates from multiple Sink threads using the same state_id are NOT dedup’d by the framework.
params
Process-time params (same identity fields as process()).

Returns

finalize_state_ids — keys the Source phase will iterate.

method initial_finalize_state

Section titled “method initial_finalize_state”
source
initial_finalize_state(
finalize_state_id: bytes,
params: TableBufferingParams[TArgs],
) -> TFinalizeState | None

Build the initial wire-serializable state for a finalize stream.

Called once per finalize_state_id at stream init time. The returned state is passed to the first finalize() tick; the framework serializes it between ticks so the stream survives worker process boundaries (HTTP transport).

Default returns None (suitable when TFinalizeState = None). Override and declare a concrete TFinalizeState subclass of ArrowSerializableDataclass to carry cursor / progress state between ticks.

method finalize

Section titled “method finalize”
source
finalize(
params: TableBufferingParams[TArgs],
finalize_state_id: bytes,
state: TFinalizeState,
out: OutputCollector,
) -> None

Produce one batch’s worth of output for finalize_state_id.

Called repeatedly by the framework (one call per tick). Each call should either:

  • out.emit(batch) to produce one output batch and mutate state in place — state is wire-serialized after the call so the next tick (possibly on a different worker) resumes from the updated value.
  • out.finish() to signal EOS for this finalize_state_id.

Mirrors TableFunctionGenerator.process exactly — the only difference is the parameterization by finalize_state_id instead of free function arguments.

method on_cancel

Section titled “method on_cancel”
source
on_cancel(
params: TableBufferingParams[TArgs],
finalize_state_id: bytes,
state: TFinalizeState,
) -> None

No-op default; runtime docstring set below via func.doc.

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.

class TableBufferingParams

Section titled “class TableBufferingParams”
source

Bases: ProcessParams[TArgs]

Description

Params for TableBufferingFunction callbacks.

Adds identity fields that the buffered API needs to scope worker-owned storage and coordinate cross-process state. Other function shapes (TableFunctionGenerator, TableInOutGenerator, aggregates) keep using the plain ProcessParams they always have.

Attributes

attribute execution_id

Section titled “attribute execution_id”

bytes

Stable across coordinator + secondary workers for one DuckDB query execution. Key worker-owned storage by this.

attribute attach_id

Section titled “attribute attach_id”

bytes

Catalog attach identity; pin attach-time config lookups by this.

attribute transaction_id

Section titled “attribute transaction_id”

bytes | None

Hex-encoded VGI transaction id when running inside a DuckDB transaction, None otherwise.

attribute function_name

Section titled “attribute function_name”

str

Convenience accessor — same as init_call.function_name.

attribute worker_path

Section titled “attribute worker_path”

str | None

Subprocess path / unix:// / launch: argv. For diagnostics.

attribute client_log

Section titled “attribute client_log”

Callable[…, None]

In-band log sink — emits a 0-row log batch on the RPC response stream, which DuckDB surfaces as a row in duckdb_logs() with type='VGI'. Use this from process() and combine() (unary RPCs with no OutputCollector); the streaming finalize(... out) callback should use out.client_log(...) instead. The worker handler wires this to ctx.client_log before invoking the user callback; the default no-op is a safety net for unit-test callers that build TableBufferingParams outside the RPC path.

Inherited members (16)