Worker & serving

On this page

Building a worker and putting it on a transport.

trait ExchangeStream

source

pub trait ExchangeStream {
  fn header(&self) -> Option<&RecordBatch>;

  fn exchange(&mut self, input: &RecordBatch) -> Result<Option<(RecordBatch, Metadata)>>;

  fn close(&mut self) -> Result<()>;

  fn cancel(&mut self) -> Result<()>;
}

Description

An open exchange stream: the client sends batches, the worker answers.

trait ProducerStream

source

pub trait ProducerStream {
  fn header(&self) -> Option<&RecordBatch>;

  fn tick(&mut self) -> Result<Option<(RecordBatch, Metadata)>>;

  fn cancel(&mut self) -> Result<()>;
}

Description

An open producer stream: the worker emits, the client ticks.

trait VgiTransport

source

pub trait VgiTransport: Send {
  fn call_unary(&mut self, method: &str, params: &RecordBatch) -> Result<RecordBatch>;

  fn open_producer<'a>(
      &'a mut self,
      method: &str,
      params: &RecordBatch,
      has_header: bool,
  ) -> Result<Box<dyn ProducerStream + 'a>>;

  fn open_exchange<'a>(
      &'a mut self,
      method: &str,
      params: &RecordBatch,
      has_header: bool,
  ) -> Result<Box<dyn ExchangeStream + 'a>>;

  fn label(&self) -> &str;
}

Description

Something that can carry the VGI protocol.

struct Dispatcher

source

pub struct Dispatcher {
  pub catalog_name: String,
  pub scalars: HashMap<String, Vec<Arc<dyn ScalarFunction>>>,
  pub tables: HashMap<String, Vec<Arc<dyn TableFunction>>>,
  pub tableinouts: HashMap<String, Vec<Arc<dyn TableInOutFunction>>>,
  pub buffering: HashMap<String, Vec<Arc<dyn TableBufferingFunction>>>,
  pub aggregates: HashMap<String, Vec<Arc<dyn AggregateFunction>>>,
  pub store: Arc<dyn FunctionStorage>,
  pub catalog: catalog::CatalogModel,
  pub secondary: Vec<catalog::CatalogModel>,
  pub(crate) secondary_functions: Vec<Vec<String>>,
  pub(crate) hidden_functions: std::collections::HashSet<String>,
  pub(crate) scopes: HashMap<(FnKind, String), Vec<FunctionScope>>,
  pub secret_types: Vec<catalog::SecretTypeSpec>,
  pub settings: Vec<catalog::SettingSpec>,
  pub attach_catalogs: Vec<crate::protocol::dtos::AttachCatalogInfo>,
  pub copy_from_formats: Vec<Arc<dyn crate::copy_from::CopyFromFunction>>,
  pub copy_to_formats: Vec<Arc<dyn crate::copy_to::CopyToFunction>>,
  exec_counter: AtomicU64,
}

Description

Shared dispatch state. Cloned (as Arc) into every RPC handler closure.

Methods

method decode_init_state

source

pub fn decode_init_state(&self, bytes: &[u8]) -> Result<vgi_rpc::stream::StreamStateKind>

Rebuild a stateless exchange stream from its HTTP-continuation blob. Registered as the init method’s state decoder so a pooled HTTP worker can resume a scalar / table-in-out exchange from an AEAD token.

method handle_catalog_catalogs

source

pub fn handle_catalog_catalogs(&self, _req: &Request) -> Result<Option<RecordBatch>>

catalog_catalogs — discovery: advertise this worker’s catalog plus its version metadata so clients can inspect before attaching.

method handle_catalog_copy_from_formats

source

pub fn handle_catalog_copy_from_formats(&self, req: &Request) -> Result<Option<RecordBatch>>

catalog_copy_from_formats — advertise the worker’s custom COPY ... FROM / COPY ... TO formats. Catalog-level (not schema-scoped). Only the primary catalog owns these formats; secondaries advertise none.

A reader advertises direction="from" and a writer direction="to". When the same format_name is registered on both sides, the two are paired into a single direction="both" entry — the extension registers one DuckDB CopyFunction per format name and keeps the first registration, so emitting two entries would silently drop the second direction. Pairing requires the reader and writer to share a handler_name (the wire carries one handler per format, and the extension hands it to both sides); the advertised options are the union of the two directions’ argument specs.

