perf(rpc): rewrite SerializeToIpcBytes around the payload API

SerializeToIpcBytes was the canonical "Arrow batch -> std::vector<uint8_t>"
helper used by every aggregate, table_in_out, table_buffering, bind and
init code path. The body wrote into a BufferOutputStream via MakeStreamWriter
(which geometrically grows the underlying heap buffer through realloc +
memcpy as bytes accumulate), then Finish()'d to a Buffer, then copied that
Buffer into a freshly allocated std::vector. Two effectively redundant
allocations and one full copy per call.

Drive arrow::ipc's payload API by hand instead:

  GetSchemaPayload      -> schema FlatBuffer once
  CollectDictionaries / -> dictionary batch payloads (empty for non-dict
  GetDictionaryPayload    schemas, no extra cost)
  GetRecordBatchPayload -> batch FlatBuffer + body buffers once
  GetPayloadSize        -> exact size for each
  std::vector<uint8_t>     allocated once at the precomputed total
  WriteIpcPayload          writes each piece into a FixedSizeBufferWriter
                           slice of the destination

Single Assemble pass, no realloc chain, no final copy.

Dictionary handling is the subtle part. The Arrow IPC streaming format
requires a dictionary-batch message between the schema and record-batch
messages for every column that uses a dictionary encoding (DuckDB enums,
some struct fields). MakeStreamWriter's IpcFormatWriter handles this via
WriteDictionaries(batch); replicating it here is mandatory or every
enum-bearing table fails to round-trip. The first attempt at this commit
skipped the dictionary messages and broke ~20 integration tests
(filter_pushdown/enums, settings/*, catalog/multi_branch_*, aggregate/
nest_tensor) with "Tried reading schema message, was null or length 0" on
the worker side. The CollectDictionaries + GetDictionaryPayload loop fixes
that; the full integration suite (178 cases, 7670 assertions) passes again.

Verified byte-identical output against the old MakeStreamWriter path on
BIGINT, string, multi-column, empty-struct, list, struct-of-int+string, and
dictionary-of-int8+utf8 schemas.

Measured win (mimalloc pool, 15-iter steady state, python+subprocess):

  aggregate_sum           t=1, 100k rows    +5.3%
  aggregate_sum           t=1, 1M  rows     +3.3%
  aggregate_sum           t=4, 100k rows    +5.0%
  aggregate_sum           t=4, 1M  rows     +1.9%
  table_in_out_sum_all    t=4, 100k rows   +13.2%
  table_in_out_sum_all    t=4, 1M  rows    +11.8%
  table_in_out_echo       t=4, 1M  rows, 1KB payload  +4.0%
  (others within +-2% noise floor)

The table_in_out_sum_all numbers are the headline: that function calls
SerializeToIpcBytes per input batch on every duckdb worker thread, so
eliminating one Assemble pass plus the growing-buffer realloc chain
compounds across threads. Cold paths (bind, init) also benefit but their
share of the workload is negligible.

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

Author: rustyconover

src/vgi_rpc_types.cpp

@@ -257,27 +257,77 @@ BuildOptionalStringMapScalar(const std::optional<std::vector<std::pair<std::stri
 // ============================================================================
 
 std::vector<uint8_t> SerializeToIpcBytes(const std::shared_ptr<arrow::RecordBatch> &batch) {
-	auto sink_result = arrow::io::BufferOutputStream::Create();
-	if (!sink_result.ok()) {
-		throw IOException("Failed to create buffer: " + sink_result.status().ToString());
-	}
-	auto sink = sink_result.ValueUnsafe();
-
-	auto writer_result = arrow::ipc::MakeStreamWriter(sink, batch->schema());
-	if (!writer_result.ok()) {
-		throw IOException("Failed to create IPC writer: " + writer_result.status().ToString());
-	}
-	auto writer = writer_result.ValueUnsafe();
-
-	CheckStatus(writer->WriteRecordBatch(*batch), "write batch to IPC");
-	CheckStatus(writer->Close(), "close IPC writer");
+	// Single-allocation path: drive the payload-level Arrow APIs by hand so
+	// RecordBatchSerializer::Assemble runs ONCE, GetPayloadSize tells us the
+	// exact total bytes, and we allocate the destination std::vector at the
+	// correct size up front. Skips the realloc chain that
+	// BufferOutputStream + MakeStreamWriter incur and the extra
+	// vector-from-buffer copy the prior implementation did at the end.
+	// Wire bytes are identical to what MakeStreamWriter+WriteRecordBatch+Close
+	// would have produced (same primitive Arrow calls underneath).
+	//
+	// Dictionary columns require an extra dictionary-batch message between
+	// the schema and record batch messages (IpcFormatWriter::WriteDictionaries
+	// in MakeStreamWriter's implementation). We replicate that ordering
+	// explicitly via CollectDictionaries + GetDictionaryPayload so enum/dict
+	// schemas (e.g. DuckDB enums) round-trip correctly.
+	const auto &options = arrow::ipc::IpcWriteOptions::Defaults();
+	arrow::ipc::DictionaryFieldMapper mapper(*batch->schema());
+
+	arrow::ipc::IpcPayload schema_payload;
+	CheckStatus(arrow::ipc::GetSchemaPayload(*batch->schema(), options, mapper, &schema_payload),
+	            "build schema payload");
+
+	// Collect dictionary payloads (empty for non-dict schemas → no extra cost).
+	auto dictionaries_result = arrow::ipc::CollectDictionaries(*batch, mapper);
+	if (!dictionaries_result.ok()) {
+		throw IOException("Arrow collect dictionaries failed: %s",
+		                  dictionaries_result.status().ToString());
+	}
+	const auto dictionaries = std::move(dictionaries_result).ValueUnsafe();
+	std::vector<arrow::ipc::IpcPayload> dict_payloads(dictionaries.size());
+	for (size_t i = 0; i < dictionaries.size(); ++i) {
+		CheckStatus(arrow::ipc::GetDictionaryPayload(dictionaries[i].first, dictionaries[i].second,
+		                                              options, &dict_payloads[i]),
+		            "build dictionary payload");
+	}
+
+	arrow::ipc::IpcPayload batch_payload;
+	CheckStatus(arrow::ipc::GetRecordBatchPayload(*batch, /*custom_metadata=*/nullptr, options, &batch_payload),
+	            "build record-batch payload");
+
+	// EOS marker: 4-byte continuation token 0xFFFFFFFF + 4-byte zero length.
+	// Matches arrow::ipc PayloadStreamWriter::WriteEOS (non-legacy format).
+	static constexpr uint8_t kEosMarker[8] = {0xff, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00};
+	const int64_t schema_size = arrow::ipc::GetPayloadSize(schema_payload, options);
+	std::vector<int64_t> dict_sizes(dict_payloads.size());
+	int64_t dicts_total = 0;
+	for (size_t i = 0; i < dict_payloads.size(); ++i) {
+		dict_sizes[i] = arrow::ipc::GetPayloadSize(dict_payloads[i], options);
+		dicts_total += dict_sizes[i];
+	}
+	const int64_t batch_size = arrow::ipc::GetPayloadSize(batch_payload, options);
+	const int64_t total = schema_size + dicts_total + batch_size + static_cast<int64_t>(sizeof(kEosMarker));
+
+	std::vector<uint8_t> out(static_cast<size_t>(total));
+	int32_t mlen = 0; // discarded; only here to satisfy the API
+	auto write_payload_at = [&](const arrow::ipc::IpcPayload &p, int64_t offset, int64_t size, const char *what) {
+		auto slice = std::make_shared<arrow::MutableBuffer>(out.data() + offset, size);
+		auto sink = std::make_shared<arrow::io::FixedSizeBufferWriter>(slice);
+		CheckStatus(arrow::ipc::WriteIpcPayload(p, options, sink.get(), &mlen), what);
+	};
 
-	auto finish_result = sink->Finish();
-	if (!finish_result.ok()) {
-		throw IOException("Failed to finish IPC buffer: " + finish_result.status().ToString());
-	}
-	auto buffer = finish_result.ValueUnsafe();
-	return std::vector<uint8_t>(buffer->data(), buffer->data() + buffer->size());
+	int64_t cursor = 0;
+	write_payload_at(schema_payload, cursor, schema_size, "write schema payload");
+	cursor += schema_size;
+	for (size_t i = 0; i < dict_payloads.size(); ++i) {
+		write_payload_at(dict_payloads[i], cursor, dict_sizes[i], "write dictionary payload");
+		cursor += dict_sizes[i];
+	}
+	write_payload_at(batch_payload, cursor, batch_size, "write record-batch payload");
+	cursor += batch_size;
+	std::memcpy(out.data() + cursor, kEosMarker, sizeof(kEosMarker));
+	return out;
 }
 
 std::shared_ptr<arrow::RecordBatch> DeserializeFromIpcBytes(const uint8_t *data, size_t len) {