Client

On this page

Calling a VGI worker from Rust, without DuckDB in the middle.

trait CatalogAuth

source

pub trait CatalogAuth: Send + Sync {
  fn bearer_token(&self) -> Option<String>;

  fn handle_unauthorized(&self, challenge: Option<&OAuthChallenge>) -> Result<String>;

  fn is_explicitly_configured(&self) -> bool;

  fn identity(&self) -> Identity;
}

Description

A credential holder for one attached catalog.

trait HttpTransport

source

pub trait HttpTransport: Send + Sync {
  fn get(&self, url: &str) -> Result<String>;
  fn post_form(&self, url: &str, form: &[(&str, &str)]) -> Result<(u16, String)>;
}

Description

Something that can perform the HTTP calls discovery and grants need.

A trait so the flows are testable against a mock provider without a network.

trait UserInteraction

source

pub trait UserInteraction: Send + Sync {
  fn prompt_device_code(&self, info: &DeviceCodePrompt);

  fn still_waiting(&self, _elapsed: Duration) {}

  fn authenticated(&self) {}
}

Description

How a human completes an interactive login.

Required, not optional. The C++ extension routes device-code prompts through DuckDB’s log manager, so on any client that does not render those logs a login silently appears to hang — the single worst UX trap in that implementation. Making this a mandatory constructor argument means a client cannot accidentally have nowhere to show the prompt.

constant AUTH_REASON_HEADER

source

pub const AUTH_REASON_HEADER: &str = “VGI-Auth-Reason”;

Description

Header carrying the reason code.

struct AnonymousAuth

source

pub struct AnonymousAuth;

Description

No credential at all.

enum ArgValue

source

pub enum ArgValue {
  Int(i64),
  Float(f64),
  Text(String),
  Bool(bool),
  Null(DataType),
  Placeholder(DataType),
}

Description

One argument value.

[ArgValue::Placeholder] is how a column argument is expressed: the type is stated but the value is null, because the data arrives per-row later.

struct Arguments

source

pub struct Arguments {
  positional: Vec<ArgValue>,
  named: Vec<(String, ArgValue)>,
}

Description

A call’s arguments, positional and named.

Methods

method from_scan_arguments

source

pub fn from_scan_arguments(bytes: &[u8]) -> Result<Self>

Decode the arguments a worker attached to a catalog table’s scan function.

A different encoding from [Arguments::to_ipc]

These are not round-tripped bind arguments. ScanFunctionResult declares positional_arguments and named_arguments, and the wire form is a flat batch whose columns are the arguments, read from row 0:

• a column named arg_<N> is positional argument N
• any other column name is a named argument

There is no args struct wrapper here, which is what [Arguments::to_ipc] produces. Forwarding these bytes straight into BindRequest.arguments therefore fails on the worker with Field "args" does not exist in schema — they have to be decoded and re-encoded. The DuckDB extension does the same (DecodeScanArguments).

Empty bytes, or a batch with no rows, means “no arguments”.

method is_empty

source

pub fn is_empty(&self) -> bool

Whether any argument was supplied.

method named

source

pub fn named(mut self, name: impl Into<String>, value: impl Into<ArgValue>) -> Self

Set a named argument. A repeated name replaces the earlier value.

method new

source

pub fn new() -> Self

No arguments.

method positional

source

pub fn positional(mut self, value: impl Into<ArgValue>) -> Self

Append a positional argument.

method to_ipc

source

pub fn to_ipc(&self) -> Result<Bytes>

Encode to the IPC blob a BindRequest carries.

An empty argument list encodes to empty bytes, which the worker reads as “no arguments” — see Arguments::parse.

enum AuthReason

source

pub enum AuthReason {
  MissingCredential,
  InvalidCredential,
  ExpiredCredential,
  InsufficientScope,
  ProxyRequired,
  Unauthorized,
}

Description

Why a request was refused.

A closed set — an unrecognised code from a newer peer maps to [AuthReason::Unauthorized] rather than failing, so a client keeps working against a service that has learned a new code.

Methods

method as_str

source

pub fn as_str(self) -> &’static str

The wire spelling.

method is_retryable_after_refresh

source

pub fn is_retryable_after_refresh(self) -> bool

Whether refreshing the credential and retrying could plausibly help.

Only expired_credential qualifies. Retrying an invalid_credential unchanged is explicitly forbidden by the spec, and re-presenting the same identity for insufficient_scope cannot change the answer.

method parse

source

pub fn parse(s: &str) -> Self

Parse a wire code, mapping anything unrecognised to Unauthorized.

struct AuthenticatedHttpTransport

source

pub struct AuthenticatedHttpTransport {
  base_url: String,
  auth: Arc<dyn CatalogAuth>,
  probe: Box<dyn OAuthHttp>,
  client: Option<HttpClient>,
  built_with: Option<String>,
  label: String,
}

Description

An HTTP transport that authenticates.

Methods

method new

source

pub fn new(base_url: impl Into<String>, auth: Arc<dyn CatalogAuth>) -> Self

Build a transport that presents auth’s credential.

method with_probe

source