method handle_empty_items

source

pub fn handle_empty_items(&self, _req: &Request) -> Result<Option<RecordBatch>>

Empty ItemsResult for the contents/get methods not yet implemented.

method handle_read_only

source

pub fn handle_read_only(&self, _req: &Request) -> Result<Option<RecordBatch>>

Every catalog-mutating DDL RPC ends here: the example catalog is read-only, so the request is accepted (proving the wire contract is intact) and rejected with a clear catalog is read-only error.

method handle_table_column_statistics_get

source

pub fn handle_table_column_statistics_get(&self, req: &Request) -> Result<Option<RecordBatch>>

Per-column optimizer statistics for a table. Returns the sparse-union IPC batch (result-wrapped), empty when the table declares no stats.

method handle_table_function_cardinality

source

pub fn handle_table_function_cardinality(
  &self,
  req: &Request,
  ctx: &CallContext,
) -> Result<Option<RecordBatch>>

Per-call cardinality for a function-backed table scan.

method handle_table_function_dynamic_to_string

source

pub fn handle_table_function_dynamic_to_string(
  &self,
  req: &Request,
) -> Result<Option<RecordBatch>>

Post-execution profiling info (EXPLAIN ANALYZE Extra Info).

method handle_table_function_statistics

source

pub fn handle_table_function_statistics(
  &self,
  req: &Request,
  ctx: &CallContext,
) -> Result<Option<RecordBatch>>

Per-call statistics for a function-backed table scan (e.g. sequence).

method handle_table_scan_branches_get

source

pub fn handle_table_scan_branches_get(&self, req: &Request) -> Result<Option<RecordBatch>>

Multi-branch scan resolution. A single-source table returns one branch wrapping its scan function; the list must be non-empty.

method handle_table_scan_function_get

source

pub fn handle_table_scan_function_get(&self, req: &Request) -> Result<Option<RecordBatch>>

Lazy scan-function resolution for non-inlined function-backed tables. Returns a FLAT ScanFunctionResult batch (no {result} envelope).

method handle_void

source

pub fn handle_void(&self, _req: &Request) -> Result<Option<RecordBatch>>

Void result (commit / rollback / detach / drop).

method hide_function

source

pub fn hide_function(&mut self, name: impl Into<String>)

Hide name from catalog_schema_contents_functions without unregistering it. The function stays bindable — a function-backed catalog table still resolves its scan — but DuckDB never creates a SQL callable for it, so the table is the only entry point.

method register_aggregate_scoped

source

pub fn register_aggregate_scoped(
  &mut self,
  f: Arc<dyn AggregateFunction>,
  scope: FunctionScope,
)

Register an aggregate declared in a specific catalog schema.

method register_attach_catalog

source

pub fn register_attach_catalog(&mut self, info: crate::protocol::dtos::AttachCatalogInfo)

Advertise a companion catalog for the client to ATTACH at VGI-attach time (surfaced in catalog_attach.attach_catalogs; lakehouse federation).

method register_buffering_scoped

source

pub fn register_buffering_scoped(
  &mut self,
  f: Arc<dyn TableBufferingFunction>,
  scope: FunctionScope,
)

Register a table-buffering function declared in a specific catalog schema.

method register_copy_from

source

pub fn register_copy_from(&mut self, f: Arc<dyn crate::copy_from::CopyFromFunction>)

Record a custom COPY ... FROM format reader for advertisement via catalog_copy_from_formats. The reader must also be registered as a table function under its handler name (done by Worker::register_copy_from).

method register_copy_to

source

pub fn register_copy_to(&mut self, f: Arc<dyn crate::copy_to::CopyToFunction>)

Record a custom COPY ... TO format writer for advertisement via catalog_copy_from_formats (direction="to"). The writer must also be registered as a table-buffering function under its handler name (done by Worker::register_copy_to).

method register_scalar_scoped

source

pub fn register_scalar_scoped(&mut self, f: Arc<dyn ScalarFunction>, scope: FunctionScope)

Register a scalar declared in a specific catalog schema.

method register_secondary_catalog

source

