fix(table-in-out): a 0-row batch is not end-of-stream in FINALIZE

A multi-batch streaming FINALIZE returned only its FIRST batch over HTTP, and
did so silently: the rows that arrived were correct, so nothing but a count
revealed the missing ones. Measured against the Python worker, a flush of 200
batches returned 1 row with the right SUM.

VgiTableInOutFinalize ended the stream on `!batch || batch->num_rows() == 0`.
Null is end-of-stream; ZERO ROWS is not — it is a producer turn that carried no
data. The FINALIZE init response buffers its batches and the last one is empty,
so the first read returned the data batch and the second returned that empty
one, ending the stream while `stream_state_token_` was still live. The
continuation request was never issued at all: 2 reads, 1 emitted batch and 0
producer exchanges, against 6 reads and 5 batches for the same query over
subprocess, where the empty batch never arrives because the IPC reader returns
null at EOS. That is why the bug was invisible on the transport everything is
developed against.

Now: read until a non-empty batch or a null. Skipping empties inside the call
rather than returning HAVE_MORE_OUTPUT keeps the decision here, so it does not
depend on how DuckDB treats an empty output chunk from FinalExecute. The loop
terminates on null — over HTTP once a response carries no continuation token,
over subprocess at IPC EOS.

Adds multi_batch_finalize.test, backed by the new `multi_batch_finish` fixture
in vgi-python and vgi-rust. It emits one batch per input row so a wrong SUM
means a batch's CONTENTS were lost and a wrong COUNT means a whole BATCH was —
the second is what catches a truncated flush. Every existing finalize fixture
in every SDK emits exactly one batch, which is the easy case (one batch fits in
one lock-step HTTP turn and never needs a continuation), and that is precisely
why nothing caught this. Green on both transports; the whole HTTP table_in_out
suite still passes.

Pins the CI fixtures ref to v0.29.1, the vgi-python tag that has the fixture.

Author: rustyconover

.github/workflows/integration.yml

@@ -47,7 +47,7 @@ jobs:
     env:
       VGI_INSTALL_FIXTURES: "1"
       VGI_FIXTURES_DIR: /tmp/vgi-python
-      VGI_FIXTURES_REF: ${{ inputs.fixtures_ref || 'v0.29.0' }}
+      VGI_FIXTURES_REF: ${{ inputs.fixtures_ref || 'v0.29.1' }}
       GEN: ninja
       USE_MERGED_VCPKG_MANIFEST: "1"
       VCPKG_BINARY_SOURCES: "clear;x-gha,readwrite"

CLAUDE.md

@@ -250,7 +250,7 @@ accumulated unnoticed, so it is wired shut:
   works for the Linux in-docker leg, where the env vars are forwarded into the
   container and the install happens inside it.
 - `VGI_FIXTURES_REF` pins the worker revision to a **vgi-python release tag**
-  (currently `v0.29.0`), not `main`. **Bump it in lockstep with the protocol
+  (currently `v0.29.1`), not `main`. **Bump it in lockstep with the protocol
   version** — it must name a release speaking the same `VGI_PROTOCOL_VERSION` as
   `src/`, or ATTACH fails with a message naming both versions (the right failure,
   but it fails the build). Override per-run with the `fixtures_ref`

Makefile

@@ -423,7 +423,7 @@ VGI_RUST_DIR ?= $(HOME)/Development/vgi-rust
 # conformance lane, so it gets the same two gates the rust/go/java/typescript
 # lanes have.
 #
-# Python runs 303 today.
+# Python runs 304 today.
 #
 # VGI_TEST_DEDICATED_WORKER is DECLARED as an expected skip rather than wired.
 # Two tests want a non-launcher worker, and pointing the var at the raw `uv run`

src/vgi_table_in_out_impl.cpp

@@ -1165,10 +1165,38 @@ OperatorFinalizeResultType VgiTableInOutFinalize(ExecutionContext &context, Tabl
 	        {{"conn", conn->GetConnIdHex()},
 	         {"worker_path", bind_data.worker_path()},
 	         {"function_name", bind_data.function_name}});
