@@ -0,0 +1,77 @@
+# Copyright 2026 Query Farm LLC - https://query.farm
+
+# ruff: noqa: S101, D101, D102, D103
+"""`Client.table_function(batch_metadata_callback=...)`.
+
+`_table_function_parallel` previously discarded `AnnotatedBatch.custom_metadata`
+unconditionally (`if output.batch.num_rows > 0: output_queue.put(output.batch)`)
+and only forwarded the bare `pa.RecordBatch` — so a worker's `vgi.cache.*`
+cacheability advertisement (`vgi/cache_control.py`), which rides
+`custom_metadata`, was structurally unreachable through the public
+`table_function()` generator. `batch_metadata_callback` mirrors the existing
+`bind_result_callback` pattern to fix that additively, with no change to the
+generator's yield type. Exercised against `cacheable_numbers`
+(`vgi/_test_fixtures/table/cache.py`), which advertises `vgi.cache.ttl` on its
+first batch.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pyarrow as pa
+import pytest
+
+from vgi.arguments import Arguments
+
+DATA = "data"
+
+
+@pytest.fixture
+def client(client_transport: Any) -> Any:
+    with client_transport() as c:
+        c.catalog_attach(name="example", data_version_spec=None, implementation_version=None)
+        yield c
+
+
+def test_callback_receives_cache_control_metadata(client: Any) -> None:
+    metas: list[pa.KeyValueMetadata | None] = []
+    batches = list(
+        client.table_function(
+            function_name="cacheable_numbers",
+            schema_name=DATA,
+            arguments=Arguments(named={"n": pa.scalar(5)}),
+            batch_metadata_callback=metas.append,
+        )
+    )
+    assert sum(b.num_rows for b in batches) == 5
+    assert len(metas) == len(batches)
+    assert metas[0] is not None
+    assert dict(metas[0])[b"vgi.cache.ttl"] == b"300"
+
+
+def test_callback_is_optional_and_defaults_to_none(client: Any) -> None:
+    """No `batch_metadata_callback` -> unchanged behavior, no crash."""
+    batches = list(
+        client.table_function(
+            function_name="cacheable_numbers",
+            schema_name=DATA,
+            arguments=Arguments(named={"n": pa.scalar(3)}),
+        )
+    )
+    assert sum(b.num_rows for b in batches) == 3
+
+
+def test_callback_sees_none_for_a_batch_without_custom_metadata(client: Any) -> None:
+    """A function that never advertises cache control still drives the callback once per batch, with `None`."""
+    metas: list[pa.KeyValueMetadata | None] = []
+    list(
+        client.table_function(
+            function_name="sequence",
+            schema_name="main",
+            arguments=Arguments(positional=[pa.scalar(4)]),
+            batch_metadata_callback=metas.append,
+        )
+    )
+    assert metas
+    assert all(m is None for m in metas)
@@ -0,0 +1,67 @@
+# Copyright 2026 Query Farm LLC - https://query.farm
+
+# ruff: noqa: S101, D101, D102, D103
+"""`Client.table_scan_branches_get`.
+
+`Client` had no wrapper for `catalog_table_scan_branches_get` at all —
+grepped the whole `vgi/client/` package for "branch" and the only hit was an
+unrelated code comment ("skip the subprocess branch"). A multi-branch table
+was therefore invisible to any non-DuckDB caller, not just unsupported by a
+higher-level connector. Exercised against `data.multi_branch_numbers` (two
+`sequence(50)` arms — `vgi/_test_fixtures/worker.py`) and, for the legacy
+fallback, `data.numbers` (an ordinary single-branch table whose worker only
+implements the older `table_scan_function_get`).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from vgi.catalog import ScanBranch, ScanBranchesResult
+
+DATA = "data"
+
+
+@pytest.fixture
+def client(client_transport: Any) -> Any:
+    with client_transport() as c:
+        c.catalog_attach(name="example", data_version_spec=None, implementation_version=None)
+        yield c
+
+
+def _attach_opaque(client: Any) -> Any:
+    # catalog_attach() was already called by the fixture above (its return
+    # value discarded) — re-attach here to get the opaque data this method
+    # needs, mirroring how CatalogClientMixin methods are used standalone.
+    return client.catalog_attach(name="example", data_version_spec=None, implementation_version=None).attach_opaque_data
+
+
+def test_multi_branch_table_returns_all_branches(client: Any) -> None:
+    result = client.table_scan_branches_get(
+        attach_opaque_data=_attach_opaque(client),
+        schema_name=DATA,
+        name="multi_branch_numbers",
+    )
+    assert isinstance(result, ScanBranchesResult)
+    assert len(result.branches) == 2
+    for branch in result.branches:
+        assert isinstance(branch, ScanBranch)
+        assert branch.function_name == "sequence"
+        assert branch.source_table is None  # function branch, not catalog-table
+
+
+def test_single_branch_table_falls_back_to_legacy_rpc(client: Any) -> None:
+    """`data.numbers` only implements `table_scan_function_get` — the fallback wraps it as one branch."""
+    result = client.table_scan_branches_get(
+        attach_opaque_data=_attach_opaque(client),
+        schema_name=DATA,
+        name="numbers",
+    )
+    assert len(result.branches) == 1
+    assert result.branches[0].function_name == client.table_scan_function_get(
+        attach_opaque_data=_attach_opaque(client),
+        schema_name=DATA,
+        name="numbers",
+    ).function_name
@@ -0,0 +1,85 @@
+# Copyright 2026 Query Farm LLC - https://query.farm
+
+# ruff: noqa: S101, D101, D102, D103
+"""`Client.table_get`/`table_function(at_unit=..., at_value=...)`.
+
+Both methods previously dropped `at_unit`/`at_value` on the floor even though
+the underlying wire RPCs (`catalog_table_get`, `BindRequest.at_unit`/
+`.at_value`) already carried them — `table_scan_function_get` was the only
+`Client` method that exposed time travel at all. Two distinct worker
+patterns are exercised, both real fixtures, proving this isn't just plumbing
+that happens to compile:
+
+* `data.versioned_data` reads its version through a pre-resolved bind
+  argument (`table_scan_function_get(at_unit=...)` bakes it into the
+  returned `ScanFunctionResult.arguments`) — `table_get(at_unit=...)` is
+  what needed fixing here, to see the schema at that version at all.
+* `data.tt_pushdown_fn` (`tt_pushdown_scan`) reads `at_unit`/`at_value`
+  **directly off the init request** (`ProcessParams.at_unit`/`.at_value`,
+  see `vgi/_test_fixtures/table/tt_pushdown.py`) — this is the pattern that
+  needed `table_function(at_unit=...)` specifically; no amount of
+  pre-resolved arguments would reach it.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pyarrow as pa
+import pytest
+
+from vgi.arguments import Arguments
+
+DATA = "data"
+
+
+@pytest.fixture
+def client(client_transport: Any) -> Any:
+    with client_transport() as c:
+        c.catalog_attach(name="example", data_version_spec=None, implementation_version=None)
+        yield c
+
+
+def _attach_opaque(client: Any) -> Any:
+    result = client.catalog_attach(name="example", data_version_spec=None, implementation_version=None)
+    return result.attach_opaque_data
+
+
+def test_table_get_at_unit_returns_the_versioned_schema(client: Any) -> None:
+    opaque = _attach_opaque(client)
+    v1 = client.table_get(
+        attach_opaque_data=opaque, schema_name=DATA, name="versioned_data", at_unit="VERSION", at_value="1"
+    )
+    live = client.table_get(attach_opaque_data=opaque, schema_name=DATA, name="versioned_data")
+
+    v1_schema = pa.ipc.read_schema(pa.py_buffer(v1.columns))
+    live_schema = pa.ipc.read_schema(pa.py_buffer(live.columns))
+    assert v1_schema.names == ["id"]  # version 1: (id int64) only
+    assert live_schema.names == ["id", "score"]  # current: version 3
+
+
+def test_table_get_with_no_at_clause_is_unaffected(client: Any) -> None:
+    """The new `at_unit`/`at_value` params default to `None` — unchanged behavior without them."""
+    info = client.table_get(attach_opaque_data=_attach_opaque(client), schema_name=DATA, name="numbers")
+    assert info is not None
+
+
+def test_table_function_at_unit_reaches_a_worker_that_reads_it_from_init(client: Any) -> None:
+    """`tt_pushdown_scan` reads `at_unit`/`at_value` off the init request, not a resolved bind arg."""
+    v1_batches = list(
+        client.table_function(function_name="tt_pushdown_scan", schema_name=DATA, at_unit="VERSION", at_value="1")
+    )
+    live_batches = list(client.table_function(function_name="tt_pushdown_scan", schema_name=DATA))
+
+    assert sum(b.num_rows for b in v1_batches) == 5
+    assert sum(b.num_rows for b in live_batches) == 10
+
+
+def test_table_function_with_no_at_clause_is_unaffected(client: Any) -> None:
+    """The new `at_unit`/`at_value` params default to `None` — unchanged behavior without them."""
+    batches = list(
+        client.table_function(
+            function_name="sequence", schema_name="main", arguments=Arguments(positional=[pa.scalar(5)])
+        )
+    )
+    assert sum(b.num_rows for b in batches) == 5
@@ -59,6 +59,8 @@ class CatalogEnabledClient(CatalogClientMixin, Client):
     MacroInfo,
     MacroType,
     OnConflict,
+    ScanBranch,
+    ScanBranchesResult,
     ScanFunctionResult,
     SchemaInfo,
     SchemaObjectType,
@@ -603,6 +605,8 @@ def table_get(
         transaction_opaque_data: TransactionOpaqueData | None = None,
         schema_name: str,
         name: str,
+        at_unit: str | None = None,
+        at_value: str | None = None,
     ) -> TableInfo | None:
         """Get information about a table.
 
@@ -611,6 +615,10 @@ def table_get(
             transaction_opaque_data: Optional transaction ID for transactional reads.
             schema_name: The schema containing the table.
             name: The table name.
+            at_unit: Optional time travel unit (e.g. 'timestamp', 'version') —
+                the schema at a past point may differ from the live one.
+                `None` for the live schema.
+            at_value: Optional time travel value, paired with `at_unit`.
 
         Returns:
             [`TableInfo`][] for the table, or None if not found.
@@ -621,6 +629,8 @@ def table_get(
                 attach_opaque_data=attach_opaque_data,
                 schema_name=schema_name,
                 name=name,
+                at_unit=at_unit,
+                at_value=at_value,
                 transaction_opaque_data=transaction_opaque_data,
             ).to_optional()
 
@@ -774,6 +784,82 @@ def table_scan_function_get(
             batch, _ = deserialize_record_batch(result_bytes)
             return ScanFunctionResult.deserialize(batch)
 
+    def table_scan_branches_get(
+        self,
+        *,
+        attach_opaque_data: AttachOpaqueData,
+        transaction_opaque_data: TransactionOpaqueData | None = None,
+        schema_name: str,
+        name: str,
+        at_unit: str | None = None,
+        at_value: str | None = None,
+    ) -> ScanBranchesResult:
+        """Get the list of scan branches for a (possibly multi-source) table.
+
+        Multi-branch tables compose a logical scan from N physical sources
+        (canonical case: Kafka hot tier + Iceberg cold tier). The VGI DuckDB
+        extension's optimizer rewrites the placeholder scan into
+        ``LogicalSetOperation(UNION_ALL, ...)``, one arm per branch — a
+        caller of this method is expected to do the equivalent (e.g.
+        `pl.concat` over one scan per branch in vgi-polars).
+
+        Falls back to :meth:`table_scan_function_get`, wrapped as a
+        single-branch `ScanBranchesResult`, when the worker raises
+        `MethodNotImplementedError` for the additive `catalog_table_scan_
+        branches_get` RPC — mirroring exactly how the C++ extension reads
+        that same fallback, so a worker written before this RPC existed
+        still works, just as a one-branch table.
+
+        Args:
+            attach_opaque_data: The attachment ID from catalog_attach.
+            transaction_opaque_data: Optional transaction ID for transactional reads.
+            schema_name: The schema containing the table.
+            name: The table name.
+            at_unit: Optional time travel unit (e.g., 'timestamp', 'version').
+                The C++ extension refuses `AT(...)` on tables with more than
+                one branch at bind time — a worker returning multiple
+                branches should expect this to always be `None`.
+            at_value: Optional time travel value.
+
+        Returns:
+            `ScanBranchesResult` with one or more `ScanBranch` entries plus
+            the union of required DuckDB extensions across all branches.
+
+        """
+        with self._catalog_connect() as proxy:
+            try:
+                result_bytes = proxy.catalog_table_scan_branches_get(
+                    attach_opaque_data=attach_opaque_data,
+                    schema_name=schema_name,
+                    name=name,
+                    at_unit=at_unit,
+                    at_value=at_value,
+                    transaction_opaque_data=transaction_opaque_data,
+                )
+            except RpcError as e:
+                if "MethodNotImplemented" not in type(e).__name__ and "not implemented" not in str(e).lower():
+                    raise
+                scan_fn = self.table_scan_function_get(
+                    attach_opaque_data=attach_opaque_data,
+                    transaction_opaque_data=transaction_opaque_data,
+                    schema_name=schema_name,
+                    name=name,
+                    at_unit=at_unit,
+                    at_value=at_value,
+                )
+                return ScanBranchesResult(
+                    branches=[
+                        ScanBranch(
+                            function_name=scan_fn.function_name,
+                            positional_arguments=list(scan_fn.positional_arguments),
+                            named_arguments=dict(scan_fn.named_arguments),
+                        )
+                    ],
+                    required_extensions=list(scan_fn.required_extensions),
+                )
+            batch, _ = deserialize_record_batch(result_bytes)
+            return ScanBranchesResult.deserialize(batch)
+
     def table_comment_set(
         self,
         *,