feat(copy-from): worker-side COPY ... FROM custom format support

Add the protocol surface and worker API for catalogs to advertise custom
COPY ... FROM formats (FROM only):

- protocol.py: CopyFromContext on BindRequest (additive, no version bump);
  catalog_copy_from_formats RPC + CopyFromFormatsResponse.
- catalog_interface.py: CopyFromFormatInfo + copy_from_formats() (default empty
  on the base; ReadOnlyCatalogInterface introspects registered CopyFromFunction
  subclasses, reusing the function-argument schema for options).
- copy_from_function.py: CopyFromFunction base (producer-mode reader binding to
  the COPY target schema; reads file_path from the copy_from context).
- worker.py / meta_worker.py: catalog_copy_from_formats handler + dispatch entry.
- _test_fixtures/copy_from.py: example_lines toy delimited reader.
- codegen/_common.py: register CopyFromFormatInfo for schema generation.
- tests + protocol-inventory allowlist entry.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rustyconover

tests/conformance/test_protocol_inventory.py

@@ -180,6 +180,15 @@ class NotExposed:
     "catalog_schema_contents_views": ("schema_contents",),
     "catalog_schema_contents_functions": ("schema_contents",),
     "catalog_schema_contents_macros": ("schema_contents",),
+    "catalog_copy_from_formats": NotExposed(
+        reason=(
+            "COPY ... FROM format discovery. Consumed today by the C++ extension, "
+            "which calls this at ATTACH to register a DuckDB CopyFunction per "
+            "advertised format; covered by C++ integration/copy_from/*.test. Client "
+            "wrappers for the other VGI languages are planned to follow the "
+            "Python/C++ implementation."
+        )
+    ),
     "catalog_schema_contents_indexes": NotExposed(
         reason=(
             "DuckDB-only metadata path. Indexes are catalog-planner territory; "

tests/test_copy_from_function.py

@@ -0,0 +1,147 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Worker-side unit tests for CopyFromFunction and the example_lines format."""
+
+from __future__ import annotations
+
+import tempfile
+
+import pyarrow as pa
+import pytest
+
+from vgi._test_fixtures.copy_from import ExampleLinesCopyFromArgs, ExampleLinesCopyFromFunction
+from vgi._test_fixtures.worker import ExampleCatalog
+from vgi.arguments import Arguments
+from vgi.invocation import FunctionType
+from vgi.protocol import BindRequest, CopyFromContext
+
+
+class _CollectOut:
+    """Minimal OutputCollector stand-in for read()."""
+
+    def __init__(self) -> None:
+        self.batches: list[pa.RecordBatch] = []
+
+    def emit(self, batch: pa.RecordBatch, **_kwargs: object) -> None:
+        self.batches.append(batch)
+
+    def finish(self) -> None:  # pragma: no cover - read() never calls finish itself
+        pass
+
+
+def _write(text: str) -> str:
+    """Write ``text`` to a throwaway file and return its path."""
+    with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as fp:
+        fp.write(text)
+        return fp.name
+
+
+EXPECTED = pa.schema([("a", pa.int64()), ("b", pa.string())])
+
+
+def test_on_bind_binds_to_expected_schema() -> None:
+    """on_bind binds output to the COPY target schema."""
+    cf = CopyFromContext(format="example_lines", file_path="/x", expected_schema=EXPECTED)
+    br = BindRequest(
+        function_name="example_lines_copy_reader",
+        arguments=Arguments(named={"null_string": pa.scalar("NA")}),
+        function_type=FunctionType.TABLE,
+        copy_from=cf,
+    )
+    resp = ExampleLinesCopyFromFunction.bind(br)
+    assert resp.output_schema.equals(EXPECTED)
+
+
+def test_on_bind_without_copy_from_context_raises() -> None:
+    """on_bind rejects a non-COPY invocation."""
+    br = BindRequest(
+        function_name="example_lines_copy_reader",
+        arguments=Arguments(named={"null_string": pa.scalar("NA")}),
+        function_type=FunctionType.TABLE,
+    )
+    with pytest.raises(ValueError, match="COPY FROM format reader"):
+        ExampleLinesCopyFromFunction.bind(br)
+
+
+def test_read_parses_and_coerces_with_null_string() -> None:
+    """read() parses, null-maps, and casts to the target schema."""
+    path = _write("1,foo\n2,NA\n3,baz\n")
+    out = _CollectOut()
+    ExampleLinesCopyFromFunction.read(
+        path=path,
+        options=ExampleLinesCopyFromArgs(null_string="NA"),
+        expected_schema=EXPECTED,
+        params=None,
+        out=out,
+    )
+    table = pa.Table.from_batches(out.batches)
+    assert table.schema.equals(EXPECTED)
+    assert table.to_pydict() == {"a": [1, 2, 3], "b": ["foo", None, "baz"]}
+
+
+def test_read_custom_delimiter_and_skip_rows() -> None:
+    """read() honors delimiter and skip_rows options."""
+    path = _write("# header\n1|a\n2|b\n")
+    out = _CollectOut()
+    ExampleLinesCopyFromFunction.read(
+        path=path,
+        options=ExampleLinesCopyFromArgs(null_string="NA", delimiter="|", skip_rows=1),
+        expected_schema=EXPECTED,
+        params=None,
+        out=out,
+    )
+    assert pa.Table.from_batches(out.batches).to_pydict() == {"a": [1, 2], "b": ["a", "b"]}
+
+
+def test_read_on_error_fail_vs_skip() -> None:
+    """on_error 'fail' raises; 'skip' drops the bad row."""
+    path = _write("1,a\nBADROW\n3,c\n")
+    with pytest.raises(ValueError, match="example_lines: row has"):
+        ExampleLinesCopyFromFunction.read(
+            path=path,
+            options=ExampleLinesCopyFromArgs(null_string="NA"),  # on_error defaults to "fail"
+            expected_schema=EXPECTED,
+            params=None,
+            out=_CollectOut(),
+        )
+
+    out = _CollectOut()
+    ExampleLinesCopyFromFunction.read(
+        path=path,
+        options=ExampleLinesCopyFromArgs(null_string="NA", on_error="skip"),
+        expected_schema=EXPECTED,
+        params=None,
+        out=out,
+    )
+    assert pa.Table.from_batches(out.batches).num_rows == 2
+
+
+def test_catalog_advertises_copy_format() -> None:
+    """The example catalog advertises the example_lines format."""
+    formats = ExampleCatalog().copy_from_formats(attach_opaque_data=b"", transaction_opaque_data=None)
+    by_name = {f.format_name: f for f in formats}
+    assert "example_lines" in by_name
+    fmt = by_name["example_lines"]
+    assert fmt.handler == "example_lines_copy_reader"
+    assert fmt.direction == "from"
+    assert fmt.comment == "Toy delimited-text reader for tests"
+    assert fmt.tags.get("category") == "copy_from"
+    opt_schema = pa.ipc.read_schema(pa.py_buffer(fmt.options))
+    assert set(opt_schema.names) == {"delimiter", "null_string", "skip_rows", "on_error"}
+    assert opt_schema.field("null_string").metadata[b"vgi_doc"] == b"Token parsed as SQL NULL"
+
+
+def test_bind_request_copy_from_wire_roundtrip() -> None:
+    """copy_from survives a BindRequest wire round-trip."""
+    cf = CopyFromContext(format="example_lines", file_path="/p", expected_schema=EXPECTED)
+    br = BindRequest(
+        function_name="h",
+        arguments=Arguments(named={"null_string": pa.scalar("NA")}),
+        function_type=FunctionType.TABLE,
+        copy_from=cf,
+    )
+    restored = BindRequest.deserialize_from_bytes(br.serialize_to_bytes())
+    assert restored.copy_from is not None
+    assert restored.copy_from.format == "example_lines"
+    assert restored.copy_from.file_path == "/p"
+    assert restored.copy_from.expected_schema.equals(EXPECTED)

vgi/_test_fixtures/copy_from.py

@@ -0,0 +1,102 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Fixture ``COPY ... FROM`` format reader for VGI integration tests.
+
+``ExampleLinesCopyFromFunction`` registers the SQL format ``example_lines`` — a
+toy delimited-text reader. It exercises the full COPY-FROM path plus the option
+machinery: a defaulted option (``delimiter``), an ``INTEGER`` option with a range
+constraint (``skip_rows``), a required option (``null_string``), and an
+enum/``choices`` option (``on_error``).
+
+Usage::
+
+    CREATE TABLE t (a INTEGER, b VARCHAR);
+    COPY t FROM '/path/data.txt' (FORMAT example_lines, null_string 'NA');
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Annotated, ClassVar
+
+import pyarrow as pa
+
+from vgi.arguments import Arg
+from vgi.copy_from_function import CopyFromFunction
+
+if TYPE_CHECKING:
+    from vgi.table_function import ProcessParams
+    from vgi_rpc.rpc import OutputCollector
+
+__all__ = ["ExampleLinesCopyFromFunction"]
+
+
+@dataclass(slots=True, frozen=True, kw_only=True)
+class ExampleLinesCopyFromArgs:
+    """Options for the ``example_lines`` COPY format."""
+
+    null_string: Annotated[str, Arg("null_string", doc="Token parsed as SQL NULL")]
+    delimiter: Annotated[str, Arg("delimiter", default=",", doc="Field separator")] = ","
+    skip_rows: Annotated[
+        int, Arg("skip_rows", default=0, ge=0, doc="Leading lines to skip before data")
+    ] = 0
+    on_error: Annotated[
+        str,
+        Arg(
+            "on_error",
+            default="fail",
+            choices=["fail", "skip"],
+            doc="Behavior on a row whose column count does not match the target",
+        ),
+    ] = "fail"
+
+
+class ExampleLinesCopyFromFunction(CopyFromFunction[ExampleLinesCopyFromArgs]):
+    """Toy delimited-text ``COPY ... FROM`` reader (test fixture)."""
+
+    COPY_FROM_FORMAT: ClassVar[str] = "example_lines"
+    COPY_FROM_COMMENT: ClassVar[str | None] = "Toy delimited-text reader for tests"
+
+    class Meta:
+        name = "example_lines_copy_reader"
+        description = "Read a delimited text file into the COPY target table"
+        categories = ["copy", "test"]
+        tags = {"category": "copy_from", "stability": "test"}
+
+    @classmethod
+    def read(
+        cls,
+        *,
+        path: str,
+        options: ExampleLinesCopyFromArgs,
+        expected_schema: pa.Schema,
+        params: ProcessParams[ExampleLinesCopyFromArgs],
+        out: OutputCollector,
+    ) -> None:
+        """Parse ``path`` line-by-line and emit one batch matching ``expected_schema``."""
+        with open(path, encoding="utf-8") as fh:
+            lines = fh.read().splitlines()
+        lines = lines[options.skip_rows :]
+
+        ncols = len(expected_schema)
+        rows: list[list[str]] = []
+        for line in lines:
+            if line == "":
+                continue
+            cells = line.split(options.delimiter)
+            if len(cells) != ncols:
+                if options.on_error == "skip":
+                    continue
+                raise ValueError(
+                    f"example_lines: row has {len(cells)} fields, expected {ncols}: {line!r}"
+                )
+            rows.append(cells)
+
+        # Column-major string arrays, NULL where the cell equals null_string,
+        # then cast each column to the target type (DuckDB inserts no cast).
+        columns = list(zip(*rows)) if rows else [() for _ in range(ncols)]
+        arrays = []
+        for idx, field in enumerate(expected_schema):
+            raw = [None if v == options.null_string else v for v in columns[idx]]
+            arrays.append(pa.array(raw, type=pa.string()).cast(field.type))
+        out.emit(pa.RecordBatch.from_arrays(arrays, schema=expected_schema))

vgi/_test_fixtures/worker.py

@@ -59,6 +59,7 @@
     SlowCancellableFunction,
     SlowCancellableInOutFunction,
 )
+from vgi._test_fixtures.copy_from import ExampleLinesCopyFromFunction
 from vgi._test_fixtures.nest_tensor import NestTensorFunction, UnnestTensorFunction, UnnestTensorRowsFunction
 from vgi._test_fixtures.scalar import (
     AddValuesFunction,
@@ -349,6 +350,8 @@ def _build_enum_stats() -> dict[str, ColumnStatisticsInput]:
                 EchoBufferingFunction,
                 BufferEmitWideFunction,
                 SlowCancellableBufferingFunction,
+                # CopyFromFunction - custom COPY ... FROM format reader
+                ExampleLinesCopyFromFunction,
                 # TableFunctionGenerator - generate output without input
                 ConstantColumnsFunction,
                 SlowCancellableFunction,