@@ -38,6 +38,7 @@ dependencies = ["vgi-python[test-fixtures-writable]", "vgi-python[duckdb]"]
 
 [project.scripts]
 vgi-fixture-worker = "vgi._test_fixtures.worker:main"
+vgi-fixture-companion-worker = "vgi._test_fixtures.companion:main"
 vgi-fixture-http = "vgi._test_fixtures.http_server:main"
 vgi-fixture-versioned-worker = "vgi._test_fixtures.versioned:main"
 vgi-fixture-versioned-tables-worker = "vgi._test_fixtures.versioned_tables:main"
 
@@ -1110,7 +1110,7 @@ class TestArrowSchemaCorrectness:
     def test_catalog_attach_result_schema(self) -> None:
         """Verify CatalogAttachResult Arrow schema."""
         schema = CatalogAttachResult.ARROW_SCHEMA
-        assert len(schema) == 14
+        assert len(schema) == 15
         assert schema.field("attach_opaque_data").type == pa.binary()
         assert schema.field("supports_transactions").type == pa.bool_()
         assert schema.field("supports_time_travel").type == pa.bool_()
@@ -1120,6 +1120,7 @@ def test_catalog_attach_result_schema(self) -> None:
         assert schema.field("default_schema").type == pa.string()
         assert schema.field("settings").type == pa.list_(pa.binary())
         assert schema.field("secret_types").type == pa.list_(pa.binary())
+        assert schema.field("attach_catalogs").type == pa.list_(pa.binary())
         assert schema.field("resolved_data_version").type == pa.string()
         assert schema.field("resolved_implementation_version").type == pa.string()
 
 
