Raise error when table_in_out_function or scalar_function has empty input

Previously, passing an empty iterator to table_in_out_function() or
scalar_function() would silently produce no output. Now raises
ClientError with a helpful message suggesting to use table_function()
for functions that generate data without input.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: rustyconover

tests/scalar/test_client.py

@@ -116,20 +116,20 @@ def test_empty_batch(self, example_worker: str) -> None:
         assert len(outputs) == 1
         assert outputs[0].num_rows == 0
 
-    def test_empty_iterator(self, example_worker: str) -> None:
-        """Test scalar function with no input batches."""
-        with Client(example_worker) as client:
-            outputs = list(
+    def test_empty_iterator_raises(self, example_worker: str) -> None:
+        """Test scalar function with no input batches raises error."""
+        with (
+            Client(example_worker) as client,
+            pytest.raises(ClientError, match="requires at least one input batch"),
+        ):
+            list(
                 client.scalar_function(
                     function_name="double_column",
                     input=iter([]),
                     arguments=Arguments(positional=(pa.scalar("x"),)),
                 )
             )
 
-        # No input means no output
-        assert len(outputs) == 0
-
     def test_scalar_function_not_started_raises(self, example_worker: str) -> None:
         """Calling scalar_function before start should raise ClientError."""
         client = Client(example_worker)

tests/table_in_out/test_client.py

@@ -58,6 +58,19 @@ def test_table_in_out_function_not_started_raises(
 class TestEdgeCases:
     """Tests for edge cases and boundary conditions."""
 
+    def test_empty_iterator_raises(self, example_worker: str) -> None:
+        """Empty iterator (no batches) should raise ClientError."""
+        with (
+            Client(example_worker) as client,
+            pytest.raises(ClientError, match="requires at least one input batch"),
+        ):
+            list(
+                client.table_in_out_function(
+                    function_name="echo",
+                    input=iter([]),
+                )
+            )
+
     def test_empty_batch(self, example_worker: str) -> None:
         """Empty batch (zero rows) should process correctly."""
         schema = make_schema(

vgi/client/client.py

@@ -1446,7 +1446,7 @@ def table_in_out_function(
                 worker's registry.
             input: Iterator yielding input RecordBatches. Must yield at least one
                 batch. The first batch's schema is used to initialize the IPC
-                stream. If the iterator is empty, no output is produced.
+                stream. Raises ClientError if the iterator is empty.
             arguments: Optional Arguments container with positional and named
                 arguments to pass to the function. Defaults to empty Arguments().
             bind_result_callback: Optional callback invoked with the raw bind
@@ -1468,9 +1468,10 @@ def table_in_out_function(
             Final output from finalize() is always yielded last.
 
         Raises:
-            ClientError: If the client is not started, input iterator yields
-                non-RecordBatch objects, communication with the worker fails,
-                or the worker returns an unexpected status or exception.
+            ClientError: If the client is not started, input iterator is empty,
+                input iterator yields non-RecordBatch objects, communication
+                with the worker fails, or the worker returns an unexpected
+                status or exception.
 
         Example:
             >>> with Client("vgi-example-worker") as client:
@@ -1522,6 +1523,13 @@ def table_in_out_function(
             )
             return
 
+        # Input iterator was empty - table-in-out functions require input
+        raise ClientError(
+            f"table_in_out_function requires at least one input batch. "
+            f"The input iterator for function '{function_name}' was empty. "
+            f"Use table_function() for functions that generate data without input."
+        )
+
     def _table_in_out_function_parallel(
         self,
         *,
@@ -1912,7 +1920,7 @@ def scalar_function(
                 worker's registry.
             input: Iterator yielding input RecordBatches. Must yield at least one
                 batch. The first batch's schema is used to initialize the IPC
-                stream. If the iterator is empty, no output is produced.
+                stream. Raises ClientError if the iterator is empty.
             arguments: Optional Arguments container with positional and named
                 arguments to pass to the function. Defaults to empty Arguments().
             bind_result_callback: Optional callback invoked with the raw bind
@@ -1932,9 +1940,10 @@ def scalar_function(
             In parallel mode (max_processes > 1), output order is non-deterministic.
 
         Raises:
-            ClientError: If the client is not started, input iterator yields
-                non-RecordBatch objects, communication with the worker fails,
-                or the worker returns an unexpected status or exception.
+            ClientError: If the client is not started, input iterator is empty,
+                input iterator yields non-RecordBatch objects, communication
+                with the worker fails, or the worker returns an unexpected
+                status or exception.
 
         Example:
             >>> with Client("vgi-example-worker") as client:
@@ -1987,6 +1996,13 @@ def scalar_function(
             )
             return
 
+        # Input iterator was empty - scalar functions require input
+        raise ClientError(
+            f"scalar_function requires at least one input batch. "
+            f"The input iterator for function '{function_name}' was empty. "
+            f"Use table_function() for functions that generate data without input."
+        )
+
     def _scalar_function_parallel(
         self,
         *,