-	auto output_batch = conn->ReadDataBatch();
+	// EOS is a NULL batch. A 0-ROW batch is not — it is a producer turn that
+	// carried no data, and treating it as end-of-stream silently truncated every
+	// multi-batch FINALIZE over HTTP.
+	//
+	// How it presented: the FINALIZE init response buffers its batches, and the
+	// last one is empty. Over subprocess that empty batch never arrives (the IPC
+	// reader returns null at EOS), so the bug was invisible there; over HTTP the
+	// first read returned the data batch and the second returned the empty one,
+	// which ended the stream while `stream_state_token_` was still live. The
+	// continuation request was never issued at all — measured as 2 reads, 1
+	// emitted batch and 0 producer exchanges for a flush of 200. A worker whose
+	// finalize emits one batch was unaffected, which is every other fixture, in
+	// every SDK.
+	//
+	// It failed SILENTLY rather than loudly: the rows that did arrive were
+	// correct, so only a count betrayed the missing ones.
+	//
+	// Skipping empties here rather than returning HAVE_MORE_OUTPUT keeps the
+	// decision inside this call, so it does not depend on how DuckDB treats an
+	// empty output chunk from FinalExecute. The loop terminates because an
+	// exhausted producer returns null: over HTTP once the response carries no
+	// continuation token, over subprocess at IPC EOS.
+	std::shared_ptr<arrow::RecordBatch> output_batch;
+	for (;;) {
+		output_batch = conn->ReadDataBatch();
+		if (!output_batch || output_batch->num_rows() > 0) {
+			break;
+		}
+	}
 
-	if (!output_batch || output_batch->num_rows() == 0) {
-		// No more output - clean up
+	if (!output_batch) {
+		// End of stream - clean up
 		release_and_finish();
 		return OperatorFinalizeResultType::FINISHED;
 	}

test/sql/integration/table/function_registration.test

@@ -72,6 +72,12 @@ ATTACH 'example' AS example (TYPE vgi, LOCATION '${VGI_TEST_WORKER}');
 # 1 parallel-streaming-finalize fixture (substream_partial_sum, a streaming
 #   TableInOutFunction whose per-substream finalize emits this substream's partial
 #   sum — see table_in_out/parallel_finalize.test)
+# 1 multi-batch-finalize fixture (multi_batch_finish, a streaming
+#   TableInOutFunction whose finalize emits ONE BATCH PER INPUT ROW. Every other
+#   finalize fixture emits a single batch, which is the easy case: over HTTP a
+#   producer is lock-step, so one batch completes in one turn and never needs a
+#   continuation. Two or more do — and that path silently truncated to the first
+#   batch until this fixture existed. See table_in_out/multi_batch_finalize.test)
 # 7 blended ("UNNEST-style") fixtures (geo_encode + geo_encode3, arity-overloaded
 #   RowTransformFunctions sharing the name "geo_encode"; row_sum, a VARARGS blended
 #   fn; blended_drop, a 1->0 scan-mode edge-case map — positional args ARE the
@@ -132,13 +138,13 @@ ATTACH 'example' AS example (TYPE vgi, LOCATION '${VGI_TEST_WORKER}');
 #   re-emits its splits across pages); and four pair splits with another
 #   feature (split_batch_index, split_partitioned, split_dynamic_filter,
 #   split_cacheable). Back splits/*.test.
-# = 161
+# = 162
 query I
 SELECT COUNT(*)
 FROM duckdb_functions()
 WHERE database_name = 'example' AND function_type = 'table';
 ----
-161
+162
 
 # The 34 pure table functions (no TABLE parameter type)
 # Note: make_series appears 5 times (5 overloads), make_pairs appears 3 times,

test/sql/integration/table_in_out/multi_batch_finalize.test

@@ -0,0 +1,107 @@
+# name: test/sql/integration/table_in_out/multi_batch_finalize.test
+# description: A streaming table-in-out FINALIZE that emits MANY batches.
+#
+#   Every other finalize fixture in every SDK returns exactly ONE batch, and one
+#   batch is the easy case: over HTTP a producer is strictly lock-step (one batch
+#   per response), so a single-batch flush completes inside its one turn and never
+#   needs a continuation token. A flush of TWO OR MORE batches does — and in
+#   vgi-rust that path was a hard error instead of a pagination, which nothing
+#   noticed for as long as no fixture emitted a second batch.
+#
+#   `multi_batch_finish` emits one batch per input row the substream saw: `rows`
+#   batches, the first carrying that substream's `total` and the rest carrying 0.
+#   Splitting it that way makes the two failure modes tell themselves apart:
+#
+#     * SUM(n) wrong   -> a batch's CONTENTS were lost, duplicated or reordered.
+#     * COUNT(*) wrong -> a BATCH was lost or repeated. This is the one that
+#       catches a broken continuation, because a flush truncated after its first
+#       batch still SUMS correctly and only the count betrays it.
+#
+#   Both invariants hold however DuckDB fans the input across substreams (each
+#   substream's partials sum to the whole, and its batch count equals its row
+#   count), so nothing here assumes a thread or substream count.
+# group: [vgi_integration_table_in_out]
+
+require-env VGI_TEST_WORKER
+
+require vgi
+
+# HTTP transport ATTACH needs httpfs loaded (else a "VGI HTTP transport requires
+# the httpfs extension" error — containing "HTTP" — makes the harness silently
+# skip the whole file). Harmless on subprocess.
+require httpfs
+
+statement ok
+ATTACH 'example' AS ex (TYPE vgi, LOCATION '${VGI_TEST_WORKER}');
+
+# ---------------------------------------------------------------------------
+# Single substream, many batches — the case the HTTP continuation must serve.
+# threads=1 keeps it to ONE substream, so one finalize really does flush 200
+# batches rather than 200 substreams flushing one batch each.
+# ---------------------------------------------------------------------------
+
+statement ok
+SET threads=1;
+
+# 200 input rows -> 200 finalize batches from a single flush.
+query II
+SELECT COUNT(*), SUM(n) FROM ex.multi_batch_finish((SELECT i AS n FROM range(1, 201) t(i)));
+----
+200	20100
+
+# Two rows is the smallest input that needs a continuation at all: one batch
+# fits in a single HTTP turn, two do not.
+query II
+SELECT COUNT(*), SUM(n) FROM ex.multi_batch_finish((SELECT i AS n FROM range(1, 3) t(i)));
+----
+2	3
+
+# One row still works — the single-batch flush keeps its one-turn drain.
+query II
+SELECT COUNT(*), SUM(n) FROM ex.multi_batch_finish((SELECT i AS n FROM range(1, 2) t(i)));
+----
+1	1
+
+# Empty input: no rows accumulated, so no batches to flush.
+query I
+SELECT COUNT(*) FROM ex.multi_batch_finish((SELECT i AS n FROM range(1, 1) t(i)));
+----
+0
+
+# ---------------------------------------------------------------------------
+# Parallel — several substreams, each flushing several batches. The invariants
+# are per-substream sums and counts, so they hold at any fan-out.
+# ---------------------------------------------------------------------------
+
+statement ok
+SET threads=8;
+
+statement ok
+CREATE TABLE big AS SELECT i AS n FROM range(1, 5001) t(i);
+
+# sum(1..5000) = 12502500, and one output row per input row.
+query II
+SELECT COUNT(*), SUM(n) FROM ex.multi_batch_finish((SELECT n FROM big));
+----
+5000	12502500
+
+# UNION ALL: each branch is its own substream, so this is several independent
+# multi-batch flushes unioned together.
+query II
+SELECT COUNT(*), SUM(n) FROM (
+    SELECT n FROM ex.multi_batch_finish((SELECT n FROM big))
+    UNION ALL
+    SELECT n FROM ex.multi_batch_finish((SELECT n FROM big))
+);
+----
+10000	25005000
+
+# pool false → a fresh worker per substream acquire, so a flush that depended on
+# state left behind on a shared worker would show up here.
+statement ok
+ATTACH 'example' AS ex2 (TYPE vgi, LOCATION '${VGI_TEST_WORKER}', pool false);
+
+query II
+SELECT COUNT(*), SUM(n) FROM ex2.multi_batch_finish((SELECT n FROM big));
+----
+5000	12502500