pub fn with_probe(
  base_url: impl Into<String>,
  auth: Arc<dyn CatalogAuth>,
  probe: Box<dyn OAuthHttp>,
) -> Self

As [Self::new], with a custom transport for the challenge probe.

The probe talks to the worker to read its WWW-Authenticate header, so tests can supply a canned answer.

struct BearerAuth

source

pub struct BearerAuth {
  token: String,
}

Description

A static bearer token.

Cannot recover from a 401: there is no refresh, no discovery, and retrying the same token unchanged is exactly what the spec forbids. So a rejection is terminal and says so.

Methods

method new

source

pub fn new(token: impl Into<String>) -> Self

Hold a static token.

struct BindSpec

source

pub struct BindSpec {
  pub function_name: String,
  pub function_type: FunctionType,
  pub schema_name: Option<String>,
  pub arguments: Arguments,
  pub raw_arguments: Option<Bytes>,
  pub settings: Option<Bytes>,
  pub at: Option<At>,
}

Description

What to bind.

Methods

method in_schema

source

pub fn in_schema(mut self, schema: impl Into<String>) -> Self

Set the owning schema.

method table

source

pub fn table(function_name: impl Into<String>) -> Self

A table-function bind with no arguments.

method with_arguments

source

pub fn with_arguments(mut self, args: Arguments) -> Self

Set the call arguments.

method with_raw_arguments

source

pub fn with_raw_arguments(mut self, args: Bytes) -> Self

Use argument bytes the worker already encoded, verbatim.

Takes precedence over [Self::with_arguments].

struct BoundFunction

source

pub struct BoundFunction {
  bind_call: Bytes,
  response: BindResponse,
  output_schema: SchemaRef,
}

Description

A bound function, ready to scan.

Holds the serialized bind request as well as the response, because init echoes the whole bind call back — the worker re-reads it rather than keeping per-bind state.

Methods

method opaque_data

source

pub fn opaque_data(&self) -> &Bytes

The worker’s opaque bind state.

method output_schema

source

pub fn output_schema(&self) -> &SchemaRef

The scan’s output schema, resolved at bind time.

method required_secret_types

source

pub fn required_secret_types(&self) -> &[String]

Secret types the worker asked the client to resolve, if any.

A non-empty list means the worker wants a second bind carrying resolved secrets. This client does not yet drive that two-phase bind.

struct CacheKey

source

pub struct CacheKey {
  pub identity_scope: String,
  pub worker_label: String,
  pub function: String,
  pub arguments: Vec<u8>,
  pub projection: Option<Vec<i64>>,
  pub filters: Option<Vec<u8>>,
  pub catalog_version: i64,
  pub at: Option<(String, String)>,
}

Description

Everything that makes one scan’s result different from another’s.

Field order here is the definition of cache identity; adding a dimension the worker can vary on without adding it here is how a cache serves wrong rows.

struct CacheLimits

source

pub struct CacheLimits {
  pub max_entry_bytes: usize,
  pub max_total_bytes: usize,
  pub max_entries: usize,
  pub default_ttl: Duration,
}

Description

Caps.

struct CacheStats

source

pub struct CacheStats {
  pub hits: u64,
  pub misses: u64,
  pub inserts: u64,
  pub evictions_lru: u64,
  pub evictions_ttl: u64,
  pub refusals: u64,
  pub entries: usize,
  pub total_bytes: usize,
}

Description

Counters, for diagnosis.

struct CachedEntry

source

pub struct CachedEntry {
  batches: Vec<RecordBatch>,
  rows: usize,
  bytes: usize,
  stored_at: Instant,
  ttl: Option<Duration>,
  pub etag: Option<String>,
  pub revalidatable: bool,
  hits: u64,
}

Description

A stored result.

Methods

method batches

source

pub fn batches(&self) -> &[RecordBatch]

The cached batches.

method bytes

source

pub fn bytes(&self) -> usize

Approximate bytes held.

method hits

source

pub fn hits(&self) -> u64

How many times this entry has been served.

method is_stale_at

source

pub fn is_stale_at(&self, now: Instant) -> bool

Whether the entry is past its freshness lifetime.

An entry with no TTL never goes stale — the worker said so.

method rows

source

pub fn rows(&self) -> usize

Total rows across every batch.

constant DEFAULT_IDLE_TIMEOUT

source

pub const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(5);

Description

How long an idle connection is kept before eviction.

Matches the extension’s vgi_worker_pool_idle_limit_seconds default.

constant DEFAULT_MAX_IDLE

source

pub const DEFAULT_MAX_IDLE: usize = 256;

Description

How many connections the pool holds before it starts closing them.

Matches the extension’s vgi_worker_pool_max default.

struct DeviceCodePrompt

source

pub struct DeviceCodePrompt {
  pub verification_uri: String,
  pub user_code: String,
  pub verification_uri_complete: Option<String>,
  pub resource_name: Option<String>,
}

Description

What to show a user starting a device-code login.

struct DiscoveredEndpoints

source

