feat: decode union-typed table varargs as TaggedUnion

Table-function varargs were always collected as raw pa.Scalar objects,
which drops the active-member discriminator of a union value. Keyed on a
declared union arrow_type, decode each vararg via _scalar_to_py() so it
arrives as a TaggedUnion (matching how non-vararg union args resolve);
the raw-scalar contract is untouched for every other type.

Add the union_varargs fixture exercising this end-to-end (echoes each
union vararg's active tag + value). Incidental: tidy the SecretsAccessor
.to_dict() return annotation and reflow copy_from.py to formatter width.

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

Author: rustyconover

vgi/_test_fixtures/copy_from.py

@@ -25,9 +25,10 @@
 from vgi.copy_from_function import CopyFromFunction
 
 if TYPE_CHECKING:
-    from vgi.table_function import ProcessParams
     from vgi_rpc.rpc import OutputCollector
 
+    from vgi.table_function import ProcessParams
+
 __all__ = ["ExampleLinesCopyFromFunction"]
 
 
@@ -37,9 +38,7 @@ class ExampleLinesCopyFromArgs:
 
     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
+    skip_rows: Annotated[int, Arg("skip_rows", default=0, ge=0, doc="Leading lines to skip before data")] = 0
     on_error: Annotated[
         str,
         Arg(
@@ -87,9 +86,7 @@ def read(
             if len(cells) != ncols:
                 if options.on_error == "skip":
                     continue
-                raise ValueError(
-                    f"example_lines: row has {len(cells)} fields, expected {ncols}: {line!r}"
-                )
+                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,

vgi/_test_fixtures/table/__init__.py

@@ -78,6 +78,7 @@
     MakePairsStrFunction,
     RepeatValueIntFunction,
     RepeatValueStrFunction,
+    UnionVarargsFunction,
 )
 from vgi._test_fixtures.table.partition_columns import (
     CountryPartitionedSalesFunction,
@@ -199,6 +200,7 @@
     "RegionYearPartitionedFunction",
     "RepeatValueIntFunction",
     "RepeatValueStrFunction",
+    "UnionVarargsFunction",
     "RFF_MULTI_COLUMNS",
     "RFF_NESTED_COLUMNS",
     "RFF_NONE_COLUMNS",

vgi/_test_fixtures/table/pairs.py

@@ -14,7 +14,7 @@
 from vgi._test_fixtures.table._common import (
     _cardinality_from_count,
 )
-from vgi.arguments import Arg
+from vgi.arguments import Arg, TaggedUnion
 from vgi.invocation import BindResponse
 from vgi.metadata import FunctionExample
 from vgi.schema_utils import schema
@@ -447,3 +447,109 @@ def process(
         data = {f"v{i}": col for i, col in enumerate(state.rows)}
         out_schema = schema({f"v{i}": pa.string() for i in range(len(state.rows))})
         out.emit(pa.RecordBatch.from_pydict(data, schema=out_schema))
+
+
+# ============================================================================
+
+# Sparse union shared by every union_varargs argument. DuckDB only ever emits
+# sparse unions (+us:) over Arrow, so this round-trips end-to-end.
+UNION_VARARGS_TYPE = pa.sparse_union([pa.field("i", pa.int64()), pa.field("s", pa.string())])
+
+UNION_VARARGS_SCHEMA = schema(idx=pa.int64(), tag=pa.string(), value=pa.string())
+
+
+@dataclass(kw_only=True)
+class UnionVarargsArgs:
+    """Arguments for union_varargs."""
+
+    configs: Annotated[
+        tuple[TaggedUnion, ...],
+        Arg(
+            0,
+            varargs=True,
+            arrow_type=UNION_VARARGS_TYPE,
+            doc="Union values whose active member tag is echoed back",
+        ),
+    ]
+
+
+@dataclass(kw_only=True)
+class UnionVarargsState(ArrowSerializableDataclass):
+    """State for union_varargs."""
+
+    idx: list[int] = field(default_factory=list)
+    tags: list[str | None] = field(default_factory=list)
+    values: list[str] = field(default_factory=list)
+    done: bool = False
+
+
+@init_single_worker
+@bind_fixed_schema
+class UnionVarargsFunction(TableFunctionGenerator[UnionVarargsArgs, UnionVarargsState]):
+    """Echo the active member tag and value of each union vararg.
+
+    USE CASE
+    --------
+    Exercises union-typed varargs: each argument arrives as a
+    [`TaggedUnion`][vgi.arguments.TaggedUnion] so the active member
+    discriminator (which a plain ``Scalar.as_py()`` would drop) is preserved.
+    Emits one row per vararg with its positional index, the active member
+    name, and the member value stringified into a single fixed column.
+
+    SCHEMA
+    ------
+    Fixed: ``{"idx": int64, "tag": string, "value": string}``.
+
+    Example:
+        SELECT * FROM union_varargs(
+            union_value(i := 1)::UNION(i INT, s VARCHAR),
+            union_value(s := 'x')::UNION(i INT, s VARCHAR))
+        Returns: (0, 'i', '1'), (1, 's', 'x')
+
+    Attributes:
+        FIXED_SCHEMA: The fixed Arrow output schema this function always produces.
+
+    """
+
+    FIXED_SCHEMA: ClassVar[pa.Schema] = UNION_VARARGS_SCHEMA
+
+    class Meta:
+        """Function metadata."""
+
+        name = "union_varargs"
+        description = "Echo the active member tag and value of each union vararg"
+        categories = ["generator", "utility"]
+        examples = [
+            FunctionExample(
+                sql=(
+                    "SELECT * FROM union_varargs("
+                    "union_value(i := 1)::UNION(i INT, s VARCHAR), "
+                    "union_value(s := 'x')::UNION(i INT, s VARCHAR))"
+                ),
+                description="Echo the tag and value of two union arguments",
+            ),
+        ]
+
+    @classmethod
+    def initial_state(cls, params: ProcessParams[UnionVarargsArgs]) -> UnionVarargsState:
+        """Decompose each union vararg into (idx, tag, value) rows."""
+        configs = params.args.configs
+        return UnionVarargsState(
+            idx=list(range(len(configs))),
+            tags=[cfg.tag for cfg in configs],
+            values=[str(cfg.value) for cfg in configs],
+        )
+
+    @classmethod
+    def process(cls, params: ProcessParams[UnionVarargsArgs], state: UnionVarargsState, out: OutputCollector) -> None:
+        """Emit one row per union vararg."""
+        if state.done:
+            out.finish()
+            return
+        state.done = True
+        out.emit(
+            pa.RecordBatch.from_pydict(
+                {"idx": state.idx, "tag": state.tags, "value": state.values},
+                schema=UNION_VARARGS_SCHEMA,
+            )
+        )

vgi/_test_fixtures/worker.py

@@ -144,6 +144,7 @@
     MakeSeriesRangeFunction,
     MakeSeriesStepFunction,
     MissingBatchIndexTagFunction,
+    MultiSecretDemoFunction,
     NamedParamsEchoFunction,
     NestedSequenceFunction,
     NonMonotoneBatchIndexFunction,
@@ -171,7 +172,6 @@
     RffStructScanFunction,
     RowIdSequenceFunction,
     SampleEchoFunction,
-    MultiSecretDemoFunction,
     ScopedSecretDemoFunction,
     SecretDemoFunction,
     SequenceFunction,
@@ -181,6 +181,7 @@
     TenThousandFunction,
     TxCachedValueFunction,
     TypedProbeFunction,
+    UnionVarargsFunction,
     ValuePruneFunction,
     VersionedConstraintsScanFunction,
     VersionedDataFunction,
@@ -378,6 +379,7 @@ def _build_enum_stats() -> dict[str, ColumnStatisticsInput]:
                 MakePairsStrFunction,
                 RepeatValueIntFunction,
                 RepeatValueStrFunction,
+                UnionVarargsFunction,
                 NamedParamsEchoFunction,
                 NestedSequenceFunction,
                 ProfilingDemoFunction,

vgi/table_function.py

@@ -38,6 +38,7 @@
     TableInput,
     _accepts_none,
     _extract_setting_secret_params,
+    _scalar_to_py,
 )
 from vgi.function_storage import BoundStorage, TransactionBoundStorage, attach_catalog_bytes
 from vgi.invocation import (
@@ -240,7 +241,7 @@ def pending_lookups(self) -> list[SecretLookupEntry]:
         """Return the list of pending secret lookups."""
         return list(self._pending_lookups)
 
-    def to_dict(self) -> "ResolvedSecrets":
+    def to_dict(self) -> ResolvedSecrets:
         """Return all resolved secrets keyed by secret name.
 
         Resolved secrets are keyed by their unique DuckDB secret name, so several
@@ -316,9 +317,7 @@ def secret_type(self, name: str) -> str | None:
 
     def of_type(self, secret_type: str) -> list[dict[str, Any]]:
         """Every resolved secret whose ``type`` field matches ``secret_type``."""
-        return [
-            f for f in self.values() if _secret_scalar_str(f.get("type")) == secret_type
-        ]
+        return [f for f in self.values() if _secret_scalar_str(f.get("type")) == secret_type]
 
     def for_scope(self, path: str) -> dict[str, Any] | None:
         """The secret whose ``scope`` is the longest prefix of ``path``.
@@ -338,9 +337,7 @@ def field_for(self, path: str, field: str) -> Any | None:
         fields = self.for_scope(path)
         return None if fields is None else fields.get(field)
 
-    def _select_for_scope(
-        self, path: str, secret_type: str | None
-    ) -> dict[str, Any] | None:
+    def _select_for_scope(self, path: str, secret_type: str | None) -> dict[str, Any] | None:
         best: dict[str, Any] | None = None
         best_len = -1
         fallback: dict[str, Any] | None = None
@@ -763,9 +760,21 @@ def _parse_arguments(args_class: type[TArgs], arguments: Arguments) -> TArgs:
             for meta in get_args(hint)[1:]:
                 if isinstance(meta, Arg):
                     if meta.varargs:
-                        # Varargs: collect remaining positional args as raw pa.Scalar objects
+                        # Varargs: collect remaining positional args as raw pa.Scalar
+                        # objects (e.g. constant_columns reads .type / pa.repeat off
+                        # them). Union-typed varargs are the exception: decode each
+                        # scalar to a TaggedUnion so the active member discriminator
+                        # is preserved — matching how non-vararg union args resolve
+                        # via Arguments.get()/_scalar_to_py(). Keyed on the declared
+                        # arrow_type so the raw-scalar contract is untouched otherwise.
                         assert isinstance(meta.position, int)
-                        kwargs[attr_name] = tuple(arguments.positional[meta.position :])
+                        varargs_scalars = arguments.positional[meta.position :]
+                        if meta.arrow_type is not None and pa.types.is_union(meta.arrow_type):
+                            kwargs[attr_name] = tuple(
+                                _scalar_to_py(s) if s is not None else None for s in varargs_scalars
+                            )
+                        else:
+                            kwargs[attr_name] = tuple(varargs_scalars)
                     else:
                         value = arguments.get(meta.position, default=meta.default)
                         # Reject SQL NULL for non-Optional Args. Without this,