Worker & serving

On this page

Registering functions and running a worker over each transport.

struct AttachDecision

source

type AttachDecision struct {
ResolvedDataVersion    string
ResolvedImplementationVersion  string
AttachOpaqueData    []byte
}

Description

AttachDecision is the custom response returned by an AttachValidator. Any non-empty ResolvedDataVersion / ResolvedImplementationVersion value is forwarded to the client so it appears as a duckdb_databases().tag. If AttachOpaqueData is nil, the worker falls back to the catalog name.

func type AttachScanBranchesGetHandler

source

type AttachScanBranchesGetHandler func(attachOpaqueData []byte, schemaName, name string, atUnit, atValue *string) (result *ScanBranchesResult, handled bool, err error)

Description

AttachScanBranchesGetHandler is the attach-opaque-data-aware handler for catalog_table_scan_branches_get. Return (result, true) to serve a multi-branch table; (nil, false) to fall through (the C++ extension then falls back to catalog_table_scan_function_get).

func type AttachScanFunctionGetHandler

source

type AttachScanFunctionGetHandler func(attachOpaqueData []byte, schemaName, name string, atUnit, atValue *string) (result *ScanFunctionResult, handled bool, err error)

Description

AttachScanFunctionGetHandler is the attach-opaque-data-aware version of ScanFunctionGetHandler. Return (result, true) to override; (nil, false) to fall through.

func type AttachTableGetHandler

source

type AttachTableGetHandler func(attachOpaqueData []byte, schemaName, name string, atUnit, atValue *string) (data []byte, handled bool, err error)

Description

AttachTableGetHandler is the attach-opaque-data-aware version of TableGetHandler. Return (data, true) to override the default; return (nil, false) to fall through. Used by version-aware workers where the attach_opaque_data encodes the resolved version.

func type AttachValidator

source

type AttachValidator func(req *CatalogAttachRequestWire, callCtx *vgirpc.CallContext) (*AttachDecision, error)

Description

AttachValidator is invoked by the default catalog_attach handler. It may inspect the requested data_version_spec / implementation_version, perform validation, and return resolved values. Returning an error causes the ATTACH to fail with that message — the test suite treats a “ValueError” RPC type as a user error the extension should surface verbatim.

func type AttachWriteFunctionGetHandler

source

type AttachWriteFunctionGetHandler func(op WriteOp, attachOpaqueData []byte, schemaName, name string) (result *ScanFunctionResult, handled bool, err error)

Description

AttachWriteFunctionGetHandler is the attach-opaque-data-aware version of catalog_table_{insert,update,delete}_function_get. Op is WriteOpInsert, WriteOpUpdate, or WriteOpDelete. Return (result, true) to route; (nil, false) to fall through to the built-in writable-catalog path or the read-only rejection.

func type CatalogVersionHook

source

type CatalogVersionHook func(attachOpaqueData []byte, callCtx *vgirpc.CallContext) error

Description

CatalogVersionHook runs on every catalog_version RPC before the response is returned. Returning a non-nil error causes the RPC to fail with that message (wrapped as a ValueError). The hook can inspect call-context cookies to assert the HTTP cookie jar is round-tripping correctly — a useful regression check for versioned workers that set a sticky cookie at ATTACH time.

struct GlobalInitResponse

source

type GlobalInitResponse struct {
// ExecutionID uniquely identifies this execution.
ExecutionID  []byte
// MaxWorkers is the maximum number of parallel workers (0 = default/4).
MaxWorkers  int64
// OpaqueData is optional data passed to secondary inits.
OpaqueData  []byte
}

Description

GlobalInitResponse is returned by a function’s OnInit method.

Methods

function DefaultInit

source

func DefaultInit() (*GlobalInitResponse, error)

DefaultInit returns a standard single-worker GlobalInitResponse (MaxWorkers: 1). Use this in OnInit for table functions that don’t need parallel execution.

struct InitParams

source

type InitParams struct {
// FunctionName is the name of the function.
FunctionName  string
// FunctionType is the type of the function.
FunctionType  FunctionType
// Args are the parsed function arguments.
Args  *Arguments
// OutputSchema is the output schema resolved during bind.
OutputSchema  *arrow.Schema
// InputSchema is the input table schema (nil for table functions).
InputSchema  *arrow.Schema
// ProjectionIDs are the projected column indices (nil = all columns).
ProjectionIDs  []int32
// Phase is the table-in-out init phase.
Phase  Phase
// ExecutionID is the execution ID for secondary inits.
ExecutionID  []byte
// BindOpaqueData is opaque data from the bind phase.
BindOpaqueData  []byte
// InitOpaqueData is opaque data from a previous global init (secondary inits).
InitOpaqueData  []byte
// Settings is a map of DuckDB setting names to their scalar values.
Settings  map[string]interface{}
// Secrets is a map of secret names to their value maps.
Secrets  Secrets
// IsSecondary is true if this is a secondary init (worker init).
IsSecondary  bool
// PushdownFilters is the pushdown filter batch (nil if none).
PushdownFilters  arrow.RecordBatch
// JoinKeys maps keys_column name -> Arrow array carrying the join keys
// referenced by FilterJoinKeys entries in PushdownFilters.
JoinKeys  map[string]arrow.Array
// OrderByHint, when non-nil, carries an ORDER BY + LIMIT pushdown
// hint set by DuckDB's RowGroupPruner optimizer.
OrderByHint  *OrderByHint
// TableSampleHint, when non-nil, carries a TABLESAMPLE pushdown hint.
TableSampleHint  *TableSampleHint
// Storage provides shared execution storage for cross-phase data.
Storage  *ExecutionStorage
}

Description

InitParams holds the parameters available during the init phase.

struct InitRecipe

source

type InitRecipe struct {
BindCall    BindRequestWire
OutputSchemaIPC    []byte
FunctionName    string
FunctionType    FunctionType
ProjectionIDs    []int32
ExecutionID    []byte
BindOpaqueData    []byte
InitOpaqueData    []byte
PushdownFilterIPC  []byte
Phase      Phase
IsSecondary    bool
// SubstreamID is the client-minted per-substream id folded into the recipe
// (and so into the HTTP state token) so every rehydrated tick of one
// substream sees the same id. Nil when the client did not supply one.
SubstreamID  []byte
// ShardKey is the per-attach Durable Object routing key (att-<hex uuid>),
// derived once at init from the unwrapped attach UUID and carried through
// serialization so a rehydrated process/finalize turn routes storage to the
// same DO without re-opening the auth-scoped seal. "" for non-attach paths.
ShardKey  string
}

Description

InitRecipe carries all serializable data needed to reconstruct ProcessParams after a state token round-trip through HTTP transport. It captures the raw IPC bytes from the init request so that the rehydration path can replay the same bind/init logic without the original wire message.

Methods

function decodeInitRecipe

source

func decodeInitRecipe(data []byte) (*InitRecipe, error)

struct OrderByHint

source

type OrderByHint struct {
ColumnName  string
Direction  OrderByDirection  // "" if unspecified
NullOrder  OrderByNullOrder  // "" if unspecified
RowLimit  int64      // -1 if unbounded
}

Description

OrderByHint is an ORDER BY + LIMIT hint pushed by the optimizer.

struct ProcessParams

source

type ProcessParams struct {
// FunctionName is the name of the function.
FunctionName  string
// FunctionType is the type of the function.
FunctionType  FunctionType
// Args are the parsed function arguments.
Args  *Arguments
// OutputSchema is the output schema (may be projected).
OutputSchema  *arrow.Schema
// InputSchema is the source/input table schema (nil for table functions
// with no input). For a COPY ... TO sink it carries the source columns, so a
// CopyToFunction can write a header even when zero rows are buffered.
InputSchema  *arrow.Schema
// ProjectionIDs are the projected column indices (nil = all columns).
ProjectionIDs  []int32
// Settings is a map of DuckDB setting names to their scalar values.
Settings  map[string]interface{}
// Secrets is a map of secret names to their value maps.
Secrets  Secrets
// ExecutionID is the execution identifier.
ExecutionID  []byte
// SubstreamID is the stable client-minted id for this streaming
// table-in-out substream. Present (identical across init / every Process /
// finalize) when the client fanned this function out across per-substream
// workers; use it to key per-substream accumulated state in shared storage
// so a finalize that lands on a different HTTP backend than the Process
// calls still finds it. Nil for the serial path, non-table-in-out
// functions, or an old client that did not supply one. Mirrors vgi-python's
// ProcessParams.substream_id.
SubstreamID  []byte
// AttachScope is the per-ATTACH plaintext (the catalog's attach_opaque_data
// with the framework UUID stripped). Stable across the queries of one ATTACH
// session; used to scope persistent state via Storage.AttachStore(scope).
// Nil when the call has no attach context.
AttachScope  []byte
// InitOpaqueData is the opaque data from the init response.
InitOpaqueData  []byte
// PushdownFilters is the pushdown filter batch (nil if none).
PushdownFilters  arrow.RecordBatch
// JoinKeys maps keys_column name -> Arrow array carrying the join keys
// referenced by FilterJoinKeys entries in PushdownFilters.
JoinKeys  map[string]arrow.Array
// AtUnit/AtValue carry the AT (TIMESTAMP|VERSION ...) time-travel clause for
// this scan, threaded onto the bind request embedded in init. Both nil when
// the scan has no AT clause. Function-backed time-travel tables resolve the
// version from these at NewState.
AtUnit  *string
AtValue  *string
// CopyFrom carries the COPY ... FROM context when this scan was opened by a
// COPY-FROM statement (nil otherwise). A CopyFromFunction reads FilePath /
// ExpectedSchema here in Process. Mirrors Python's
// ProcessParams.init_call.bind_call.copy_from.
CopyFrom  *CopyFromContext
// CopyTo carries the COPY ... TO context when this sink was opened by a
// COPY-TO statement (nil otherwise). A CopyToFunction reads Format /
// FilePath here in Process (per-shard write) and Combine (terminal write).
// Mirrors Python's ProcessParams.init_call.bind_call.copy_to.
CopyTo  *CopyToContext
// CurrentPushdownFilters is the filter state for the *current* Produce
// tick. It starts at the init-time pushdown filters and is replaced
// whenever DuckDB's dynamic filter tightens (DynamicFilter pushdown).
// Functions that want to react to filter updates per batch should read
// this field; functions that only care about static filters should use
// PushdownFilters.
CurrentPushdownFilters  *PushdownFilters
// OrderByHint, when non-nil, carries an ORDER BY + LIMIT pushdown hint.
OrderByHint  *OrderByHint
// TableSampleHint, when non-nil, carries a TABLESAMPLE pushdown hint.
TableSampleHint  *TableSampleHint
// Storage provides shared execution storage for cross-phase data.
Storage  *ExecutionStorage
// Auth is the authentication context for the current request.
// Always non-nil; unauthenticated requests receive vgirpc.Anonymous().
Auth  *vgirpc.AuthContext
// BatchIndex is the DuckDB per-chunk batch index threaded into a
// table_buffering_process call when the function declares
// RequiresInputBatchIndex. Nil otherwise.
BatchIndex  *int64
// IfNoneMatch is the conditional-revalidation validator carrying the
// client's stored ETag. It is set when the client holds a
// stale-but-revalidatable cached result and asks the worker to confirm
// freshness cheaply. A function that advertised CacheControl.Revalidatable
// compares it against its current validator and, when unchanged, emits a
// 0-row batch tagged CacheControl{NotModified: true} instead of
// re-streaming. Nil on a normal call.
IfNoneMatch  *string
// IfModifiedSince is the conditional-revalidation validator carrying the
// client's stored Last-Modified. Companion to IfNoneMatch. Nil otherwise.
IfModifiedSince  *string
// clientLog forwards an in-band log message to the client (surfaced in
// duckdb_logs() with type='VGI'). Set by the framework on the unary
// table-buffering RPCs, which have no streaming OutputCollector. Nil when
// in-band logging is unavailable.
clientLog  func(level vgirpc.LogLevel, msg string)
}

Description

ProcessParams holds the parameters available during the process phase.

Methods

method ClientLog

source

func (p *ProcessParams) ClientLog(level vgirpc.LogLevel, msg string)

ClientLog emits an in-band log message to the client, if the framework wired a logging sink for this call (e.g. from table_buffering_process/combine).

func type SchemaContentsHandler

source

type SchemaContentsHandler func(attachOpaqueData []byte, schemaName string) ([]SerializedSchemaItem, bool)

Description

SchemaContentsHandler lets callers override catalog_schema_contents_tables on a per-attach-opaque-data basis (e.g. return different tables per resolved data version). Return nil to fall through to the default registered-tables behaviour.

struct SecretTypeSpec

source

type SecretTypeSpec struct {
Name    string
Description  string
Schema    *arrow.Schema  // parameter schema; use field metadata {"redact":"true"} for sensitive fields
}

Description

SecretTypeSpec describes a DuckDB secret type registered by the worker.

type SerializedSchemaItem

source

type SerializedSchemaItem []byte

Description

SerializedSchemaItem is a single pre-serialized schema item (TableInfo or ViewInfo IPC bytes).

struct SettingSpec

source

type SettingSpec struct {
Name    string
Description  string
Type    arrow.DataType
DefaultValue  interface{}  // Go value matching the Type (nil = no default)
}

Description

SettingSpec describes a DuckDB custom setting registered by the worker.

struct TableSampleHint

source

type TableSampleHint struct {
Percentage  float64
Seed    int64
}

Description

TableSampleHint is a TABLESAMPLE pushdown hint.

struct Worker

source

type Worker struct {
scalars    map[string][]ScalarFunction
tables    map[string][]TableFunction
tableInOuts  map[string][]TableInOutFunction
tableBufferings  map[string][]TableBufferingFunction
aggregates  map[string][]AggregateFunction
aggStorage  *aggregateStorage
// streamingSessions tracks per-execution_id state for streaming-partitioned
// aggregates (aggregate_streaming_open/_chunk/_close).
streamingSessions  streamingSessionStore
catalogName    string
catalogComment    string
catalogTags    map[string]string
// globalFunctionNames / globalFunctionPrefix back the protocol-1.3.0
// global_functions + global_function_prefix fields of catalog_attach: the
// subset of this catalog's functions a client may republish into its own
// global (system.main) namespace, and the prefix it publishes them under.
// The Go analogue of vgi-python's Catalog(global_functions=...,
// global_function_prefix=...).
globalFunctionNames  []string
globalFunctionPrefix  string
supportsTransactions  bool
schemaComments    map[string]string
schemaTags    map[string]map[string]string
catalog      *DefaultReadOnlyCatalog
// extraCatalogs are additional catalog names this worker accepts via
// catalog_attach. They share the worker's registered functions but
// have their own (writable) table/schema state. Indexed by name.
extraCatalogs  map[string]*WritableCatalog
// catalogAliases are extra catalog names that ATTACH against the
// worker's primary read-only catalog (no separate writable state, no
// distinct catalog implementation). Useful for fixture workers that
// publish multiple cross-language reproducer catalogs (e.g.
// projection_repro) sharing one binary.
catalogAliases  map[string]struct{}
// funcOrigins records, for every registered implementation, the catalog
// schema that declares it and the catalog it is restricted to (if any).
// Keyed by (kind, function name) with the slice index-aligned to the
// same-named slice in scalars/tables/tableInOuts/tableBufferings/
// aggregates, so two implementations sharing a name keep separate
// origins — which is exactly what makes a name registered in two schemas
// (or two catalogs) resolvable.
funcOrigins  map[funcKey][]funcOrigin
// catalogAliasInfos carries per-alias discovery metadata (data version,
// etc.) for aliases that should advertise themselves distinctly in
// catalog_catalogs and mint a random per-ATTACH scope at attach (so two
// ATTACHes of the same alias are isolated). Keyed by catalog name.
catalogAliasInfos    map[string]CatalogInfo
storages      sync.Map  // map[hex execution ID string]*ExecutionStorage
bufferingParams      sync.Map  // map[hex execution ID string]*bufferingParamsEntry
fsOnce        sync.Once
fs        FunctionStorage
fsErr        error
settings      []SettingSpec
catalogTables      map[string][]CatalogTable  // schema_name → tables
catalogViews      map[string][]CatalogView  // schema_name → views
catalogMacros      map[string][]CatalogMacro  // schema_name → macros
dynamicSchemas      map[string]string    // schema_name → comment (for SchemaContentsHandler-only schemas)
scanFunctionGetHandler    ScanFunctionGetHandler
tableGetHandler      TableGetHandler
catalogInfoOverride    *CatalogInfo
attachValidator      AttachValidator
schemaContentsHandler    SchemaContentsHandler
attachTableGetHandler    AttachTableGetHandler
attachScanFunctionGetHandler  AttachScanFunctionGetHandler
attachScanBranchesGetHandler  AttachScanBranchesGetHandler
attachWriteFunctionGetHandler  AttachWriteFunctionGetHandler
catalogVersionHook    CatalogVersionHook
authenticateFunc    vgirpc.AuthenticateFunc
oauthMetadata      *vgirpc.OAuthResourceMetadata
oauthPkce      *vgirpc.OAuthPkceConfig
secretTypes      []SecretTypeSpec
attachCatalogs      []AttachCatalogInfo
attachOptions      []AttachOptionSpec
// catalogAttachOptions holds per-alias-catalog option specs
// (WithAttachOptionsForCatalog); a catalog absent here falls back to
// attachOptions.
catalogAttachOptions  map[string][]AttachOptionSpec
logLevel    slog.Level  // slog.LevelInfo (0) by default — Info level is intentional.
logHandler    slog.Handler  // nil means default TextHandler to stderr
logFormat    LogFormat  // empty means text
logLoggers    []string  // empty means all known loggers
logConfigured    bool    // true once any logging WorkerOption fires
httpSigningKey    []byte    // HMAC key for HTTP state tokens (explicit via WithHttpSigningKey, else ephemeral per-process)
// sealOpaqueData gates AEAD sealing of catalog opaque-data envelopes. It is
// enabled only when an explicit signing key is configured (WithHttpSigningKey),
// matching vgi-python: an anonymous worker with an ephemeral, per-process key
// (generated only so the HTTP state-token machinery has one) does not seal —
// sealing binds opaque-data to a principal, and there is none without auth, so
// it would only add a cross-implementation incompatibility (the published
// extension round-trips plaintext opaque-data, not the longer sealed envelope).
sealOpaqueData  bool

// copyFromFormats holds the custom COPY ... FROM formats advertised by this
// worker (one per RegisterCopyFrom). Surfaced via catalog_copy_from_formats
// so the VGI extension can register a DuckDB CopyFunction per entry.
copyFromFormats  []copyFromFormatRecord
}

Description

Worker is the main VGI worker that hosts functions and serves RPC.

Methods

function NewWorker

source

func NewWorker(opts …WorkerOption) *Worker

NewWorker creates a new VGI worker.

method RegisterAggregate

source

func (w *Worker) RegisterAggregate(f AggregateFunction)

RegisterAggregate registers an aggregate function in the catalog’s default schema. Multiple registrations with the same Name() are kept as overloads — distinguished by their ArgumentSpecs at catalog-discovery time.

method RegisterAggregateInSchema

source

func (w *Worker) RegisterAggregateInSchema(schemaName string, f AggregateFunction)

RegisterAggregateInSchema registers an aggregate function in a named catalog schema. See RegisterScalarInSchema for why the schema is part of the identity.

method RegisterCatalogMacro

source

func (w *Worker) RegisterCatalogMacro(schemaName string, macro CatalogMacro)

RegisterCatalogMacro registers a macro in the given schema of the catalog.

method RegisterCatalogSchema

source

func (w *Worker) RegisterCatalogSchema(name, comment string)

RegisterCatalogSchema declares a schema that exists in the catalog but whose tables are produced dynamically by a SchemaContentsHandler. Use when there are no registered CatalogTable/View/Macro entries to “anchor” the schema (otherwise it wouldn’t appear in catalog_schemas).

method RegisterCatalogTable

source

func (w *Worker) RegisterCatalogTable(schemaName string, table CatalogTable)

RegisterCatalogTable registers a table in the given schema of the catalog.

If table.Function is set, the function is also auto-registered into the dispatch table (deduped by name) so the scan resolves without a separate RegisterTable call — mirroring vgi-python’s _build_registry, which auto-scans each Table.function.

method RegisterCatalogView

source

func (w *Worker) RegisterCatalogView(schemaName string, view CatalogView)

RegisterCatalogView registers a view in the given schema of the catalog.

method RegisterCopyFrom

source

func (w *Worker) RegisterCopyFrom(f CopyFromFunction)

RegisterCopyFrom registers a custom COPY … FROM format. The function is registered as an ordinary producer-mode table function (so it appears in duckdb_functions and reuses the table scan path) AND advertised via catalog_copy_from_formats so the VGI extension registers a DuckDB CopyFunction for it. Mirrors vgi-python registering a CopyFromFunction subclass in the catalog’s function list.

method RegisterCopyTo

source

func (w *Worker) RegisterCopyTo(f CopyToFunction)

RegisterCopyTo registers a custom COPY … TO format. The function is registered as a table-buffering function (so init/process/combine reuse the buffered Sink+Combine path) AND advertised via catalog_copy_from_formats with direction=“to” so the VGI extension registers a DuckDB CopyFunction for it. Mirrors vgi-python registering a CopyToFunction subclass in the catalog’s function list.

method RegisterScalar

source

func (w *Worker) RegisterScalar(f ScalarFunction)

RegisterScalar registers a scalar function in the catalog’s default schema.

method RegisterScalarForCatalog

source

func (w *Worker) RegisterScalarForCatalog(catalogName string, f ScalarFunction)

RegisterScalarForCatalog registers a scalar function scoped to a single catalog name, in that catalog’s default schema. The function only surfaces in — and is only dispatchable from — ATTACHes of the named catalog, so two catalogs served by one worker may each declare their own implementation of the same function name.

method RegisterScalarInSchema

source

func (w *Worker) RegisterScalarInSchema(schemaName string, f ScalarFunction)

RegisterScalarInSchema registers a scalar function in a named catalog schema.

The same name may be registered in more than one schema; a schema-qualified call (db.schema.fn(...)) then dispatches to the implementation declared in the schema it names, rather than colliding with the other as an overload.

method RegisterTable

source

func (w *Worker) RegisterTable(f TableFunction)

RegisterTable registers a table function in the catalog’s default schema.

method RegisterTableBuffering

source

func (w *Worker) RegisterTableBuffering(f TableBufferingFunction)

RegisterTableBuffering registers a table-buffering function in the catalog’s default schema.

method RegisterTableBufferingForCatalog

source

func (w *Worker) RegisterTableBufferingForCatalog(catalogName string, f TableBufferingFunction)

RegisterTableBufferingForCatalog registers a table-buffering function scoped to a single catalog (visible only under that ATTACH). See RegisterTableForCatalog for the rationale.

method RegisterTableBufferingInSchema

source

func (w *Worker) RegisterTableBufferingInSchema(schemaName string, f TableBufferingFunction)

RegisterTableBufferingInSchema registers a table-buffering function in a named catalog schema. See RegisterScalarInSchema for why the schema is part of the identity.

method RegisterTableForCatalog

source

func (w *Worker) RegisterTableForCatalog(catalogName string, f TableFunction)

RegisterTableForCatalog registers a table function scoped to a single catalog name. The function is invokable as normal but only surfaces in catalog_schema_contents_functions / duckdb_functions for ATTACH calls against the named catalog. Use for fixture functions that should only be visible to their own reproducer catalog (e.g. proj_repro_* under “projection_repro“).

method RegisterTableInOut

source

func (w *Worker) RegisterTableInOut(f TableInOutFunction)

RegisterTableInOut registers a table-in-out function in the catalog’s default schema.

method RegisterTableInOutForCatalog

source

func (w *Worker) RegisterTableInOutForCatalog(catalogName string, f TableInOutFunction)

RegisterTableInOutForCatalog is the catalog-scoped sibling of RegisterTableInOut; the function is invokable as normal but only surfaces under the named catalog’s function listing. See RegisterTableForCatalog for the rationale (per-catalog function inventories that fixture tests assert on).

method RegisterTableInOutInSchema

source

func (w *Worker) RegisterTableInOutInSchema(schemaName string, f TableInOutFunction)

RegisterTableInOutInSchema registers a table-in-out function in a named catalog schema. See RegisterScalarInSchema for why the schema is part of the identity.

method RegisterTableInSchema

source

func (w *Worker) RegisterTableInSchema(schemaName string, f TableFunction)

RegisterTableInSchema registers a table function in a named catalog schema. See RegisterScalarInSchema for why the schema is part of the identity.

method RegisterTableUnlisted

source

func (w *Worker) RegisterTableUnlisted(f TableFunction)

RegisterTableUnlisted registers a table function that is invokable but never appears in any catalog’s function listing (duckdb_functions, describe.json). Use it for a function that exists only to back a catalog table: the table’s scan resolves it by name, but exposing it as a callable table function too would be redundant surface area.

method RegisterWritableCatalog

source

func (w *Worker) RegisterWritableCatalog(c *WritableCatalog)

RegisterWritableCatalog adds a writable catalog this worker handles alongside its primary read-only catalog. The catalog accepts ATTACH requests by its name and supports DDL/DML operations on user-created tables.

method RunHttp

source

func (w *Worker) RunHttp(addr string) error

RunHttp runs the worker serving RPC over HTTP. It listens on the given address (e.g. “127.0.0.1:0” for a random port) and prints “PORT:<n>” to stdout so callers can discover the assigned port.

method RunStdio

source

func (w *Worker) RunStdio()

RunStdio runs the worker serving RPC over stdin/stdout.

method RunTcp

source

func (w *Worker) RunTcp(host string, port int, idleTimeout time.Duration) error

RunTcp runs the worker serving RPC over a raw TCP socket — the launcher transport’s AF_INET sibling. It binds host:port (host “” defaults to 127.0.0.1; port 0 picks a free port), prints the readiness marker “TCP:<host>:<port>” with the actual bound port to stdout once listening (then writes nothing further to stdout), and self-shuts-down after idleTimeout with no active connections (<=0 disables the timeout).

Raw TCP framing carries no auth/encryption — bind loopback / a trusted network only; use RunHttp for untrusted networks.

method RunUnix

source

func (w *Worker) RunUnix(path string, idleTimeout time.Duration) error

RunUnix runs the worker serving RPC over an AF_UNIX socket at path — the “launcher” transport. It prints the readiness marker “UNIX:<path>” to stdout once the socket is listening (then writes nothing further to stdout), and self-shuts-down after idleTimeout with no active connections (<=0 disables the timeout). Mutually exclusive with HTTP.

method SetAuthenticate

source

func (w *Worker) SetAuthenticate(fn vgirpc.AuthenticateFunc)

SetAuthenticate sets an authentication callback for HTTP mode. When set, every HTTP request is validated and the resulting AuthContext is available via ProcessParams.Auth and BindParams.Auth.

method SetOAuthPkce

source

func (w *Worker) SetOAuthPkce(cfg vgirpc.OAuthPkceConfig)

SetOAuthPkce enables the browser-based OAuth PKCE login flow for HTTP mode. It requires SetAuthenticate and SetOAuthResourceMetadata (with a ClientID) to also be configured; RunHttp applies it after both. When enabled, the server serves /_oauth/callback, /_oauth/logout, and the /_oauth/token exchange proxy, and redirects unauthenticated browser GETs to the authorization server.

method SetOAuthResourceMetadata

source

func (w *Worker) SetOAuthResourceMetadata(m *vgirpc.OAuthResourceMetadata)

SetOAuthResourceMetadata configures OAuth Protected Resource Metadata (RFC 9728) for HTTP mode. When set, the server exposes a well-known endpoint and includes a WWW-Authenticate header on 401 responses.

method SetScanFunctionGetHandler

source

func (w *Worker) SetScanFunctionGetHandler(h ScanFunctionGetHandler)

SetScanFunctionGetHandler sets a handler for resolving scan functions for tables that are not directly backed by a registered Function.

method SetTableGetHandler

source

func (w *Worker) SetTableGetHandler(h TableGetHandler)

SetTableGetHandler sets a handler for customizing catalog_table_get responses. Return non-nil bytes to override the default lookup; return nil to fall through.

method attachOptionsFor

source

func (w *Worker) attachOptionsFor(catalogName string) []AttachOptionSpec

attachOptionsFor returns the option specs governing an ATTACH of catalogName: the per-catalog set when one is registered, otherwise the worker-wide set.

method attachScopeForPtr

source

func (w *Worker) attachScopeForPtr(sealed *[]byte, cc *vgirpc.CallContext, fallback []byte) []byte

attachScopeForPtr resolves the per-ATTACH plaintext scope (catalog bytes with the framework UUID stripped) from a sealed wire attach value, using the live call context so it is correct on both subprocess (pass-through) and HTTP (AEAD) transports. Returns fallback when the value is absent or can’t be opened (e.g. cold-load best-effort scope on the rehydrate path).

method buildBindArgs

source

func (w *Worker) buildBindArgs(ct *CatalogTable) *Arguments

buildBindArgs creates an Arguments struct from CatalogTable.FuncArgs for use in OnBind calls to derive output schemas.

method buildScanResultFromTable

source

func (w *Worker) buildScanResultFromTable(ct *CatalogTable) *ScanFunctionResult

buildScanResultFromTable creates a ScanFunctionResult from a function-backed CatalogTable.

method buildServer

source

func (w *Worker) buildServer(transport serverTransport) *vgirpc.Server

method candidatesFor

source

func (w *Worker) candidatesFor(name string, ft FunctionType) []candidate

candidatesFor gathers every registration of name, preferring the registry matching the requested function type and falling back to the others (DuckDB reports table-in-out and table-buffering functions as TABLE, and old clients may send an unrecognized type).

method catalogOfAttach

source

func (w *Worker) catalogOfAttach(attachOpaqueData []byte) string

catalogOfAttach names the catalog an unwrapped attach plaintext belongs to. Every function is homed in exactly one catalog, so both the function listing and bind dispatch need one catalog name per request — this is where it comes from.

Writable catalogs mint their own “writable:<name>” plaintext, so their name is recovered through that mapping; every other catalog carries its name up front. A plaintext naming no catalog this worker serves — an attach validator that replaced it with its own encoding (a resolved data version, encoded ATTACH options), or no attachment at all — resolves to the worker’s own catalog, which is the only catalog such a worker can mean.

method catalogOfAttachPtr

source

func (w *Worker) catalogOfAttachPtr(attachOpaqueData *[]byte, cc *vgirpc.CallContext) string

catalogOfAttachPtr names the catalog for a request that carries the attach value as an optional wire field (the unary aggregate / table-buffering RPCs). It opens the envelope down to the catalog’s own plaintext first, then applies catalogOfAttach. Returns the worker’s own catalog when there is no attachment or it cannot be opened — the same single-home rule the bind path uses.

method clearTransactionState

source

func (w *Worker) clearTransactionState(txID []byte)

clearTransactionState best-effort clears per-transaction K/V storage when a transaction commits or rolls back.

method findCatalogTable

source

func (w *Worker) findCatalogTable(schemaName, name string) *CatalogTable

findCatalogTable returns the registered CatalogTable for (schema, name) or nil.

method findWritableTable

source

func (w *Worker) findWritableTable(schemaName, tableName string) (*WritableCatalog, *writableTable, error)

findWritableTable looks up a (schema, table) across all writable catalogs. Reads from the SQLite store so DuckDB-spawned worker subprocesses see the same state as the process that ran CREATE TABLE.

method functionStorage

source

func (w *Worker) functionStorage() (FunctionStorage, error)

functionStorage returns the worker’s shared FunctionStorage backend, constructed on first call. Today this is always a SQLite backend at the per-user default path; an env-driven selector (“VGI_WORKER_SHARED_STORAGE”) can be added later to pick alternative backends (Cloudflare DO, etc.).

method getArgSpecs

source

func (w *Worker) getArgSpecs(fn interface{}) []ArgSpec

getArgSpecs returns the ArgSpecs for a resolved function.

method getFunctionMetadata

source

func (w *Worker) getFunctionMetadata(fn interface{}) FunctionMetadata

getFunctionMetadata returns the FunctionMetadata for a resolved function.

method getOrCreateStorage

source

func (w *Worker) getOrCreateStorage(ctx context.Context, executionID []byte, shardKey string) (*ExecutionStorage, error)

