feat(table-in-out): blended ("UNNEST-style") RowTransformFunction core (Phase B)

A blended table-in-out function's positional args ARE its per-row input columns
(real typed args, no synthetic TABLE placeholder), so ONE registration serves
every call shape:
  geo_encode(52.0, 13.0)            -- literal  -> one input row
  FROM t, geo_encode(t.x, t.y)      -- columns   -> streaming input
  LATERAL geo_encode(t.x, t.y)

The signal is worker-advertised input_from_args (VgiFunctionInfo), parsed off the
catalog function info. Registration enters the in_out_function branch on it even
without a TABLE arg (positional_types are real value types). Bind builds the
worker input schema from the DECLARED positional arg names (the worker reads
columns by name) + the input-provided types, ignoring input_table_names (empty in
the literal shape), and sets single_row_scan for the childless call.

Critical fix — the literal shape is driven by PhysicalTableScan, which acts as a
SOURCE: it re-invokes the callback with the SAME cardinality-1 input chunk and
decides flow SOLELY on chunk.size() (discarding our returned OperatorResultType).
The new single_row_scan branch in VgiTableInOutFunction writes the one synthesized
input row once, CloseInputWriter()s so the worker reaches EOS, then drains to EOS
inside the call (skipping empty-but-not-EOS batches), returning a 0-row chunk ONLY
at true EOS — otherwise the query infinite-loops. Cancel-not-pool the scan-mode
worker on EOS.

Validated end to end: literal / column / LATERAL / projection / int->DOUBLE cast /
1->0 filter all work. No finalize (map-shaped; the worker side rejects a finalize
override on a blended fn). Inline named args (f(x,y, opt:=v)) are NOT yet supported
(a follow-up binder patch). Worker-side RowTransformFunction + fixture land in
vgi-python.

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

Author: rustyconover

src/generated/vgi_protocol_schemas.hpp

@@ -2,7 +2,7 @@
 // GENERATED by vgi.codegen.cpp_schemas. DO NOT EDIT BY HAND.
 //
 // Generator: vgi-gen-cpp-schemas v1
-// Content hash: 28aa7d751da6
+// Content hash: c025cbc6995f
 //
 // To regenerate:
 //   uv run --project ~/Development/vgi-python vgi-gen-cpp-schemas \
@@ -119,6 +119,7 @@ inline const std::shared_ptr<arrow::Schema> &FunctionInfoSchema() {
 	    arrow::field("source_order_dependent", arrow::boolean(), /*nullable=*/false),
 	    arrow::field("sink_order_dependent", arrow::boolean(), /*nullable=*/false),
 	    arrow::field("requires_input_batch_index", arrow::boolean(), /*nullable=*/false),
+	    arrow::field("input_from_args", arrow::boolean(), /*nullable=*/false),
 	    arrow::field("required_settings", arrow::list(arrow::utf8()), /*nullable=*/false),
 	    arrow::field("required_secrets", arrow::list(arrow::struct_({arrow::field("secret_type", arrow::utf8(), /*nullable=*/false), arrow::field("scope", arrow::utf8(), /*nullable=*/true), arrow::field("secret_name", arrow::utf8(), /*nullable=*/true)})), /*nullable=*/false),
 	});

src/include/vgi_catalog_metadata.hpp

@@ -540,6 +540,14 @@ struct VgiFunctionInfo {
 	// with sink_order_dependent.
 	bool requires_input_batch_index = false;
 
+	// Blended ("UNNEST-style") table-in-out: the function's positional args ARE
+	// its per-row input columns (real typed args, no synthetic TABLE placeholder),
+	// so ONE registration serves f(52,13) (literal), FROM t, f(t.x,t.y) (columns),
+	// and LATERAL f(t.x,t.y). The registration enters the in-out branch on this
+	// flag even without a TABLE-typed arg, and bind builds the input schema from
+	// the declared arg names + drives the literal single-row scan-mode.
+	bool input_from_args = false;
+
 	// Settings required by this function (must be set before invocation)
 	std::vector<std::string> required_settings;
 

src/include/vgi_table_in_out_impl.hpp

@@ -112,6 +112,17 @@ struct VgiTableInOutBindData : public TableFunctionData {
 	// own worker), but Execute must keep the substream connection open at input-EOS
 	// so VgiTableInOutFinalize can reuse it (a no-finalize map releases it there).
 	bool has_finalize = false;
+
+	// Blended ("UNNEST-style") table-in-out (Phase B). input_from_args: this is a
+	// blended function (positional args are input columns). single_row_scan: this
+	// particular bind is the childless call shape (literal f(52,13) or a
+	// pure-varargs childless call) that DuckDB drives via PhysicalTableScan — a
+	// SOURCE that re-feeds its cardinality-1 input chunk until the callback returns
+	// a 0-row chunk. Execute uses the write-once -> CloseInputWriter -> drain-to-EOS
+	// scan-mode branch for this shape. False for the streaming column/LATERAL shape
+	// (PhysicalTableInOutFunction, which advances on NEED_MORE_INPUT).
+	bool input_from_args = false;
+	bool single_row_scan = false;
 };
 
 // ============================================================================
@@ -219,6 +230,12 @@ struct VgiTableInOutLocalState : public ArrowScanLocalState {
 	// finalize is independent per substream, so the "init once" latch must live on
 	// the local state, not the global one (which the serial path uses).
 	bool finalize_sent = false;
+
+	// Phase B (blended literal scan-mode): whether the single synthesized input row
+	// has been written + the input writer closed. Single-thread-safe:
+	// PhysicalTableScan::ParallelSource() is false for in-out functions, so the
+	// literal scan runs on one local state.
+	bool input_submitted = false;
 };
 
 // ============================================================================
@@ -238,6 +255,16 @@ struct VgiTableInOutBindParams {
 	// per-substream fan-out. Threaded to bind_data.parallel_safe.
 	int32_t max_workers = 0;
 
+	// Blended ("UNNEST-style") table-in-out (Phase B): positional args ARE the
+	// per-row input columns. When true, bind builds the worker input schema from
+	// `positional_input_names` (the DECLARED arg names — the worker reads columns
+	// by those names) + `input.input_table_types`, ignoring input_table_names
+	// (empty in the literal shape), and sets single_row_scan for the childless
+	// (literal / pure-varargs) call so Execute uses the write-once scan-mode.
+	bool input_from_args = false;
+	std::vector<std::string> positional_input_names;
+	bool has_varargs = false;
+
 	// Routes through to bind_data so the OptimizerExtension can recognize a
 	// LogicalGet of a buffered table function and rewrite it.
 	bool table_buffering = false;

src/storage/vgi_table_function_set.cpp

@@ -229,6 +229,16 @@ static unique_ptr<FunctionData> VgiCatalogTableInOutFunctionBind(ClientContext &
 	// A3 serial opt-out: a worker that declares Meta.max_workers=1 keeps its
 	// streaming table-in-out on a single shared worker (MaxThreads()=1).
 	params.max_workers = vgi_info.function_info().max_workers.value_or(0);
+	// Blended (Phase B): the DECLARED positional arg names are the worker's input
+	// column names (it reads batch.column("<name>")). Re-parse the arguments schema
+	// to recover them (+ varargs flag) so bind can build the input schema by name.
+	params.input_from_args = vgi_info.function_info().input_from_args;
+	if (params.input_from_args && vgi_info.function_info().arguments_schema) {
+		auto blended_args =
+		    vgi::ParseFunctionArgumentSchema(context, vgi_info.function_info().arguments_schema);
+		params.positional_input_names = blended_args.positional_names;
+		params.has_varargs = blended_args.has_varargs;
+	}
 
 	// Validate required settings
 	const auto &required_settings = vgi_info.function_info().required_settings;
@@ -317,8 +327,21 @@ void VgiTableFunctionSet::LoadEntries(ClientContext &context, const std::lock_gu
 				arg_types = vgi::ParseFunctionArgumentSchema(context, func_info.arguments_schema);
 			}
 
-			// Check if this is a table-in-out function (has TABLE input argument)
-			if (arg_types.HasTableInput()) {
+			// Check if this is a table-in-out function: either it has a TABLE input
+			// argument (classic), or it is a blended ("UNNEST-style") function whose
+			// positional args ARE its per-row input columns (input_from_args). Both
+			// register through the in_out_function path — a blended function's
+			// positional_types are real value types (no TABLE marker), so it also
+			// serves the literal f(52,13) and column FROM t, f(t.x,t.y) shapes.
+			if (arg_types.HasTableInput() || func_info.input_from_args) {
+				// A blended function must not also declare a finalize (map-shaped).
+				// The worker's resolve_metadata already rejects this; assert as a
+				// defense against a hand-rolled/old worker that advertises both.
+				if (func_info.input_from_args && func_info.has_finalize) {
+					throw InvalidInputException(
+					    "Function '%s' advertises input_from_args (blended) AND has_finalize; a blended "
+					    "RowTransformFunction is map-shaped and cannot have a finalize.", func_info.name);
+				}
 				// Create a table-in-out function
 				// Table-in-out functions use LogicalType::TABLE in args and have an in_out_function
 				TableFunction table_func(arg_types.positional_types, nullptr, VgiCatalogTableInOutFunctionBind,

src/vgi_catalog_api.cpp

@@ -2645,6 +2645,11 @@ VgiFunctionInfo ParseFunctionInfo(const std::shared_ptr<arrow::RecordBatch> &bat
 	// so DuckDB threads source-position metadata to every Sink call.
 	info.requires_input_batch_index = row["requires_input_batch_index"].value_or(false);
 
+	// input_from_args — optional bool (defaults to false for older workers). A
+	// blended ("UNNEST-style") table-in-out whose positional args ARE its per-row
+	// input columns.
+	info.input_from_args = row["input_from_args"].value_or(false);
+
 	// Required settings for this function (list of strings)
 	info.required_settings = row["required_settings"].value_or(std::vector<std::string> {});
 

src/vgi_table_in_out_impl.cpp

@@ -77,6 +77,8 @@ unique_ptr<FunctionData> VgiTableInOutBindData::Copy() const {
 	copy->requires_input_batch_index = requires_input_batch_index;
 	copy->parallel_safe = parallel_safe;
 	copy->has_finalize = has_finalize;
+	copy->input_from_args = input_from_args;
+	copy->single_row_scan = single_row_scan;
 	return copy;
 }
 
@@ -165,30 +167,75 @@ unique_ptr<FunctionData> VgiTableInOutBind(ClientContext &context, TableFunction
 	// case is honored, matching the pure-table path's use of the same field.)
 	bind_data->parallel_safe = (params.max_workers != 1);
 
-	// Build arguments from the regular (non-TABLE) inputs
-	// input.inputs contains positional arguments, but TABLE arguments are represented as NULL
-	// We skip NULL values since they represent the TABLE input (not a scalar argument)
-	// Named arguments are in input.named_parameters
+	bind_data->input_from_args = params.input_from_args;
+
+	// Build arguments from the regular (non-TABLE) inputs.
+	// input.inputs contains positional arguments, but TABLE arguments are represented as NULL.
+	// We skip NULL values since they represent the TABLE input (not a scalar argument).
+	// Named arguments are in input.named_parameters.
+	//
+	// BLENDED (input_from_args): the positional inputs ARE the per-row input
+	// columns (delivered by DuckDB's synthesized input chunk), NOT bind arguments —
+	// skip ALL of them. Only named args survive as bind-time scalars.
 	vector<Value> positional_args;
-	for (auto &val : input.inputs) {
-		// Skip NULL values - these represent TABLE arguments
-		if (val.IsNull()) {
-			continue;
+	if (!params.input_from_args) {
+		for (auto &val : input.inputs) {
+			// Skip NULL values - these represent TABLE arguments
+			if (val.IsNull()) {
+				continue;
+			}
+			positional_args.push_back(val);
 		}
-		positional_args.push_back(val);
 	}
 	vector<std::pair<string, Value>> named_args;
 	for (auto &[name, value] : input.named_parameters) {
 		named_args.emplace_back(name, value);
 	}
 	bind_data->arguments = BuildArgumentsFromValues(context, positional_args, named_args);
 
-	// Build the input schema from the table input types/names
-	bind_data->input_schema = BuildArrowSchemaFromDuckDB(context, input.input_table_types, input.input_table_names);
+	// Build the input schema.
+	//
+	// BLENDED: use the DECLARED positional arg names (the worker reads columns by
+	// those names) with the input-provided types. IGNORE input.input_table_names —
+	// it is empty in the literal shape (f(52,13) -> ["",""]) and would otherwise
+	// name the worker's input columns after the referenced columns in the column
+	// shape. For a pure-VARARGS blended function the declared names don't cover the
+	// N runtime columns, so fall back to generated col0..colN-1.
+	if (params.input_from_args) {
+		vector<string> in_names;
+		const idx_t n_cols = input.input_table_types.size();
+		if (!params.has_varargs && params.positional_input_names.size() == n_cols) {
+			for (auto &nm : params.positional_input_names) {
+				in_names.push_back(nm);
+			}
+		} else {
+			for (idx_t i = 0; i < n_cols; i++) {
+				in_names.push_back("col" + std::to_string(i));
+			}
+		}
+		bind_data->input_schema = BuildArrowSchemaFromDuckDB(context, input.input_table_types, in_names);
+		// Childless call shape (literal f(52,13) or a pure-varargs childless call)
+		// -> PhysicalTableScan, which needs the write-once scan-mode in Execute.
+		// Robust signal: all synthesized input names are empty (the column/LATERAL
+		// shape gives non-empty names and a streaming child). Do NOT key on
+		// !input.inputs.empty() (false-negatives the zero-positional childless call).
+		bool all_names_empty = true;
+		for (auto &nm : input.input_table_names) {
+			if (!nm.empty()) {
+				all_names_empty = false;
+				break;
+			}
+		}
+		bind_data->single_row_scan = all_names_empty;
+	} else {
+		bind_data->input_schema = BuildArrowSchemaFromDuckDB(context, input.input_table_types, input.input_table_names);
+	}
 
 	VGI_LOG(context, "table_in_out.bind",
 	        {{"worker_path", bind_data->worker_path()},
 	         {"function_name", bind_data->function_name},
+	         {"input_from_args", params.input_from_args ? "true" : "false"},
+	         {"single_row_scan", bind_data->single_row_scan ? "true" : "false"},
 	         {"input_columns", std::to_string(input.input_table_types.size())}});
 
 	// Create the connection and perform bind
@@ -631,6 +678,72 @@ OperatorResultType VgiTableInOutFunction(ExecutionContext &context, TableFunctio
 	}
 	IFunctionConnection &conn = parallel ? *local_state.connection : *global_state.connection;
 
+	// ------------------------------------------------------------------------
+	// Phase B — blended LITERAL scan-mode (single_row_scan).
+	// ------------------------------------------------------------------------
+	// DuckDB drives the childless call shape (f(52,13) / pure-varargs childless)
+	// through PhysicalTableScan, which acts as a SOURCE: it re-invokes this
+	// callback with the SAME cardinality-1 input chunk and decides flow SOLELY on
+	// `chunk.size()` (it DISCARDS our returned OperatorResultType). So we must
+	// return a 0-row chunk ONLY at true end-of-stream, and never mid-stream.
+	if (bind_data.single_row_scan) {
+		// 1) Drain a large (1->N) output batch that overflowed STANDARD_VECTOR_SIZE.
+		if (HasRemainingBatchData(local_state)) {
+			idx_t rows_copied =
+			    ProduceOutputFromBatch(local_state, bind_data.arrow_table, output, bind_data.projection_pushdown);
+			VGI_LOG(client_context, "table_in_out.scan_output",
+			        {{"conn", conn.GetConnIdHex()}, {"function_name", bind_data.function_name},
+			         {"output_rows", std::to_string(rows_copied)}});
+			return OperatorResultType::HAVE_MORE_OUTPUT;
+		}
+		// 2) First call: write the single synthesized input row ONCE, then close the
+		//    input writer so the worker can reach EOS (no deadlock).
+		if (!local_state.input_submitted) {
+			auto input_batch = DataChunkToArrow(client_context, input, bind_data.input_schema);
+			VGI_LOG(client_context, "table_in_out.scan_write_input",
+			        {{"conn", conn.GetConnIdHex()}, {"function_name", bind_data.function_name},
+			         {"input_rows", std::to_string(input_batch->num_rows())}});
+			conn.WriteInputBatch(input_batch);
+			conn.CloseInputWriter();
+			local_state.input_submitted = true;
+		}
+		// 3) Drain to EOS INSIDE this call: skip empty-but-not-EOS batches (worker
+		//    heartbeat / 1->0). Return a >0-row chunk on the first non-empty batch;
+		//    return a 0-row chunk (SetCardinality(0)) only on nullptr (true EOS).
+		while (true) {
+			std::shared_ptr<arrow::RecordBatch> output_batch;
+			try {
+				output_batch = conn.ReadDataBatch();
+			} catch (const IOException &) {
+				// Worker may exit right after input EOS (e.g. empty 1->0). Treat as
+				// clean end-of-stream, mirroring the scalar path's post-close read.
+				output_batch = nullptr;
+			}
+			if (!output_batch) {
+				output.SetCardinality(0);
+				// Cancel-not-pool: the "no-finalize in-out worker returns to a clean
+				// accept-loop after input-EOS" transition is unproven for scan-mode,
+				// so drop the connection instead of pooling it.
+				if (parallel) {
+					local_state.connection.reset();
+				} else {
+					global_state.stream_finished = true;
+				}
+				return OperatorResultType::FINISHED;
+			}
+			if (output_batch->num_rows() == 0) {
+				continue; // empty-but-not-EOS: keep reading, never return 0 rows mid-stream
+			}
+			LoadBatchIntoScanState(local_state, output_batch);
+			idx_t rows_copied =
+			    ProduceOutputFromBatch(local_state, bind_data.arrow_table, output, bind_data.projection_pushdown);
+			VGI_LOG(client_context, "table_in_out.scan_output",
+			        {{"conn", conn.GetConnIdHex()}, {"function_name", bind_data.function_name},
+			         {"output_rows", std::to_string(rows_copied)}});
+			return OperatorResultType::HAVE_MORE_OUTPUT;
+		}
+	}
+
 	// Continue producing rows from a batch that exceeded STANDARD_VECTOR_SIZE
 	if (HasRemainingBatchData(local_state)) {
 		idx_t rows_copied = ProduceOutputFromBatch(local_state, bind_data.arrow_table, output, bind_data.projection_pushdown);

test/landing/fixtures/describe.expected.json

@@ -3,7 +3,7 @@
     {
       "attach_options": [],
       "counts": {
-        "functions": 184,
+        "functions": 185,
         "schemas": 2,
         "tables": 59,
         "views": 3
@@ -1820,6 +1820,30 @@
               "name": "generator_exception",
               "type": "table"
             },
+            {
+              "args": [
+                {
+                  "desc": "Latitude input column",
+                  "name": "latitude",
+                  "type": "double"
+                },
+                {
+                  "desc": "Longitude input column",
+                  "name": "longitude",
+                  "type": "double"
+                },
+                {
+                  "default": "4",
+                  "desc": "Rounding precision",
+                  "name": "precision",
+                  "named": true,
+                  "type": "int64"
+                }
+              ],
+              "doc": "Blended per-row geo encoder (lat, lon -> geohash)",
+              "name": "geo_encode",
+              "type": "table"
+            },
             {
               "args": [
                 {

test/sql/integration/table/function_registration.test

@@ -72,6 +72,9 @@ 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 blended ("UNNEST-style") fixture (geo_encode, a RowTransformFunction whose
+#   positional args ARE its per-row input columns — one registration serves
+#   literal / column / LATERAL; see table_in_out/blended.test)
 # 12 result-cache fixtures (cacheable_numbers, cache_nonce, cache_no_store,
 #   cache_scoped_txn, cache_big, cache_multicol, cache_revalidatable, cache_whoami,
 #   cache_versioned_scan, cache_projection, cache_poison, cache_external_fail —
@@ -90,13 +93,13 @@ ATTACH 'example' AS example (TYPE vgi, LOCATION '${VGI_TEST_WORKER}');
 #   filter_bytes keys distinct entries, backs cache/filter_pushdown_keys.test, ALSO a data
 #   Table (surfaces as a table); cache_partitioned — single-value partition_values through
 #   the spill blob, backs cache/spill_partition_values.test)
-# = 122
+# = 123
 query I
 SELECT COUNT(*)
 FROM duckdb_functions()
 WHERE database_name = 'example' AND function_type = 'table';
 ----
-122
+123
 
 # The 34 pure table functions (no TABLE parameter type)
 # Note: make_series appears 5 times (5 overloads), make_pairs appears 3 times,