test(copy-to): add header_repeat + fail_on_value writer fixture options

Extend the example_lines_out COPY-TO fixture to back the expanded shared SLT
suite:
- header_repeat (int, ge=0, le=3, default 1): when header=true, write the header
  line that many times — exercises ranged-option coercion + worker-side ge/le.
- fail_on_value (str, default ''): write() raises mid-sink when a cell equals it
  — exercises the in-flight failure/teardown path.
Ordered variant inherits both. Unit test updated for the 6-option set.

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

Author: rustyconover

tests/test_copy_to_function.py

@@ -110,4 +110,11 @@ def test_catalog_advertises_copy_to_format() -> None:
     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"}
+    assert set(opt_schema.names) == {
+        "delimiter",
+        "null_string",
+        "header",
+        "header_repeat",
+        "on_exists",
+        "fail_on_value",
+    }

vgi/_test_fixtures/copy_to.py

@@ -42,6 +42,10 @@ class ExampleLinesCopyToArgs:
     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
+    header_repeat: Annotated[
+        int,
+        Arg("header_repeat", default=1, ge=0, le=3, doc="When header=true, write the header line this many times"),
+    ] = 1
     on_exists: Annotated[
         str,
         Arg(
@@ -51,6 +55,10 @@ class ExampleLinesCopyToArgs:
             doc="Behavior when the destination file already exists",
         ),
     ] = "overwrite"
+    fail_on_value: Annotated[
+        str,
+        Arg("fail_on_value", default="", doc="If non-empty, fail mid-write when a cell equals this value"),
+    ] = ""
 
 
 class ExampleLinesCopyToFunction(CopyToFunction[ExampleLinesCopyToArgs]):
@@ -75,6 +83,13 @@ def write(
         params: TableBufferingParams[ExampleLinesCopyToArgs],
     ) -> None:
         """Buffer one input batch as an IPC blob in execution-scoped storage."""
+        # Mid-sink failure trigger: raise during a process() call when a cell
+        # matches fail_on_value. Exercises the in-flight teardown/recovery path.
+        if options.fail_on_value:
+            for col in batch.columns:
+                for value in col.to_pylist():
+                    if value is not None and str(value) == options.fail_on_value:
+                        raise ValueError(f"example_lines_out: fail_on_value hit: {options.fail_on_value!r}")
         sink = pa.BufferOutputStream()
         with pa.ipc.new_stream(sink, batch.schema) as writer:
             writer.write_batch(batch)
@@ -100,24 +115,30 @@ def close(
         def fmt(value: object) -> str:
             return options.null_string if value is None else str(value)
 
+        def write_header(fh: object, names: list[str]) -> None:
+            # header=true writes the column-name line `header_repeat` times.
+            if options.header:
+                for _ in range(options.header_repeat):
+                    fh.write(options.delimiter.join(names) + "\n")  # type: ignore[attr-defined]
+
         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")
+                if not wrote_header:
+                    write_header(fh, list(table.schema.names))
                     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
+            # Empty COPY with header=true still emits the header row(s). We need the
             # source column names; they ride the bind's input_schema.
-            if options.header and not wrote_header:
+            if 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")
+                    write_header(fh, list(in_schema.names))
         return rows_written