getOrCreateStorage returns or creates an ExecutionStorage for the given execution ID. The wrapper binds against the worker’s shared FunctionStorage backend (initialized lazily on first call) so every execution in the worker uses one backend — one SQLite DB shared across all executions and across worker subprocesses spawned for parallel scans.

method handleAggregateBind

source

func (w *Worker) handleAggregateBind(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateBindRequestWire) (AggregateBindResponseWire, error)

method handleAggregateCombine

source

func (w *Worker) handleAggregateCombine(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateCombineRequestWire) (AggregateCombineResponseWire, error)

method handleAggregateDestructor

source

func (w *Worker) handleAggregateDestructor(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateDestructorRequestWire) (AggregateDestructorResponseWire, error)

method handleAggregateFinalize

source

func (w *Worker) handleAggregateFinalize(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateFinalizeRequestWire) (AggregateFinalizeResponseWire, error)

method handleAggregateStreamingChunk

source

func (w *Worker) handleAggregateStreamingChunk(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateStreamingChunkRequestWire) (AggregateStreamingChunkResponseWire, error)

method handleAggregateStreamingClose

source

func (w *Worker) handleAggregateStreamingClose(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateStreamingCloseRequestWire) (AggregateStreamingCloseResponseWire, error)

method handleAggregateStreamingOpen

source

func (w *Worker) handleAggregateStreamingOpen(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateStreamingOpenRequestWire) (AggregateStreamingOpenResponseWire, error)

method handleAggregateUpdate

source

func (w *Worker) handleAggregateUpdate(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateUpdateRequestWire) (AggregateUpdateResponseWire, error)

method handleAggregateWindow

source

func (w *Worker) handleAggregateWindow(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateWindowRequestWire) (AggregateWindowResponseWire, error)

method handleAggregateWindowBatch

source

func (w *Worker) handleAggregateWindowBatch(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateWindowBatchRequestWire) (AggregateWindowBatchResponseWire, error)

handleAggregateWindowBatch serves the batched window RPC (one call per Evaluate() instead of one per output row). UNREACHABLE under the current build: the C++ extension wires SetWindowBatchCallback only under #ifdef DUCKDB_HAS_AGGREGATE_WINDOW_BATCH, which is never defined because the pinned DuckDB lacks aggregate_window_batch_t. Until DuckDB ships that API and the extension enables it, only the per-row handleAggregateWindow runs. Kept symmetric with the per-row path so it works the moment it lands.

method handleAggregateWindowDestructor

source

func (w *Worker) handleAggregateWindowDestructor(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateWindowDestructorRequestWire) (AggregateWindowDestructorResponseWire, error)

method handleAggregateWindowInit

source

func (w *Worker) handleAggregateWindowInit(ctx context.Context, callCtx *vgirpc.CallContext, req AggregateWindowInitRequestWire) (AggregateWindowInitResponseWire, error)

method handleBind

source

func (w *Worker) handleBind(ctx context.Context, callCtx *vgirpc.CallContext, req BindRequestWire) (resp BindResponseWire, err error)

handleBind processes a bind RPC request.

method handleCardinality

source

func (w *Worker) handleCardinality(ctx context.Context, callCtx *vgirpc.CallContext, req CardinalityRequestWire) (card TableCardinality, err error)

handleCardinality processes a table_function_cardinality RPC request.

method handleInit

source

func (w *Worker) handleInit(ctx context.Context, callCtx *vgirpc.CallContext, req InitRequestWire) (result *vgirpc.StreamResult, err error)

handleInit processes an init RPC request and returns a StreamResult.

method handleTableBufferingCombine

source

func (w *Worker) handleTableBufferingCombine(ctx context.Context, cc *vgirpc.CallContext, req TableBufferingCombineRequestWire) (TableBufferingCombineResponseWire, error)

method handleTableBufferingDestructor

source

func (w *Worker) handleTableBufferingDestructor(ctx context.Context, cc *vgirpc.CallContext, req TableBufferingDestructorRequestWire) (TableBufferingDestructorResponseWire, error)

method handleTableBufferingProcess

source

func (w *Worker) handleTableBufferingProcess(ctx context.Context, cc *vgirpc.CallContext, req TableBufferingProcessRequestWire) (TableBufferingProcessResponseWire, error)

method handleTableFunctionDynamicToString

source

func (w *Worker) handleTableFunctionDynamicToString(ctx context.Context, callCtx *vgirpc.CallContext, req TableFunctionDynamicToStringRequestWire) (TableFunctionDynamicToStringResponseWire, error)

method handleTableFunctionStatistics

source

func (w *Worker) handleTableFunctionStatistics(ctx context.Context, callCtx *vgirpc.CallContext, req CardinalityRequestWire) (out []byte, err error)

handleTableFunctionStatistics processes a table_function_statistics RPC request, returning serialized per-column statistics IPC bytes (or nil when unknown).

method handleWritableAttach

source

func (w *Worker) handleWritableAttach(req CatalogAttachRequestWire, c *WritableCatalog) (CatalogAttachResultWire, error)

handleWritableAttach serves catalog_attach for a writable catalog.

method initScalar

source

func (w *Worker) initScalar(ctx context.Context, fn ScalarFunction, initParams *InitParams, processParams *ProcessParams, outputSchema *arrow.Schema, recipe *InitRecipe) (*vgirpc.StreamResult, error)

method initTable

source

func (w *Worker) initTable(ctx context.Context, fn TableFunction, initParams *InitParams, processParams *ProcessParams, outputSchema *arrow.Schema, autoProjectIDs []int32, recipe *InitRecipe) (*vgirpc.StreamResult, error)

method initTableBuffering

source

func (w *Worker) initTableBuffering(ctx context.Context, fn TableBufferingFunction, initParams *InitParams, processParams *ProcessParams, outputSchema *arrow.Schema, phase Phase, recipe *InitRecipe, finalizeStateID *[]byte) (*vgirpc.StreamResult, error)

initTableBuffering handles the two buffering init phases. TABLE_BUFFERING is the sink init: it persists the InitRecipe so process/combine can cold-load it, and returns an empty stream (the sink ingests via the unary process RPC). TABLE_BUFFERING_FINALIZE opens a producer that emits all batches the function returns for one finalize_state_id.

method initTableInOut

source

func (w *Worker) initTableInOut(ctx context.Context, fn TableInOutFunction, initParams *InitParams, processParams *ProcessParams, outputSchema *arrow.Schema, phase Phase, recipe *InitRecipe) (*vgirpc.StreamResult, error)

method loadAggArgs

source

func (w *Worker) loadAggArgs(funcName string, execID []byte, shardKey string) *Arguments

loadAggArgs returns the bind-time arguments stashed by handleAggregateBind.

method loadBufferingParams

source

func (w *Worker) loadBufferingParams(executionID []byte, shardKey string) (TableBufferingFunction, *ProcessParams, error)

loadBufferingParams returns the (function, ProcessParams) pair for a process/combine RPC. It caches the decoded template per execution_id so only the first call pays the storage scan + gob decode + schema/arg re-parse; every later batch of the same execution gets a cheap shallow copy. The cache is evicted by the destructor RPC (end-of-query cleanup).

method loadCachedPartition

source

func (w *Worker) loadCachedPartition(funcName string, execID []byte, partitionID int64, shardKey string) (*WindowPartition, interface{}, error)

loadCachedPartition rebuilds a WindowPartition from the gob-encoded payload stored at WindowInit time, plus the optional gob-encoded WindowInit state.

method lookupAggregate

source

func (w *Worker) lookupAggregate(name string, schemaName *string, attach *[]byte, cc *vgirpc.CallContext) (AggregateFunction, error)

lookupAggregate resolves the aggregate one unary RPC names.

Every aggregate RPC re-resolves by name — they are stateless requests with no bound connection, so the request itself carries the whole identity. Since protocol 1.2.0 that includes schema_name, without which an aggregate declared in two schemas would run whichever implementation the by-name lookup found first: bind could resolve correctly and update/finalize then silently return the other schema’s answer.

method lookupFor

source

func (w *Worker) lookupFor(req *BindRequestWire, params *BindParams) functionLookup

lookupFor builds the (catalog, schema, name) resolution key for a bind_call whose params were parsed on the live path — where parseBindRequest has already opened the attachment down to the catalog’s own plaintext, so the catalog name reads straight off it.

method lookupTable

