Buffering functions

On this page

Sink every row, combine, then stream the result back.

interface TableBufferingFunction

source

type TableBufferingFunction interface {
// Name returns the function name used in SQL.
Name() string
// Metadata returns descriptive metadata (PartitionKind, ordering flags...).
Metadata() FunctionMetadata
// ArgumentSpecs returns the function's argument specifications.
ArgumentSpecs() []ArgSpec
// OnBind resolves the output schema given the bind parameters.
OnBind(params *BindParams) (*BindResponse, error)
// Process buffers one input batch and returns an opaque state_id naming
// where it was stored.
Process(ctx context.Context, params *ProcessParams, batch arrow.RecordBatch) ([]byte, error)
// Combine receives every state_id from every Process call (unordered) and
// returns the finalize_state_ids the source phase will iterate.
Combine(ctx context.Context, params *ProcessParams, stateIDs [][]byte) ([][]byte, error)
// Finalize returns the batches to emit for one finalize_state_id.
Finalize(ctx context.Context, params *ProcessParams, finalizeStateID []byte) ([]arrow.RecordBatch, error)
}

Description

TableBufferingFunction is a sink→source (“table-buffering”) VGI function. It buffers all input during a sink phase, reshapes it once at end-of-input, then streams results during a source phase. Mirrors vgi-python’s TableBufferingFunction.

Lifecycle (all keyed by the execution_id assigned at sink init):

• Process: called once per input batch; persists the batch to

execution-scoped storage and returns an opaque state_id.

• Combine: called once after all Process calls with every returned

state_id; returns the finalize_state_ids that drive the source phase.

• Finalize: called once per finalize_state_id; returns the batches to emit

for that partition.

Cross-process state MUST live in params.Storage (the sqlite-backed state log) — Process and Finalize may run in different worker processes.

interface TableBufferingFunctionWithCardinality

source

type TableBufferingFunctionWithCardinality interface {
TableBufferingFunction
// Cardinality estimates the function's output row count for the optimizer.
Cardinality(params *BindParams) (*TableCardinality, error)
}

Description

TableBufferingFunctionWithCardinality lets a buffering function declare a cardinality estimate (e.g. a reducer that always emits one row).

struct bufferingParamsEntry

source

type bufferingParamsEntry struct {
fn  TableBufferingFunction
params  ProcessParams  // template — copied per call
}

Description

bufferingParamsEntry caches the decoded (function, ProcessParams template) for one buffering execution. The InitRecipe is written once at sink init and never changes, so the decode + rebuild is invariant across the execution’s process/combine RPCs. Each call gets a shallow copy of params so the per-call fields (Auth/AttachScope/clientLog/BatchIndex) never race; the shared maps and Storage are read-only during Process.

function encodeInitRecipe

source

func encodeInitRecipe(r *InitRecipe) ([]byte, error)