group-commit patches

scripts/benchmark/README.md

@@ -121,22 +121,43 @@ optional. Reports `spans_per_sec`, `rows_ingested` (checked against
 `total_spans_sent` — any mismatch is a real bug, not a benchmark artifact),
 and writes JSON to `benchmark/results/`.
 
-This benchmark caught a real concurrency bug: multiple exporter processes
-racing to `INSERT` into a table that doesn't exist yet each open their own
-transaction, and DuckDB's catalog allows only one of them to `CREATE TABLE`
-— the rest saw a `TransactionException` ("write-write conflict") surfaced as
-an HTTP 400, which OTLP exporters correctly do not retry (4xx = client
-error), silently dropping that batch. A blind bounded retry closed most of
-the gap but re-parsed the whole payload on every attempt — under a big
-enough pile-up (16-way concurrent cold start) that pushed request latency
-past clients' own read timeouts instead. Fixed properly with
-`RawIngestSerialized` (`raw_ingest.cpp`): a table's first-ever insert in this
-process queues behind an in-process lock instead of racing at all, while
-every request afterward (the steady-state case) never touches the lock —
-just a cached membership check. 30/30 clean in stress testing at 16-way
-concurrency, no throughput change. Regression-tested in
-`test/http/raw_api_compat.sh`'s "concurrent create" section (fires 8
-concurrent requests at a brand-new table).
+This benchmark caught two real concurrency issues under sustained 16-32-way
+concurrent OTLP/HTTP load, both fixed in `raw_ingest.cpp`:
+
+- **Concurrent first-insert races.** Multiple exporter processes racing to
+  `INSERT` into a table that doesn't exist yet each open their own
+  transaction, and DuckDB's catalog allows only one of them to `CREATE TABLE`
+  — the rest saw a `TransactionException` ("write-write conflict") surfaced
+  as an HTTP 400, which OTLP exporters correctly do not retry (4xx = client
+  error), silently dropping that batch. Fixed with `RawIngestSerialized`: a
+  table's first-ever insert in this process queues behind an in-process lock
+  instead of racing at all; every request afterward (the steady-state case)
+  never touches the lock, just a cached membership check.
+- **Per-commit WAL/fsync serialization tail latency.** Even with the race
+  fixed, DuckDB's single-writer WAL still serializes every commit's fsync
+  regardless of how many independent Connections are committing — no single
+  commit was ever slow in isolation (~200-450ms), but under a big enough
+  pile-up (16+ concurrent committers) the cumulative queueing occasionally
+  pushed one unlucky request's total latency past a 10s client read timeout.
+  Fixed with `RawIngestGroupCommit`: concurrent requests to the same table
+  coalesce into one shared commit (a "leader" merges every currently-queued
+  request's already-parsed payload via `MergeParsedPayloads` and runs a
+  single transaction for all of them). Lingering to let a batch form is
+  gated on an `active` in-flight counter, not unconditional: a lone,
+  uncontended request (no sibling already in flight for the same table)
+  skips the linger and the whole coalescing dance entirely, going straight
+  through at the same latency as a direct, non-batched call. The first cut
+  of this fix lingered unconditionally and regressed *every* request's
+  latency ~7x (single-digit ms -> ~30ms) even with nobody to batch with —
+  caught by explicitly re-benchmarking the solo/no-contention case after
+  the fix, not just the concurrent stress test that motivated it.
+
+Verified both ends: 45/45 clean in stress testing (16-way and 32-way
+concurrency, 0 failures) with throughput ~130-140k spans/sec while
+contention is genuinely engaged, *and* solo-request latency back to the
+same ~4-6ms baseline as before any of this (measured directly, sequential
+uncontended requests). Regression-tested in `test/http/raw_api_compat.sh`'s
+"concurrent create" section (fires 8 concurrent requests at a brand-new table).
 
 ## VARIANT vs RawDuck (branch `feat/variant-benchmark`)
 

src/include/raw_functions.hpp

@@ -38,6 +38,14 @@ struct RawIngestStats {
 };
 RawIngestStats RawIngestPayload(ClientContext &context, const string &target, const string &payload,
                                 const RawParseOptions &options);
+// Direct (non-SQL) transaction control: skips the parser/binder/optimizer/
+// task-scheduled executor that Connection::BeginTransaction/Commit/Rollback
+// each go through for a plain "BEGIN TRANSACTION"/"COMMIT"/"ROLLBACK" -
+// measured to be a substantial share of per-request cost under concurrent
+// HTTP ingest. Semantically identical for RawDuck's usage (see raw_ingest.cpp).
+void RawBeginTransaction(ClientContext &context);
+void RawCommitTransaction(ClientContext &context);
+void RawRollbackTransaction(ClientContext &context);
 // Ingest a payload that's already been parsed (RawParsedPayload::Process).
 // Lets a caller retry just the catalog/append step — e.g. after a concurrent
 // CREATE/ALTER conflict — without re-parsing and re-shredding the same bytes