source

func (w *Worker) lookupTable(name string) (TableFunction, error)

lookupTable resolves a registered table function by name (single-overload lookup; the dynamic_to_string call always knows the resolved name).

method openAttach

source

func (w *Worker) openAttach(envelope []byte, cc *vgirpc.CallContext) ([]byte, error)

openAttach opens an attach_opaque_data envelope, returning the catalog’s own bytes — the framework UUID prefix is stripped. This is what catalog handlers and function bodies see (the catalog never knows about the shard UUID); storage routing uses openAttachFull to reach the UUID instead.

method openAttachFull

source

func (w *Worker) openAttachFull(envelope []byte, cc *vgirpc.CallContext) ([]byte, error)

openAttachFull opens an attach_opaque_data envelope, returning the full framework plaintext uuid(16) || catalog_bytes (not stripped). Storage shards on the leading UUID, so the function-execution paths use this to derive the shard key. Pass-through when there is no signing key.

method openTransaction

source

func (w *Worker) openTransaction(envelope, attachEnvelope []byte, cc *vgirpc.CallContext) ([]byte, error)

openTransaction opens a transaction_opaque_data envelope. attachEnvelope is the (sealed) attach_opaque_data the same call carried — it must match the attach the transaction was minted under, or the open fails.

method originOf

source

func (w *Worker) originOf(kind funcKind, name string, idx int) funcOrigin

originOf returns the home recorded for the idx’th registration of (kind, name). A registration made by a test poking the registry maps directly falls back to the worker’s own catalog and default schema.

method parseBindRequest

source

func (w *Worker) parseBindRequest(req BindRequestWire, callCtx *vgirpc.CallContext) (*BindParams, error)

parseBindRequest converts a wire bind request into BindParams.

method rebuildProcessParams

source

func (w *Worker) rebuildProcessParams(recipe *InitRecipe) (interface{}, *ProcessParams, error)

rebuildProcessParams reconstructs ProcessParams from an InitRecipe. It also returns the resolved function (via overload resolution) to avoid a redundant second resolution by the caller.

method recordOrigin

source

func (w *Worker) recordOrigin(kind funcKind, name string, o funcOrigin)

recordOrigin appends the home for a registration, keeping the funcOrigins slice index-aligned with the registry slice it mirrors. An empty catalog resolves to the worker’s own catalog, so every entry names one.

method registerAggregateRPCs

source

func (w *Worker) registerAggregateRPCs(s *vgirpc.Server)

method registerAggregateStreamingRPCs

source

func (w *Worker) registerAggregateStreamingRPCs(s *vgirpc.Server)

method registerCatalogMethods

source

func (w *Worker) registerCatalogMethods(s *vgirpc.Server)

method registerDynamicToStringRPCs

source

func (w *Worker) registerDynamicToStringRPCs(s *vgirpc.Server)

method registerTableBufferingRPCs

source

func (w *Worker) registerTableBufferingRPCs(s *vgirpc.Server)

method registerWritableFunctions

source

func (w *Worker) registerWritableFunctions(catalogName string)

registerWritableFunctions registers the four generic writable table functions for one writable catalog. Called automatically per RegisterWritableCatalog.

One set is registered per catalog rather than one shared set, because every function is homed in exactly one catalog: a bind arriving through catalog X resolves only functions homed in X, so a single shared registration would be reachable from just one of the worker’s writable catalogs.

method rehydrateFinalize

source

func (w *Worker) rehydrateFinalize(s *FinalizeProducerState) error

method rehydrateScalar

source

func (w *Worker) rehydrateScalar(s *ScalarExchangeState) error

method rehydrateState

source

func (w *Worker) rehydrateState(state interface{}, method string) error

rehydrateState reconstructs non-serializable fields on a deserialized stream state. This is the RehydrateFunc callback for the HTTP server.

method rehydrateTableInOut

source

func (w *Worker) rehydrateTableInOut(s *TableInOutExchangeState) error

method rehydrateTableProducer

source

func (w *Worker) rehydrateTableProducer(s *TableProducerState) error

method resolveAggregate

source

func (w *Worker) resolveAggregate(name, schema, catalog string) (AggregateFunction, error)

resolveAggregate resolves an aggregate RPC to a single registered implementation.

Every aggregate RPC (bind / update / combine / finalize / destructor, the four window calls, the three streaming calls) re-resolves the function by name: they are stateless unary requests with no bound connection, so the request is the only carrier of identity. Before protocol 1.2.0 none of them carried a schema, so an aggregate name declared in two schemas ran whichever the by-name lookup found first — bind could resolve correctly and update/finalize then return the other schema’s answer.

Aggregates do not disambiguate by argument signature (overloads of one aggregate share state semantics, so any of them drives dispatch); the home is the whole of the resolution.

method resolveFunction

source

func (w *Worker) resolveFunction(lk functionLookup) (interface{}, error)

resolveFunction resolves a call to a single registered implementation.

Scoping by home (see scopeToHome) runs first, narrowing “every registration of this name” to “the registrations the caller could possibly mean”. Only then do argument signatures pick between same-schema overloads.

method resolveScanFunction

source

func (w *Worker) resolveScanFunction(req TableScanFunctionGetRequestWire) (*ScanFunctionResult, error)

resolveScanFunction resolves the scan function backing a catalog table, following the same precedence as catalog_table_scan_function_get: writable catalog, registered function-backed table, attach-aware handler, then plain handler. Returns an RpcError when no scan function is available.

method scopeToHome

source

func (w *Worker) scopeToHome(cands []candidate, name, schema, catalog string) ([]candidate, error)

scopeToHome narrows candidates to the registrations the caller could possibly mean, by their recorded home:

• Catalog: a call arriving through catalog X can only reach functions homed

in X. This is what lets one worker process serve two catalogs that each
declare the same function name.

• Schema: a schema-qualified call is exact — only functions declared in that

schema survive, so a name registered in two schemas resolves to the one
the caller named instead of colliding. Naming a schema that does not hold
the function reports where it does live rather than the generic
unknown-function list.

A caller that names no schema keeps every candidate; the caller decides whether a surviving cross-schema tie is an error (see crossSchemaAmbiguity).

method sealAttach

source

func (w *Worker) sealAttach(plaintext []byte, cc *vgirpc.CallContext) ([]byte, error)

sealAttach seals a plaintext attach value into an envelope bound to the caller’s identity.

method serializeCatalogTable

source

func (w *Worker) serializeCatalogTable(schemaName string, ct *CatalogTable) ([]byte, error)

serializeCatalogTable converts a CatalogTable into serialized TableInfo bytes.

method serializedGlobalFunctions

source

func (w *Worker) serializedGlobalFunctions() (SerializedItems, error)

registerCatalogMethods registers all catalog RPC methods on the server. serializedGlobalFunctions returns the IPC-serialized FunctionInfo of every function named by WithGlobalFunctions, in declaration order, resolved against the catalog’s default schema. Names are carried unprefixed — the client applies global_function_prefix itself. Unknown or unlisted names contribute nothing, so declaring a function that was never registered is inert rather than fatal.

method shardKeyForAttach

source

func (w *Worker) shardKeyForAttach(sealed []byte, cc *vgirpc.CallContext) (string, error)

shardKeyForAttach unwraps the sealed attach and derives its shard key. An empty/absent attach yields “” (non-sharding backends ignore the key; the CfDo backend rejects an empty key server-side, the “must not happen” case).

method shardKeyForAttachPtr

source

func (w *Worker) shardKeyForAttachPtr(sealed *[]byte, cc *vgirpc.CallContext) (string, error)

shardKeyForAttachPtr is shardKeyForAttach for a nilable wire field.

method unwrapReqOpaque

source

func (w *Worker) unwrapReqOpaque(reqPtr any, cc *vgirpc.CallContext) error

unwrapReqOpaque unwraps the AttachOpaqueData ([]byte) and, if present, TransactionOpaqueData (*[]byte) fields of a catalog request struct in place, so handler bodies always see plaintext. The transaction envelope is opened with the sealed attach value as part of its AAD, so it stays bound to its parent attach. A no-op when the worker has no signing key.

method writableByAttachOpaqueData

source

func (w *Worker) writableByAttachOpaqueData(attachOpaqueData []byte) *WritableCatalog

