Cache results on the client

Caching is advertised, not requested. The worker attaches vgi.cache.* metadata to the batches it emits, and the client — the DuckDB extension — decides what to do with it. Nothing is cached unless you say so, which is the right default for a function that might not be pure.

The whole worker

cache.rs

// Copyright 2025, 2026 Query Farm LLC - https://query.farm

//! The result-caching example for the vgi-rust documentation.
//!
//! Caching is advertised, not requested: the worker attaches `vgi.cache.*`
//! metadata to the FIRST data batch it emits, and the client (the DuckDB
//! extension) decides what to do with it. Nothing is cached unless you say so.
//!
//! ```text
//! cargo build --release --bin cache
//! # then, in a Haybarn shell:
//! ATTACH 'rates' (TYPE vgi, LOCATION './target/release/cache');
//! SELECT * FROM rates.rates();   -- repeat calls inside the TTL never land here
//! SELECT hits, misses, inserts FROM vgi_result_cache_stats();
//! SELECT * FROM rates.upstream_calls();   -- proves the worker was not re-run
//! ```

use std::collections::HashMap;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::Arc;

use arrow_array::{ArrayRef, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use vgi::cache_control::{CacheControl, ConditionalRequest};
use vgi::catalog::CatalogModel;
use vgi::function::{ArgSpec, BindParams, BindResponse, FunctionMetadata, ProcessParams};
use vgi::table_function::{TableFunction, TableProducer};
use vgi::vgi_rpc::OutputCollector;
use vgi::{Result, RpcError};

/// Counts real invocations, so the caching can be observed rather than assumed.
/// A worker is one process, so a plain atomic is enough.
static UPSTREAM_CALLS: AtomicI64 = AtomicI64::new(0);

const TTL_SECONDS: i64 = 300;

/// A strong validator for the payload below. Anything opaque and stable works —
/// a content hash, a database version, an upstream ETag — as long as it changes
/// exactly when the payload does.
const ETAG: &str = "\"rates-v1\"";

fn rates_schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("pair", DataType::Utf8, true),
        Field::new("rate", DataType::Int64, true),
    ]))
}

struct RatesProducer {
    schema: SchemaRef,
    done: bool,
    /// Set by `on_conditional_request` before the first `next_batch`.
    if_none_match: Option<String>,
    /// Set alongside the batch; the framework reads it back through
    /// `last_metadata` and puts it on the wire.
    meta: Option<HashMap<String, String>>,
}

impl TableProducer for RatesProducer {
    /// The client already has a payload and is asking whether it is still good.
    /// It only ever asks because the result advertised `with_revalidatable`.
    fn on_conditional_request(&mut self, request: &ConditionalRequest) {
        self.if_none_match = request.if_none_match.clone();
    }

    fn next_batch(&mut self, _out: &mut OutputCollector) -> Result<Option<RecordBatch>> {
        if self.done {
            return Ok(None);
        }
        self.done = true;

        let mut cc = CacheControl::ttl(TTL_SECONDS)
            .with_etag(ETAG)
            .with_revalidatable()
            // Grace windows: serve stale immediately while refreshing in the
            // background, and keep serving stale if a refresh RPC fails.
            .with_stale_while_revalidate(60)
            .with_stale_if_error(3600);

        let still_fresh = self.if_none_match.as_deref() == Some(ETAG);
        if still_fresh {
            cc = cc.with_not_modified();
        }

        // Metadata rides the FIRST data batch. It cannot go on the schema — the
        // IPC stream fixes that when the stream opens, before this runs.
        self.meta = Some(cc.to_metadata());

        let (pairs, rates): (Vec<&str>, Vec<i64>) = if still_fresh {
            // A zero-row batch carrying not_modified is the 304 equivalent:
            // keep what you have. The client reuses its stored rows.
            (Vec::new(), Vec::new())
        } else {
            UPSTREAM_CALLS.fetch_add(1, Ordering::Relaxed);
            (vec!["EURUSD", "GBPUSD", "USDJPY"], vec![108, 127, 15_700])
        };

        let cols: Vec<ArrayRef> = vec![
            Arc::new(StringArray::from(pairs)),
            Arc::new(Int64Array::from(rates)),
        ];
        RecordBatch::try_new(self.schema.clone(), cols)
            .map(Some)
            .map_err(|e| RpcError::runtime_error(e.to_string()))
    }

    fn last_metadata(&self) -> Option<HashMap<String, String>> {
        self.meta.clone()
    }
}

struct Rates;

impl TableFunction for Rates {
    fn name(&self) -> &str {
        "rates"
    }
    fn metadata(&self) -> FunctionMetadata {
        FunctionMetadata {
            description: "Exchange rates from a slow upstream, cached on the client".to_string(),
            ..Default::default()
        }
    }
    fn argument_specs(&self) -> Vec<ArgSpec> {
        Vec::new()
    }
    fn on_bind(&self, _params: &BindParams) -> Result<BindResponse> {
        Ok(BindResponse {
            output_schema: rates_schema(),
            opaque_data: Vec::new(),
        })
    }
    fn producer(&self, params: &ProcessParams) -> Result<Box<dyn TableProducer>> {
        Ok(Box::new(RatesProducer {
            schema: params.output_schema.clone(),
            done: false,
            if_none_match: None,
            meta: None,
        }))
    }
}