@@ -49,6 +57,14 @@ RawIngestStats RawIngestParsedPayload(ClientContext &context, const string &targ
 // (see RawTableCreationCache in raw_ingest.cpp) instead of retrying blind.
 RawIngestStats RawIngestSerialized(Connection &conn, const string &target, shared_ptr<RawParsedPayload> parsed,
                                    const string &payload, const RawParseOptions &options);
+// HTTP/programmatic ingest entry point: coalesces concurrent requests to the
+// same table into one shared commit (see RawCommitCoordinator). Preferred
+// over RawIngestSerialized directly for the HTTP path — a failed batch is
+// rolled back exactly once by whichever request led it, so a caller must
+// NOT call Rollback itself after this throws (it may not have even opened a
+// transaction, if it was a waiter rather than the leader).
+RawIngestStats RawIngestGroupCommit(Connection &conn, const string &target, shared_ptr<RawParsedPayload> parsed,
+                                    const string &payload, const RawParseOptions &options);
 
 TableFunction GetRawServeFunction();
 TableFunction GetRawServeStopFunction();

src/raw_api.cpp

@@ -263,7 +263,7 @@ void HandleIngest(const duckdb_httplib::Request &req, duckdb_httplib::Response &
 		auto options = otlp_signal.empty() ? RequestParseOptions(*conn.context, req, body)
 		                                   : ResolveTransform(*conn.context, "otlp-" + otlp_signal, "");
 		auto parsed = RawParsedPayload::Process(body, options);
-		auto stats = RawIngestSerialized(conn, table, std::move(parsed), body, options);
+		auto stats = RawIngestGroupCommit(conn, table, std::move(parsed), body, options);
 		JsonDoc json;
 		auto root = duckdb_yyjson::yyjson_mut_obj(json.doc);
 		if (!otlp_signal.empty()) {
@@ -282,7 +282,10 @@ void HandleIngest(const duckdb_httplib::Request &req, duckdb_httplib::Response &
 		duckdb_yyjson::yyjson_mut_obj_add_uint(json.doc, root, "inserted", stats.rows);
 		Respond(res, 200, json, root);
 	} catch (std::exception &ex) {
-		conn.Rollback();
+		// no Rollback here: RawIngestGroupCommit already rolled back a
+		// failed batch on whichever request led it, and a waiter (or a
+		// parse failure before we ever got there) never opened a
+		// transaction on this connection to begin with.
 		RespondError(res, 400, ErrorData(ex).RawMessage());
 	}
 }
@@ -323,13 +326,13 @@ void HandleOtlpProtobuf(const duckdb_httplib::Request &req, duckdb_httplib::Resp
 	}
 	try {
 		auto parsed = RawParsedPayload::Process(payload, options);
-		auto stats = RawIngestSerialized(conn, table, std::move(parsed), payload, options);
+		auto stats = RawIngestGroupCommit(conn, table, std::move(parsed), payload, options);
 		res.status = 200;
 		res.set_content(
 		    RawOtlpProtobufResponse(signal, stats.errors, stats.errors ? "some records could not be parsed" : ""),
 		    "application/x-protobuf");
 	} catch (std::exception &ex) {
-		conn.Rollback();
+		// no Rollback here: see the equivalent comment in HandleIngest.
 		RespondOtlpStatus(res, 400, ErrorData(ex).RawMessage());
 	}
 }

src/raw_async.cpp

@@ -148,15 +148,15 @@ class RawAsyncBuffers : public ObjectCacheEntry {
 			return 0;
 		}
 		Connection conn(*db_locked);
-		conn.BeginTransaction();
+		RawBeginTransaction(*conn.context);
 		idx_t flushed = 0;
 		try {
 			for (auto &payload : buffer.payloads) {
 				flushed += RawIngestPayload(*conn.context, target, payload.first, payload.second).rows;
 			}
-			conn.Commit();
+			RawCommitTransaction(*conn.context);
 		} catch (...) {
-			conn.Rollback();
+			RawRollbackTransaction(*conn.context);
 			// fire-and-forget semantics: the batch is dropped, like
 			// ClickHouse async inserts without wait_for_async_insert
 			return 0;

src/raw_grpc.cpp

@@ -110,18 +110,18 @@ grpc::Status ExportSignal(const grpc::ServerContext &context, const google::prot
 		RawAsyncEnqueue(*conn.context, table, payload, parse_options);
 		return grpc::Status::OK;
 	}
-	conn.BeginTransaction();
+	RawBeginTransaction(*conn.context);
 	try {
 		auto parse_options = ResolveTransform(*conn.context, "otlp-" + signal, "");
 		auto stats = RawIngestPayload(*conn.context, table, payload, parse_options);
-		conn.Commit();
+		RawCommitTransaction(*conn.context);
 		rejected = stats.errors;
 		if (rejected > 0) {
 			error_message = "some records could not be parsed";
 		}
 		return grpc::Status::OK;
 	} catch (std::exception &ex) {
-		conn.Rollback();
+		RawRollbackTransaction(*conn.context);
 		return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, ErrorData(ex).RawMessage());
 	}
 }