writableByAttachOpaqueData returns the writable catalog whose attach_opaque_data matches. Attach IDs are deterministic (“writable:<name>”) so DuckDB-spawned worker processes resolve to the same catalog without sharing in-memory state.

method writableSchemaContentsTables

source

func (w *Worker) writableSchemaContentsTables(c *WritableCatalog, schemaName string) ([][]byte, error)

method writableSchemaCreate

source

func (w *Worker) writableSchemaCreate(c *WritableCatalog, name string, onConflict onConflictAction, comment *string) error

method writableSchemaDrop

source

func (w *Worker) writableSchemaDrop(c *WritableCatalog, name string, ignoreNotFound, cascade bool) error

method writableSchemaGet

source

func (w *Worker) writableSchemaGet(c *WritableCatalog, name string) ([][]byte, error)

method writableSchemas

source

func (w *Worker) writableSchemas(c *WritableCatalog) ([][]byte, error)

method writableTableCreate

source

func (w *Worker) writableTableCreate(c *WritableCatalog, req TableCreateRequestWire) error

method writableTableDrop

source

func (w *Worker) writableTableDrop(c *WritableCatalog, schemaName, name string, ignoreNotFound, cascade bool) error

method writableTableGet

source

func (w *Worker) writableTableGet(c *WritableCatalog, schemaName, tableName string) ([][]byte, error)

func type WorkerOption

source

type WorkerOption func(*Worker)

Description

WorkerOption configures a Worker.

Methods

function WithAttachCatalogs

source

func WithAttachCatalogs(catalogs …AttachCatalogInfo) WorkerOption

WithAttachCatalogs advertises companion catalogs (lakehouse federation) that the client should ATTACH when this VGI catalog attaches. Surfaced via catalog_attach.attach_catalogs; the C++ extension attaches each at VGI-attach time.

function WithAttachOptions

source

func WithAttachOptions(opts …AttachOptionSpec) WorkerOption

WithAttachOptions declares ATTACH-time options accepted by the worker. These are advertised via catalog_catalogs so DuckDB can validate option names/types at ATTACH time. Values passed at ATTACH arrive as the CatalogAttachRequestWire.Options RecordBatch.

function WithAttachOptionsForCatalog

source

func WithAttachOptionsForCatalog(catalogName string, opts …AttachOptionSpec) WorkerOption

WithAttachOptionsForCatalog declares ATTACH-time options for one alias catalog (see WithCatalogAliasInfo) rather than for the worker as a whole.

A worker that serves several catalogs rarely wants one option set spanning all of them: an option that is Required on a gated catalog must not gate the others too. Options registered here are advertised on that catalog’s vgi_catalogs() row only, and enforced only when that catalog is the ATTACH target. The primary catalog keeps using WithAttachOptions.

function WithAttachScanBranchesGetHandler

source

func WithAttachScanBranchesGetHandler(h AttachScanBranchesGetHandler) WorkerOption

WithAttachScanBranchesGetHandler installs an attach-opaque-data-aware scan_branches_get handler for multi-branch (UNION-of-sources) tables.

function WithAttachScanFunctionGetHandler

source

func WithAttachScanFunctionGetHandler(h AttachScanFunctionGetHandler) WorkerOption

WithAttachScanFunctionGetHandler installs an attach-opaque-data-aware scan_function_get handler.

function WithAttachTableGetHandler

source

func WithAttachTableGetHandler(h AttachTableGetHandler) WorkerOption

WithAttachTableGetHandler installs an attach-opaque-data-aware table_get handler.

function WithAttachValidator

source

func WithAttachValidator(v AttachValidator) WorkerOption

WithAttachValidator installs a custom attach validator. Required by the versioned/versioned-tables example workers.

function WithAttachWriteFunctionGetHandler

source

func WithAttachWriteFunctionGetHandler(h AttachWriteFunctionGetHandler) WorkerOption

WithAttachWriteFunctionGetHandler installs an attach-opaque-data-aware handler that resolves the worker function backing INSERT/UPDATE/DELETE on a given (attach_opaque_data, schema, table). Use this for fixture catalogs that publish writable tables outside of the built-in WritableCatalog path (e.g. schema_reconcile, which has its own per-table SQLite store and strict-schema validators).

function WithCatalogAliasInfo

source

func WithCatalogAliasInfo(name string, info CatalogInfo) WorkerOption

WithCatalogAliasInfo registers a catalog alias that advertises itself distinctly in catalog discovery and is isolated per ATTACH. Unlike a plain WithCatalogAliases entry (which shares the primary catalog’s identity and a stable name-based attach scope), an alias registered here:

• appears as its own row in vgi_catalogs() carrying info.DataVersionSpec; and
• mints a random per-ATTACH scope at attach time (info.Name + NUL + random),

so two ATTACHes of the same alias never share attach-scoped state.

info.Name should equal name. Functions meant to surface only under this alias must be registered with Register*ForCatalog(name, …). Used by the accumulate fixture, whose per-ATTACH row collections must be isolated.

function WithCatalogAliases

source

func WithCatalogAliases(names …string) WorkerOption

WithCatalogAliases adds extra catalog names this worker accepts via catalog_attach. Each alias resolves to the same primary read-only catalog (same registered functions, no separate state). Useful for fixture workers that publish multiple cross-language reproducer catalogs out of one binary (projection_repro, schema_reconcile, …).

function WithCatalogComment

source

func WithCatalogComment(comment string) WorkerOption

WithCatalogComment sets the comment reported by catalog_attach (surfaces in duckdb_databases().comment).

function WithCatalogInfo

source

func WithCatalogInfo(info CatalogInfo) WorkerOption

WithCatalogInfo overrides the CatalogInfo record returned by catalog_catalogs (used by vgi_catalogs(location) for discovery). Set it to advertise implementation_version / data_version_spec for versioned workers.

function WithCatalogName

source

func WithCatalogName(name string) WorkerOption

WithCatalogName sets the catalog name.

function WithCatalogTags

source

func WithCatalogTags(tags map[string]string) WorkerOption

WithCatalogTags sets tags reported by catalog_attach (duckdb_databases().tags).

function WithCatalogVersionHook

source

func WithCatalogVersionHook(h CatalogVersionHook) WorkerOption

WithCatalogVersionHook installs a hook that runs on every catalog_version RPC. Use it to assert invariants like cookie presence on HTTP transport.

function WithFunctionStorage

source

func WithFunctionStorage(s FunctionStorage) WorkerOption

WithFunctionStorage injects an explicit FunctionStorage backend. When unset, the worker defaults to a local SQLite backend at the per-user state path. Use this to wire a Cloudflare Durable Object client or any other backend implementing the FunctionStorage interface; combine with vgi/storage/resolve.FromEnv to get vgi-python-style env-driven selection (VGI_WORKER_SHARED_STORAGE=sqlite|cloudflare-do).

function WithGlobalFunctionPrefix

source

func WithGlobalFunctionPrefix(prefix string) WorkerOption

WithGlobalFunctionPrefix sets the prefix a client prepends when publishing this catalog’s global functions (see WithGlobalFunctions) — e.g. prefix “vgi_example” publishes global_scalar as vgi_example_global_scalar. Reported by catalog_attach as global_function_prefix (protocol 1.3.0). Mirrors vgi-python’s Catalog(global_function_prefix=…).

function WithGlobalFunctions

source

func WithGlobalFunctions(names …string) WorkerOption

WithGlobalFunctions names the registered functions this catalog advertises for publication into the client’s global function namespace (DuckDB’s system.main), in the order they should be published. Each name is resolved against the catalog’s default schema; the serialized FunctionInfo of every match is carried in catalog_attach’s global_functions field (protocol 1.3.0), under the unprefixed registered name — the prefix set by WithGlobalFunctionPrefix is the client’s to apply.

Advertising a function does not change where the worker registers it: it stays an ordinary member of its schema and is still callable as catalog.schema.name. Mirrors vgi-python’s Catalog(global_functions=[…]).

function WithHttpSigningKey

source

func WithHttpSigningKey(key []byte) WorkerOption

WithHttpSigningKey configures the HMAC key used by the HTTP transport to sign opaque state tokens. When unset, the HTTP server generates a random key at startup — fine for ephemeral workers, but in-flight state tokens become invalid on restart. Pass a stable key (≥16 bytes, typically from VGI_SIGNING_KEY) for production deployments.