/// Reports how many times the upstream was actually hit, so a query can prove
/// the cache engaged rather than take it on faith.
struct UpstreamCalls;

struct CallsProducer {
    schema: SchemaRef,
    done: bool,
}

impl TableProducer for CallsProducer {
    fn next_batch(&mut self, _out: &mut OutputCollector) -> Result<Option<RecordBatch>> {
        if self.done {
            return Ok(None);
        }
        self.done = true;
        let col: ArrayRef = Arc::new(Int64Array::from(vec![
            UPSTREAM_CALLS.load(Ordering::Relaxed)
        ]));
        RecordBatch::try_new(self.schema.clone(), vec![col])
            .map(Some)
            .map_err(|e| RpcError::runtime_error(e.to_string()))
    }
}

impl TableFunction for UpstreamCalls {
    fn name(&self) -> &str {
        "upstream_calls"
    }
    fn metadata(&self) -> FunctionMetadata {
        FunctionMetadata {
            description: "How many times rates() actually computed a result".to_string(),
            ..Default::default()
        }
    }
    fn argument_specs(&self) -> Vec<ArgSpec> {
        Vec::new()
    }
    fn on_bind(&self, _params: &BindParams) -> Result<BindResponse> {
        Ok(BindResponse {
            output_schema: Arc::new(Schema::new(vec![Field::new(
                "calls",
                DataType::Int64,
                true,
            )])),
            opaque_data: Vec::new(),
        })
    }
    fn producer(&self, params: &ProcessParams) -> Result<Box<dyn TableProducer>> {
        Ok(Box::new(CallsProducer {
            schema: params.output_schema.clone(),
            done: false,
        }))
    }
}

fn main() {
    let mut worker = vgi::Worker::new();
    worker.register_table(Rates);
    worker.register_table(UpstreamCalls);
    worker.set_catalog(CatalogModel {
        name: "rates".to_string(),
        comment: Some("Documentation example: advertising a cacheable result".to_string()),
        ..Default::default()
    });
    worker.run();
}

Metadata rides the first data batch

A producer does not emit metadata directly. It stashes it and the framework reads it back through last_metadata after each batch:

fn next_batch(&mut self, _out: &mut OutputCollector) -> Result<Option<RecordBatch>> {
  // … build the batch …
  self.meta = Some(cc.to_metadata());
  Ok(Some(batch))
}

fn last_metadata(&self) -> Option<HashMap<String, String>> {
  self.meta.clone()
}

It cannot go on the schema. The IPC stream fixes the schema when the stream opens, before next_batch has produced anything — so batch metadata is the only channel that exists at the point where you know what you are returning.

Set it on the first data batch. Later batches carrying it is not an error, but nothing reads them.

Freshness

Builder	Meaning
`CacheControl::ttl(n)`	Lifetime in seconds from full-result receipt. Skew-immune, and it wins over `expires`.
`with_expires`	An absolute RFC 3339 UTC deadline.
`CacheControl::no_store()`	Explicit “never cache”. Overrides any freshness key.
`with_transaction_scope`	Reuse only inside the transaction that produced it.
`with_stale_while_revalidate`	Grace window to serve stale immediately while refreshing in the background.
`with_stale_if_error`	Grace window to keep serving stale when a refresh fails.

A ttl or an expires is what makes a result cacheable at all.

Validators, and the 304 equivalent

with_etag plus with_revalidatable tells the client it may ask “is this still good?” instead of paying for a recompute. The question arrives through a hook on the producer, not on the params:

fn on_conditional_request(&mut self, request: &ConditionalRequest) {
  self.if_none_match = request.if_none_match.clone();
}

It is called before the first next_batch, so the producer can decide there. A zero-row batch carrying with_not_modified is the 304: keep what you have. The client reuses its stored rows without a restream.

Only advertise `with_revalidatable` if answering is genuinely cheap

The flag is what gates whether the client ever sends a conditional request at all. Setting it on a function whose freshness check costs as much as recomputing turns one round trip into two.

Prove it engaged

Guessing is not good enough here, and the worker itself can settle it. cache.rs counts real invocations in an atomic:

SELECT count(*) FROM rates.rates();
SELECT count(*) FROM rates.rates();
SELECT count(*) FROM rates.rates();
SELECT count(*) FROM rates.rates();

SELECT hits, misses, inserts FROM vgi_result_cache_stats();
SELECT * FROM rates.upstream_calls();

Output

hits	misses	inserts
3	2	1

Output

calls
1

Four queries, one upstream call. That second result is the one that matters: it is measured inside the worker, so it cannot be explained away by the extension’s own bookkeeping.

Per-value memoization

with_per_value additionally memoizes each distinct input tuple’s output — for a scalar, or a blended table-in-out called through a correlated LATERAL.

Off by default, and it should usually stay off

A per-value serve is not free: the client pays a key probe, a decode, and a per-value assembly step, and that only pays back when it costs less than the worker call it replaces. For a cheap map — arithmetic, a string tweak, a lookup in memory — the engine measures it at roughly 50× slower than just calling the worker.

Turn it on when a single call is genuinely heavy and repeats across rows: a model inference, a geocode, a rate-limited HTTP fetch. Only you know which side of that line your function is on, which is why the engine will not guess.

Next steps

Every field → Cache control.
Caching a whole table → Expose a catalog.