pub struct DiscoveredEndpoints {
  pub token_endpoint: String,
  pub device_authorization_endpoint: Option<String>,
  pub authorization_endpoint: Option<String>,
  pub device_client_id: Option<String>,
  pub device_client_secret: Option<String>,
  pub client_id: Option<String>,
  pub client_secret: Option<String>,
  pub scope: String,
  pub resource_name: Option<String>,
  pub use_id_token_as_bearer: bool,
}

Description

Everything needed to run a flow, resolved from the challenge.

struct Exchange

source

pub struct Exchange<'a> {
  stream: Box<dyn ExchangeStream + 'a>,
  header: GlobalInitResponse,
  schema: SchemaRef,
  parent_rows: Option<Vec<i32>>,
  closed: bool,
}

Description

An open exchange: send input batches, read answers.

Methods

method cancel

source

pub fn cancel(&mut self) -> Result<()>

Ask the worker to stop early.

method close

source

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

Signal input EOS.

method execution_id

source

pub fn execution_id(&self) -> &Bytes

The worker-minted id for this exchange.

method parent_rows

source

pub fn parent_rows(&self) -> Option<&[i32]>

Per-output-row provenance from the most recent answer, if the worker supplied it.

A vgi_rpc.parent_row array says, for each output row, which input row produced it — how a 1→N or 1→0 transform stays attributable. Its absence means an identity 1→1 map, in which case output rows correspond to input rows positionally.

method schema

source

pub fn schema(&self) -> &SchemaRef

The output schema, as resolved at bind.

method send

source

pub fn send(&mut self, input: &RecordBatch) -> Result<Option<RecordBatch>>

Send one input batch and read the worker’s answer.

Ok(None) means the worker ended the stream. A zero-row answer is returned as an empty batch rather than None, because “no rows for this input” is a real answer in exchange mode and callers of a 1→0 transform need to see it.

enum FunctionType

source

pub enum FunctionType {
  Table,
  TableBuffering,
  TableInOut,
  Scalar,
  Aggregate,
}

Description

Which flavour of function is being bound.

Methods

method as_str

source

pub fn as_str(self) -> &’static str

The wire spelling.

enum Identity

source

pub enum Identity {
  Anonymous,
  OAuth {
      issuer: String,
      subject: String,
  },
  Bearer(String),
  Unresolved,
}

Description

The identity a set of credentials resolves to.

Methods

method fingerprint

source

pub fn fingerprint(&self, salt: &[u8]) -> String

The cache-isolation fingerprint.

An empty string means do not cache — see the module docs.

method is_cacheable

source

pub fn is_cacheable(&self) -> bool

Whether results for this identity may be cached at all.

enum Ineligible

source

pub enum Ineligible {
  NotCacheable,
  NoFreshness,
  IdentityUnresolved,
  EntryTooLarge,
  TransactionScoped,
}

Description

Why a scan was not cached.

enum NullOrder

source

pub enum NullOrder {
  First,
  Last,
}

Description

Where nulls sort.

struct OAuthAuth

source

pub struct OAuthAuth {
  http: Box<dyn HttpTransport>,
  interaction: Box<dyn UserInteraction>,
  flow_timeout: Duration,
  wait_timeout: Duration,
  state: Mutex<State>,
  cv: Condvar,
}

Description

An OAuth identity that can refresh and, when it must, run an interactive flow.

Single-flight

Several threads hitting an expired token all get a 401 at once. One becomes the leader and performs the exchange; the rest wait and reuse its result, so there is exactly one token exchange rather than N.

The wait is bounded. The C++ implementation waits without a timeout, so every other thread blocks for the full duration of a device-code poll — up to two minutes of a human typing a code. A bounded wait turns that into a clear error instead of an apparent hang.

Methods

method clear

source

pub fn clear(&self)

Forget every token, as for a logout.

method is_authenticated

source

pub fn is_authenticated(&self) -> bool

Whether a usable token is held right now.

method new

source

pub fn new(http: Box<dyn HttpTransport>, interaction: Box<dyn UserInteraction>) -> Self

A fresh OAuth identity with nothing seeded.

interaction is required rather than optional: a device-code prompt with nowhere to go is a login that appears to hang forever.

method with_flow_timeout

source

pub fn with_flow_timeout(mut self, t: Duration) -> Self

How long a human has to complete an interactive flow.

method with_refresh_token

source

pub fn with_refresh_token(self, token: impl Into<String>) -> Self

Seed a refresh token, so the first call can refresh instead of prompting.

method with_wait_timeout

source

pub fn with_wait_timeout(mut self, t: Duration) -> Self

How long a non-leader thread waits for the leader’s exchange.

struct OAuthChallenge

source

pub struct OAuthChallenge {
  pub resource_metadata: String,
  pub client_id: Option<String>,
  pub client_secret: Option<String>,
  pub device_code_client_id: Option<String>,
  pub device_code_client_secret: Option<String>,
  pub use_id_token_as_bearer: bool,
}

Description

A parsed Bearer challenge.

Methods

method parse

source

pub fn parse(header: &str) -> Option<Self>

Parse a WWW-Authenticate header value.

Returns None unless this is a Bearer challenge carrying resource_metadata — anything else is not a challenge this client can act on.