Has no effect when running in stdio mode.

function WithLogFormat

source

func WithLogFormat(format LogFormat) WorkerOption

WithLogFormat selects the stderr log format. Default is text. Ignored when WithLogHandler is also set (the custom handler wins).

function WithLogHandler

source

func WithLogHandler(h slog.Handler) WorkerOption

WithLogHandler sets a custom slog.Handler for all logging. When set, WithLogLevel/WithLogFormat/WithLoggers are ignored — the handler controls its own level and formatting. The package’s named loggers are re-bound to the supplied handler at worker startup.

function WithLogLevel

source

func WithLogLevel(level slog.Level) WorkerOption

WithLogLevel sets the minimum log level for the default handler. The zero value (slog.LevelInfo) logs Info and above.

function WithLoggers

source

func WithLoggers(names …string) WorkerOption

WithLoggers restricts which named loggers emit records. Use to silence noisy subsystems or focus on one (e.g. WithLoggers(“vgi.catalog”)). Empty means all known loggers. Unknown names log a warning but are honored.

function WithSchemaComments

source

func WithSchemaComments(comments map[string]string) WorkerOption

WithSchemaComments overrides the default comment for built-in schemas (“main” and “data”). Other schemas retain their auto-generated comments.

function WithSchemaContentsHandler

source

func WithSchemaContentsHandler(h SchemaContentsHandler) WorkerOption

WithSchemaContentsHandler installs a handler that can replace the tables returned for a given (attach_opaque_data, schema) pair.

function WithSchemaTags

source

func WithSchemaTags(tags map[string]map[string]string) WorkerOption

WithSchemaTags sets schema-level tags surfaced via duckdb_schemas().tags (SchemaInfo.tags). Keyed by schema name. The metadata-quality linter expects vgi.description_llm / vgi.description_md tags on each schema. Tags merge with any previously configured tags for the same schema; later keys win.

function WithSecretTypes

source

func WithSecretTypes(types …SecretTypeSpec) WorkerOption

WithSecretTypes registers secret types that will be sent to DuckDB during catalog_attach.

function WithSettings

source

func WithSettings(settings …SettingSpec) WorkerOption

WithSettings adds custom DuckDB settings to the worker.

function WithSupportsTransactions

source

func WithSupportsTransactions(v bool) WorkerOption

WithSupportsTransactions makes the catalog report supports_transactions=true on attach, so DuckDB threads a transaction_opaque_data through bind/scan inside BEGIN/COMMIT (needed by transaction-scoped storage like tx_cached_value).

struct funcKey

source

type funcKey struct {
kind  funcKind
name  string
}

Description

funcKey identifies one registry slice: (kind, function name).

type funcKind

source

type funcKind uint8

Description

funcKind distinguishes the five registries a function can land in. It is part of the funcOrigins key so an index in one registry never aliases an index in another.

struct funcOrigin

source

type funcOrigin struct {
// catalog is the catalog that owns the function. Never empty: a
// registration that names none is homed in the worker's own catalog.
catalog  string
// schema is the catalog schema declaring the function, lowercased.
schema  string
// unlisted hides the function from every function listing without
// changing its home — it exists to back a catalog table, whose scan
// resolves it by name, so publishing it as a callable function too would
// be redundant surface area. Dispatch is unaffected.
unlisted  bool
}

Description

funcOrigin is the single home of one registered implementation: exactly one (catalog, schema) pair. There is no “unscoped” home — a function is never visible in every schema, nor reachable from every catalog.

A function name is not a unique key: the same name may be registered in two schemas of one catalog, or in two catalogs served by one worker process. Registration is the only place that knows which, so the home is recorded here and is what both the catalog listing and bind dispatch resolve against — a bind naming schema data can only reach an implementation homed in data.

type serverTransport

source

type serverTransport int

Description

buildServer creates and configures the vgirpc.Server with all handlers registered. Shared between RunStdio and RunHttp. serverTransport selects how the worker serves and how execution-storage cleanup is handled.

struct storageCleanupHook

source

type storageCleanupHook struct {
worker    *Worker
pendingKeys  []string  // keys from previous dispatch, candidates for cleanup
staleKeys  []string  // snapshot of pendingKeys at current dispatch start
isHTTP    bool    // when true, skip storage cleanup (no reliable stream-end signal)
}

Description

storageCleanupHook implements vgirpc.DispatchHook to clean up storage entries when execution streams complete.

Cleanup is deferred by one dispatch cycle: keys tracked during a dispatch become “pending” at its end, and are only cleaned up in the next dispatch’s OnDispatchEnd — but only if the new dispatch did NOT reuse them. This handles multi-worker (primary + secondary inits share an execution ID) and table-in-out (INPUT + FINALIZE share an execution ID).

Methods

method OnDispatchEnd

source

func (h *storageCleanupHook) OnDispatchEnd(ctx context.Context, token vgirpc.HookToken, info vgirpc.DispatchInfo, stats *vgirpc.CallStatistics, err error)

OnDispatchEnd cleans up stale storage entries that were not reused in this dispatch and defers this dispatch’s keys to the next cycle. In HTTP mode it does nothing, since there is no reliable stream-end signal.

method OnDispatchStart

source

func (h *storageCleanupHook) OnDispatchStart(ctx context.Context, info vgirpc.DispatchInfo) (context.Context, vgirpc.HookToken)

OnDispatchStart begins tracking storage for an “init” dispatch: it snapshots the previous cycle’s pending keys as stale candidates and installs a storageTracker in the context so keys touched during this dispatch can be recorded.

struct storageTracker

source

type storageTracker struct {
keys []string
}

Description

storageTracker records execution ID hex keys used during a single dispatch.

Methods

method track

source

func (t *storageTracker) track(key string)

track records a hex key, deduplicating against previously tracked keys.

struct storageTrackerKeyType

source

type storageTrackerKeyType struct{}

Description

storageTrackerKeyType is the context key type for the storage tracker.

function buildDefaultValueBatch

source

func buildDefaultValueBatch(mem memory.Allocator, schema *arrow.Schema, dt arrow.DataType, val interface{}) (arrow.RecordBatch, error)

buildDefaultValueBatch creates a 1-row batch with the default value.

function coldAttachScope

source

func coldAttachScope(raw []byte) []byte

coldAttachScope derives the per-ATTACH plaintext scope on the cold-load path, where parseBindRequest ran with no call context and so left the attach value raw (the framework uuid(16) || plaintext, pass-through on subprocess). Strip the UUID prefix to match what OnBind saw. On HTTP, where the raw value is a sealed envelope this can’t unwrap, the buffering RPC handlers re-derive the scope from the live request instead; this is the subprocess best-effort.

function gobDecode

source

func gobDecode(data []byte) (interface{}, error)

gobDecode gob-decodes bytes back into an interface{}.

function gobEncode

source

func gobEncode(v interface{}) ([]byte, error)

gobEncode gob-encodes a value to bytes.

function makeClientBundleHandler

source

func makeClientBundleHandler() http.HandlerFunc

makeClientBundleHandler returns the GET {prefix}/vgi-client.js handler.

function makeLandingHandler

source

func makeLandingHandler(name, serverID string, oauth bool) http.HandlerFunc

makeLandingHandler returns the GET {prefix}/ handler: HTML for browsers, a JSON status document for health checks and for the page’s own identity read.

function newExecutionID

source

func newExecutionID() []byte

newExecutionID generates a UUID-based execution ID.

function serializeSecretTypeSpec

source

func serializeSecretTypeSpec(spec SecretTypeSpec) ([]byte, error)

serializeSecretTypeSpec serializes a SecretTypeSpec to Arrow IPC bytes. Format: RecordBatch with schema {name: string, description: string, parameters_schema: binary}.

function serializeSettingSpec

source

func serializeSettingSpec(spec SettingSpec) ([]byte, error)

serializeSettingSpec serializes a SettingSpec to Arrow IPC bytes. Format: RecordBatch with schema {name: string, description: string, type: binary, default_value: binary?}.

function vgiGoVersion

source

func vgiGoVersion() string

vgiGoVersion returns the build version of the module that embeds this package, falling back to “dev”. Surfaced as the status document’s version.

function writeJSON

source

func writeJSON(rw http.ResponseWriter, v any)