Cache results on the client

How a worker tells the DuckDB client that a result can be reused, so the next query answers without calling you at all. Worth doing when your worker is slower than the query around it — a remote API, a rate-limited service, an expensive model.

Prerequisites

• A working table, scalar, or buffering function (see Function patterns).
• vgi-python 0.14.0+ for table functions, 0.16.1+ for scalar functions and partition_scope, 0.18.0+ for per_value.

The model

Caching is advertised, not requested. The worker attaches vgi.cache.* metadata to the first data batch it emits, and the client decides what to do with it. Nothing is cached unless you say so.

The vocabulary is deliberately HTTP’s (RFC 9111/9110), because the problem is the same one: a freshness lifetime, a reuse scope, validators for revalidating cheaply, and grace windows for serving stale.

Here is the whole thing — a table function standing in for a slow upstream, advertising a five-minute TTL and a validator:

cache_worker.py

Copy Copied!

# /// script
# requires-python = ">=3.13"
# dependencies = ["vgi-python"]
# ///
"""A table function that advertises its result as cacheable, and revalidates it.

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.

This worker exposes ``rates()``, standing in for a slow upstream — a remote API,
a rate-limited service — whose answer is worth reusing. It shows the whole
vocabulary:

- a **freshness lifetime** (``ttl``), so repeat queries inside the window never
  reach the worker at all;
- a **validator** (``etag``) plus ``revalidatable=True``, which is what lets the
  client ask "is this still good?" instead of paying for a recompute;
- the **304-equivalent** reply — a 0-row batch carrying
  ``CacheControl(not_modified=True)`` — which tells the client its stored payload
  is still fresh.

``out`` is typed as vgi-rpc's ``OutputCollector``, which knows nothing about
caching; the object the framework actually passes is a VGI wrapper that accepts
``cache_control=``. Cast to :class:`vgi.protocol.VgiOutputCollector` to reach it
with types intact — every cache-aware function does this.

    ATTACH 'rates' (TYPE vgi, LOCATION 'uv run cache_worker.py');
    SELECT * FROM rates.rates();   -- second call inside the TTL never lands here
"""

from dataclasses import dataclass
from typing import cast

import pyarrow as pa

from vgi import Worker
from vgi.cache_control import CacheControl
from vgi.catalog import Catalog, Schema
from vgi.protocol import VgiOutputCollector
from vgi.table_function import (
    OutputCollector,
    ProcessParams,
    TableFunctionGenerator,
    bind_fixed_schema,
    init_single_worker,
)

_SCHEMA = pa.schema([("currency", pa.string()), ("rate", pa.float64())])

# Stands in for whatever makes your upstream's answer change. A real worker
# would derive this from the source — a last-modified header, a version column,
# a content hash — and it is the ONLY thing revalidation compares.
_DATA_VERSION = '"rates-v3"'

_ROWS = {"currency": ["EUR", "GBP", "JPY"], "rate": [1.09, 1.27, 0.0067]}


@dataclass(slots=True, frozen=True, kw_only=True)
class RatesArgs:
    """``rates()`` takes no arguments — the whole table is the result."""


@bind_fixed_schema
@init_single_worker
class Rates(TableFunctionGenerator[RatesArgs, None]):
    """Emit the rate table once, advertising it as cacheable for 5 minutes."""

    FIXED_SCHEMA = _SCHEMA

    class Meta:
        """Function metadata."""

        name = "rates"
        description = "Exchange rates from a slow upstream, cacheable for 5 minutes"

    @classmethod
    def process(cls, params: ProcessParams[RatesArgs, None], state: None, out: OutputCollector) -> None:
        """Answer the scan — or, if the client's copy is still current, say so."""
        vgi_out = cast(VgiOutputCollector, out)

        # Conditional request: the client holds a stale-but-revalidatable copy
        # and is asking whether it may keep it. Both validators are None on a
        # normal call, so this branch is simply skipped.
        if params.if_none_match == _DATA_VERSION:
            vgi_out.emit(
                pa.RecordBatch.from_pylist([], schema=_SCHEMA),  # zero rows: "keep what you have"
                cache_control=CacheControl(
                    ttl=300,
                    etag=_DATA_VERSION,
                    revalidatable=True,
                    not_modified=True,
                ),
            )
            out.finish()
            return

        # Normal path: stream the result and advertise how it may be reused.
        # The metadata rides on the FIRST data batch; attaching it later has no
        # effect, because by then the client has decided how to treat the stream.
        vgi_out.emit(
            pa.RecordBatch.from_pydict(_ROWS, schema=_SCHEMA),
            cache_control=CacheControl(
                ttl=300,  # reusable for 5 minutes without asking
                etag=_DATA_VERSION,  # ...and after that, cheap to revalidate
                revalidatable=True,  # gates whether the client ever asks
                stale_while_revalidate=60,  # serve stale while refreshing behind it
                stale_if_error=600,  # serve stale rather than fail
            ),
        )
        out.finish()