pub fn register_secondary_catalog(
  &mut self,
  model: catalog::CatalogModel,
  functions: Vec<String>,
)

Add a secondary catalog (served alongside the primary, MetaWorker-style), declaring the worker-global function names it owns (so its function listing is scoped and the primary hides them).

method register_table_if_absent

source

pub fn register_table_if_absent(&mut self, f: Arc<dyn TableFunction>)

Register f only if no table function with its name is registered yet. Used by Worker::set_catalog to auto-register catalog tables’ embedded scan_function_impl without clobbering an explicit register_table.

method register_table_in_out_scoped

source

pub fn register_table_in_out_scoped(
  &mut self,
  f: Arc<dyn TableInOutFunction>,
  scope: FunctionScope,
)

Register a table-in-out function declared in a specific catalog schema.

method register_table_scoped

source

pub fn register_table_scoped(&mut self, f: Arc<dyn TableFunction>, scope: FunctionScope)

Register a table (producer) function declared in a specific catalog schema.

struct ExchangeBlob

source

pub struct ExchangeBlob {
  pub kind: String, // "scalar" | "table_in_out"
  pub function_name: String,
  pub output_schema: Vec<u8>,
  pub input_schema: Vec<u8>, // empty = none
  pub arguments: Vec<u8>,
  pub settings: Vec<u8>,
  pub secrets: Vec<u8>,
  pub execution_id: Vec<u8>,
  pub substream_id: Vec<u8>,
  pub init_opaque: Vec<u8>,
  pub pushdown_filters: Vec<u8>, // empty = none
  pub auto_apply: bool,
  pub inner_resume: Vec<u8>,
  pub at_unit: String,
  pub at_value: String,
  pub catalog_name: String,
  pub schema_name: String,
}

Description

Serializable rebuild info for an exchange stream, so HTTP continuations can reconstruct the state from an AEAD token on any pooled worker.

struct FunctionScope

source

pub struct FunctionScope {
  pub catalog: String,
  pub schema: String,
}

Description

Where a function instance is declared: the VGI catalog that owns it and the schema within that catalog. Every registered function has exactly one.

A function name is not a unique key — a worker may declare the same name in two schemas of one catalog, or (serving several catalogs from one process) in the same schema name of two different catalogs. The home is what breaks the tie, and the bind request carries the caller’s half of it: the schema on BindRequest::schema_name and the catalog inside attach_opaque_data.

There is deliberately no “unscoped” state. A function with no home would be advertised everywhere and would match any call, which makes example.data.f() and example.main.f() indistinguishable — the exact ambiguity this type exists to remove. Registering without naming a home (Worker::register_scalar and friends) still yields one: the worker’s own catalog and its default schema (main), which is where DuckDB registers such functions anyway.

Methods

method new

source

pub fn new(catalog: impl Into<String>, schema: impl Into<String>) -> Self

Declare a function into schema of catalog.

struct HttpTransport

source

pub struct HttpTransport {
  client: HttpClient,
  label: String,
}

Description

An HTTP transport.

Methods

method new

source

pub fn new(client: HttpClient, label: impl Into<String>) -> Self

Wrap an already-connected HttpClient.

struct StreamTransport

source

pub struct StreamTransport {
  client: RpcClient,
  label: String,
}

Description

A byte-stream transport: subprocess, AF_UNIX, or TCP.

Methods

method new

source

pub fn new(client: RpcClient, label: impl Into<String>) -> Self

Wrap an already-connected [RpcClient].

struct Worker

source

pub struct Worker {
  disp: Dispatcher,
  server_id: Option<String>,
}

Description

VGI wire protocol version advertised to the C++ extension.

Enforced as an exact major+minor match at the dispatch boundary (carried in vgi_rpc.protocol_version custom metadata), so this must track A VGI worker: the process DuckDB launches and talks to.

Build one with [Worker::new], register one or more functions (register_scalar, register_table, register_aggregate, …) and/or a catalog (set_catalog), then call run to serve. run does not return — it serves until DuckDB disconnects.

Examples

use vgi::Worker;
fn main() {
  let mut worker = Worker::new();
  worker.register_scalar(UpperCase);
  worker.run(); // never returns
}

Methods

method build_server

source

pub fn build_server(self) -> RpcServer