struct OrderBy

source

pub struct OrderBy {
  pub column: String,
  pub direction: SortDirection,
  pub null_order: NullOrder,
  pub limit: Option<i64>,
}

Description

An ORDER BY (+ optional LIMIT) pushed into the scan.

struct PoolConfig

source

pub struct PoolConfig {
  pub idle_timeout: Duration,
  pub max_idle: usize,
}

Description

Tunables for a [WorkerPool].

Methods

method disabled

source

pub fn disabled() -> Self

A configuration that never pools — every acquire opens a fresh worker.

The equivalent of the extension’s pool false ATTACH option, and the way to prove a test does not depend on connection reuse.

struct PoolStats

source

pub struct PoolStats {
  pub hits: u64,
  pub misses: u64,
  pub evicted_idle: u64,
  pub evicted_full: u64,
  pub discarded_poisoned: u64,
  pub idle: u64,
}

Description

Cumulative pool counters.

struct PooledClient

source

pub struct PooledClient {
  pool: WorkerPool,
  location: VgiLocation,
  client: Option<VgiClient>,
  poisoned: bool,
}

Description

A checked-out connection that returns itself to the pool on drop.

Methods

method poison

source

pub fn poison(&mut self)

Do not return this connection to the pool; close it on drop.

Call after any protocol-level failure. A VGI connection carries session state — bind results, transaction tokens, an open stream — so one that failed mid-protocol is not safe to hand to the next caller even though the socket may still be writable.

method with

source

pub fn with<T>(&mut self, f: impl FnOnce(&mut VgiClient) -> Result<T>) -> Result<T>

Run f, poisoning the connection if it fails.

The ergonomic form of poison: every fallible use of a pooled connection should go through this, so “did anyone forget to poison” is not a thing to audit.

struct ProviderMetadata

source

pub struct ProviderMetadata {
  pub token_endpoint: Option<String>,
  pub authorization_endpoint: Option<String>,
  pub device_authorization_endpoint: Option<String>,
}

Description

The subset of OIDC provider metadata this client needs.

struct ResourceMetadata

source

pub struct ResourceMetadata {
  pub authorization_servers: Vec<String>,
  pub scopes_supported: Vec<String>,
  pub resource_name: Option<String>,
  pub client_id: Option<String>,
  pub client_secret: Option<String>,
  pub device_code_client_id: Option<String>,
  pub device_code_client_secret: Option<String>,
  pub token_endpoint: Option<String>,
  pub use_id_token_as_bearer: bool,
}

Description

RFC 9728 protected-resource metadata, plus the VGI extensions.

struct ResultCache

source

pub struct ResultCache {
  limits: CacheLimits,
  inner: Mutex<Inner>,
}

Description

The cache.

Methods

method eligibility

source

pub fn eligibility(
  &self,
  control: Option<&CacheControl>,
  identity_scope: Option<&str>,
  bytes: usize,
) -> Result<Duration, Ineligible>

Decide whether a result may be stored, given what the worker advertised.

identity_scope is None when the caller’s identity is configured but unresolved, which is refused rather than cached under a guess.

method flush_all

source

pub fn flush_all(&self) -> usize

Drop everything.

method flush_scope

source

pub fn flush_scope(&self, identity_scope: &str) -> usize

Drop everything for one catalog identity.

Scoped by prefix so one tenant’s flush cannot touch another’s entries even when they name the same catalog.

method get

source

pub fn get(&self, key: &CacheKey) -> Option<CachedEntry>

Look up a fresh entry, counting the hit or miss.

A stale entry is dropped rather than returned, unless it is revalidatable — those survive so a conditional request can slide them.

method get_for_revalidation

source

pub fn get_for_revalidation(&self, key: &CacheKey) -> Option<CachedEntry>

Look up an entry even if stale, for conditional revalidation.

Separate from [Self::get] because get drops a stale entry, which would throw away exactly the bytes a 304 lets us reuse.

method insert

source

pub fn insert(
  &self,
  key: CacheKey,
  batches: Vec<RecordBatch>,
  ttl: Duration,
  control: Option<&CacheControl>,
)

Store a result.

method new

source

pub fn new(limits: CacheLimits) -> Self

A cache with the given caps.

method reap

source

pub fn reap(&self) -> usize

Drop every stale entry, returning how many went.

A revalidatable entry is kept: it is meant to read as stale, and reaping it would throw away the bytes a 304 exists to reuse.

method slide

source

pub fn slide(&self, key: &CacheKey, ttl: Duration) -> bool

Slide a revalidated entry’s lifetime forward.

What a 304 buys: the stored bytes stay, and only the clock moves.

method stats

source

pub fn stats(&self) -> CacheStats

A snapshot of the counters.

struct Sample

source

pub struct Sample {
  pub percentage: f64,
  pub seed: Option<i64>,
}

Description

A TABLESAMPLE pushed into the scan.

struct Scan

source

pub struct Scan<'a> {
  stream: Box<dyn ProducerStream + 'a>,
  header: GlobalInitResponse,
  schema: SchemaRef,
  last_cache_control: Option<CacheControl>,
  finished: bool,
}