class CacheWorker(Worker):
    """A worker exposing the ``rates`` catalog."""

    catalog = Catalog(
        name="rates",
        schemas=[Schema(name="main", functions=[Rates])],
    )


if __name__ == "__main__":
    CacheWorker().run()

Copy Copied!

ATTACH 'rates' (TYPE vgi, LOCATION 'uv run cache_worker.py');
SELECT * FROM rates.rates();   -- repeat calls inside the TTL never reach the worker

Running that SELECT four times in one session calls the worker’s process() once.

Cast to VgiOutputCollector to reach cache_control

The out handed to process() is typed as vgi-rpc’s OutputCollector, whose emit() takes only (batch, metadata). The object the framework actually passes is a VGI wrapper that also accepts cache_control=, batch_index= and partition_values=. Reach it by casting — the base type can’t carry the wider signature without breaking process() override compatibility everywhere else:

Copy Copied!

from typing import cast
from vgi.protocol import VgiOutputCollector

cast(VgiOutputCollector, out).emit(batch, cache_control=CacheControl(ttl=300))

CacheControl renders to the wire keys for you. If you’re mirroring another implementation you can pass them directly instead — metadata is on the base emit(), so this form needs no cast:

Copy Copied!

out.emit(first_batch, metadata={"vgi.cache.ttl": "300"})

First batch only

The metadata rides on the first batch of the result. Attaching it to a later batch has no effect — by then the client has already decided how to treat the stream.

Freshness

Presence of ttl or expires is what makes a result cacheable at all. no_store overrides either.

Field	Meaning
ttl	Lifetime in whole seconds, measured from full-result receipt. Skew-immune, and wins over expires.
expires	Absolute RFC 3339 UTC deadline. Lifetime is expires - now at receipt.
no_store	Explicit “never cache”. Overrides any freshness key.
scope	“catalog” (default) reuses across transactions within the calling catalog identity; “transaction” reuses only inside the same transaction.

Prefer ttl unless your upstream genuinely publishes an absolute deadline — it doesn’t depend on the client’s clock agreeing with yours.

Revalidation

A TTL alone means the result is recomputed from scratch once it expires. If you can check freshness more cheaply than you can recompute, advertise a validator and set revalidatable:

Copy Copied!

cast(VgiOutputCollector, out).emit(
  first_batch,
  cache_control=CacheControl(
      ttl=300,
      etag='"rates-v3"',        # strong validator
      revalidatable=True,       # "ask me instead of recomputing"
  ),
)

revalidatable is what gates whether the client ever sends a conditional request at all. When it does, the validators it holds arrive on the scan params:

Copy Copied!

@classmethod
def process(cls, params, state, out) -> None:
  vgi_out = cast(VgiOutputCollector, out)
  if params.if_none_match == _DATA_VERSION:
      # Nothing changed — 304-equivalent. Emit ZERO rows and say so.
      vgi_out.emit(
          pa.RecordBatch.from_pylist([], schema=_SCHEMA),
          cache_control=CacheControl(ttl=300, etag=_DATA_VERSION, not_modified=True),
      )
      out.finish()
      return
  ...  # changed: stream the result normally

params.if_none_match and params.if_modified_since are both None on a normal call. Answering with not_modified=True on a 0-row batch tells the client its stored payload is still good, and it reuses that instead of re-streaming yours.

