@@ -0,0 +1,84 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Tests for the wide-schema benchmark fixture.
+
+Guards the fixture the width benchmarks depend on: if ``wide`` stops
+producing the schema it claims, a profile taken against it measures
+something other than what it says on the tin.
+"""
+
+from __future__ import annotations
+
+import pyarrow as pa
+import pytest
+from vgi_rpc.rpc import RpcServer
+
+from vgi._test_fixtures.table.wide import DEFAULT_COLUMNS, WideFunction
+from vgi.arguments import Arguments
+from vgi.protocol import BindRequest, FunctionType, InitRequest, VgiProtocol
+from vgi.worker import Worker
+
+
+class _WideWorker(Worker):
+    """Worker exposing only the wide fixture, so it stays off ExampleWorker."""
+
+    functions = [WideFunction]
+
+
+def _scan(columns: int, count: int, batch_size: int) -> list[pa.RecordBatch]:
+    from vgi_rpc.http import http_connect, make_sync_client
+
+    key = b"\x11" * 32
+    worker = _WideWorker(quiet=True)
+    worker._signing_key = key
+    client = make_sync_client(RpcServer(VgiProtocol, worker, enable_describe=False), token_key=key)
+    with http_connect(VgiProtocol, client=client, compression_level=None) as proxy:  # type: ignore[type-abstract]
+        bind = BindRequest(
+            function_name="wide",
+            arguments=Arguments(
+                positional=(pa.scalar(count),),
+                named={"columns": pa.scalar(columns), "batch_size": pa.scalar(batch_size)},
+            ),
+            function_type=FunctionType.TABLE,
+            input_schema=pa.schema([]),
+        )
+        resp = proxy.bind(request=bind)
+        assert len(resp.output_schema) == columns
+        stream = proxy.init(
+            request=InitRequest(bind_call=bind, output_schema=resp.output_schema, bind_opaque_data=resp.opaque_data)
+        )
+        try:
+            return [ab.batch for ab in stream]
+        finally:
+            stream.close()
+
+
+@pytest.mark.parametrize("columns", [1, 12, 2000])
+def test_wide_emits_the_requested_width(columns: int) -> None:
+    """The bound schema and every emitted batch carry exactly ``columns``."""
+    batches = _scan(columns=columns, count=8, batch_size=4)
+    assert batches, "expected at least one batch"
+    assert sum(b.num_rows for b in batches) == 8
+    for batch in batches:
+        assert batch.num_columns == columns
+        assert batch.schema.names[0] == "c0"
+        assert batch.schema.names[-1] == f"c{columns - 1}"
+
+
+def test_wide_survives_multi_turn_state_round_trip() -> None:
+    """Rows stay contiguous across turns, so the cursor round-trips at width.
+
+    Each turn re-serializes state and rebuilds a zero-row sentinel batch of
+    the full width; this is the path the width benchmark exercises.
+    """
+    batches = _scan(columns=DEFAULT_COLUMNS, count=20, batch_size=4)
+    values = [v for b in batches for v in b.column(0).to_pylist()]
+    assert values == list(range(20))
+    assert all(b.num_columns == DEFAULT_COLUMNS for b in batches)
+
+
+def test_wide_is_not_registered_on_the_example_worker() -> None:
+    """Keep it off ExampleWorker: that count is asserted by the C++ suite."""
+    from vgi._test_fixtures.worker import ExampleWorker
+
+    assert WideFunction not in ExampleWorker.functions
@@ -0,0 +1,113 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Very wide output schemas, for measuring per-column framework cost.
+
+Most fixtures emit one or two columns, which is the cheapest possible case
+for anything the framework does per *column* rather than per batch — building
+the zero-row sentinel batch that carries a continuation token, scanning field
+metadata for partition columns, converting a row dict through Arrow. A
+one-column benchmark reports those as noise even when they dominate a real
+workload.
+
+This function makes the width a parameter so that cost is visible. Measured
+on the zero-row sentinel alone: 3.9us at one column, 33.9us at twelve.
+
+Deliberately **not** registered on ``ExampleWorker``. Adding a function there
+changes the registered-function count that the C++ integration suite asserts
+(see ``function_registration.test``), and this exists to be pointed at by a
+benchmark, not to be part of the example surface.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Annotated, Any
+
+import numpy as np
+import pyarrow as pa
+from vgi_rpc.rpc import OutputCollector
+
+from vgi._test_fixtures.table._common import CountdownState
+from vgi.arguments import Arg
+from vgi.invocation import BindResponse
+from vgi.table_function import (
+    BindParams,
+    ProcessParams,
+    TableFunctionGenerator,
+    init_single_worker,
+)
+
+DEFAULT_COLUMNS = 2000
+"""Wide enough that per-column work dominates per-batch work."""
+
+
+@dataclass(frozen=True)
+class WideArgs:
+    """Arguments for :class:`WideFunction`."""
+
+    count: Annotated[int, Arg(0, doc="Number of rows to generate", ge=0)]
+    columns: Annotated[
+        int,
+        Arg("columns", default=DEFAULT_COLUMNS, doc="Number of output columns", ge=1),
+    ] = DEFAULT_COLUMNS
+    batch_size: Annotated[int, Arg("batch_size", default=64, doc="Rows per batch", ge=1)] = 64
+
+
+@init_single_worker
+class WideFunction(TableFunctionGenerator[WideArgs, CountdownState]):
+    """Emit ``columns`` int64 columns per row.
+
+    SQL::
+
+        SELECT * FROM wide(1000, columns := 2000, batch_size := 64);
+
+    The schema is derived from the ``columns`` argument at bind time rather
+    than fixed, so one function covers the whole width range.
+
+    Attributes:
+        FunctionArguments: The argument dataclass bound to this function.
+
+    """
+
+    FunctionArguments = WideArgs
+
+    class Meta:
+        """Metadata for WideFunction."""
+
+        name = "wide"
+        description = "Generates a configurable number of int64 columns (benchmark fixture)"
+        categories = ["test", "benchmark"]
+
+    @classmethod
+    def on_bind(cls, params: BindParams[WideArgs]) -> BindResponse:
+        """Derive an ``n``-column int64 schema from the ``columns`` argument."""
+        return BindResponse(output_schema=pa.schema([(f"c{i}", pa.int64()) for i in range(params.args.columns)]))
+
+    @classmethod
+    def initial_state(cls, params: ProcessParams[WideArgs]) -> CountdownState:
+        """Start with the full row count outstanding."""
+        return CountdownState(remaining=params.args.count)
+
+    @classmethod
+    def process(
+        cls,
+        params: ProcessParams[WideArgs],
+        state: CountdownState,
+        out: OutputCollector,
+    ) -> None:
+        """Emit one batch per tick until the row count is exhausted."""
+        if state.remaining <= 0:
+            out.finish()
+            return
+        rows = min(params.args.batch_size, state.remaining)
+        base = np.arange(state.current_index, state.current_index + rows, dtype=np.int64)
+        # Same column values throughout: this fixture measures per-column
+        # framework overhead, not data generation.
+        column: Any = pa.array(base)
+        batch = pa.RecordBatch.from_arrays(
+            [column] * len(params.output_schema),
+            schema=params.output_schema,
+        )
+        state.remaining -= rows
+        state.current_index += rows
+        out.emit(batch)