Description

An open producer stream.

Methods

method cache_control

source

pub fn cache_control(&self) -> Option<&CacheControl>

Cache directives the worker advertised on the most recent batch.

Workers state these on the first data batch, so this is populated after the first [Scan::next_batch] and stays until superseded.

method cancel

source

pub fn cancel(&mut self) -> Result<()>

Ask the worker to stop early.

method collect

source

pub fn collect(&mut self) -> Result<Vec<RecordBatch>>

Collect every remaining batch.

method execution_id

source

pub fn execution_id(&self) -> &Bytes

The worker-minted id for this scan. Pass it to parallel connections.

method max_workers

source

pub fn max_workers(&self) -> i64

How many connections the worker will accept for this scan.

Advisory: it is an upper bound, not a requirement.

method next_batch

source

pub fn next_batch(&mut self) -> Result<Option<RecordBatch>>

Pull the next batch, or None at end of stream.

method schema

source

pub fn schema(&self) -> &SchemaRef

The output schema, as resolved at bind.

struct ScanOptions

source

pub struct ScanOptions {
  pub projection: Option<Vec<i64>>,
  pub pushdown_filters: Option<Vec<u8>>,
  pub join_keys: Option<Vec<Vec<u8>>>,
  pub order_by: Option<OrderBy>,
  pub sample: Option<Sample>,
  pub execution_id: Option<Bytes>,
  pub substream_id: Option<Bytes>,
}

Description

Everything decided between bind and the first row.

enum SortDirection

source

pub enum SortDirection {
  Ascending,
  Descending,
}

Description

Sort direction for an ORDER BY pushdown.

struct StderrInteraction

source

pub struct StderrInteraction;

Description

Writes prompts to stderr.

A reasonable default for a CLI. A server-side embedder should supply its own so the prompt reaches an operator rather than a log nobody tails.

struct TokenSet

source

pub struct TokenSet {
  pub access_token: String,
  pub id_token: Option<String>,
  pub refresh_token: Option<String>,
  pub expires_at: Option<Instant>,
  pub use_id_token: bool,
  pub identity: Option<(String, String)>,
}

Description

A token set, with the client-side expiry it was received with.

Methods

method bearer

source

pub fn bearer(&self) -> &str

The value to put in the Authorization header.

method is_valid_at

source

pub fn is_valid_at(&self, now: Instant, skew: Duration) -> bool

Whether the token is still usable, allowing a skew margin.

The margin is why this is not a bare now < expires_at: refreshing a little early costs one extra exchange, while refreshing a little late costs a wasted round trip that re-uploads the whole request body.

struct Unauthorized

source

pub struct Unauthorized {
  pub reason: AuthReason,
  pub detail: String,
  pub proxy_hint: Option<String>,
}

Description

A parsed 401.

Methods

method message

source

pub fn message(&self) -> String

A message suitable for surfacing to a caller.

Includes proxy_hint when present — the spec requires it reach whoever sees the error, and it is usually the only actionable part.

method parse

source

pub fn parse(body: &str, headers: &HashMap<String, String>) -> Self

Parse a 401 from its body and headers.

Never fails. A body that is not the VGI envelope degrades to Unauthorized with a bounded excerpt as detail, because a 401 from a gateway or SSO portal is still a 401 and must not become a parse error.

struct UreqTransport

source

pub struct UreqTransport;

Description

A ureq-backed transport.

struct VgiClient

source

pub struct VgiClient {
  transport: Box<dyn VgiTransport>,
}

Description

A connection to a VGI worker.

One client owns one connection. The worker is single-threaded per connection, so a caller that wants parallelism opens several — which is also how a scan fans out across the worker’s advertised max_workers.

Methods

method aggregate_bind

source

pub fn aggregate_bind(
  &mut self,
  cat: &AttachedCatalog,
  spec: &BindSpec,
  input_schema: &Schema,
) -> Result<BoundAggregate>

Bind an aggregate, minting an execution to fold into.

method aggregate_combine

source

pub fn aggregate_combine(
  &mut self,
  cat: &AttachedCatalog,
  agg: &BoundAggregate,
  merge_batch: &RecordBatch,
) -> Result<()>

Merge another execution’s partial state into this one.

This is what makes parallel aggregation possible: each worker folds its own slice into its own execution, then the partials are merged.

method aggregate_destroy

source

pub fn aggregate_destroy(&mut self, agg: &BoundAggregate) -> Result<()>

Release the worker’s state for this execution.

Best-effort: a worker that never allocated anything answers happily. Note the destructor carries no attach handle — the execution id alone identifies what to free.

method aggregate_finalize

source

pub fn aggregate_finalize(
  &mut self,
  cat: &AttachedCatalog,
  agg: &BoundAggregate,
  group_ids: &[i64],
) -> Result<RecordBatch>

Ask for the results of a set of groups.

The answer has one row per requested group id, in the order asked.

method aggregate_update

source

pub fn aggregate_update(
  &mut self,
  cat: &AttachedCatalog,
  agg: &BoundAggregate,
  batch: &RecordBatch,
) -> Result<()>

Fold a batch of grouped values into the aggregate’s state.

batch must carry the [GROUP_COLUMN_NAME] column; build it with [with_group_ids].

method attach

source

pub fn attach(&mut self, name: &str, options: AttachOptions) -> Result<AttachedCatalog>

Attach a catalog by name, returning the handle every later call needs.

method begin_transaction

source

pub fn begin_transaction(&mut self, cat: &mut AttachedCatalog) -> Result<()>

Open a transaction, threading its handle onto later reads.

A worker may legitimately return no handle — supports_transactions is advisory and some catalogs treat every read as its own snapshot. In that case this is a no-op and cat.transaction() stays None.

method bind

source

pub fn bind(&mut self, cat: &AttachedCatalog, spec: &BindSpec) -> Result<BoundFunction>

Resolve a function’s output schema before any data moves.

method bind_with_input

source

pub fn bind_with_input(
  &mut self,
  cat: &AttachedCatalog,
  spec: &BindSpec,
  input_schema: &Schema,
) -> Result<BoundFunction>

Bind a function that takes input rows.

The input schema is what makes this an exchange rather than a producer: the worker sees it on the bind and again on the init, and its presence is what selects exchange mode.

method buffering_begin

source

pub fn buffering_begin(&mut self, bound: &BoundFunction) -> Result<Bytes>

Open the TABLE_BUFFERING init that mints an execution id.

Runs once before any chunk is sent; peers reuse the id it returns.

method buffering_combine

source

pub fn buffering_combine(
  &mut self,
  cat: &AttachedCatalog,
  spec: &BindSpec,
  execution_id: &Bytes,
  state_ids: Vec<Bytes>,
) -> Result<Vec<Bytes>>

Collapse the per-chunk state ids into the ids the finalize phase drains.

method buffering_finalize

source

pub fn buffering_finalize<'a>(
  &'a mut self,
  bound: &BoundFunction,
  execution_id: &Bytes,
  finalize_state_id: &Bytes,
) -> Result<Scan<'a>>

Drain one finalize state id of a buffered function.

method buffering_process

source

pub fn buffering_process(
  &mut self,
  cat: &AttachedCatalog,
  spec: &BindSpec,
  execution_id: &Bytes,
  input: &RecordBatch,
  batch_index: Option<i64>,
) -> Result<Bytes>

Send one input chunk to a buffered function, returning the worker’s opaque state id.

Unlike the scan phases, the buffering RPCs re-resolve the function by (schema, name) rather than echoing the bind back — so they take the catalog handle and the function’s coordinates directly.

The state id bytes are chosen by the worker and round-tripped without inspection; the common pattern is for every chunk of one execution to answer with the same id so they land in one bucket.

method catalog_version

source

pub fn catalog_version(&mut self, cat: &AttachedCatalog) -> Result<i64>

The catalog’s current version counter.

method catalogs

source

pub fn catalogs(&mut self) -> Result<Vec<CatalogInfo>>

List the catalogs this worker serves.

method commit

source

pub fn commit(&mut self, cat: &mut AttachedCatalog) -> Result<()>

Commit the open transaction, if any.

method connect_http

source

pub fn connect_http(base_url: &str) -> Result<Self>

Connect to a worker serving VGI over HTTP.

method connect_http_with_auth

source

pub fn connect_http_with_auth(
  base_url: &str,
  auth: std::sync::Arc<dyn crate::auth::CatalogAuth>,
) -> Self

Connect over HTTP, presenting a credential and recovering from a 401.

The credential is whatever CatalogAuth holds: a static bearer token, or an OAuth identity that can refresh and run an interactive flow.

method connect_location

source

pub fn connect_location(location: &VgiLocation) -> Result<Self>

Connect to wherever a LOCATION string points.

The one entry point that turns the extension’s LOCATION spelling into a connection, so a caller (or a test corpus) names a worker the same way whichever client is reading the string. See [VgiLocation] for the scheme table.

method connect_subprocess

source

pub fn connect_subprocess<S: AsRef<OsStr>>(cmd: &[S]) -> Result<Self>

Spawn a worker as a child process and talk over its stdin/stdout.

method connect_tcp

source

pub fn connect_tcp(host: &str, port: u16) -> Result<Self>

Connect to a worker listening on TCP.

method connect_to

source

pub fn connect_to(location: &str) -> Result<Self>

Parse a LOCATION string and connect to it.

method detach

source

pub fn detach(&mut self, cat: &AttachedCatalog) -> Result<()>

Release an attach. The handle is dead afterwards.

method finalize_table_in_out

source

pub fn finalize_table_in_out<'a>(
  &'a mut self,
  bound: &BoundFunction,
  execution_id: &Bytes,
) -> Result<Scan<'a>>

Run the FINALIZE phase of a table-in-out function.

This is a producer stream, not a continuation of the exchange: the init carries phase = FINALIZE and no input schema, which is what puts the worker back in tick mode.

method functions

source