Use etag when you have an opaque version token; last_modified (RFC 3339 UTC) is the weaker fallback when you only have a timestamp.

Serving stale

Two grace windows let the client answer immediately instead of blocking on you:

• stale_while_revalidate — seconds it may serve the stale result while revalidating in the background.
• stale_if_error — seconds it may serve the stale result if a revalidation RPC fails.

Both are the difference between a slow upstream being a latency problem and being an availability problem.

Caching a slice instead of the whole scan

Two opt-ins cache at a finer grain than “the whole result”. Both are additive to the whole-scan cache, not replacements for it.

Per-partition

For a SINGLE_VALUE_PARTITIONS table function, partition_scope=True also caches the result split by partition value — one entry per distinct partition-value tuple — so a later =/IN-filtered scan reuses the per-partition entries rather than re-running the whole scan.

Copy Copied!

cast(VgiOutputCollector, out).emit(
  first_batch, cache_control=CacheControl(ttl=600, partition_scope=True)
)

Per-value

For an exchange-mode map — a scalar function, or a blended table-in-out called via correlated LATERAL — per_value=True memoizes each distinct input tuple’s output, so the same value is served without the worker on a later chunk or query.

Leave per_value off unless one call is genuinely expensive

A per-value serve costs a cache probe, a decode and an assembly step per distinct value. That only pays back when it is cheaper than calling you. For an arithmetic map it measures roughly 50x slower than simply answering the call. Turn it on for model inference, geocoding, or a rate-limited remote fetch — not for arithmetic.

Checking that it worked

Caching is advertised, so a mistake is silent: everything still returns the right answer, just without the reuse. The extension exposes the counters directly — ask it rather than guessing.

Copy Copied!

SELECT hits, misses, inserts, entries, total_bytes FROM vgi_result_cache_stats();

Copy Copied!

┌──────┬────────┬─────────┬─────────┬─────────────┐
│ hits │ misses │ inserts │ entries │ total_bytes │
├──────┼────────┼─────────┼─────────┼─────────────┤
│    2 │      2 │       1 │       1 │         448 │
└──────┴────────┴─────────┴─────────┴─────────────┘

Run your query several times and watch hits climb while inserts stays put. If inserts is 0 the result was never considered cacheable — check that the metadata is on the first batch and that ttl or expires is actually set, since neither is optional for a result to be cached at all.

Function	What it gives you
vgi_result_cache_stats()	Counters: hits, misses, inserts, evictions, entries, bytes — plus separate exchange-mode and per-partition tallies.
vgi_result_cache()	One row per cached entry: catalog, function, key hash, scope, versions.
vgi_result_cache_flush()	Drop everything — the quickest way to get a clean measurement.
vgi_result_cache_reap()	Evict what has expired, without waiting for the reaper.

EXPLAIN ANALYZE also annotates the scan with Cache: hit (memory) or Cache: miss, which is often the fastest way to see what one particular query did.

Wire keys

CacheControl.to_metadata() renders these; the C++ extension reads them by exact string. Booleans render as "1" and are omitted when false, unset optionals are omitted entirely, and scope is always emitted so the client never infers the default.

Key	Field
vgi.cache.ttl	ttl (integer seconds)
vgi.cache.expires	expires (RFC 3339 UTC)
vgi.cache.no_store	no_store
vgi.cache.scope	scope — catalog | transaction
vgi.cache.etag	etag
vgi.cache.last_modified	last_modified
vgi.cache.revalidatable	revalidatable
vgi.cache.stale_while_revalidate	stale_while_revalidate
vgi.cache.stale_if_error	stale_if_error
vgi.cache.not_modified	not_modified
vgi.cache.partition_scope	partition_scope
vgi.cache.per_value	per_value

Request-side validators travel the other way on the input batch’s metadata, as vgi.cache.if_none_match and vgi.cache.if_modified_since; the framework unpacks them onto params for you.

Validation CacheControl checks itself

scope must be catalog or transaction, and ttl / stale_while_revalidate / stale_if_error must be non-negative — both raise ValueError at construction, not at emit time.

Next steps

• API Reference — cache_control — every field and constant.
• Integrate with the optimizer — the other half of making a worker cheap to query.