Add table function cardinality support and empty batch handling

- Table functions now emit an empty batch if process() yields nothing,
  ensuring clients don't block waiting for data
- Client filters out empty batches before yielding to callers
- Worker sends cardinality info in bind result for TableFunctionGenerator
- SequenceFunction now has configurable batch_size parameter (default 1000)
- Added numpy for efficient range generation in SequenceFunction
- Updated protocol documentation with empty output handling details

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

Author: rustyconover

CLAUDE.md

@@ -390,14 +390,15 @@ class SequenceFunction(TableFunctionGenerator):
         max_workers = 1
 
     count: Annotated[int, Arg(0, doc="Number of integers")]
+    batch_size: Annotated[int, Arg(1, default=1000, doc="Batch size for output")]
 
     @property
     def output_schema(self) -> pa.Schema:
         return pa.schema([("n", pa.int64())])
 
     def process(self):
-        for start in range(0, self.count, 1000):
-            end = min(start + 1000, self.count)
+        for start in range(0, self.count, self.batch_size):
+            end = min(start + self.batch_size, self.count)
             yield Output(pa.RecordBatch.from_pydict(
                 {"n": list(range(start, end))}, schema=self.output_schema
             ))

docs/protocol.md

@@ -92,7 +92,7 @@ During data transfer, each direction uses a **single long-lived IPC stream** con
 | 3 | Client → Worker | Handshake | InitInput | 1 |
 | 4 | Worker → Client | Handshake | InitResult | 1 |
 | 5 | Client → Worker | Data | Input batches | 0..N |
-| 6 | Worker → Client | Data | Output batches with status | 1..N |
+| 6 | Worker → Client | Data | Output batches with status | 1..N (always ≥1) |
 
 **Notes:**
 - Stream 5 is absent for Table functions (no input)
@@ -182,6 +182,11 @@ Client                                  Worker
   └───────────────────────────────────────┘
 ```
 
+**Empty output handling:**
+- If `process()` yields no batches, the worker emits an empty batch with `FINISHED` status
+- This ensures the client receives a completion signal and doesn't block waiting for data
+- The client filters out empty batches before yielding to callers (protocol detail not exposed to users)
+
 ### Table-In-Out Function Protocol
 
 Table-in-out functions transform input batches with an optional finalize phase.
@@ -451,12 +456,14 @@ Workers support multiple invocations within a single process. After completing o
 - Schema validation before sending to client
 - Row count validation for scalar functions
 - Automatic error wrapping with stack traces
+- Table functions always emit at least one batch (empty if no output) to signal completion
 
 **Client responsibilities:**
 - Spawn worker subprocess with correct command
 - Send Invocation as first message
 - Handle all status values correctly
 - Close stdin (not just IPC stream) to signal worker shutdown
+- Filter out empty batches from table functions before yielding to callers
 
 **Worker responsibilities:**
 - Parse Invocation and lookup function

pyproject.toml

@@ -4,7 +4,13 @@ version = "0.1.0"
 description = "Vector Gateway Interface - Connect DuckDB to external programs via Apache Arrow"
 readme = "README.md"
 requires-python = ">=3.12.4"
-dependencies = ["click", "pyarrow", "structlog", "platformdirs"]
+dependencies = [
+    "click",
+ "pyarrow",
+ "structlog",
+ "platformdirs",
+ "numpy>=2.4.1",
+]
 
 [project.optional-dependencies]
 dev = ["mypy", "pyarrow-stubs", "pytest", "pytest-cov", "pytest-examples", "pytest-mypy", "pytest-ruff", "pytest-xdist", "ruff"]

tests/table/generator/test_sequence_function.py

@@ -6,6 +6,7 @@
 
 from tests.conftest import make_invocation
 from vgi.arguments import Arguments
+from vgi.client import Client
 from vgi.examples.table import SequenceFunction
 from vgi.testing import assert_table_function_output, batch
 
@@ -79,3 +80,58 @@ def test_large_sequence_batches(
         table = pa.Table.from_batches(outputs)
         assert table.num_rows == 2500
         assert table.column("n").to_pylist() == list(range(2500))
+
+    def test_custom_batch_size(self, run_table_function_mode: RunnerWithMode) -> None:
+        """Custom batch size should control output batch sizes."""
+        runner, mode = run_table_function_mode
+        # Generate 250 values with batch size of 100
+        outputs, logs = runner(SequenceFunction, (250, 100))
+
+        # Should produce 3 batches: 100, 100, 50
+        assert len(outputs) == 3
+        assert outputs[0].num_rows == 100
+        assert outputs[1].num_rows == 100
+        assert outputs[2].num_rows == 50
+
+        table = pa.Table.from_batches(outputs)
+        assert table.column("n").to_pylist() == list(range(250))
+
+    def test_batch_size_larger_than_count(
+        self, run_table_function_mode: RunnerWithMode
+    ) -> None:
+        """Batch size larger than count should produce single batch."""
+        runner, mode = run_table_function_mode
+        outputs, logs = runner(SequenceFunction, (50, 1000))
+
+        assert len(outputs) == 1
+        assert outputs[0].num_rows == 50
+        assert outputs[0].column("n").to_pylist() == list(range(50))
+
+
+class TestSequenceFunctionClient:
+    """Tests for SequenceFunction via Client (wire protocol)."""
+
+    def test_cardinality_returned_in_bind_result(self) -> None:
+        """Cardinality should be returned in bind_result via Client."""
+        bind_results: list[pa.RecordBatch] = []
+
+        def capture_bind_result(result: pa.RecordBatch) -> None:
+            bind_results.append(result)
+
+        with Client("vgi-example-worker") as client:
+            list(
+                client.table_function(
+                    function_name="sequence",
+                    arguments=Arguments(positional=(pa.scalar(100),)),
+                    bind_result_callback=capture_bind_result,
+                )
+            )
+
+        assert len(bind_results) == 1
+        bind_result = bind_results[0]
+
+        # Verify cardinality fields are present and correct
+        assert "cardinality_estimated" in bind_result.schema.names
+        assert "cardinality_max" in bind_result.schema.names
+        assert bind_result.column("cardinality_estimated")[0].as_py() == 100
+        assert bind_result.column("cardinality_max")[0].as_py() == 100