pub fn functions(
  &mut self,
  cat: &AttachedCatalog,
  schema: &str,
  kind: FunctionKind,
) -> Result<Vec<FunctionInfo>>

Functions of one kind in a schema.

method label

source

pub fn label(&self) -> &str

A short label for this connection, for error messages and logs.

method macros

source

pub fn macros(
  &mut self,
  cat: &AttachedCatalog,
  schema: &str,
  kind: MacroKind,
) -> Result<Vec<MacroInfo>>

Macros of one kind in a schema.

method new

source

pub fn new(transport: Box<dyn VgiTransport>) -> Self

Build a client over any transport.

method open_exchange

source

pub fn open_exchange<'a>(
  &'a mut self,
  bound: &BoundFunction,
  opts: &ScanOptions,
) -> Result<Exchange<'a>>

Open an exchange over a bound input-taking function.

method rollback

source

pub fn rollback(&mut self, cat: &mut AttachedCatalog) -> Result<()>

Roll back the open transaction, if any.

method scan

source

pub fn scan<’a>(&’a mut self, bound: &BoundFunction, opts: &ScanOptions) -> Result<Scan<’a>>

Open a scan over a bound function.

method schema_get

source

pub fn schema_get(&mut self, cat: &AttachedCatalog, name: &str) -> Result<Option<SchemaInfo>>

One schema by name, or None when the catalog has no such schema.

method schemas

source

pub fn schemas(&mut self, cat: &AttachedCatalog) -> Result<Vec<SchemaInfo>>

Every schema in the catalog.

method streaming_chunk

source

pub fn streaming_chunk(
  &mut self,
  cat: &AttachedCatalog,
  session: &StreamingAggregate,
  input: &RecordBatch,
) -> Result<RecordBatch>

Feed one input chunk, receiving the same number of output rows.

method streaming_close

source

pub fn streaming_close(
  &mut self,
  cat: &AttachedCatalog,
  session: &StreamingAggregate,
) -> Result<()>

End a streaming session and free its state.

method streaming_open

source

pub fn streaming_open(
  &mut self,
  cat: &AttachedCatalog,
  spec: &BindSpec,
  input_schema: &Schema,
  output_schema: &Schema,
  partition_key_count: i64,
  order_key_count: i64,
) -> Result<StreamingAggregate>

Open a streaming-aggregate session.

partition_key_count and order_key_count say how many leading columns of the input are the PARTITION BY and ORDER BY keys respectively; the worker uses them to detect partition boundaries as chunks arrive.

method table_get

source

pub fn table_get(
  &mut self,
  cat: &AttachedCatalog,
  schema: &str,
  name: &str,
  at: Option<&At>,
) -> Result<Option<TableInfo>>

One table by name, optionally at a past version.

method table_scan_function

source

pub fn table_scan_function(
  &mut self,
  cat: &AttachedCatalog,
  table: &TableInfo,
  at: Option<&At>,
) -> Result<ScanFunctionResult>

How to scan a catalog table: which function to bind, with what arguments.

A VGI catalog table is not storage the client reads directly — it is a function call the worker chose, so scanning one means binding that function with the worker’s own arguments. Those arguments arrive already IPC-encoded and are forwarded verbatim ([BindSpec::with_raw_arguments]) rather than decoded and rebuilt, since they may carry types this client does not model.

The worker may inline the answer on [TableInfo::scan_function] to save a round trip, or leave it empty, in which case this fires catalog_table_scan_function_get. Both are normal; inlining is an optimisation, not a different kind of table.

method tables

source

pub fn tables(&mut self, cat: &AttachedCatalog, schema: &str) -> Result<Vec<TableInfo>>

Tables in a schema.

method views

source

pub fn views(&mut self, cat: &AttachedCatalog, schema: &str) -> Result<Vec<ViewInfo>>

Views in a schema.

method window_destroy

source

pub fn window_destroy(&mut self, part: &WindowPartition) -> Result<()>

Drop a cached window partition.

method window_evaluate

source

pub fn window_evaluate(
  &mut self,
  part: &WindowPartition,
  row: i64,
  frames: &[(i64, i64)],
) -> Result<RecordBatch>

Evaluate one output row over its frames.

frames are (start, end) row offsets within the partition. A row usually has one frame; several appear for frame types that union disjoint ranges.

method window_evaluate_batch

source

pub fn window_evaluate_batch(
  &mut self,
  part: &WindowPartition,
  row_idx: i64,
  frames_per_row: &[i64],
  frames: &[(i64, i64)],
) -> Result<RecordBatch>

Evaluate frames_per_row.len() consecutive output rows in one call.

The frame arrays are flattened: frames_per_row[i] says how many of the frame_starts/frame_ends entries belong to row row_idx + i. Batching here is what keeps a window query from costing one RPC per row.

method window_init

source

pub fn window_init(
  &mut self,
  agg: &BoundAggregate,
  partition_id: i64,
  partition_batch: &RecordBatch,
) -> Result<WindowPartition>

Ship a whole window partition to the worker.

partition_batch is every row of the partition, in window order.

enum VgiLocation

source