Build the configured [RpcServer], registering every VGI method.

method hide_function

source

pub fn hide_function(&mut self, name: impl Into<String>)

Hide an already-registered function from the catalog’s advertised function list. It stays bindable, so a function-backed catalog table can still resolve it as a scan function, but the client creates no SQL callable for it — use this when the table is the only intended entry point.

method new

source

pub fn new() -> Self

Create a worker.

The catalog name DuckDB sees in ATTACH 'name' (TYPE vgi, …) defaults to example and can be overridden with the VGI_WORKER_CATALOG_NAME environment variable. (In SQL you qualify functions by the alias you give ATTACH, not by this internal name.)

method register_aggregate

source

pub fn register_aggregate(&mut self, f: impl crate::aggregate::AggregateFunction + ’static)

Register an aggregate function.

method register_aggregate_in

source

pub fn register_aggregate_in(
  &mut self,
  catalog: &str,
  schema: &str,
  f: impl crate::aggregate::AggregateFunction + 'static,
)

Register an aggregate function declared in schema of catalog. See register_scalar_in.

method register_attach_catalog

source

pub fn register_attach_catalog(&mut self, info: crate::protocol::dtos::AttachCatalogInfo)

Advertise a companion catalog for the client to ATTACH at VGI-attach time (surfaced via catalog_attach.attach_catalogs; lakehouse federation).

method register_buffering

source

pub fn register_buffering(
  &mut self,
  f: impl crate::buffering::TableBufferingFunction + 'static,
)

Register a table-buffering function.

method register_buffering_in

source

pub fn register_buffering_in(
  &mut self,
  catalog: &str,
  schema: &str,
  f: impl crate::buffering::TableBufferingFunction + 'static,
)

Register a table-buffering function declared in schema of catalog. See register_scalar_in.

method register_copy_from

source

pub fn register_copy_from(&mut self, f: impl crate::copy_from::CopyFromFunction + ’static)

Register a custom COPY ... FROM format reader.

The reader is exposed two ways: as a producer-mode table function (so the whole table bind/init/scan path is reused) and as an advertised COPY ... FROM format via catalog_copy_from_formats. Users then run COPY target FROM 'path' (FORMAT <alias>.<format>, opt val, ...). See [crate::copy_from::CopyFromFunction].

method register_copy_to

source

pub fn register_copy_to(&mut self, f: impl crate::copy_to::CopyToFunction + ’static)

Register a custom COPY ... TO format writer.

The writer is exposed two ways: as a table-buffering (Sink+Combine) function (so the whole buffering RPC path is reused — write() per shard, close() for the terminal destination write; no Source phase) and as an advertised COPY ... TO format via catalog_copy_from_formats (direction="to"). Users then run COPY (source) TO 'path' (FORMAT <alias>.<format>, opt val, ...). See [crate::copy_to::CopyToFunction].

method register_scalar

source

pub fn register_scalar(&mut self, f: impl ScalarFunction + ’static)

Register a scalar function.

It is declared in this worker’s own catalog, in the main schema — every function has exactly one home. Use register_scalar_in to place it in a different catalog schema.

method register_scalar_in

source

pub fn register_scalar_in(
  &mut self,
  catalog: &str,
  schema: &str,
  f: impl ScalarFunction + 'static,
)

Register a scalar function declared in schema of catalog.

A function name is not a unique key: the same name may be declared in two schemas of one catalog, or in two catalogs served by one worker process. The home is what tells them apart — the function is advertised only in that schema, and only a bind naming that schema resolves to it. See FunctionScope.

method register_secondary_catalog

source

pub fn register_secondary_catalog(
  &mut self,
  model: crate::catalog::CatalogModel,
  functions: Vec<String>,
)

Add a secondary catalog served alongside the primary (MetaWorker model): advertised by catalog_catalogs and attachable by its name. functions names the worker-global functions it owns (scopes its function listing).

method register_secret_type

source

pub fn register_secret_type(&mut self, spec: crate::catalog::SecretTypeSpec)

Register a secret type (surfaced via catalog_attach).

method register_setting

source

pub fn register_setting(&mut self, spec: crate::catalog::SettingSpec)

Register a custom setting (surfaced via catalog_attach).

method register_table

source

pub fn register_table(&mut self, f: impl crate::table_function::TableFunction + ’static)

Register a table (producer) function.

method register_table_in

source

pub fn register_table_in(
  &mut self,
  catalog: &str,
  schema: &str,
  f: impl crate::table_function::TableFunction + 'static,
)

Register a table (producer) function declared in schema of catalog. See register_scalar_in.

method register_table_in_out

source

pub fn register_table_in_out(
  &mut self,
  f: impl crate::table_in_out::TableInOutFunction + 'static,
)

Register a table-in-out function.

method register_table_in_out_in

source

pub fn register_table_in_out_in(
  &mut self,
  catalog: &str,
  schema: &str,
  f: impl crate::table_in_out::TableInOutFunction + 'static,
)

Register a table-in-out function declared in schema of catalog. See register_scalar_in.

method run

source

pub fn run(self)

Parse argv and serve over the selected transport, blocking until the connection closes.

DuckDB launches the worker with the right flags; you normally just call run() from main. The transport is chosen from argv:

• (none) — stdio (the default).
• --unix <path> — Unix-socket launcher transport (--idle-timeout <secs> optional; Unix only).
• --tcp [<host>:]<port> — TCP launcher transport (raw Arrow-IPC framing, no auth/TLS; host defaults to 127.0.0.1, port 0 auto-selects; --idle-timeout <secs> optional).
• --http — HTTP transport (Arrow-IPC over HTTP). Bearer auth is enabled by setting VGI_BEARER_TOKENS (token=principal,…).

method serve_reader_writer

source

pub fn serve_reader_writer<R: std::io::Read, W: std::io::Write>(self, mut r: R, mut w: W)

Serve the worker’s RPC protocol over an arbitrary byte stream (used by the SAB transport and native tests). Blocking; consumes the worker.

method server_id

source

pub fn server_id(mut self, id: impl Into<String>) -> Self

Override the server id.

method set_catalog

source

pub fn set_catalog(&mut self, model: crate::catalog::CatalogModel)

Install the declarative catalog (views / macros / tables).

Any catalog table built with [crate::catalog::CatTable::with_function] carries an embedded scan function; these are auto-registered into the dispatch table here (deduped by name), so a function-backed table needs no separate [Worker::register_table] call — parity with the Go CatalogTable.Function ergonomics.

function serve_http

source

pub fn serve_http(
  server: Arc<RpcServer>,
  authenticate: Option<vgi_rpc::Authenticate>,
  landing_info: Option<vgi_rpc::http::LandingInfo>,
)

Description

Serve over HTTP: bind a TCP port, announce it with PORT:<n>, and serve the axum router. An optional authenticate callback enables bearer auth.

Gated on transport-http (tokio/axum) so the crate stays wasm-buildable.

function serve_stdio

source

pub fn serve_stdio(server: Arc<RpcServer>)

Description

Serve a single sequential Arrow-IPC stream over stdin/stdout until EOF.

function serve_tcp

source

pub fn serve_tcp(server: Arc<RpcServer>, host: &str, port: u16, idle_timeout: f64)

Description

Bind a TCP socket, announce it with TCP:<host>:<port> (the actual bound port, so port == 0 ephemeral binds are discoverable), and serve each inbound connection on a worker thread. idle_timeout (seconds, 0 = never) self-shuts the worker after that long without a new connection, matching the launcher protocol.

Raw TCP framing carries no authentication or TLS — bind loopback / a trusted network only; use [serve_http] for untrusted networks.

Native only: this thread-per-connection variant needs std::thread, which wasm lacks. A single-thread wasm serve_tcp (wasip2 std::net, no threads) is added separately for the in-browser/wasmtime shared-worker path.

function serve_unix

source

pub fn serve_unix(server: Arc<RpcServer>, path: &str, idle_timeout: f64)

Description

Bind an AF_UNIX socket, announce it with UNIX:<path>, and serve each inbound connection on a worker thread. idle_timeout (seconds, 0 = never) self-shuts the worker after that long without a new connection, matching the launcher protocol.

Unix-only: the launcher transport relies on AF_UNIX sockets. On other platforms the worker falls back to stdio/HTTP (see [crate::Worker::run]).