fix: build empty batches for union and extension schemas

pa.array([], type=...) has no builder for union or extension types, so the
ungrouped-empty aggregate client result, the filter_v2 bind check and the
dynamic aggregate fixture failed for such an output schema. Use
vgi_rpc.utils.empty_batch / pa.nulls(0, ...).

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

Author: rustyconover

vgi/_test_fixtures/aggregate/dynamic.py

@@ -232,7 +232,9 @@ def _slice_to_frame(
                     t = t.filter(null_mask)
             slices.append(t)
         if not slices:
-            return pa.table({c: pa.array([], type=cols[i].type) for i, c in enumerate(col_names)})
+            # pa.nulls(0, ...), not pa.array([], ...): the latter has no
+            # builder for union or extension types.
+            return pa.table({c: pa.nulls(0, type=cols[i].type) for i, c in enumerate(col_names)})
         return pa.concat_tables(slices)
 
     @staticmethod

vgi/client/aggregate.py

@@ -50,6 +50,7 @@
 
 import pyarrow as pa
 from vgi_rpc.rpc import RpcError
+from vgi_rpc.utils import empty_batch
 
 from vgi.aggregate_function import GROUP_COLUMN_NAME
 from vgi.arguments import Arguments
@@ -783,12 +784,9 @@ def aggregate_function(
                 batch = next(batches, None)
 
             if not group_keys:
-                results = [
-                    pa.RecordBatch.from_arrays(
-                        [pa.array([], type=field.type) for field in session.output_schema],
-                        schema=session.output_schema,
-                    )
-                ]
+                # empty_batch, not pa.array([], type=...): that has no builder
+                # for union or extension types.
+                results = [empty_batch(session.output_schema)]
             else:
                 results = [
                     session.finalize(range(start, min(start + finalize_chunk_size, len(group_keys))))

vgi/filter_v2.py

@@ -16,6 +16,7 @@
 from typing import Literal as TypingLiteral
 
 import pyarrow as pa
+from vgi_rpc.utils import empty_batch
 
 FILTER_ENCODING = "vgi.filters.v2"
 FILTER_VERSION = "2"
@@ -765,10 +766,9 @@ def _predicate(self, obj: dict[str, object], where: str, *, update: bool = False
             and not isinstance(expression, RuntimeFilter)
             and not _REPLAYING_ACCEPTED_STATE.get()
         ):
-            empty = pa.RecordBatch.from_arrays(
-                [pa.array([], type=field.type) for field in self.output_schema],
-                schema=self.output_schema,
-            )
+            # empty_batch, not pa.array([], type=...): that has no builder
+            # for union or extension types.
+            empty = empty_batch(self.output_schema)
             try:
                 _evaluate_expression(expression, empty, self.context)
             except Exception as exc: