diff --git a/scripts/benchmark/README.md b/scripts/benchmark/README.md index 3ef4aef..1b337a4 100644 --- a/scripts/benchmark/README.md +++ b/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`) diff --git a/src/include/raw_functions.hpp b/src/include/raw_functions.hpp index 83fb761..28fd0ef 100644 --- a/src/include/raw_functions.hpp +++ b/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 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 parsed, + const string &payload, const RawParseOptions &options); TableFunction GetRawServeFunction(); TableFunction GetRawServeStopFunction(); diff --git a/src/raw_api.cpp b/src/raw_api.cpp index d03a6f9..4a3e752 100644 --- a/src/raw_api.cpp +++ b/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()); } } diff --git a/src/raw_async.cpp b/src/raw_async.cpp index d906d9d..2da61af 100644 --- a/src/raw_async.cpp +++ b/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; diff --git a/src/raw_grpc.cpp b/src/raw_grpc.cpp index d54c5d9..7fbc74e 100644 --- a/src/raw_grpc.cpp +++ b/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()); } } diff --git a/src/raw_ingest.cpp b/src/raw_ingest.cpp index 2d72271..06e4b0a 100644 --- a/src/raw_ingest.cpp +++ b/src/raw_ingest.cpp @@ -16,6 +16,7 @@ #include "duckdb/main/connection.hpp" #include "duckdb/main/database.hpp" #include "duckdb/main/database_manager.hpp" +#include "duckdb/main/valid_checker.hpp" #include "duckdb/parser/expression/cast_expression.hpp" #include "duckdb/parser/expression/columnref_expression.hpp" #include "duckdb/parser/expression/function_expression.hpp" @@ -35,6 +36,7 @@ #include "duckdb/transaction/duck_transaction.hpp" #include "duckdb/transaction/meta_transaction.hpp" +#include #include #include #include @@ -102,6 +104,111 @@ class RawTableCreationCache : public ObjectCacheEntry { mutex creation_lock; }; +//===--------------------------------------------------------------------===// +// Group commit: coalesces concurrent HTTP/programmatic requests to the same +// table into a single underlying transaction. DuckDB's single-writer WAL +// serializes every commit's fsync regardless of how many independent +// Connections are involved -- N concurrent requests each committing on their +// own pay N times the WAL-write/fsync latency, and under high fan-in that +// tail can exceed a client's read timeout even though no single commit is +// slow in isolation (observed directly: 16-way concurrent first-time OTLP +// ingest occasionally exceeded a 10s client timeout with every individual +// commit under half a second). One request becomes the "leader" for a +// table's currently-queued batch, merges every queued request's +// already-parsed payload (MergeParsedPayloads — the same merge already used +// for small-batch coalescing on the bulk-file path) and runs ONE +// RawIngestSerialized call for all of them, then wakes every waiter with its +// own row/error count against the batch's shared result. +//===--------------------------------------------------------------------===// + +class RawCommitCoordinator : public ObjectCacheEntry { +public: + static string ObjectType() { + return "rawduck_commit_coordinator"; + } + string GetObjectType() override { + return ObjectType(); + } + optional_idx GetEstimatedCacheMemory() const override { + return optional_idx(); + } + + struct PendingItem { + shared_ptr parsed; + string payload_text; + idx_t rows = 0; + idx_t parse_errors = 0; + bool done = false; + bool failed = false; + string error_message; + RawIngestStats stats; + }; + + mutex lock; + std::condition_variable cv; + // FIFO of requests waiting for the next combined commit, per table. + unordered_map>> pending; + // tables currently being drained/committed by a leader + case_insensitive_set_t committing; + // requests currently inside RawIngestGroupCommit for this table (from + // the moment they enter to the moment they return/throw) — the signal + // used to decide whether lingering is worth it at all. Unlike + // `pending`, this is incremented before the mutex serializes anything, + // so a request that becomes leader can see that siblings are already + // under way even though none of them have reached `pending` yet. + unordered_map active; +}; + +//===--------------------------------------------------------------------===// +// Direct (non-SQL) transaction control. Connection::BeginTransaction() / +// Commit() / Rollback() each run a full SQL statement ("BEGIN TRANSACTION" / +// "COMMIT" / "ROLLBACK") through the parser, binder, optimizer, and the +// task-scheduled executor -- real, measured overhead (profiling showed +// Commit() alone accounting for a substantial share of per-request cost +// under concurrent HTTP ingest). These call the same ClientContext:: +// transaction primitives PhysicalTransaction's operator uses, skipping the +// SQL front end entirely. Semantically identical for RawDuck's usage: no +// read-only/custom-invalidation-policy modifiers (matching a plain "BEGIN +// TRANSACTION"), and immediate_transaction_mode defaults false and RawDuck +// never sets it, so that setting's extra eager-attach behavior never +// applies to these transactions either way. +//===--------------------------------------------------------------------===// + +void RawBeginTransaction(ClientContext &context) { + if (!context.transaction.IsAutoCommit()) { + throw TransactionException("cannot start a transaction within a transaction"); + } + context.transaction.SetAutoCommit(false); +} + +void RawRollbackTransaction(ClientContext &context) { + auto &txn = context.transaction; + if (txn.IsAutoCommit()) { + throw TransactionException("cannot rollback - no transaction is active"); + } + auto &valid_checker = ValidChecker::Get(txn.ActiveTransaction()); + if (valid_checker.IsInvalidated()) { + ErrorData error(ExceptionType::TRANSACTION, valid_checker.InvalidatedMessage()); + txn.Rollback(error); + } else { + txn.Rollback(nullptr); + } +} + +void RawCommitTransaction(ClientContext &context) { + auto &txn = context.transaction; + if (txn.IsAutoCommit()) { + throw TransactionException("cannot commit - no transaction is active"); + } + if (ValidChecker::IsInvalidated(txn.ActiveTransaction())) { + // mirrors PhysicalTransaction: an invalidated transaction can't be + // committed, so treat this exactly like a ROLLBACK instead. + RawRollbackTransaction(context); + return; + } + txn.Commit(); +} + static uint64_t HashPayloadShape(const vector &columns) { uint64_t shape = 0xcbf29ce484222325ULL; for (auto &column : columns) { @@ -1285,24 +1392,148 @@ RawIngestStats RawIngestSerialized(Connection &conn, const string &target, share } } - conn.BeginTransaction(); + RawBeginTransaction(*conn.context); try { auto stats = RawIngestParsedPayload(*conn.context, target, parsed, payload, options); - conn.Commit(); + RawCommitTransaction(*conn.context); mark_known(); return stats; } catch (TransactionException &) { - conn.Rollback(); + RawRollbackTransaction(*conn.context); } // final attempt: no catch here. Any failure is left open for the // caller's own catch/rollback, same as every other error path. - conn.BeginTransaction(); + RawBeginTransaction(*conn.context); auto stats = RawIngestParsedPayload(*conn.context, target, parsed, payload, options); - conn.Commit(); + RawCommitTransaction(*conn.context); mark_known(); return stats; } +// HTTP/programmatic ingest entry point (replaces calling RawIngestSerialized +// directly): coalesces this request with any others concurrently targeting +// the same table into one shared commit. See RawCommitCoordinator. +// +// Waiters never touch their own Connection's transaction state at all — the +// leader runs the entire Begin/Ingest/Commit/Rollback sequence on its own +// `conn`, so a failed batch is rolled back exactly once by the leader. +// Callers must not call Rollback themselves after this throws. +RawIngestStats RawIngestGroupCommit(Connection &conn, const string &target, shared_ptr parsed, + const string &payload, const RawParseOptions &options) { + auto &coord = *ObjectCache::GetObjectCache(*conn.context) + .GetOrCreate(RawCommitCoordinator::ObjectType()); + + auto item = make_shared_ptr(); + item->parsed = parsed; + item->payload_text = payload; + item->rows = parsed->payload.rows.size(); + item->parse_errors = parsed->payload.parse_errors; + + unique_lock guard(coord.lock); + // Mark ourselves "in flight" for this target before anything else: this + // is what lets a brand-new leader tell a genuine concurrent burst apart + // from a lone request with nobody to batch with, and it's set before + // `pending` is touched so it catches siblings that haven't reached + // `pending` yet either (all of it happens under the same lock, so the + // order within this one line doesn't race). + auto active_at_entry = ++coord.active[target]; + struct ActiveGuard { + RawCommitCoordinator &coord; + const string ⌖ + ~ActiveGuard() { + lock_guard g(coord.lock); + auto it = coord.active.find(target); + if (it != coord.active.end() && --it->second == 0) { + coord.active.erase(it); + } + } + } active_guard {coord, target}; + + coord.pending[target].push_back(item); + coord.cv.notify_all(); // wake a lingering leader if the batch just grew + if (coord.committing.count(target) > 0) { + // someone else is already leading a batch for this table: wait for + // OUR item to be resolved by that (or a subsequent) leader round + coord.cv.wait(guard, [&] { return item->done; }); + } else { + coord.committing.insert(target); + // Only linger if a sibling request for this table is *already* + // under way (active_at_entry > 1, counting ourselves): a lone, + // uncontended request has nothing to batch with, and waiting here + // regardless was a measured ~7x latency regression for that common + // case (single-digit ms -> ~30ms per request) for no throughput + // benefit — nobody was going to join anyway. When contention is + // real, wake immediately once a decent batch has formed rather + // than always paying the full window, bounded so a slow-to-arrive + // burst still resolves promptly. + if (active_at_entry > 1) { + constexpr idx_t LINGER_TARGET_BATCH = 3; + coord.cv.wait_for(guard, std::chrono::milliseconds(25), + [&] { return coord.pending[target].size() >= LINGER_TARGET_BATCH; }); + } + while (!coord.pending[target].empty()) { + auto batch = std::move(coord.pending[target]); + coord.pending[target].clear(); + guard.unlock(); + + // merge every queued request's payload into one combined ingest. + // A representative payload TEXT (NDJSON: one request's text per + // line) rides along for the non-native (DuckLake) fallback path, + // which re-parses raw text via raw_records() rather than reusing + // the already-parsed structure — an empty string there would + // silently insert zero rows for a merged batch. + shared_ptr combined = batch.front()->parsed; + string combined_text = batch.front()->payload_text; + for (idx_t i = 1; i < batch.size(); i++) { + MergeParsedPayloads(*combined, std::move(*batch[i]->parsed)); + combined_text += "\n"; + combined_text += batch[i]->payload_text; + } + + RawIngestStats batch_stats; + bool ok = true; + string error_message; + try { + batch_stats = RawIngestSerialized(conn, target, combined, combined_text, options); + } catch (std::exception &ex) { + ok = false; + error_message = ErrorData(ex).RawMessage(); + // RawIngestSerialized always leaves a failed attempt's + // transaction open for the caller to roll back exactly once + // (matching every other ingest error path) — this IS that + // caller now that the leader owns the whole batch's commit. + RawRollbackTransaction(*conn.context); + } + for (auto &b : batch) { + b->done = true; + b->failed = !ok; + b->error_message = error_message; + if (ok) { + // aggregate stats (created/columns_added/columns_widened) + // are batch-wide by nature; rows/errors are reported + // per-request against each request's own input, matching + // what a non-batched call would have returned. + b->stats = batch_stats; + b->stats.rows = b->rows; + b->stats.errors = b->parse_errors; + } + } + + guard.lock(); + coord.cv.notify_all(); + // loop: more requests may have queued up while we were + // committing: drain them too instead of waking a new leader. + } + coord.committing.erase(target); + } + guard.unlock(); + + if (item->failed) { + throw InvalidInputException(item->error_message); + } + return item->stats; +} + // Streaming handle over RawIngestor for the INSERT-syntax path namespace { class RawStreamIngestorImpl : public RawStreamIngestor {