@@ -0,0 +1,113 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Worker-side unit tests for CopyToFunction and the example_lines_out format."""
+
+from __future__ import annotations
+
+import tempfile
+import types
+
+import pyarrow as pa
+
+from vgi._test_fixtures.copy_to import ExampleLinesCopyToArgs, ExampleLinesCopyToFunction
+from vgi._test_fixtures.worker import ExampleCatalog
+
+SCHEMA = pa.schema([("a", pa.int64()), ("b", pa.string())])
+
+
+class _Store:
+    """Minimal in-memory BoundStorage stub (append + ordered log scan)."""
+
+    def __init__(self) -> None:
+        self.log: list[tuple[int, bytes]] = []
+
+    def state_append(self, ns: bytes, key: bytes, val: bytes) -> None:
+        self.log.append((len(self.log), val))
+
+    def state_log_scan(self, ns: bytes, key: bytes, after_id: int = -1, limit: int | None = None) -> list:
+        rows = [(i, v) for (i, v) in self.log if i > after_id]
+        return rows if limit is None else rows[:limit]
+
+
+def _tmp_path() -> str:
+    with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as fh:
+        return fh.name
+
+
+def _read(path: str) -> str:
+    with open(path, encoding="utf-8") as fh:
+        return fh.read()
+
+
+def _params(store: _Store) -> types.SimpleNamespace:
+    bind_call = types.SimpleNamespace(input_schema=SCHEMA)
+    init_call = types.SimpleNamespace(bind_call=bind_call)
+    return types.SimpleNamespace(storage=store, init_call=init_call, execution_id=b"x", args=None)
+
+
+def test_write_then_close_round_trips_with_null_string() -> None:
+    """write() buffers shards; close() concatenates them to a delimited file."""
+    store = _Store()
+    params = _params(store)
+    opts = ExampleLinesCopyToArgs(null_string="NA")
+    out_name = _tmp_path()
+
+    ExampleLinesCopyToFunction.write(
+        batch=pa.record_batch({"a": [1, 2], "b": ["foo", None]}, schema=SCHEMA),
+        options=opts,
+        file_path=out_name,
+        params=params,
+    )
+    ExampleLinesCopyToFunction.write(
+        batch=pa.record_batch({"a": [3], "b": ["baz"]}, schema=SCHEMA),
+        options=opts,
+        file_path=out_name,
+        params=params,
+    )
+    n = ExampleLinesCopyToFunction.close(options=opts, file_path=out_name, params=params)
+    assert n == 3
+    assert _read(out_name) == "1,foo\n2,NA\n3,baz\n"
+
+
+def test_close_honors_delimiter_and_header() -> None:
+    """Non-default delimiter + header row are applied."""
+    store = _Store()
+    params = _params(store)
+    opts = ExampleLinesCopyToArgs(null_string="NA", delimiter="|", header=True)
+    out_name = _tmp_path()
+    ExampleLinesCopyToFunction.write(
+        batch=pa.record_batch({"a": [1], "b": ["x"]}, schema=SCHEMA),
+        options=opts,
+        file_path=out_name,
+        params=params,
+    )
+    n = ExampleLinesCopyToFunction.close(options=opts, file_path=out_name, params=params)
+    assert n == 1
+    assert _read(out_name) == "a|b\n1|x\n"
+
+
+def test_close_empty_input_with_header_writes_header_only() -> None:
+    """An empty COPY with header=true still emits the header row (0 data rows)."""
+    store = _Store()
+    params = _params(store)
+    out_name = _tmp_path()
+    n = ExampleLinesCopyToFunction.close(
+        options=ExampleLinesCopyToArgs(null_string="NA", header=True),
+        file_path=out_name,
+        params=params,
+    )
+    assert n == 0
+    assert _read(out_name) == "a,b\n"
+
+
+def test_catalog_advertises_copy_to_format() -> None:
+    """The example catalog advertises example_lines_out with direction='to'."""
+    formats = ExampleCatalog().copy_from_formats(attach_opaque_data=b"", transaction_opaque_data=None)
+    by = {(f.direction, f.format_name): f for f in formats}
+    assert ("to", "example_lines_out") in by
+    fmt = by[("to", "example_lines_out")]
+    assert fmt.handler == "example_lines_writer"
+    assert fmt.comment == "Toy delimited-text writer for tests"
+    assert fmt.tags.get("category") == "copy_to"
+    opt_schema = pa.ipc.read_schema(pa.py_buffer(fmt.options))
+    assert set(opt_schema.names) == {"delimiter", "null_string", "header", "on_exists"}
@@ -0,0 +1,139 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Fixture ``COPY ... TO`` format writer for VGI integration tests.
+
+``ExampleLinesCopyToFunction`` registers the SQL format ``example_lines_out`` — a
+toy delimited-text writer, the symmetric counterpart of the ``example_lines``
+reader. It exercises the COPY-TO Sink+Combine path plus the option machinery: a
+required option (``null_string``), a defaulted option (``delimiter``), a BOOLEAN
+option (``header``), and an enum/``choices`` option (``on_exists``).
+
+Shards are buffered in ``params.storage`` (``execution_id``-scoped) by ``write()``
+and concatenated to the destination by ``close()`` — the cross-process-safe
+pattern, so it works under pool rotation / HTTP.
+
+Usage::
+
+    COPY (SELECT * FROM t) TO '/path/out.txt' (FORMAT 'acme.example_lines_out', 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_to_function import CopyToFunction
+
+if TYPE_CHECKING:
+    from vgi.table_buffering_function import TableBufferingParams
+
+__all__ = ["ExampleLinesCopyToFunction", "ExampleLinesOrderedCopyToFunction"]
+
+_SHARD_NS = b"copy_to_shard"
+
+
+@dataclass(slots=True, frozen=True, kw_only=True)
+class ExampleLinesCopyToArgs:
+    """Options for the ``example_lines_out`` COPY format."""
+
+    null_string: Annotated[str, Arg("null_string", doc="Token written for SQL NULL")]
+    delimiter: Annotated[str, Arg("delimiter", default=",", doc="Field separator")] = ","
+    header: Annotated[bool, Arg("header", default=False, doc="Write a header row of column names")] = False
+    on_exists: Annotated[
+        str,
+        Arg(
+            "on_exists",
+            default="overwrite",
+            choices=["overwrite", "error"],
+            doc="Behavior when the destination file already exists",
+        ),
+    ] = "overwrite"
+
+
+class ExampleLinesCopyToFunction(CopyToFunction[ExampleLinesCopyToArgs]):
+    """Toy delimited-text ``COPY ... TO`` writer (test fixture)."""
+
+    COPY_TO_FORMAT: ClassVar[str] = "example_lines_out"
+    COPY_TO_COMMENT: ClassVar[str | None] = "Toy delimited-text writer for tests"
+
+    class Meta:
+        name = "example_lines_writer"
+        description = "Write the COPY source to a delimited text file"
+        categories = ["copy", "test"]
+        tags = {"category": "copy_to", "stability": "test"}
+
+    @classmethod
+    def write(
+        cls,
+        *,
+        batch: pa.RecordBatch,
+        options: ExampleLinesCopyToArgs,
+        file_path: str,
+        params: TableBufferingParams[ExampleLinesCopyToArgs],
+    ) -> None:
+        """Buffer one input batch as an IPC blob in execution-scoped storage."""
+        sink = pa.BufferOutputStream()
+        with pa.ipc.new_stream(sink, batch.schema) as writer:
+            writer.write_batch(batch)
+        # state_append is atomic + race-safe across parallel sink threads/workers.
+        params.storage.state_append(_SHARD_NS, b"", sink.getvalue().to_pybytes())
+
+    @classmethod
+    def close(
+        cls,
+        *,
+        options: ExampleLinesCopyToArgs,
+        file_path: str,
+        params: TableBufferingParams[ExampleLinesCopyToArgs],
+    ) -> int:
+        """Concatenate every shard and write the delimited destination file (once)."""
+        import os
+
+        if options.on_exists == "error" and os.path.exists(file_path):
+            raise FileExistsError(f"example_lines_out: destination already exists: {file_path}")
+
+        shards = params.storage.state_log_scan(_SHARD_NS, b"", after_id=-1)
+
+        def fmt(value: object) -> str:
+            return options.null_string if value is None else str(value)
+
+        rows_written = 0
+        with open(file_path, "w", encoding="utf-8") as fh:
+            wrote_header = False
+            for _log_id, blob in shards:
+                table = pa.ipc.open_stream(blob).read_all()
+                if options.header and not wrote_header:
+                    fh.write(options.delimiter.join(table.schema.names) + "\n")
+                    wrote_header = True
+                for row in table.to_pylist():
+                    fh.write(options.delimiter.join(fmt(row[name]) for name in table.schema.names) + "\n")
+                    rows_written += 1
+            # Empty COPY with header=true still emits the header row. We need the
+            # source column names; they ride the bind's input_schema.
+            if options.header and not wrote_header:
+                assert params.init_call is not None
+                in_schema = params.init_call.bind_call.input_schema
+                if in_schema is not None:
+                    fh.write(options.delimiter.join(in_schema.names) + "\n")
+        return rows_written
+
+
+class ExampleLinesOrderedCopyToFunction(ExampleLinesCopyToFunction):
+    """Ordered variant of :class:`ExampleLinesCopyToFunction`.
+
+    ``Meta.ordered = True`` makes the extension use a single-threaded sink, so the
+    worker receives every batch in source order and writes the file in order.
+    """
+
+    COPY_TO_FORMAT: ClassVar[str] = "example_lines_ordered_out"
+    COPY_TO_COMMENT: ClassVar[str | None] = "Toy delimited-text writer (ordered, single-thread sink)"
+
+    class Meta:
+        name = "example_lines_ordered_writer"
+        description = "Write the COPY source to a delimited file, preserving source order"
+        categories = ["copy", "test"]
+        tags = {"category": "copy_to", "stability": "test"}
+        sink_order_dependent = True  # ordered COPY TO → single-thread sink
@@ -60,6 +60,7 @@
     SlowCancellableInOutFunction,
 )
 from vgi._test_fixtures.copy_from import ExampleLinesCopyFromFunction
+from vgi._test_fixtures.copy_to import ExampleLinesCopyToFunction, ExampleLinesOrderedCopyToFunction
 from vgi._test_fixtures.nest_tensor import NestTensorFunction, UnnestTensorFunction, UnnestTensorRowsFunction
 from vgi._test_fixtures.scalar import (
     AddValuesFunction,
@@ -354,6 +355,9 @@ def _build_enum_stats() -> dict[str, ColumnStatisticsInput]:
                 SlowCancellableBufferingFunction,
                 # CopyFromFunction - custom COPY ... FROM format reader
                 ExampleLinesCopyFromFunction,
+                # CopyToFunction - custom COPY ... TO format writer
+                ExampleLinesCopyToFunction,
+                ExampleLinesOrderedCopyToFunction,
                 # TableFunctionGenerator - generate output without input
                 ConstantColumnsFunction,
                 SlowCancellableFunction,
 
@@ -705,18 +705,25 @@ class CopyFromFormatInfo(CatalogObject, ArrowSerializableDataclass):
             :attr:`FunctionInfo.arguments`); each field's metadata carries the
             option type / default / ``vgi_doc`` description. The reserved
             ``file_path`` positional is excluded.
-        direction: ``"from"`` — the only direction supported today. Reserved
-            ``"to"`` for a future ``COPY ... TO``; surfaced so the C++
-            ``vgi_copy_formats()`` diagnostic can split FROM vs TO.
+        direction: ``"from"`` | ``"to"`` | ``"both"`` — which COPY direction(s)
+            this format serves; surfaced so the C++ ``vgi_copy_formats()``
+            diagnostic can split FROM vs TO and so registration wires the right
+            callbacks.
         description: Intrinsic documentation from the handler's
             ``Meta.description``.
+        ordered: COPY ... TO only — when true the worker requires rows in source
+            order, so the extension uses a single-threaded sink
+            (``REGULAR_COPY_TO_FILE``) instead of the default parallel sharded
+            write. Set via ``Meta.ordered = True`` on a ``CopyToFunction``.
+            Ignored for readers.
     """
 
     format_name: str
     handler: str
     options: SerializedSchema
     direction: str = "from"
     description: str = ""