pub enum VgiLocation {
  Subprocess(Vec<String>),
  Http(String),
  Unix(PathBuf),
  Tcp {
      host: String,
      port: u16,
  },
  Launch(Vec<String>),
}

Description

A parsed LOCATION.

Methods

method label

source

pub fn label(&self) -> String

A short human-readable label, for errors and plan display.

method parse

source

pub fn parse(location: &str) -> Result<Self>

Parse a LOCATION string.

struct WorkerPool

source

pub struct WorkerPool {
  inner: Arc<PoolInner>,
}

Description

A pool of worker connections, keyed by location.

Cheap to clone — clones share one underlying pool.

Methods

method acquire

source

pub fn acquire(&self, location: &VgiLocation) -> Result<PooledClient>

Take a connection for location, reusing an idle one when possible.

The connection returns to the pool when the guard drops, unless it was poisoned.

method flush

source

pub fn flush(&self) -> usize

Drop every idle connection, closing those workers.

Returns how many were closed. Checked-out connections are unaffected — they return to an empty pool as usual.

method new

source

pub fn new(config: PoolConfig) -> Self

Build a pool.

method reap

source

pub fn reap(&self) -> usize

Close connections that have been idle past the timeout.

Called opportunistically on acquire; expose it so a long-lived host can reclaim workers without waiting for the next query.

method stats

source

pub fn stats(&self) -> PoolStats

Current counters.

function call

source

pub fn call<P, R>(tr: &mut dyn VgiTransport, method: &str, params: P) -> Result<R>
where
  P: VgiArrow,
  R: VgiArrow,

Description

Call a method that returns a single typed DTO.

function call_items

source

pub fn call_items<P, I>(tr: &mut dyn VgiTransport, method: &str, params: P) -> Result<Vec<I>>
where
  P: VgiArrow,
  I: VgiArrow,

Description

Call a method that returns ItemsResult, decoding each item.

Catalog discovery is all this shape: the outer DTO holds a list of binary blobs, each an IPC batch of one SchemaInfo / TableInfo / FunctionInfo.

function call_items_raw

source

pub fn call_items_raw<I: VgiArrow>(
  tr: &mut dyn VgiTransport,
  method: &str,
  params: &arrow_array::RecordBatch,
) -> Result<Vec<I>>

Description

Decode an ItemsResult given a pre-built params batch.

function call_raw

source

pub fn call_raw<R: VgiArrow>(
  tr: &mut dyn VgiTransport,
  method: &str,
  params: &arrow_array::RecordBatch,
) -> Result<R>

Description

Call a method that returns a single typed DTO, given a pre-built params batch.

The batch form exists for the handful of methods whose params carry no columns at all (catalog_catalogs), which have no generated struct because VgiArrow cannot derive on a field-less type.

function call_unit

source

pub fn call_unit<P: VgiArrow>(tr: &mut dyn VgiTransport, method: &str, params: P) -> Result<()>

Description

Call a method that returns nothing.

Void methods (catalog_detach, the transaction enders, the DDL family) register an empty result schema, so the worker replies with a batch that has no columns at all — not a {result: binary} envelope wrapping an empty inner batch. See register_void in the worker’s protocol::register.

Both shapes are accepted: a zero-column response is the canonical void reply, and a wrapped one is unwrapped and discarded so a worker that chooses to send an envelope still interoperates. Anything else is an error, so a method that unexpectedly returns data does not pass silently.

function device_code_flow

source

pub fn device_code_flow(
  http: &dyn HttpTransport,
  endpoints: &DiscoveredEndpoints,
  interaction: &dyn UserInteraction,
  timeout: Duration,
) -> Result<TokenSet>

Description

Run the RFC 8628 device-code flow to completion.

function discover

source

pub fn discover(
  http: &dyn HttpTransport,
  challenge: &OAuthChallenge,
) -> Result<DiscoveredEndpoints>

Description

Walk the discovery chain from a challenge to concrete endpoints.

function enforce_https

source

pub fn enforce_https(url: &str) -> Result<()>

Description

Reject any URL that is not https, allowing loopback http for local testing.

function envelope

source

pub fn envelope<T: VgiArrow>(inner: T) -> Result<Bytes>

Description

IPC-encode a DTO for carriage in a wrapped {request: binary} envelope.

Methods whose params are a single request column expect the real request dataclass serialized as an Arrow IPC stream inside that cell.

function identity_scope

source

pub fn identity_scope(catalog: &str, identity: &Identity, salt: &[u8]) -> Option<String>

Description

Scope a cache key to a catalog and the identity reading it.

Returns None when the identity is unresolved, which callers must treat as “do not cache” rather than substituting a default.

function is_invalid_grant

source

pub fn is_invalid_grant(err: &RpcError) -> bool

Description

Whether a refresh failure means the refresh token itself is dead.

invalid_grant is the provider saying “this token is no good” — the next attempt must fall through to an interactive flow rather than retrying.

function refresh

source

pub fn refresh(
  http: &dyn HttpTransport,
  endpoints: &DiscoveredEndpoints,
  refresh_token: &str,
) -> Result<TokenSet>

Description

Exchange a refresh token for a fresh access token.