@@ -0,0 +1,165 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Test fixture: a VGI catalog that advertises a companion catalog + a
+multi-branch table whose branches are catalog-table branches on that companion.
+
+Exercises the lakehouse-federation feature end-to-end:
+
+* ``catalog_attach`` advertises ``attach_catalogs=[AttachCatalogInfo(...)]`` so
+  the C++ extension attaches a companion catalog (DuckLake / attached DuckDB /
+  …) on the client at VGI-attach time.
+* the ``data.hot_cold`` table's two branches are **catalog-table branches**
+  (``function_name=""`` + ``source_*``) that scan ``<alias>.main.events`` in the
+  companion, split hot/cold by ``branch_filter``.
+
+Config via env (so the test controls the companion location):
+
+* ``VGI_TEST_COMPANION_TARGET``  — ATTACH target, e.g.
+  ``ducklake:sqlite:/scratch/lake.sqlite`` or ``/scratch/companion.duckdb``.
+  When unset, no companion is advertised (the table still exists but binding a
+  branch against the missing companion errors — used by the opt-out test).
+* ``VGI_TEST_COMPANION_ALIAS``   — companion alias (default ``vgi_companion``).
+* ``VGI_TEST_COMPANION_DBTYPE``  — explicit db_type (default: infer from target
+  scheme; set ``duckdb`` for a plain ``.duckdb`` file so the scheme allowlist
+  admits it).
+* ``VGI_TEST_COMPANION_HIDDEN``  — ``1`` to attach the companion hidden.
+
+Run with::
+
+    VGI_TEST_COMPANION_TARGET=ducklake:sqlite:/tmp/lake.sqlite \
+        vgi-fixture-companion-worker
+"""
+
+from __future__ import annotations
+
+import os
+
+import pyarrow as pa
+
+from vgi.catalog import (
+    AttachCatalogInfo,
+    Catalog,
+    ReadOnlyCatalogInterface,
+    ScanBranch,
+    ScanBranchesResult,
+    Schema,
+    Table,
+)
+from vgi.worker import Worker
+
+_ALIAS = os.environ.get("VGI_TEST_COMPANION_ALIAS", "vgi_companion")
+_TARGET = os.environ.get("VGI_TEST_COMPANION_TARGET", "")
+_DBTYPE = os.environ.get("VGI_TEST_COMPANION_DBTYPE", "")
+_HIDDEN = os.environ.get("VGI_TEST_COMPANION_HIDDEN", "") == "1"
+
+# Optional SECOND companion that fails (disallowed scheme) AFTER the first has
+# already attached — exercises the extension's partial-failure cleanup (the
+# first companion must not leak). Enabled with VGI_TEST_COMPANION_POISON=1.
+_POISON = os.environ.get("VGI_TEST_COMPANION_POISON", "") == "1"
+
+
+def _cols(**fields: pa.DataType) -> pa.Schema:
+    return pa.schema([pa.field(n, t, nullable=True) for n, t in fields.items()])
+
+
+_COMPANION_CATALOG = Catalog(
+    name="companion",
+    default_schema="data",
+    comment="VGI companion-catalog federation fixture",
+    schemas=[
+        Schema(
+            name="data",
+            comment="Multi-branch tables backed by a companion catalog",
+            tables=[
+                Table(
+                    name="hot_cold",
+                    columns=_cols(id=pa.int64(), val=pa.string()),
+                    comment="Two catalog-table branches on the companion's events table (hot/cold via branch_filter)",
+                ),
+            ],
+        ),
+    ],
+)
+
+
+class CompanionCatalog(ReadOnlyCatalogInterface):
+    """Advertises a companion catalog + a catalog-table-branch multi-branch table."""
+
+    catalog = _COMPANION_CATALOG
+
+    # Advertised only when a target is configured; empty otherwise (opt-out /
+    # missing-companion tests).
+    attach_catalogs = (
+        (
+            [
+                AttachCatalogInfo(
+                    alias=_ALIAS,
+                    target=_TARGET,
+                    db_type=_DBTYPE,
+                    required=True,
+                    hidden=_HIDDEN,
+                )
+            ]
+            + (
+                # Second entry with a disallowed scheme ⇒ the extension throws
+                # AFTER the first attached ⇒ partial-failure cleanup must release
+                # the first.
+                [AttachCatalogInfo(alias="vgi_companion_poison", target="ftp://nope/x", required=True)]
+                if _POISON
+                else []
+            )
+        )
+        if _TARGET
+        else []
+    )
+
+    def table_scan_branches_get(
+        self,
+        *,
+        attach_opaque_data,  # noqa: ANN001
+        transaction_opaque_data,  # noqa: ANN001
+        schema_name: str,
+        name: str,
+        at_unit: str | None = None,
+        at_value: str | None = None,
+    ) -> ScanBranchesResult:
+        del attach_opaque_data, transaction_opaque_data, at_unit, at_value
+        if schema_name.lower() == "data" and name.lower() == "hot_cold":
+            # Both arms scan the SAME companion table with disjoint branch_filters.
+            # A query predicate disjoint with an arm's filter prunes that arm's
+            # companion scan from the plan entirely.
+            def _arm(branch_filter: str) -> ScanBranch:
+                return ScanBranch(
+                    function_name="",
+                    positional_arguments=[],
+                    named_arguments={},
+                    branch_filter=branch_filter,
+                    source_catalog=_ALIAS,
+                    source_schema="main",
+                    source_table="events",
+                )
+
+            # DuckLake builds its scan from parquet_scan's bind, so the branch
+            # needs parquet loaded before it binds. Declaring it here makes the
+            # rewriter auto-load it (works even where extension autoload is off).
+            return ScanBranchesResult(
+                branches=[_arm("id < 100"), _arm("id >= 100")],
+                required_extensions=["parquet"],
+            )
+        msg = f"Unknown multi-branch table: {schema_name}.{name}"
+        raise ValueError(msg)
+
+
+class CompanionCatalogWorker(Worker):
+    """Worker serving :class:`CompanionCatalog`."""
+
+    catalog_interface = CompanionCatalog
+
+
+def main() -> None:
+    """Console-script entry point (vgi-fixture-companion-worker)."""
+    CompanionCatalogWorker.main()
+
+
+if __name__ == "__main__":
+    main()
@@ -11,6 +11,7 @@
     CatalogAttachResult,
     CatalogDataVersionRelease,
     CatalogExample,
+    AttachCatalogInfo,
     CatalogInfo,
     CatalogInterface,
     CatalogObject,
@@ -68,6 +69,7 @@
     "TableInfo",
     "ViewInfo",
     "FunctionInfo",
+    "AttachCatalogInfo",
     "CopyFromFormatInfo",
     "ScanBranch",
     "ScanBranchesResult",
 
@@ -176,6 +176,55 @@ class CatalogInfo(ArrowSerializableDataclass):
     source_url: str | None = None
 
 
+@dataclass(frozen=True)
+class AttachCatalogInfo(ArrowSerializableDataclass):
+    """A companion catalog the client should ATTACH when this VGI catalog attaches.
+
+    Advertised via :attr:`CatalogAttachResult.attach_catalogs`. The VGI DuckDB
+    extension attaches each entry at VGI-attach time (into the client's
+    DatabaseManager) so that multi-branch catalog-table branches — and direct
+    user queries — can resolve tables in a companion lakehouse (DuckLake /
+    iceberg / postgres / …) without the client hand-attaching anything.
+
+    Trust: companion attach is remote-influenced, so the extension applies a
+    scheme allowlist and a never-clobber conflict policy (it will never replace
+    a catalog it did not itself create). See ``docs/companion_catalogs.md``.
+
+    Attributes:
+        alias: Catalog name to attach as. Also the ``source_catalog`` a
+            catalog-table :class:`ScanBranch` references. **Namespace this by
+            your catalog identity** (e.g. ``"acme_lake"``) so two workers don't
+            both claim ``"lake"`` on the same client — collisions are rejected,
+            never silently merged.
+        target: The ATTACH target — a path or DSN, e.g.
+            ``"ducklake:sqlite:/data/meta.sqlite"`` or
+            ``"postgres:dbname=… host=…"``.
+        db_type: DuckDB storage/db type (e.g. ``"ducklake"``, ``"postgres"``).
+            Empty ⇒ the extension infers it from the ``target`` scheme prefix.
+        options: Extra ATTACH options forwarded verbatim (e.g. DuckLake
+            ``DATA_PATH``). Keys are matched case-insensitively by DuckDB.
+        hidden: When true, attach hidden (excluded from ``duckdb_databases()``);
+            still resolvable by qualified name and by branches. Use for
+            branch-only companions the user shouldn't see.
+        required: When true, a failure to attach (unreachable / conflict) fails
+            the whole VGI ATTACH loudly. When false, the failure is logged and
+            skipped; branches referencing it then error at bind.
+        secret_ref: Optional name of a credential to pre-resolve (via the
+            catalog's Orchard secret provider) and inject into the companion's
+            ATTACH options — for metadata connections (e.g. a postgres DSN)
+            where the catch-all path-keyed secret lookup isn't enough. Empty ⇒
+            rely on the automatic catch-all provider for data-file creds.
+    """
+
+    alias: str
+    target: str
+    db_type: str = ""
+    options: dict[str, str] = field(default_factory=dict)
+    hidden: bool = False
+    required: bool = False
+    secret_ref: str = ""
+
+
 @dataclass(frozen=True)
 class CatalogAttachResult(ArrowSerializableDataclass):
     """Result from attaching to a catalog.
@@ -199,6 +248,11 @@ class CatalogAttachResult(ArrowSerializableDataclass):
             Each ExtensionOption is serialized as bytes for Arrow compatibility.
         secret_types: Secret types registered with DuckDB's SecretManager. Each
             SecretTypeSpec is serialized as bytes for Arrow compatibility.
+        attach_catalogs: Companion catalogs the client should ATTACH alongside
+            this VGI catalog (lakehouse federation). Each :class:`AttachCatalogInfo`
+            is serialized as bytes for Arrow compatibility. Attached at
+            VGI-attach time; detached (refcounted) on DETACH. See
+            ``AttachCatalogInfo`` and ``docs/companion_catalogs.md``.
         comment: Optional comment describing this catalog/database.
         tags: Optional key-value tags associated with this catalog/database.
         supports_column_statistics: Whether any tables in this catalog can
@@ -221,6 +275,7 @@ class CatalogAttachResult(ArrowSerializableDataclass):
     default_schema: str = "main"
     settings: list[bytes] = field(default_factory=list)
     secret_types: list[bytes] = field(default_factory=list)
+    attach_catalogs: list[bytes] = field(default_factory=list)
     comment: str | None = None
     tags: dict[str, str] = field(default_factory=dict)
     supports_column_statistics: bool = False
@@ -858,11 +913,33 @@ def deserialize(cls, batch: pa.RecordBatch) -> Self:
 class ScanBranch:
     """One physical source backing a multi-branch scan.
 
+    A branch is one of two **kinds**:
+
+    * **Function branch** (the default) — ``function_name`` names a DuckDB table
+      function bound with ``positional_arguments``/``named_arguments``.
+    * **Catalog-table branch** — ``function_name`` is empty (``""``) and
+      ``source_table`` is set; the branch scans the base table
+      ``source_catalog.source_schema.source_table`` in an *attached* catalog
+      (typically an :class:`AttachCatalogInfo` companion, e.g. a DuckLake
+      table). The extension binds it via the catalog's own scan function, so a
+      companion's snapshot/pruning semantics are honored.
+
+    The discriminator is simply "``source_table`` present ⇒ catalog-table kind".
+
     Attributes:
-        function_name: The DuckDB function to call for this branch
+        function_name: The DuckDB function to call for a *function* branch
             (e.g., ``"read_parquet"``, ``"iceberg_scan"``, or a VGI
             table function). The C++ rewriter resolves this name against
-            DuckDB's function catalog and binds it at optimize time.
+            DuckDB's function catalog and binds it at optimize time. Empty
+            string for a catalog-table branch.
+        source_catalog: Catalog-table branch only — the attached catalog name
+            (matches an :attr:`AttachCatalogInfo.alias`). ``None`` for function
+            branches.
+        source_schema: Catalog-table branch only — the schema of the source
+            table. ``None`` for function branches.
+        source_table: Catalog-table branch only — the base table name; its
+            presence selects the catalog-table kind. ``None`` for function
+            branches.
         positional_arguments: Positional arguments as PyArrow scalars,
             passed through to the function's ``bind``.
         named_arguments: Named arguments as PyArrow scalars.
@@ -892,13 +969,19 @@ class ScanBranch:
     named_arguments: dict[str, pa.Scalar]  # type: ignore[type-arg]
     branch_filter: str | None = None
     writable: bool = False
+    source_catalog: str | None = None
+    source_schema: str | None = None
+    source_table: str | None = None
 
     ARROW_SCHEMA: ClassVar[pa.Schema] = pa.schema(
         [
             pa.field("function_name", pa.string(), nullable=False),
             pa.field("arguments", pa.binary(), nullable=False),
             pa.field("branch_filter", pa.string(), nullable=True),
             pa.field("writable", pa.bool_(), nullable=False),
+            pa.field("source_catalog", pa.string(), nullable=True),
+            pa.field("source_schema", pa.string(), nullable=True),
+            pa.field("source_table", pa.string(), nullable=True),
         ]  # type: ignore[arg-type]
     )
 
@@ -925,6 +1008,9 @@ def to_row_dict(self) -> dict[str, Any]:
             "arguments": serialize_record_batch_bytes(argument_batch),
             "branch_filter": self.branch_filter,
             "writable": self.writable,
+            "source_catalog": self.source_catalog,
+            "source_schema": self.source_schema,
+            "source_table": self.source_table,
         }
 
     def serialize(self) -> bytes:
@@ -966,6 +1052,9 @@ def deserialize(cls, batch: pa.RecordBatch) -> Self:
             branch_filter=cast("str | None", branch_filter_value) if branch_filter_value is not None else None,
             # writable is non-nullable on the wire — trust the schema.
             writable=bool(row["writable"]),
+            source_catalog=cast("str | None", row.get("source_catalog")),
+            source_schema=cast("str | None", row.get("source_schema")),
+            source_table=cast("str | None", row.get("source_table")),
         )
 
 
@@ -2258,6 +2347,7 @@ class ReadOnlyCatalogInterface(CatalogInterface):
     settings: list["SettingSpec"] = []
     secret_types: list["SecretTypeSpec"] = []
     attach_option_specs: list["AttachOptionSpec"] = []
+    attach_catalogs: list["AttachCatalogInfo"] = []
 
     # NEW: Optional Catalog object for declarative definition
     catalog: "Catalog | None" = None
@@ -2427,6 +2517,7 @@ def catalog_attach(
         # Serialize settings and secret types for the attach result
         serialized_settings = [s.serialize() for s in self.settings]
         serialized_secret_types = [st.serialize() for st in self.secret_types]
+        serialized_attach_catalogs = [c.serialize_to_bytes() for c in self.attach_catalogs]
 
         # Auto-derive supports_time_travel and supports_column_statistics from tables
         self._build_registries()
@@ -2444,6 +2535,7 @@ def catalog_attach(
             default_schema=self._default_schema_name,
             settings=serialized_settings,
             secret_types=serialized_secret_types,
+            attach_catalogs=serialized_attach_catalogs,
             comment=self.catalog.comment if self.catalog is not None else None,
             tags=dict(self.catalog.tags) if self.catalog is not None else {},
             supports_column_statistics=has_column_statistics,
 
@@ -23,6 +23,7 @@
 from vgi_rpc.rpc._types import MethodType, rpc_methods  # type: ignore[attr-defined]
 
 from vgi.catalog.catalog_interface import (
+    AttachCatalogInfo,
     CatalogInfo,
     CatalogObject,
     CopyFromFormatInfo,
@@ -58,6 +59,7 @@
     ScanFunctionResult,  # catalog_table_{scan,insert,update,delete}_function_get
     ScanBranchesResult,  # catalog_table_scan_branches_get (top-level wrapper)
     ScanBranch,  # one entry inside ScanBranchesResult.branches (binary blob)
+    AttachCatalogInfo,  # one entry inside CatalogAttachResult.attach_catalogs (binary blob)
 )