+    ordered: bool = False
 
 
 @dataclass(frozen=True)
@@ -2863,17 +2870,20 @@ def copy_from_formats(
         attach_opaque_data: AttachOpaqueData,
         transaction_opaque_data: TransactionOpaqueData | None,
     ) -> list[CopyFromFormatInfo]:
-        """Advertise every ``CopyFromFunction`` registered in this catalog.
+        """Advertise every custom COPY format registered in this catalog.
 
         Introspects the catalog's function list for
-        :class:`vgi.copy_from_function.CopyFromFunction` subclasses and converts
-        each into a :class:`CopyFromFormatInfo`. The option schema reuses the same
-        argument serialization as :meth:`_function_to_info`, so option types /
-        defaults / ``doc`` descriptions surface identically to
-        ``vgi_function_arguments()``.
+        :class:`vgi.copy_from_function.CopyFromFunction` (``direction='from'``) and
+        :class:`vgi.copy_to_function.CopyToFunction` (``direction='to'``) subclasses
+        and converts each into a :class:`CopyFromFormatInfo`. (The RPC name is
+        historical; it returns *all* directions, distinguished by ``direction``.)
+        The option schema reuses the same argument serialization as
+        :meth:`_function_to_info`, so option types / defaults / ``doc`` descriptions
+        surface identically to ``vgi_function_arguments()``.
         """
         from vgi.argument_spec import argument_specs_to_schema, extract_argument_specs
         from vgi.copy_from_function import CopyFromFunction
+        from vgi.copy_to_function import CopyToFunction
         from vgi.metadata import resolve_metadata
 
         self._build_registries()
@@ -2883,23 +2893,45 @@ def copy_from_formats(
         seen: set[str] = set()
         for (_, _), func_classes in self._function_registry.items():
             for func_cls in func_classes:
-                if not (isinstance(func_cls, type) and issubclass(func_cls, CopyFromFunction)):
+                if not isinstance(func_cls, type):
                     continue
-                fmt = getattr(func_cls, "COPY_FROM_FORMAT", None)
-                if not fmt or fmt in seen:
+                # A format is identified by COPY_FROM_FORMAT (readers) or
+                # COPY_TO_FORMAT (writers); direction comes from the matching
+                # *_DIRECTION marker.
+                is_to = False
+                if issubclass(func_cls, CopyFromFunction):
+                    fmt = getattr(func_cls, "COPY_FROM_FORMAT", None)
+                    direction = getattr(func_cls, "COPY_FROM_DIRECTION", "from")
+                    comment = getattr(func_cls, "COPY_FROM_COMMENT", None)
+                elif issubclass(func_cls, CopyToFunction):
+                    fmt = getattr(func_cls, "COPY_TO_FORMAT", None)
+                    direction = getattr(func_cls, "COPY_TO_DIRECTION", "to")
+                    comment = getattr(func_cls, "COPY_TO_COMMENT", None)
+                    is_to = True
+                else:
                     continue
-                seen.add(fmt)
+                # Key by (format, direction): a from-format and a to-format may
+                # legitimately share a name; the extension scopes by alias anyway.
+                dedup_key = f"{direction}:{fmt}"
+                if not fmt or dedup_key in seen:
+                    continue
+                seen.add(dedup_key)
                 meta = resolve_metadata(func_cls)
+                # COPY TO writers that require source order declare it via
+                # Meta.sink_order_dependent = True (single-thread sink ingest);
+                # the extension maps that to REGULAR_COPY_TO_FILE. Ignored for readers.
+                ordered = bool(is_to and meta.sink_order_dependent)
                 args_schema = argument_specs_to_schema(extract_argument_specs(func_cls))
                 formats.append(
                     CopyFromFormatInfo(
-                        comment=getattr(func_cls, "COPY_FROM_COMMENT", None),
+                        comment=comment,
                         tags=meta.tags,
                         format_name=fmt,
                         handler=meta.name,
                         options=SerializedSchema(args_schema.serialize().to_pybytes()),
-                        direction=getattr(func_cls, "COPY_FROM_DIRECTION", "from"),
+                        direction=direction,
                         description=meta.description or "",
+                        ordered=ordered,
                     )
                 )
         return formats