diff --git a/.github/workflows/MainDistributionPipeline.yml b/.github/workflows/MainDistributionPipeline.yml index e7c45f9..88cf03d 100644 --- a/.github/workflows/MainDistributionPipeline.yml +++ b/.github/workflows/MainDistributionPipeline.yml @@ -24,10 +24,10 @@ concurrency: jobs: duckdb-stable-build: name: Build extension binaries - uses: duckdb/extension-ci-tools/.github/workflows/_extension_distribution.yml@v1.5-variegata + uses: duckdb/extension-ci-tools/.github/workflows/_extension_distribution.yml@v1.5.5 with: duckdb_version: v1.5.5 - ci_tools_version: v1.5-variegata + ci_tools_version: v1.5.5 extension_name: rawduck # Windows is excluded until the VS2026 runner image is compatible with # duckdb v1.5's vendored fmt (stdext::checked_array_iterator was @@ -36,9 +36,9 @@ jobs: code-quality-check: name: Code Quality Check - uses: duckdb/extension-ci-tools/.github/workflows/_extension_code_quality.yml@v1.5-variegata + uses: duckdb/extension-ci-tools/.github/workflows/_extension_code_quality.yml@v1.5.5 with: duckdb_version: v1.5.5 - ci_tools_version: v1.5-variegata + ci_tools_version: v1.5.5 extension_name: rawduck format_checks: 'format;tidy' diff --git a/.github/workflows/QuackIntegration.yml b/.github/workflows/QuackIntegration.yml index aa2d250..735536b 100644 --- a/.github/workflows/QuackIntegration.yml +++ b/.github/workflows/QuackIntegration.yml @@ -21,10 +21,10 @@ concurrency: jobs: quack-integration: name: rawduck:quack (linux_amd64) - uses: duckdb/extension-ci-tools/.github/workflows/_extension_distribution.yml@v1.5-variegata + uses: duckdb/extension-ci-tools/.github/workflows/_extension_distribution.yml@v1.5.5 with: duckdb_version: v1.5.5 - ci_tools_version: v1.5-variegata + ci_tools_version: v1.5.5 extension_name: rawduck opt_in_archs: 'linux_amd64' use_merged_vcpkg_manifest: '1' diff --git a/.github/workflows/Release.yml b/.github/workflows/Release.yml index 6a7acf0..811a5a5 100644 --- a/.github/workflows/Release.yml +++ b/.github/workflows/Release.yml @@ -53,10 +53,10 @@ env: jobs: build: name: Build extension binaries - uses: duckdb/extension-ci-tools/.github/workflows/_extension_distribution.yml@v1.5-variegata + uses: duckdb/extension-ci-tools/.github/workflows/_extension_distribution.yml@v1.5.5 with: duckdb_version: v1.5.5 - ci_tools_version: v1.5-variegata + ci_tools_version: v1.5.5 extension_name: rawduck # build exactly the four published platforms: linux/osx x amd64/arm64. # windows stays excluded for the VS2026/fmt incompatibility (see diff --git a/AGENTS.md b/AGENTS.md index 09b92fb..dce5deb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -11,9 +11,11 @@ real typed columns at ingest (one-time cost) so every query runs at native colum instead of storing opaque JSON strings and paying `->>` extraction on every scan (45–265× slower, see BENCHMARK.md). -Pinned to DuckDB **v1.5.5**: the `duckdb/` submodule commit and the versions in -`.github/workflows/MainDistributionPipeline.yml` must stay in sync. Build with `GEN=ninja make -release`; test with `./build/release/test/unittest --test-dir . "test/sql/*"`; format with +Pinned to DuckDB **v1.5.5** (latest stable as of this pin): the `duckdb/` submodule +tag, `extension-ci-tools` on the matching `v1.5.5` tip, and `duckdb_version` / +`ci_tools_version` in `.github/workflows/{MainDistributionPipeline,Release,QuackIntegration}.yml` +must stay in sync. Build with `GEN=ninja make release`; test with +`./build/release/test/unittest --test-dir . "test/sql/*"`; format with `make format-fix` (CI enforces it). ## Source map (src/) @@ -133,6 +135,8 @@ sqllogictests in `test/sql/`: `rawduck.test` (core types/records), `raw_ingest.t (stores, transactions, persistence), `raw_api.test` (server lifecycle), `ducklake.test` (`require ducklake`, skips when absent). OTEL ingest performance: `./scripts/benchmark/run_otel.sh` + `./scripts/benchmark/compare.sh` (see `scripts/benchmark/README.md` and `BENCHMARK.md`). +VARIANT vs RawDuck (v1.5.5, traces): `./scripts/benchmark/run_variant.sh` on branch +`feat/variant-benchmark`. Every feature needs: happy path, evolution interaction, error case, and—for anything that can return wrong data—a proof test (e.g. tampering with a projection to prove the rewrite engaged). `raw_ingest` output is `(table, created, columns_added, columns_widened, rows, errors)`; diff --git a/BENCHMARK.md b/BENCHMARK.md index 0d0e6e7..94fccd1 100644 --- a/BENCHMARK.md +++ b/BENCHMARK.md @@ -1,119 +1,135 @@ # RawDuck Benchmark -RawDuck's bet: shred schema-less event JSON into real typed columns at ingest so every later query -runs at native columnar speed, instead of keeping opaque JSON and paying `->>` extraction on every -scan. The primary benchmark is the realistic workload — **OTEL telemetry** (OTLP/JSON logs, metrics, -traces) — with the GH Archive run kept below as a historical wide-schema stress test. +Primary workload: **OTEL telemetry** (OTLP/JSON logs, metrics, traces). GH Archive is a +wide-schema stress test in the appendix. + +All published numbers: DuckDB **v1.5.5**, default RawDuck settings, Apple **M3 Ultra** +(32 cores, 512 GiB) unless noted. Cold ingest = first `raw_ingest_file` in a fresh database; +warm = second ingest in the same process after `DELETE` (`columns_added = 0`). Report the +best of N sessions unless noted. See `scripts/benchmark/README.md` for metric definitions. ### Harness ```sh GEN=ninja make release -./scripts/benchmark/run_otel.sh --quick # CI / remote smoke (100k) -./scripts/benchmark/compare.sh benchmark/results/smoke.json # vs committed baseline -./scripts/benchmark/run_otel.sh --records 1000000 --runs 5 # full publishable numbers -``` -Each session is **one DuckDB process**: cold ingest (schema discovery) then warm ingest (same -process, fresh timestamps, `columns_added = 0`). See `scripts/benchmark/README.md`. +# OTEL bulk ingest (NDJSON file) +./scripts/benchmark/run_otel.sh --quick +./scripts/benchmark/run_otel.sh --records 1000000 --runs 5 + +# OTEL streaming ingest (OpenTelemetry SDK → raw_serve HTTP) +./scripts/benchmark/run_otel_streaming.sh --quick +./scripts/benchmark/run_otel_streaming.sh --workers 16 --spans-per-worker 60000 -## OTEL ingestion (primary) +# VARIANT vs RawDuck (same trace dataset) +./scripts/benchmark/run_variant.sh --quick +./scripts/benchmark/run_variant.sh --records 1000000 --runs 3 +``` + +## OTEL bulk ingest -Published results — multi-core commodity hardware, DuckDB v1.5.5, 1,000,000 records per signal, -OTLP/JSON export envelopes (the exact bytes an OpenTelemetry Collector posts to an OTLP/HTTP json -endpoint), default settings (no manual tuning): +1,000,000 records per signal, OTLP/JSON export envelopes (collector POST bodies), best of 5 +sessions: -| signal | records | columns | source NDJSON | ingest | records/s | throughput | on disk | -|---|---:|---:|---:|---:|---:|---:|---:| -| traces | 1,000,000 | 15 | 435 MB | 0.74 s | 1.35M | 586 MB/s | 50 MB | -| logs | 1,000,000 | 11 | 294 MB | 0.80 s | 1.26M | 369 MB/s | 9 MB | -| metrics | 1,000,000 | 9 | 353 MB | 1.03 s | 970k | 342 MB/s | 88 MB | +| signal | records | source NDJSON | cold ingest | records/s | throughput | +|---|---:|---:|---:|---:|---:| +| traces | 1,000,000 | 435 MB | 1.19 s | 841k | 366 MB/s | +| logs | 1,000,000 | 294 MB | 0.87 s | 1.15M | 338 MB/s | +| metrics | 1,000,000 | 353 MB | 1.13 s | 889k | 314 MB/s | -3M telemetry records shredded into typed columns in **2.6 s (~1.2M records/s)**. The OTLP transform -explodes the nested `resource → scope → record` envelopes, flattens KeyValue attributes -(`http.status_code`, `service.name`, …) into typed columns, and normalizes byte ids to hex — all in -the parallel parse stage. Because each NDJSON line is a fat export envelope that explodes into many -records, the loader batches by bytes (not just line count) so the parse threads stay fed -automatically; no `batch_size` tuning is needed. +3M telemetry records in **3.2 s** (~940k records/s average). Warm ingest matches cold within +~2% on each signal. ### Query speed (1,000,000 spans) -Same spans, identical results — shredded typed columns vs the "just keep the JSON" baseline (one -JSON object per span, queried with `->>`): +Same spans — shredded typed columns vs one JSON object per span (`->>`), best of 3 runs: | query | JSON `->>` | RawDuck | speedup | |---|---:|---:|---:| -| error count by service (`status>=500`) | 71 ms | 2 ms | 36× | -| p99 latency by route | 136 ms | 5 ms | 27× | -| status-code distribution | 63 ms | 1 ms | 63× | -| storage | 232 MB | 38 MB | 6× smaller | - -This baseline is the *favorable* one. Real OTLP keeps attributes as KeyValue **arrays**, so querying -them without shredding means `UNNEST`-ing the `resource → scope → span` nesting and scanning each -attribute array by key — hundreds of times slower than reading a typed column. +| error count by service (`status>=500`) | 39 ms | 1.5 ms | 26× | +| p99 latency by route | 99 ms | 3.2 ms | 31× | +| status-code distribution | 35 ms | 2.5 ms | 14× | +| storage | 143 MB | 39.5 MB | 3.6× smaller | ### Reproduce -Generate OTLP/JSON envelopes (one `Export*ServiceRequest` per line): - -```python -# gen_otlp.py -> python3 gen_otlp.py 1000000 -import json, os, random, sys -random.seed(11) -SVC = ["checkout","cart","payments","search","auth","inventory","shipping","frontend"] -RT = ["/api/v1/orders","/api/v1/cart","/api/v1/pay","/api/v1/search","/login","/health"] -def kv(k,v): - if isinstance(v,bool): return {"key":k,"value":{"boolValue":v}} - if isinstance(v,int): return {"key":k,"value":{"intValue":str(v)}} - if isinstance(v,float):return {"key":k,"value":{"doubleValue":v}} - return {"key":k,"value":{"stringValue":str(v)}} -def res(s): return {"attributes":[kv("service.name",s),kv("deployment.environment","production"), - kv("cloud.region","us-east-1"),kv("host.name","pod-%d"%random.randint(1,400))]} -def span(ts): - st=random.choice([200,200,200,201,400,404,500]); d=random.randint(2*10**5,8*10**8) - return {"traceId":os.urandom(16).hex(),"spanId":os.urandom(8).hex(),"name":random.choice(RT), - "kind":random.randint(1,5),"startTimeUnixNano":str(ts),"endTimeUnixNano":str(ts+d), - "attributes":[kv("http.method",random.choice(["GET","POST","PUT","DELETE"])), - kv("http.route",random.choice(RT)),kv("http.status_code",st), - kv("retry",random.choice([True,False]))],"status":{"code":2 if st>=500 else 1}} -def gen(name,total,per,rec,wrap): - ts=1_700_000_000_000_000_000; w=0 - with open(name+".ndjson","w") as f: - while w1 else 1_000_000 -gen("traces",n,80,span,lambda s,r:{"resourceSpans":[{"resource":res(s),"scopeSpans":[{"spans":r}]}]}) -# logs/metrics envelopes follow the same shape with resourceLogs.scopeLogs.logRecords / -# resourceMetrics.scopeMetrics.metrics (see the repo's bench scripts for the full generator) +```sh +./scripts/benchmark/run_otel.sh --records 1000000 --runs 5 +./scripts/benchmark/run_variant.sh --records 1000000 --runs 3 # queries + storage above ``` -Ingest and query: +## OTEL streaming ingest -```sql --- bduck otel.db -.timer on -CALL raw_ingest_file('traces', 'traces.ndjson', transform := 'otlp-traces'); -- and otlp-logs / otlp-metrics -CHECKPOINT; +Real OTLP/HTTP **protobuf** traffic via the OpenTelemetry Python SDK into `raw_serve()` +(concurrent exporter processes, not bulk NDJSON): --- baseline: identical spans as a JSON blob per row -CREATE TABLE traces_json AS SELECT to_json(traces)::JSON AS j FROM traces; +| workers | spans | wall | spans/s | +|---:|---:|---:|---:| +| 4 | 20,000 | 0.75 s | 27k | +| 16 | 960,000 | 3.85 s | 250k | + +`rows_ingested` must equal `total_spans_sent` (checked by the harness). + +### Reproduce --- typed columns vs ->> extraction -SELECT "resource.service.name", count(*) FROM traces WHERE "http.status_code" >= 500 GROUP BY 1; -SELECT j->>'resource.service.name', count(*) FROM traces_json - WHERE CAST(j->>'http.status_code' AS BIGINT) >= 500 GROUP BY 1; +```sh +./scripts/benchmark/run_otel_streaming.sh --workers 16 --spans-per-worker 60000 ``` -Run each query three times (against a `-readonly` database) and report the best. +First run creates `benchmark/work/otel-streaming-venv` (OpenTelemetry SDK dependency). + +## VARIANT vs RawDuck (DuckDB v1.5.5) + +Same 1,000,000 OTLP/JSON trace spans. Paths: + +| path | definition | +|---|---| +| RawDuck | `raw_ingest_file(..., transform := 'otlp-traces')` → typed columns | +| VARIANT OTLP | SQL unnest → one `VARIANT` `{resource, span}` per span (KeyValue arrays kept) | +| JSON OTLP | same exploded shape as `JSON` | +| VARIANT-flat | `to_json(traces)::VARIANT` of shredded RawDuck rows (encode/query only) | +| JSON-flat | same shredded rows as `JSON`, queried with `->>` | + +Disk = `used_blocks × block_size` after cold `CHECKPOINT`. Parenthetical file size is after +warm re-ingest (includes free-list holes; not comparable across paths). VARIANT requires +`STORAGE_VERSION 'v1.5.0'`. + +### Ingest + storage (1,000,000 spans) -## Appendix: GH Archive (historical, wide-schema stress test) +| path | M3 Ultra | Spark GB10 aarch64 (`--threads 8`) | +|---|---|---| +| RawDuck | 0.99 s · 1.01M rec/s · 39.5 MB (108 MB file) | 1.18 s · 850k rec/s · 35.5 MB (91 MB file) | +| VARIANT-flat | encode · 35.8 MB | encode · 39.5 MB | +| VARIANT OTLP | 11.96 s · 84k · 53.5 MB (106 MB file) | 7.96 s · 126k · 54.5 MB (114 MB file) | +| JSON-flat | encode · 143 MB | encode · 142 MB | +| JSON OTLP | 4.72 s · 212k · 241 MB (484 MB file) | 5.71 s · 175k · 242 MB (484 MB file) | -One hour of real [GH Archive](https://www.gharchive.org/) data — 247,199 events / 956 MB NDJSON -exploding to a **914-column** schema. This is the worst case for shredding (extreme, sparse schema -churn), kept as a stress test rather than the representative workload. +Spark GB10: 20 cores, 122 GiB, Linux aarch64. -Published results (Apple Silicon, 10 cores, DuckDB v1.5.5): +### Queries (best of 3, ms) + +| encoding | M3 Ultra | Spark GB10 | +|---|---|---| +| | errors / p99 / status | errors / p99 / status | +| RawDuck | 1.5 / 3.2 / 2.5 | 1.3 / 4.9 / 5.1 | +| JSON-flat | 39 / 99 / 35 | 65 / 136 / 63 | +| JSON OTLP positional | 213 / 303 / 193 | 253 / 415 / 215 | +| JSON OTLP key lookup | 344 / 418 / 304 | 397 / 580 / 366 | +| VARIANT-flat | 436 / 1225 / 416 | 700 / 1988 / 696 | +| VARIANT OTLP positional | 1227 / 3479 / 1162 | 2013 / 5947 / 1852 | +| VARIANT OTLP key lookup | 1493 / 3658 / 1404 | 2107 / 5996 / 2049 | + +### Reproduce + +```sh +./scripts/benchmark/run_variant.sh --records 1000000 --runs 3 +./scripts/benchmark/run_variant.sh --records 1000000 --runs 3 --threads 8 # many-core ARM +``` + +## Appendix: GH Archive (wide-schema stress test) + +One hour of [GH Archive](https://www.gharchive.org/) data — 247,199 events / 956 MB NDJSON / +914 columns. Apple Silicon, 10 cores, DuckDB v1.5.5: | | JSON column | RawDuck | | |---|---:|---:|---| @@ -122,73 +138,23 @@ Published results (Apple Silicon, 10 cores, DuckDB v1.5.5): | distinct repos per actor | 457 ms | 10 ms | 46× | | sum of push payload sizes | 265 ms | 1 ms | 265× | | events per minute | 236 ms | 3 ms | 79× | -| ingest | 1.4 s | ~13 s | one-time cost | +| cold ingest | 1.4 s | ~13 s | one-time cost | +| warm re-ingest | — | ~4.9 s | steady state | | storage | 1.05 GB | 636 MB | 40% smaller | +### Reproduce + ```sh -curl -sL https://data.gharchive.org/2024-01-15-10.json.gz -o gh.json.gz # raw_ingest_file reads .gz directly +curl -sL https://data.gharchive.org/2024-01-15-10.json.gz -o gh.json.gz ``` ```sql --- RawDuck: one call shreds the whole hour (914 typed columns, evolution included) CALL raw_ingest_file('gh_events', 'gh.json.gz'); CHECKPOINT; - --- baseline keeps raw JSON (records='false' alone still infers a STRUCT; the columns clause keeps it raw) -CREATE TABLE gh_raw AS SELECT json - FROM read_json('gh.json.gz', format='newline_delimited', records='false', columns={json: 'JSON'}); -CHECKPOINT; -``` - -RawDuck queries / baseline (`->>`) queries: - -```sql -SELECT type, count(*) AS n FROM gh_events GROUP BY type ORDER BY n DESC; -SELECT "repo.name", count(*) AS n FROM gh_events WHERE type='PushEvent' GROUP BY 1 ORDER BY n DESC LIMIT 10; -SELECT "actor.login", count(DISTINCT "repo.name") AS r FROM gh_events GROUP BY 1 ORDER BY r DESC LIMIT 10; -SELECT sum("payload.size") FROM gh_events WHERE type='PushEvent'; -SELECT date_trunc('minute', created_at) AS m, count(*) FROM gh_events GROUP BY m ORDER BY m; - -SELECT json->>'$.type' AS type, count(*) AS n FROM gh_raw GROUP BY type ORDER BY n DESC; -SELECT json->>'$.repo.name' AS repo, count(*) AS n FROM gh_raw WHERE json->>'$.type'='PushEvent' GROUP BY 1 ORDER BY n DESC LIMIT 10; -SELECT json->>'$.actor.login' AS a, count(DISTINCT json->>'$.repo.name') AS r FROM gh_raw GROUP BY 1 ORDER BY r DESC LIMIT 10; -SELECT sum(CAST(json->>'$.payload.size' AS BIGINT)) FROM gh_raw WHERE json->>'$.type'='PushEvent'; -SELECT date_trunc('minute', CAST(json->>'$.created_at' AS TIMESTAMP)) AS m, count(*) FROM gh_raw GROUP BY m ORDER BY m; -``` - -### Warm-table ingest - -The cold GH numbers include schema discovery (CREATE + evolution sync points). Re-ingesting the same -hour into the already-evolved table runs fully parallel: **~4.9 s** — the steady-state rate once a -table's shape has stabilized (the realistic OTEL case, where the schema is stable after warmup). - -### INSERT-syntax streaming (fastest path) - -`INSERT INTO raw.ingest.t SELECT ...` streams any SQL source through a parallel zero-copy sink: - -```sql -ATTACH 'rawduck:store.db' AS raw; -INSERT INTO raw.ingest.narrow SELECT '{"a":' || range || '}' FROM range(5000000); --- 5M narrow rows in ~0.8 s (~6.1M rows/s) -``` - -### Adaptive layout and projections - -```sql -CALL raw_stats(); -CALL raw_optimize('gh_events'); -- physically reorders by hottest columns -CALL raw_project('gh_events'); -- materializes the hottest aggregation -SET rawduck_use_projections = true; -- transparent rewrite of eligible count(*) queries ``` ## Pitfalls -- Don't split NDJSON with Python's `splitlines()` — it splits on `\u2028`/`\u2029` which appear raw - inside real-world strings and corrupts records. Split on `\n` only. -- The shell reports query times with `.timer on`; dot-commands don't work via `duckdb -c`, use - `-f script.sql`. -- A shallow duckdb submodule clone without tags makes the shell report `v0.0.1`; fetch the release - tag (`git -C duckdb fetch --depth 1 origin tag v1.5.5`) or extension installs 404. -- For large imports where each line is a fat container (OTLP envelopes, CloudWatch log groups), the - loader auto-parallelizes via byte-aware batching; you only need `batch_size` to *raise* the line - cap for very small flat records. +- Split NDJSON on `\n` only (not `splitlines()` — `\u2028`/`\u2029` appear in strings). +- Use `.timer on` via `duckdb -f script.sql` (not `duckdb -c`). +- Shallow duckdb clones without tags report `v0.0.1`; fetch tag `v1.5.5`. diff --git a/docs/UPDATING.md b/docs/UPDATING.md index a3ac73e..6e371ca 100644 --- a/docs/UPDATING.md +++ b/docs/UPDATING.md @@ -4,12 +4,17 @@ will inevitably come a time when a new DuckDB is released and the extension repo as follows: - Bump submodules - - `./duckdb` should be set to latest tagged release - - `./extension-ci-tools` should be set to updated branch corresponding to latest DuckDB release. So if you're building for DuckDB `v1.1.0` there will be a branch in `extension-ci-tools` named `v1.1.0` to which you should check out. + - `./duckdb` should be set to latest tagged release (currently `v1.5.5`) + - `./extension-ci-tools` should be set to the matching branch tip. Prefer the + exact patch branch (`v1.5.5`) when it exists; otherwise the line branch + (`v1.5-variegata`). Keep the submodule SHA in sync with that tip. - Bump versions in `./github/workflows` - - `duckdb_version` input in `duckdb-stable-build` job in `MainDistributionPipeline.yml` should be set to latest tagged release - - `duckdb_version` input in `duckdb-stable-deploy` job in `MainDistributionPipeline.yml` should be set to latest tagged release - - the reusable workflow `duckdb/extension-ci-tools/.github/workflows/_extension_distribution.yml` for the `duckdb-stable-build` job should be set to latest tagged release + - `duckdb_version` / `ci_tools_version` in `MainDistributionPipeline.yml`, + `Release.yml`, and `QuackIntegration.yml` should be the latest tagged release + - the reusable workflow refs + (`duckdb/extension-ci-tools/.github/workflows/_extension_*.yml@…`) should use + the same tag/branch (e.g. `@v1.5.5`), not a moving alias unless intentional + - `DUCKDB_VERSION` in `Release.yml` must match (GitHub Pages path + CLI download) # API changes DuckDB extensions built with this extension template are built against the internal C++ API of DuckDB. This API is not guaranteed to be stable. diff --git a/extension-ci-tools b/extension-ci-tools index 7faa44c..72e76e9 160000 --- a/extension-ci-tools +++ b/extension-ci-tools @@ -1 +1 @@ -Subproject commit 7faa44c030fb3e368d9cf1de49fe884609fa72a4 +Subproject commit 72e76e99cd7fee45a99739cd118ec2db64e034ec diff --git a/scripts/benchmark/README.md b/scripts/benchmark/README.md index 2213a5c..3ef4aef 100644 --- a/scripts/benchmark/README.md +++ b/scripts/benchmark/README.md @@ -92,5 +92,104 @@ python3 scripts/benchmark/gen_otlp.py traces 1000000 benchmark/data 170008640000 | `run_otel_session.py` | Single-process cold→warm driver (host timing) | | `run_otel.sh` | Orchestrator: data gen, N sessions, JSON output (bash) | | `compare.sh` | Regression gate vs `baseline-otel.json` (bash) | +| `run_variant.sh` / `run_variant.py` | VARIANT (DuckDB v1.5) vs RawDuck ingest + query + storage | +| `run_otel_streaming.sh` / `.py` / `otel_gen_load.py` | Real OTLP/HTTP streaming ingestion via the actual OpenTelemetry SDK (see below) | Requires: bash, python3, a release build (`build/release/duckdb` + extension). + +## Realistic OTEL streaming ingestion (`run_otel_streaming.sh`) + +`run_otel.sh` and `run_variant.sh` both bulk-load one big NDJSON file — useful +for measuring the shredding/ingest engine in isolation, but not how OTEL data +actually arrives. This benchmark instead drives real traffic through +`raw_serve()`'s HTTP OTLP endpoint using the **actual OpenTelemetry Python +SDK** (`TracerProvider` + `BatchSpanProcessor` + `OTLPSpanExporter`, real +protobuf wire format — not RawDuck's own NDJSON generator), from multiple +concurrent *processes* (sidesteps Python's GIL, models multiple +services/collectors exporting to the same collector hub concurrently). + +```sh +GEN=ninja make release +./scripts/benchmark/run_otel_streaming.sh --quick # smoke: 4 workers x 5k spans +./scripts/benchmark/run_otel_streaming.sh --workers 16 --spans-per-worker 60000 # ~180k spans/sec on a 20-core arm64 box +``` + +First run creates a dedicated venv (`benchmark/work/otel-streaming-venv`) with +the OpenTelemetry SDK — a plain system-wide `pip install` is refused on +externally-managed Python installs (PEP 668), so this is required, not +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). + +## VARIANT vs RawDuck (branch `feat/variant-benchmark`) + +Compares DuckDB **VARIANT as of v1.5.5** (not the v2.0 shredded-execution preview) +against RawDuck typed columns on the same OTLP/JSON traces envelopes. + +```sh +# after: git clone --recurse-submodules … && git checkout feat/variant-benchmark +# GEN=ninja make release +./scripts/benchmark/run_variant.sh --quick +./scripts/benchmark/run_variant.sh --records 1000000 --runs 3 \ + --output "benchmark/results/variant_1m_$(hostname -s)_$(date -u +%Y%m%dT%H%M%SZ).json" +``` + +`--quick` is 100k records / 1 ingest session (sanity). Publishable numbers use +1M records and best-of-3 ingest sessions. + +DuckDB is **CPU-only** (a CUDA GPU does not accelerate this). On many-core ARM +hosts, default `threads = nproc` often makes VARIANT extract / nested `UNNEST` +thrash — pin workers: + +```sh +./scripts/benchmark/run_variant.sh --quick --threads 8 +./scripts/benchmark/run_variant.sh --records 1000000 --runs 3 --threads 8 +# still stuck on queries: skip honest KeyValue lookups +./scripts/benchmark/run_variant.sh --records 1000000 --runs 3 --threads 8 --skip-kv +``` + +### Linux / arm64 hang (DuckDB 1.5 CLI) + +If the process sits idle with no CPU at `ingest …` (especially right after +startup), that is usually the DuckDB **v1.5 CLI stdin/color-detection stall**, +not RawDuck. The harness now passes `-dark-mode`, sets `DUCKDB_NO_HIGHLIGHT=1`, +pre-creates the v1.5.0 DB with a one-shot `-c` (no interactive `ATTACH`), and +reads stdout from a dedicated thread. + +If CPU is pegged on `variant_otlp` / `variant_envelope`, that is VARIANT +shredding fat OTLP rows (slow, not a hang). Envelope ingest stays off by +default; use `--paths rawduck,variant_otlp,json_otlp,variant_flat,json_flat`. + +Send back: the JSON file. It already embeds `git_commit`, `duckdb_version`, and a +`host` block (CPU, cores, RAM, OS). Envelope ingest rows are **not** span +records — the JSON labels grain so rec/s is not mixed. + +Paths in the result: + +| path | what it measures | +|---|---| +| `rawduck` | `raw_ingest_file` + typed columns (`otlp-traces`) | +| `variant_envelope` | *(off by default)* one VARIANT per NDJSON line. Fat OTLP envelopes can hang for hours on Linux aarch64; pass `--paths …,variant_envelope` to include. | +| `variant_otlp` | SQL unnest → one VARIANT `{resource, span}` per span (KeyValue arrays kept) | +| `json_otlp` | same exploded shape stored as JSON | +| `variant_flat` / `json_flat` | query/storage encodings of already-shredded RawDuck rows (not an ingest path) | + +Queries: error count by service, p99 latency by route, status-code distribution. +`*_pos` uses generator-stable attribute indexes; `*_kv` does honest key lookup. diff --git a/scripts/benchmark/otel_gen_load.py b/scripts/benchmark/otel_gen_load.py new file mode 100755 index 0000000..7e8fcc2 --- /dev/null +++ b/scripts/benchmark/otel_gen_load.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +"""Generate real OTLP traffic via the actual OpenTelemetry Python SDK and +export it to a RawDuck raw_serve() endpoint. + +This is deliberately the real SDK (TracerProvider, BatchSpanProcessor, +OTLPSpanExporter) rather than RawDuck's own NDJSON generator: it exercises +the exact wire format (OTLP/HTTP protobuf, the SDK default) and envelope +shape a production OpenTelemetry Collector/SDK actually sends. + +Each invocation is one process = one TracerProvider = one exporter. Run +multiple processes concurrently (see run_otel_streaming.py) to generate +load beyond what a single process's GIL allows. +""" +from __future__ import annotations + +import argparse +import random +import sys +import time + +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.trace import SpanKind, StatusCode + +SERVICES = ["checkout", "cart", "payments", "search", "auth", "inventory", "shipping", "frontend"] +ROUTES = ["/api/v1/orders", "/api/v1/cart", "/api/v1/pay", "/api/v1/search", "/login", "/health"] +METHODS = ["GET", "POST", "PUT", "DELETE"] +STATUSES = [200, 200, 200, 201, 400, 404, 500] + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--endpoint", required=True, help="http://host:port/otlp/v1/traces") + ap.add_argument("--token", required=True) + ap.add_argument("--count", type=int, required=True, help="spans to emit from this process") + ap.add_argument("--service", default=None, help="fix the service name (else random per process)") + ap.add_argument("--seed", type=int, default=0) + ap.add_argument("--batch-size", type=int, default=2048, help="OTLP export batch size") + ap.add_argument("--queue-size", type=int, default=200_000, help="BatchSpanProcessor max_queue_size") + ap.add_argument("--schedule-delay-ms", type=int, default=200) + args = ap.parse_args() + + random.seed(args.seed) + service = args.service or random.choice(SERVICES) + resource = Resource.create( + { + "service.name": service, + "deployment.environment": "production", + "cloud.region": "us-east-1", + "host.name": f"pod-{random.randint(1, 400)}", + } + ) + exporter = OTLPSpanExporter(endpoint=args.endpoint, headers={"Authorization": f"Bearer {args.token}"}) + processor = BatchSpanProcessor( + exporter, + max_queue_size=args.queue_size, + max_export_batch_size=args.batch_size, + schedule_delay_millis=args.schedule_delay_ms, + ) + provider = TracerProvider(resource=resource) + provider.add_span_processor(processor) + tracer = provider.get_tracer("rawduck-bench") + + start = time.perf_counter() + for _ in range(args.count): + route = random.choice(ROUTES) + status = random.choice(STATUSES) + with tracer.start_as_current_span(route, kind=SpanKind.SERVER) as span: + span.set_attribute("http.method", random.choice(METHODS)) + span.set_attribute("http.route", route) + span.set_attribute("http.status_code", status) + span.set_attribute("retry", random.choice([True, False])) + if status >= 500: + span.set_status(StatusCode.ERROR) + gen_elapsed = time.perf_counter() - start + + flush_start = time.perf_counter() + provider.force_flush() + provider.shutdown() + flush_elapsed = time.perf_counter() - flush_start + + total = time.perf_counter() - start + print( + f"generated={args.count} gen_s={gen_elapsed:.3f} flush_s={flush_elapsed:.3f} total_s={total:.3f}", + file=sys.stderr, + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/benchmark/otel_streaming_requirements.txt b/scripts/benchmark/otel_streaming_requirements.txt new file mode 100644 index 0000000..0dcbdcd --- /dev/null +++ b/scripts/benchmark/otel_streaming_requirements.txt @@ -0,0 +1,2 @@ +opentelemetry-sdk +opentelemetry-exporter-otlp-proto-http diff --git a/scripts/benchmark/run_otel_session.py b/scripts/benchmark/run_otel_session.py index 28ee008..afd071d 100755 --- a/scripts/benchmark/run_otel_session.py +++ b/scripts/benchmark/run_otel_session.py @@ -12,6 +12,7 @@ """ from __future__ import annotations +import os import subprocess import sys import time @@ -21,6 +22,13 @@ def escape_sql_path(path: str) -> str: return path.replace("'", "''") +# DuckDB 1.5 CLI: skip terminal color / highlight probes on piped stdin (Linux stalls). +_DUCKDB_ENV = { + **os.environ, + "DUCKDB_NO_HIGHLIGHT": "1", +} + + def _read_until_done(stdout) -> list[str]: lines: list[str] = [] while True: @@ -38,7 +46,10 @@ def _read_until_done(stdout) -> list[str]: def _exec(proc: subprocess.Popen, sql: str) -> list[str]: assert proc.stdin is not None and proc.stdout is not None - proc.stdin.write((sql.rstrip() + "\nSELECT '__bench_done__';\n").encode()) + body = sql.rstrip() + if not body.endswith(";"): + body += ";" + proc.stdin.write((body + "\nSELECT '__bench_done__';\n").encode()) proc.stdin.flush() return _read_until_done(proc.stdout) @@ -65,10 +76,13 @@ def run_session( ext_sql = escape_sql_path(ext) proc = subprocess.Popen( - [duckdb, db_path, "-unsigned", "-batch", "-csv", "-noheader"], + # -dark-mode: skip OSC-11 /dev/tty color probe (DuckDB 1.5 CLI stall on Linux pipes) + [duckdb, db_path, "-unsigned", "-batch", "-csv", "-noheader", "-dark-mode"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + env=_DUCKDB_ENV, + bufsize=0, ) assert proc.stdin is not None and proc.stdout is not None diff --git a/scripts/benchmark/run_otel_streaming.py b/scripts/benchmark/run_otel_streaming.py new file mode 100755 index 0000000..f0d2bff --- /dev/null +++ b/scripts/benchmark/run_otel_streaming.py @@ -0,0 +1,232 @@ +#!/usr/bin/env python3 +"""Realistic OTEL streaming ingestion benchmark. + +Unlike run_otel.sh (bulk NDJSON file load), this drives real OTLP/HTTP +traffic — via the actual OpenTelemetry Python SDK, protobuf wire format, +the SDK's default BatchSpanProcessor batching — into RawDuck's in-process +HTTP API (raw_serve). Concurrent worker *processes* (not threads: sidesteps +Python's GIL) each run an independent TracerProvider/exporter, modeling +multiple services/collectors exporting to the same RawDuck endpoint. + +Requires the packages in otel_streaming_requirements.txt; run_otel_streaming.sh +manages a venv for this automatically. +""" +from __future__ import annotations + +import argparse +import json +import os +import platform +import subprocess +import sys +import time +import urllib.request +from datetime import datetime, timezone +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +GEN_SCRIPT = Path(__file__).resolve().parent / "otel_gen_load.py" + +_ENV = {**os.environ, "DUCKDB_NO_HIGHLIGHT": "1"} + + +def _env_path(name: str, default: Path) -> Path: + return Path(os.environ.get(name, str(default))) + + +def duckdb_bin() -> Path: + return _env_path("DUCKDB", ROOT / "build/release/duckdb") + + +def extension_path() -> Path: + return _env_path("EXT", ROOT / "build/release/extension/rawduck/rawduck.duckdb_extension") + + +def work_dir() -> Path: + return _env_path("BENCH_WORK", ROOT / "benchmark/work") + + +def results_dir() -> Path: + return _env_path("BENCH_RESULTS", ROOT / "benchmark/results") + + +def escape_sql(s: str) -> str: + return s.replace("'", "''") + + +class Server: + """One interactive `duckdb` CLI process running raw_serve() for the + benchmark's duration. Commands are piped over stdin (the CLI never sees + EOF, so the listener thread stays alive) and responses are read back off + stdout up to a marker line — the same pattern run_variant.py's DuckSession + uses for cold->warm sessions, adapted for a long-lived server process. + """ + + def __init__(self, db_path: Path, host: str, port: int, token: str): + db_path.parent.mkdir(parents=True, exist_ok=True) + for p in (db_path, Path(str(db_path) + ".wal")): + if p.exists(): + p.unlink() + argv = [str(duckdb_bin()), str(db_path), "-unsigned", "-batch", "-dark-mode"] + self.proc = subprocess.Popen( + argv, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + env=_ENV, + bufsize=0, + text=True, + ) + self.host, self.port, self.token = host, port, token + self._send(f"LOAD '{escape_sql(str(extension_path()))}';") + self._send(f"CALL raw_serve(host := '{host}', port := {port}, token := '{token}');") + self._wait_healthy() + + def _send(self, sql: str) -> None: + assert self.proc.stdin is not None + self.proc.stdin.write(sql.rstrip() + "\n") + self.proc.stdin.flush() + + def _wait_healthy(self, timeout_s: float = 10.0) -> None: + deadline = time.monotonic() + timeout_s + url = f"http://{self.host}:{self.port}/health" + while time.monotonic() < deadline: + try: + with urllib.request.urlopen(url, timeout=1) as resp: + if resp.status == 200: + return + except Exception: + time.sleep(0.2) + raise RuntimeError("raw_serve did not become healthy in time") + + def query(self, sql: str) -> str: + assert self.proc.stdin is not None and self.proc.stdout is not None + marker = "__RAWDUCK_BENCH_DONE__" + self._send(f"{sql}\nSELECT '{marker}';") + lines = [] + while True: + line = self.proc.stdout.readline() + if not line: + break + line = line.rstrip("\n") + if marker in line: + break + lines.append(line) + return "\n".join(lines) + + def scalar_count(self, table: str) -> int: + # -csv-ish minimal parse: last non-empty, non-box-drawing line with digits + out = self.query(f"SELECT count(*) FROM {table};") + for line in reversed(out.splitlines()): + stripped = "".join(ch for ch in line if ch.isdigit()) + if stripped and stripped.isdigit(): + return int(stripped) + raise RuntimeError(f"could not parse row count from: {out!r}") + + def close(self) -> None: + try: + self._send("CALL raw_serve_stop();") + time.sleep(0.3) + assert self.proc.stdin is not None + self.proc.stdin.close() + self.proc.wait(timeout=10) + except Exception: + self.proc.kill() + + +def host_info() -> dict: + try: + cores = os.cpu_count() or 0 + except Exception: + cores = 0 + return {"host": platform.node(), "cpu": platform.machine(), "cores": cores} + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--workers", type=int, default=16, help="concurrent exporter processes") + ap.add_argument("--spans-per-worker", type=int, default=60000) + ap.add_argument("--quick", action="store_true", help="small smoke run: 4 workers x 5000 spans") + ap.add_argument("--port", type=int, default=19999) + ap.add_argument("--token", default="rt_bench_secret") + ap.add_argument("--python", default=os.environ.get("OTEL_VENV_PYTHON", sys.executable)) + ap.add_argument("--output", default=None, help="write JSON results here") + args = ap.parse_args() + + if args.quick: + args.workers, args.spans_per_worker = 4, 5000 + + if not duckdb_bin().exists() or not extension_path().exists(): + print("Build release first: GEN=ninja make release", file=sys.stderr) + return 1 + if not Path(args.python).exists(): + print(f"Missing python interpreter: {args.python}", file=sys.stderr) + return 1 + + db_path = work_dir() / f"otel_streaming_{os.getpid()}.db" + server = Server(db_path, "127.0.0.1", args.port, args.token) + try: + endpoint = f"http://127.0.0.1:{args.port}/otlp/v1/traces" + argvs = [ + [ + args.python, str(GEN_SCRIPT), + "--endpoint", endpoint, + "--token", args.token, + "--count", str(args.spans_per_worker), + "--seed", str(i), + ] + for i in range(args.workers) + ] + + start = time.perf_counter() + procs = [subprocess.Popen(argv, stderr=subprocess.PIPE, text=True) for argv in argvs] + failures = [] + for i, p in enumerate(procs): + _, err = p.communicate(timeout=300) + if p.returncode != 0: + failures.append((i, err)) + elapsed = time.perf_counter() - start + + if failures: + for i, err in failures: + print(f"worker {i} failed:\n{err}", file=sys.stderr) + return 1 + + total_spans = args.workers * args.spans_per_worker + rows = server.scalar_count("otel_traces") + spans_per_sec = total_spans / elapsed if elapsed > 0 else 0.0 + + result = { + "benchmark": "otel_streaming", + "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + "transport": "otlp/http/protobuf", + "workers": args.workers, + "spans_per_worker": args.spans_per_worker, + "total_spans_sent": total_spans, + "rows_ingested": rows, + "wall_seconds": round(elapsed, 3), + "spans_per_sec": round(spans_per_sec), + **host_info(), + } + print(json.dumps(result, indent=2)) + + if rows != total_spans: + print(f"WARNING: sent {total_spans} spans but only {rows} rows landed", file=sys.stderr) + + output = args.output + if output is None: + ts = result["timestamp"].replace(":", "") + output = str(results_dir() / f"otel_streaming_{total_spans}_{host_info()['host']}_{ts}.json") + Path(output).parent.mkdir(parents=True, exist_ok=True) + Path(output).write_text(json.dumps(result, indent=2) + "\n") + print(f"Wrote {output}", file=sys.stderr) + return 0 if rows == total_spans else 1 + finally: + server.close() + for p in (db_path, Path(str(db_path) + ".wal")): + if p.exists(): + p.unlink() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/benchmark/run_otel_streaming.sh b/scripts/benchmark/run_otel_streaming.sh new file mode 100755 index 0000000..da39b26 --- /dev/null +++ b/scripts/benchmark/run_otel_streaming.sh @@ -0,0 +1,39 @@ +#!/usr/bin/env bash +# Realistic OTEL streaming ingestion benchmark: drives real OTLP/HTTP traffic +# (actual OpenTelemetry Python SDK, protobuf wire format) into RawDuck's +# raw_serve() HTTP API with concurrent exporter processes, instead of bulk- +# loading an NDJSON file (see run_otel.sh for that). +# +# Manages a dedicated venv (not the system python: PEP 668 externally-managed +# environments reject a plain `pip install`) for the OpenTelemetry SDK deps. +# +# Examples: +# ./scripts/benchmark/run_otel_streaming.sh --quick +# ./scripts/benchmark/run_otel_streaming.sh --workers 16 --spans-per-worker 60000 +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +# shellcheck source=scripts/benchmark/lib.sh +source "${ROOT}/scripts/benchmark/lib.sh" + +bench_require_build + +VENV_DIR="$(bench_work_dir)/otel-streaming-venv" +REQS="${ROOT}/scripts/benchmark/otel_streaming_requirements.txt" + +if [[ ! -x "${VENV_DIR}/bin/python3" ]]; then + echo "Setting up OTel SDK venv at ${VENV_DIR} (one-time)..." >&2 + python3 -m venv "${VENV_DIR}" + "${VENV_DIR}/bin/pip" install --quiet --upgrade pip + "${VENV_DIR}/bin/pip" install --quiet -r "${REQS}" +fi + +export OTEL_VENV_PYTHON="${VENV_DIR}/bin/python3" +export DUCKDB="$(bench_duckdb)" +export EXT="$(bench_extension)" +export BENCH_WORK="$(bench_work_dir)" +export BENCH_RESULTS="$(bench_results_dir)" +mkdir -p "${BENCH_WORK}" "${BENCH_RESULTS}" + +exec "${VENV_DIR}/bin/python3" "${ROOT}/scripts/benchmark/run_otel_streaming.py" \ + --python "${OTEL_VENV_PYTHON}" "$@" diff --git a/scripts/benchmark/run_variant.py b/scripts/benchmark/run_variant.py new file mode 100755 index 0000000..dcc2fde --- /dev/null +++ b/scripts/benchmark/run_variant.py @@ -0,0 +1,1051 @@ +#!/usr/bin/env python3 +"""Compare DuckDB VARIANT (v1.5) against RawDuck shredded tables. + +Same OTLP/JSON NDJSON envelopes as run_otel.sh. VARIANT is measured as it +ships in DuckDB v1.5.5 (no v2.0 shredded execution / extraction pushdown). + +Ingest grain is labeled explicitly: envelope rows are not span records. + +Query encodings: + rawduck typed columns after otlp-traces shred + variant_otlp one VARIANT {resource, span} per span (KeyValue arrays kept) + variant_otlp_pos same table, positional attribute extract (generator-stable) + variant_otlp_kv same table, key lookup via list comprehension (honest OTLP) + json_otlp / _pos / _kv same shape stored as JSON + variant_flat VARIANT of the shredded RawDuck row (query/storage only) + json_flat JSON of the shredded RawDuck row (existing ->> baseline) + +VARIANT columns require storage v1.5.0; every path uses that so sizes are comparable. +""" +from __future__ import annotations + +import argparse +import json +import os +import platform +import queue +import shutil +import subprocess +import sys +import threading +import time +from datetime import datetime, timezone +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] + +# DuckDB 1.5 CLI: terminal color / highlight probes can stall piped stdin on +# Linux (and some remote TTYs). Force a non-interactive color scheme and skip +# highlight detection — see duckdb/duckdb#21243 and the CLI troubleshooting guide. +_DUCKDB_ENV = { + **os.environ, + "DUCKDB_NO_HIGHLIGHT": "1", +} + + +def escape_sql(path: str) -> str: + return path.replace("'", "''") + + +def _env_path(name: str, default: Path) -> Path: + return Path(os.environ.get(name, str(default))) + + +def duckdb_bin() -> Path: + return _env_path("DUCKDB", ROOT / "build/release/duckdb") + + +def extension_path() -> Path: + return _env_path("EXT", ROOT / "build/release/extension/rawduck/rawduck.duckdb_extension") + + +def data_dir() -> Path: + return _env_path("BENCH_DATA", ROOT / "benchmark/data") + + +def work_dir() -> Path: + return _env_path("BENCH_WORK", ROOT / "benchmark/work") + + +def results_dir() -> Path: + return _env_path("BENCH_RESULTS", ROOT / "benchmark/results") + + +def _duckdb_argv(binary: Path, db_path: Path | None = None) -> list[str]: + argv = [str(binary)] + if db_path is not None: + argv.append(str(db_path)) + # -dark-mode skips OSC-11 /dev/tty color probing; -batch forces non-TTY input. + argv.extend(["-unsigned", "-batch", "-csv", "-noheader", "-dark-mode"]) + return argv + + +def create_v15_database(binary: Path, db_path: Path) -> None: + """Create an empty storage-v1.5.0 file in a one-shot process (no interactive ATTACH).""" + db_path = Path(db_path) + db_path.parent.mkdir(parents=True, exist_ok=True) + for p in (db_path, Path(str(db_path) + ".wal")): + if p.exists(): + p.unlink() + sql = f"ATTACH '{escape_sql(str(db_path))}' AS bench (STORAGE_VERSION 'v1.5.0');" + # Prefer -c over interactive stdin: avoids the v1.5 piped-script stall path. + subprocess.run( + _duckdb_argv(binary) + ["-c", sql], + check=True, + env=_DUCKDB_ENV, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + timeout=60, + ) + + +def explode_cte(file_sql: str) -> str: + return f""" +WITH envelopes AS ( + SELECT unnest(resourceSpans) AS rs + FROM read_json('{file_sql}', format='newline_delimited') +), scopes AS ( + SELECT rs.resource AS resource, unnest(rs.scopeSpans) AS ss + FROM envelopes +), spans AS ( + SELECT resource, unnest(ss.spans) AS span + FROM scopes +) +""" + + +# Query SQL per (query, encoding). Each must return columns (k, n). +QUERY_SQL = { + "errors_by_service": { + "rawduck": """ +SELECT "resource.service.name" AS k, count(*) AS n +FROM traces +WHERE "http.status_code" >= 500 +GROUP BY 1 ORDER BY 1 +""", + "variant_otlp_pos": """ +SELECT CAST(payload.resource.attributes[1].value.stringValue AS VARCHAR) AS k, count(*) AS n +FROM t +WHERE CAST(payload.span.attributes[3].value.intValue AS BIGINT) >= 500 +GROUP BY 1 ORDER BY 1 +""", + "variant_otlp_kv": """ +SELECT CAST(( + [x->'value'->>'stringValue' FOR x IN CAST(payload.resource.attributes AS JSON[]) + IF x->>'key' = 'service.name'] +)[1] AS VARCHAR) AS k, count(*) AS n +FROM t +WHERE CAST(( + [x->'value'->>'intValue' FOR x IN CAST(payload.span.attributes AS JSON[]) + IF x->>'key' = 'http.status_code'] +)[1] AS BIGINT) >= 500 +GROUP BY 1 ORDER BY 1 +""", + "json_otlp_pos": """ +SELECT CAST(payload->'resource'->'attributes'->0->'value'->>'stringValue' AS VARCHAR) AS k, + count(*) AS n +FROM t +WHERE CAST(payload->'span'->'attributes'->2->'value'->>'intValue' AS BIGINT) >= 500 +GROUP BY 1 ORDER BY 1 +""", + "json_otlp_kv": """ +SELECT CAST(( + [x->'value'->>'stringValue' FOR x IN CAST(payload->'resource'->'attributes' AS JSON[]) + IF x->>'key' = 'service.name'] +)[1] AS VARCHAR) AS k, count(*) AS n +FROM t +WHERE CAST(( + [x->'value'->>'intValue' FOR x IN CAST(payload->'span'->'attributes' AS JSON[]) + IF x->>'key' = 'http.status_code'] +)[1] AS BIGINT) >= 500 +GROUP BY 1 ORDER BY 1 +""", + "variant_flat": """ +SELECT CAST(payload['resource.service.name'] AS VARCHAR) AS k, count(*) AS n +FROM t +WHERE CAST(payload['http.status_code'] AS BIGINT) >= 500 +GROUP BY 1 ORDER BY 1 +""", + "json_flat": """ +SELECT j->>'resource.service.name' AS k, count(*) AS n +FROM t +WHERE CAST(j->>'http.status_code' AS BIGINT) >= 500 +GROUP BY 1 ORDER BY 1 +""", + }, + "p99_by_route": { + "rawduck": """ +SELECT "http.route" AS k, + quantile_cont(("endTimeUnixNano" - "startTimeUnixNano"), 0.99)::BIGINT AS n +FROM traces +GROUP BY 1 ORDER BY 1 +""", + "variant_otlp_pos": """ +SELECT CAST(payload.span.attributes[2].value.stringValue AS VARCHAR) AS k, + quantile_cont( + CAST(payload.span.endTimeUnixNano AS BIGINT) + - CAST(payload.span.startTimeUnixNano AS BIGINT), 0.99)::BIGINT AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + "variant_otlp_kv": """ +SELECT CAST(( + [x->'value'->>'stringValue' FOR x IN CAST(payload.span.attributes AS JSON[]) + IF x->>'key' = 'http.route'] +)[1] AS VARCHAR) AS k, + quantile_cont( + CAST(payload.span.endTimeUnixNano AS BIGINT) + - CAST(payload.span.startTimeUnixNano AS BIGINT), 0.99)::BIGINT AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + "json_otlp_pos": """ +SELECT CAST(payload->'span'->'attributes'->1->'value'->>'stringValue' AS VARCHAR) AS k, + quantile_cont( + CAST(payload->'span'->>'endTimeUnixNano' AS BIGINT) + - CAST(payload->'span'->>'startTimeUnixNano' AS BIGINT), 0.99)::BIGINT AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + "json_otlp_kv": """ +SELECT CAST(( + [x->'value'->>'stringValue' FOR x IN CAST(payload->'span'->'attributes' AS JSON[]) + IF x->>'key' = 'http.route'] +)[1] AS VARCHAR) AS k, + quantile_cont( + CAST(payload->'span'->>'endTimeUnixNano' AS BIGINT) + - CAST(payload->'span'->>'startTimeUnixNano' AS BIGINT), 0.99)::BIGINT AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + "variant_flat": """ +SELECT CAST(payload['http.route'] AS VARCHAR) AS k, + quantile_cont( + CAST(payload.endTimeUnixNano AS BIGINT) + - CAST(payload.startTimeUnixNano AS BIGINT), 0.99)::BIGINT AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + "json_flat": """ +SELECT j->>'http.route' AS k, + quantile_cont( + CAST(j->>'endTimeUnixNano' AS BIGINT) + - CAST(j->>'startTimeUnixNano' AS BIGINT), 0.99)::BIGINT AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + }, + "status_dist": { + "rawduck": """ +SELECT "http.status_code"::VARCHAR AS k, count(*) AS n +FROM traces +GROUP BY 1 ORDER BY 1 +""", + "variant_otlp_pos": """ +SELECT CAST(payload.span.attributes[3].value.intValue AS VARCHAR) AS k, count(*) AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + "variant_otlp_kv": """ +SELECT CAST(( + [x->'value'->>'intValue' FOR x IN CAST(payload.span.attributes AS JSON[]) + IF x->>'key' = 'http.status_code'] +)[1] AS VARCHAR) AS k, count(*) AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + "json_otlp_pos": """ +SELECT CAST(payload->'span'->'attributes'->2->'value'->>'intValue' AS VARCHAR) AS k, count(*) AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + "json_otlp_kv": """ +SELECT CAST(( + [x->'value'->>'intValue' FOR x IN CAST(payload->'span'->'attributes' AS JSON[]) + IF x->>'key' = 'http.status_code'] +)[1] AS VARCHAR) AS k, count(*) AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + "variant_flat": """ +SELECT CAST(payload['http.status_code'] AS VARCHAR) AS k, count(*) AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + "json_flat": """ +SELECT j->>'http.status_code' AS k, count(*) AS n +FROM t +GROUP BY 1 ORDER BY 1 +""", + }, +} + +# Expected (groups, sum_n) for verification. p99 sum_n is not a row count. +QUERY_EXPECT = { + "errors_by_service": {"groups": 8, "sum_n": None, "sum_is_error_rows": True}, + "p99_by_route": {"groups": 6, "sum_n": None, "sum_is_error_rows": False}, + "status_dist": {"groups": 5, "sum_n": None, "sum_is_error_rows": False}, +} + + +class DuckSession: + """Interactive DuckDB CLI session (cold→warm needs one process). + + Hardening for Linux/arm64 (DuckDB v1.5 CLI): + - Always open an existing v1.5.0 file (pre-created via create_v15_database); + never ATTACH STORAGE_VERSION over an interactive pipe. + - Pass -dark-mode + DUCKDB_NO_HIGHLIGHT=1 so color probing cannot stall. + - Unbuffered pipes + a stdout reader thread (no select on BufferedReader). + """ + + def __init__(self, binary: Path, ext: Path, db_path: Path, *, create: bool): + self.db_path = Path(db_path) + self._err_chunks: list[bytes] = [] + self._lines: queue.Queue = queue.Queue() + self._reader_done = threading.Event() + self._last_sql = "" + + if create: + create_v15_database(binary, self.db_path) + + argv = _duckdb_argv(binary, self.db_path) + self.proc = subprocess.Popen( + argv, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=_DUCKDB_ENV, + bufsize=0, # unbuffered: avoid Linux pipe + BufferedReader stalls + ) + if self.proc.stdin is None or self.proc.stdout is None: + raise RuntimeError("failed to open duckdb pipes") + threading.Thread(target=self._drain_stderr, daemon=True).start() + threading.Thread(target=self._read_stdout, daemon=True).start() + self.exec("SET enable_progress_bar = false;") + threads = os.environ.get("DUCKDB_THREADS", "").strip() + if threads: + self.exec(f"SET threads = {int(threads)};") + self.exec(f"LOAD '{escape_sql(str(ext))}';") + + def _drain_stderr(self) -> None: + assert self.proc.stderr is not None + while True: + chunk = self.proc.stderr.read(4096) + if not chunk: + break + self._err_chunks.append(chunk) + + def _read_stdout(self) -> None: + assert self.proc.stdout is not None + try: + while True: + raw = self.proc.stdout.readline() + if not raw: + break + self._lines.put(raw.decode(errors="replace").rstrip("\n\r")) + finally: + self._reader_done.set() + self._lines.put(None) + + def _read_until_done(self, timeout_sec: float = 1800.0) -> list[str]: + lines: list[str] = [] + deadline = time.monotonic() + timeout_sec + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RuntimeError( + "timed out waiting for __bench_done__" + + (f"\nlast_sql={self._last_sql[:200]!r}" if self._last_sql else "") + + (f"\n{self.stderr_text()}" if self.stderr_text() else "") + ) + try: + text = self._lines.get(timeout=min(remaining, 1.0)) + except queue.Empty: + if self.proc.poll() is not None and self._reader_done.is_set(): + raise RuntimeError( + "duckdb process ended before __bench_done__" + + (f"\n{self.stderr_text()}" if self.stderr_text() else "") + ) + continue + if text is None: + raise RuntimeError( + "duckdb process ended before __bench_done__" + + (f"\n{self.stderr_text()}" if self.stderr_text() else "") + ) + text = text.strip() + if not text: + continue + if text == "__bench_done__" or text.startswith("__bench_done__,"): + break + lines.append(text) + return lines + + def exec(self, sql: str) -> list[str]: + assert self.proc.stdin is not None + body = sql.rstrip() + if not body.endswith(";"): + body += ";" + self._last_sql = body + payload = (body + "\nSELECT '__bench_done__';\n").encode() + self.proc.stdin.write(payload) + self.proc.stdin.flush() + return self._read_until_done() + + def stderr_text(self) -> str: + return b"".join(self._err_chunks).decode(errors="replace") + + def close(self) -> str: + if self.proc.poll() is None: + try: + assert self.proc.stdin is not None + self.proc.stdin.close() + except Exception: + pass + try: + self.proc.wait(timeout=60) + except subprocess.TimeoutExpired: + self.proc.kill() + self.proc.wait() + return self.stderr_text() + + def fail(self, message: str) -> None: + err = self.close() + if err: + sys.stderr.write(err) + if not err.endswith("\n"): + sys.stderr.write("\n") + raise RuntimeError(message) + + +def parse_count(lines: list[str]) -> int: + for line in reversed(lines): + token = line.split(",")[0].strip() + if token.lstrip("-").isdigit(): + return int(token) + raise RuntimeError(f"no count in output: {lines!r}") + + +def parse_ingest(lines: list[str]) -> tuple[int, int, int, int]: + for line in reversed(lines): + parts = line.split(",") + if len(parts) >= 4 and parts[0].lstrip("-").isdigit(): + return int(parts[0]), int(parts[1]), int(parts[2]), int(parts[3]) + raise RuntimeError(f"no ingest result row in output: {lines!r}") + + +def parse_db_size(lines: list[str]) -> dict: + """Parse total_blocks, used_blocks, free_blocks, block_size from pragma_database_size.""" + for line in reversed(lines): + parts = [p.strip() for p in line.split(",")] + if len(parts) >= 4 and all(p.lstrip("-").isdigit() for p in parts[:4]): + total, used, free, block = (int(p) for p in parts[:4]) + return { + "total_blocks": total, + "used_blocks": used, + "free_blocks": free, + "block_size": block, + "used_bytes": used * block, + } + raise RuntimeError(f"no database size row in output: {lines!r}") + + +DB_SIZE_SQL = ( + "SELECT total_blocks, used_blocks, free_blocks, block_size " + "FROM pragma_database_size() WHERE total_blocks > 0 ORDER BY used_blocks DESC LIMIT 1;" +) + + +def parse_fingerprint(lines: list[str]) -> tuple[int, int]: + """Parse `groups,sum_n` from the fingerprint SELECT.""" + for line in reversed(lines): + parts = [p.strip() for p in line.split(",")] + if len(parts) >= 2 and parts[0].lstrip("-").isdigit() and parts[1].lstrip("-").isdigit(): + return int(parts[0]), int(parts[1]) + raise RuntimeError(f"no fingerprint in output: {lines!r}") + + +def file_bytes(path: Path) -> int: + return path.stat().st_size if path.exists() else 0 + + +def ingest_sql(path_name: str, file_sql: str) -> str: + if path_name == "rawduck": + return f""" +SELECT rows, columns_added, columns_widened, errors +FROM raw_ingest_file('traces', '{file_sql}', transform := 'otlp-traces'); +CHECKPOINT; +""" + if path_name == "variant_envelope": + return f""" +CREATE TABLE IF NOT EXISTS t (payload VARIANT); +INSERT INTO t +SELECT json::VARIANT +FROM read_json('{file_sql}', format='newline_delimited', records='false', columns={{json:'JSON'}}); +CHECKPOINT; +""" + cte = explode_cte(file_sql) + if path_name == "variant_otlp": + return f""" +CREATE TABLE IF NOT EXISTS t (payload VARIANT); +INSERT INTO t +{cte} +SELECT {{'resource': resource, 'span': span}}::VARIANT AS payload FROM spans; +CHECKPOINT; +""" + if path_name == "json_otlp": + return f""" +CREATE TABLE IF NOT EXISTS t (payload JSON); +INSERT INTO t +{cte} +SELECT {{'resource': resource, 'span': span}}::JSON AS payload FROM spans; +CHECKPOINT; +""" + raise ValueError(path_name) + + +def create_empty_sql(path_name: str) -> str: + if path_name == "rawduck": + return "-- table created by raw_ingest_file" + if path_name == "variant_envelope": + return "CREATE TABLE t (payload VARIANT);" + if path_name == "variant_otlp": + return "CREATE TABLE t (payload VARIANT);" + if path_name == "json_otlp": + return "CREATE TABLE t (payload JSON);" + raise ValueError(path_name) + + +def count_sql(path_name: str) -> str: + table = "traces" if path_name == "rawduck" else "t" + return f"SELECT count(*) FROM {table};" + + +def delete_sql(path_name: str) -> str: + table = "traces" if path_name == "rawduck" else "t" + return f"DELETE FROM {table};" + + +def query_table_for(encoding: str) -> str | None: + if encoding == "rawduck": + return "rawduck" + if encoding.startswith("variant_otlp"): + return "variant_otlp" + if encoding.startswith("json_otlp"): + return "json_otlp" + if encoding == "variant_flat": + return "variant_flat" + if encoding == "json_flat": + return "json_flat" + return None + + +def run_ingest_session( + binary: Path, + ext: Path, + db_path: Path, + path_name: str, + cold_file: Path, + warm_file: Path, +) -> dict: + db_path.parent.mkdir(parents=True, exist_ok=True) + if db_path.exists(): + db_path.unlink() + wal = Path(str(db_path) + ".wal") + if wal.exists(): + wal.unlink() + + sess = DuckSession(binary, ext, db_path, create=True) + cold_sql_file = escape_sql(str(cold_file)) + warm_sql_file = escape_sql(str(warm_file)) + try: + if path_name != "rawduck": + sess.exec(create_empty_sql(path_name)) + + t0 = time.perf_counter() + cold_lines = sess.exec(ingest_sql(path_name, cold_sql_file)) + cold_sec = time.perf_counter() - t0 + if path_name == "rawduck": + cold_rows, added, widened, errors = parse_ingest(cold_lines) + else: + cold_rows = parse_count(sess.exec(count_sql(path_name))) + added = widened = errors = 0 + cold_size = parse_db_size(sess.exec(DB_SIZE_SQL)) + + sess.exec(delete_sql(path_name)) + + t1 = time.perf_counter() + warm_lines = sess.exec(ingest_sql(path_name, warm_sql_file)) + warm_sec = time.perf_counter() - t1 + if path_name == "rawduck": + warm_rows, warm_added, warm_widened, _warm_errors = parse_ingest(warm_lines) + if warm_added != 0 or warm_widened != 0: + raise RuntimeError( + f"warm ingest mutated schema (columns_added={warm_added}, " + f"columns_widened={warm_widened}); shape must be stable" + ) + else: + warm_rows = parse_count(sess.exec(count_sql(path_name))) + warm_size = parse_db_size(sess.exec(DB_SIZE_SQL)) + except Exception as exc: + sess.fail(f"{path_name} ingest failed: {exc}") + raise + err = sess.close() + if err and "Error" in err: + sys.stderr.write(err) + + return { + "path": path_name, + "grain": "envelope" if path_name == "variant_envelope" else "span", + "cold_seconds": round(cold_sec, 6), + "warm_seconds": round(warm_sec, 6), + "cold_rows": cold_rows, + "warm_rows": warm_rows, + "columns_added": added, + "columns_widened": widened, + "errors": errors, + "file_bytes": file_bytes(db_path), + "bytes": cold_size["used_bytes"], + "used_bytes": cold_size["used_bytes"], + "free_blocks": cold_size["free_blocks"], + "used_blocks": cold_size["used_blocks"], + "block_size": cold_size["block_size"], + "note_storage": ( + "bytes/used_bytes is live data after cold CHECKPOINT (before DELETE). " + "file_bytes is the file after warm re-ingest and may include free-list holes." + ), + "db": str(db_path), + "warm_size": warm_size, + } + + +def encode_flat( + binary: Path, + ext: Path, + src_db: Path, + dst_db: Path, + encoding: str, +) -> dict: + if dst_db.exists(): + dst_db.unlink() + sess = DuckSession(binary, ext, dst_db, create=True) + src = escape_sql(str(src_db)) + try: + sess.exec(f"ATTACH '{src}' AS src (READ_ONLY);") + if encoding == "variant_flat": + sql = "CREATE TABLE t AS SELECT to_json(src.traces)::VARIANT AS payload FROM src.traces; CHECKPOINT;" + elif encoding == "json_flat": + sql = "CREATE TABLE t AS SELECT to_json(src.traces)::JSON AS j FROM src.traces; CHECKPOINT;" + else: + raise ValueError(encoding) + t0 = time.perf_counter() + sess.exec(sql) + encode_sec = time.perf_counter() - t0 + rows = parse_count(sess.exec("SELECT count(*) FROM t;")) + size = parse_db_size(sess.exec(DB_SIZE_SQL)) + except Exception as exc: + sess.fail(f"{encoding} encode failed: {exc}") + raise + sess.close() + return { + "path": encoding, + "grain": "span", + "encode_seconds": round(encode_sec, 6), + "rows": rows, + "bytes": size["used_bytes"], + "used_bytes": size["used_bytes"], + "file_bytes": file_bytes(dst_db), + "used_blocks": size["used_blocks"], + "free_blocks": size["free_blocks"], + "block_size": size["block_size"], + "db": str(dst_db), + "note": "query/storage encoding of already-shredded RawDuck rows; not an OTLP ingest path", + } + + +def run_queries( + binary: Path, + ext: Path, + db_path: Path, + encoding: str, + query_runs: int, + expected_rows: int, + error_rows: int | None, +) -> dict: + sess = DuckSession(binary, ext, db_path, create=False) + out: dict = {} + try: + for qname, variants in QUERY_SQL.items(): + sql = variants.get(encoding) + if not sql: + continue + print(f" {encoding} / {qname}...", file=sys.stderr, flush=True) + wrapped = f"SELECT count(*) AS groups, coalesce(sum(n), 0)::BIGINT AS sum_n FROM ({sql}) q" + # warmup + sess.exec(wrapped) + best = None + groups = sum_n = 0 + for _ in range(query_runs): + t0 = time.perf_counter() + lines = sess.exec(wrapped) + sec = time.perf_counter() - t0 + groups, sum_n = parse_fingerprint(lines) + if best is None or sec < best: + best = sec + expect = QUERY_EXPECT[qname] + ok = groups == expect["groups"] + if qname == "status_dist": + ok = ok and sum_n == expected_rows + elif qname == "errors_by_service" and error_rows is not None: + ok = ok and sum_n == error_rows + out[qname] = { + "seconds": round(best or 0.0, 6), + "ms": round((best or 0.0) * 1000.0, 3), + "groups": groups, + "sum_n": sum_n, + "ok": ok, + } + except Exception as exc: + sess.fail(f"{encoding} query failed: {exc}") + raise + sess.close() + return out + + +def git_info() -> tuple[str, str]: + def _run(args: list[str]) -> str: + try: + return subprocess.check_output(args, cwd=ROOT, text=True).strip() + except Exception: + return "unknown" + + return _run(["git", "rev-parse", "HEAD"]), _run(["git", "rev-parse", "--abbrev-ref", "HEAD"]) + + +def duckdb_version(binary: Path) -> str: + try: + out = subprocess.check_output( + [str(binary), "-unsigned", "-csv", "-noheader", "-c", "SELECT library_version FROM pragma_version();"], + text=True, + ) + return out.strip().splitlines()[-1] + except Exception: + return "unknown" + + +def host_info() -> dict: + info: dict = { + "hostname": platform.node(), + "os": platform.system(), + "os_release": platform.release(), + "machine": platform.machine(), + "processor": platform.processor(), + "python": platform.python_version(), + "cpu_count": os.cpu_count(), + } + try: + if sys.platform == "darwin": + info["ram_bytes"] = int( + subprocess.check_output(["sysctl", "-n", "hw.memsize"], text=True).strip() + ) + info["cpu_brand"] = subprocess.check_output( + ["sysctl", "-n", "machdep.cpu.brand_string"], text=True + ).strip() + elif sys.platform.startswith("linux"): + mem_kb = None + with open("/proc/meminfo", encoding="utf-8") as fh: + for line in fh: + if line.startswith("MemTotal:"): + mem_kb = int(line.split()[1]) + break + if mem_kb is not None: + info["ram_bytes"] = mem_kb * 1024 + cpuinfo = Path("/proc/cpuinfo") + if cpuinfo.is_file(): + for line in cpuinfo.read_text(encoding="utf-8", errors="replace").splitlines(): + if line.lower().startswith("model name"): + info["cpu_brand"] = line.split(":", 1)[1].strip() + break + lscpu = subprocess.run(["lscpu"], capture_output=True, text=True) + if lscpu.returncode == 0: + info["lscpu"] = lscpu.stdout.strip() + except Exception: + pass + if info.get("ram_bytes"): + info["ram_gib"] = round(info["ram_bytes"] / (1024**3), 1) + return info + + +def ensure_traces(records: int) -> tuple[Path, Path]: + data = data_dir() + data.mkdir(parents=True, exist_ok=True) + cold = data / f"traces_{records // 1000}k.ndjson" + warm = data / f"traces_{records // 1000}k_warm.ndjson" + gen = ROOT / "scripts/benchmark/gen_otlp.py" + python = os.environ.get("PYTHON") or sys.executable + if not cold.exists(): + subprocess.check_call([python, str(gen), "traces", str(records), str(data)]) + if not warm.exists(): + subprocess.check_call( + [python, str(gen), "traces", str(records), str(data), "1700086400000000000", "_warm"] + ) + return cold, warm + + +def rec_s(rows: int, seconds: float) -> int: + return int(rows / seconds) if seconds > 0 else 0 + + +def mb_s(nbytes: int, seconds: float) -> float: + return round(nbytes / seconds / 1e6, 1) if seconds > 0 else 0.0 + + +def best_of(runs: list[dict], key: str) -> dict: + return min(runs, key=lambda r: r[key]) + + +def print_summary(doc: dict) -> None: + ingest = doc["ingest"] + print("\n== ingest (best of sessions; CHECKPOINT included) ==", file=sys.stderr) + print( + f"{'path':<20} {'grain':<10} {'rows':>10} {'cold s':>10} {'cold rec/s':>12} {'warm s':>10} {'disk':>10}", + file=sys.stderr, + ) + for name, row in ingest.items(): + rows = row.get("cold_rows") or row.get("rows") or 0 + cold = row.get("cold_seconds") + warm = row.get("warm_seconds") + disk = row["bytes"] / (1 << 20) + file_disk = row.get("file_bytes") + extra = "" + if file_disk and file_disk > row["bytes"] * 1.05: + extra = f" ({file_disk / (1 << 20):.1f} file)" + cold_s = f"{cold:.3f}" if cold is not None else "—" + warm_s = f"{warm:.3f}" if warm is not None else "—" + rate = f"{row.get('cold_records_per_sec') or rec_s(rows, cold or 0):,}" if cold else "—" + print( + f"{name:<20} {row.get('grain','?'):<10} {rows:>10,} {cold_s:>10} {rate:>12} {warm_s:>10} {disk:>8.1f} MB{extra}", + file=sys.stderr, + ) + + print("\n== queries (best of N; ms) ==", file=sys.stderr) + queries = doc["queries"] + qnames = ["errors_by_service", "p99_by_route", "status_dist"] + encodings = list(queries.keys()) + header = f"{'encoding':<22}" + "".join(f"{q[:16]:>16}" for q in qnames) + print(header, file=sys.stderr) + for enc in encodings: + cells = [] + for q in qnames: + cell = queries[enc].get(q) + if not cell: + cells.append(f"{'—':>16}") + else: + flag = "" if cell["ok"] else "!" + cells.append(f"{cell['ms']:.1f}{flag}" .rjust(16)) + print(f"{enc:<22}" + "".join(cells), file=sys.stderr) + print( + "\nVARIANT here is DuckDB v1.5.5 (persisted shredded VARIANT). " + "v2.0 extraction pushdown / shredded execution is not in this pin.", + file=sys.stderr, + ) + + +def merge_best_ingest(acc: dict | None, row: dict) -> dict: + if acc is None: + return dict(row) + if row["cold_seconds"] < acc["cold_seconds"]: + for k in ( + "cold_seconds", + "cold_rows", + "columns_added", + "columns_widened", + "errors", + "bytes", + "used_bytes", + "file_bytes", + "free_blocks", + "used_blocks", + "block_size", + "warm_size", + "note_storage", + "db", + ): + acc[k] = row[k] + if row["warm_seconds"] < acc["warm_seconds"]: + acc["warm_seconds"] = row["warm_seconds"] + acc["warm_rows"] = row["warm_rows"] + return acc + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--records", type=int, default=1_000_000) + parser.add_argument("--runs", type=int, default=3) + parser.add_argument("--query-runs", type=int, default=3) + parser.add_argument("--quick", action="store_true", help="100k records, 1 ingest run") + parser.add_argument("--output", type=str, default="", help="JSON output path (default: benchmark/results/variant___.json)") + parser.add_argument("--skip-kv", action="store_true", help="skip honest KeyValue-lookup queries") + parser.add_argument( + "--threads", + type=int, + default=0, + help="DuckDB worker threads (0 = DuckDB default / all cores). Pin this on many-core ARM.", + ) + parser.add_argument( + "--paths", + type=str, + default="rawduck,variant_otlp,json_otlp,variant_flat,json_flat", + ) + args = parser.parse_args() + if args.quick: + args.records = 100_000 + args.runs = 1 + if args.threads > 0: + os.environ["DUCKDB_THREADS"] = str(args.threads) + + binary = duckdb_bin() + ext = extension_path() + if not binary.is_file() or not os.access(binary, os.X_OK): + print("Build release first: GEN=ninja make release", file=sys.stderr) + return 1 + if not ext.is_file(): + print(f"Missing extension: {ext}", file=sys.stderr) + return 1 + + cold_file, warm_file = ensure_traces(args.records) + src_bytes = file_bytes(cold_file) + work = work_dir() / f"variant_{os.getpid()}" + work.mkdir(parents=True, exist_ok=True) + results_dir().mkdir(parents=True, exist_ok=True) + + ingest_paths = [p.strip() for p in args.paths.split(",") if p.strip()] + encode_paths = [p for p in ingest_paths if p in ("variant_flat", "json_flat")] + timed_ingest = [p for p in ingest_paths if p not in ("variant_flat", "json_flat")] + + commit, branch = git_info() + version = duckdb_version(binary) + host = host_info() + ingest_best: dict[str, dict] = {} + last_dbs: dict[str, Path] = {} + + for r in range(1, args.runs + 1): + for path_name in timed_ingest: + db = work / f"{path_name}_run{r}.db" + print(f"ingest {path_name} run {r}/{args.runs}...", file=sys.stderr) + row = run_ingest_session(binary, ext, db, path_name, cold_file, warm_file) + ingest_best[path_name] = merge_best_ingest(ingest_best.get(path_name), row) + last_dbs[path_name] = db + + encode_info: dict[str, dict] = {} + raw_db = last_dbs.get("rawduck") + if encode_paths and raw_db is None: + print("variant_flat/json_flat require the rawduck path", file=sys.stderr) + return 1 + for encoding in encode_paths: + assert raw_db is not None + db = work / f"{encoding}.db" + print(f"encode {encoding}...", file=sys.stderr) + encode_info[encoding] = encode_flat(binary, ext, raw_db, db, encoding) + last_dbs[encoding] = db + + # Error-row count from rawduck (or first successful errors query). + error_rows = None + query_encodings = [] + for enc_group, src in ( + ("rawduck", "rawduck"), + ("variant_otlp_pos", "variant_otlp"), + ("variant_otlp_kv", "variant_otlp"), + ("json_otlp_pos", "json_otlp"), + ("json_otlp_kv", "json_otlp"), + ("variant_flat", "variant_flat"), + ("json_flat", "json_flat"), + ): + if src in last_dbs and any(enc_group == p or p == src for p in ingest_paths): + if args.skip_kv and enc_group.endswith("_kv"): + continue + query_encodings.append((enc_group, last_dbs[src])) + + queries: dict[str, dict] = {} + for encoding, db in query_encodings: + print(f"query {encoding}...", file=sys.stderr) + q = run_queries(binary, ext, db, encoding, args.query_runs, args.records, error_rows) + if error_rows is None and "errors_by_service" in q and q["errors_by_service"]["ok"]: + error_rows = q["errors_by_service"]["sum_n"] + elif error_rows is not None and "errors_by_service" in q: + q["errors_by_service"]["ok"] = ( + q["errors_by_service"]["ok"] and q["errors_by_service"]["sum_n"] == error_rows + ) + queries[encoding] = q + + ingest_out = {} + for name, row in ingest_best.items(): + recs = args.records if row["grain"] == "span" else row["cold_rows"] + ingest_out[name] = { + **row, + "source_bytes": src_bytes, + "cold_records_per_sec": rec_s(recs, row["cold_seconds"]), + "warm_records_per_sec": rec_s( + args.records if row["grain"] == "span" else row["warm_rows"], + row["warm_seconds"], + ), + "cold_mb_per_sec": mb_s(src_bytes, row["cold_seconds"]), + "warm_mb_per_sec": mb_s(src_bytes, row["warm_seconds"]), + } + if row["grain"] == "envelope": + ingest_out[name]["note"] = ( + "row grain is OTLP export envelopes, not exploded spans; " + "do not compare rec/s to RawDuck without dividing by spans-per-line" + ) + for name, row in encode_info.items(): + ingest_out[name] = row + + ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + host_slug = host.get("hostname", "host").split(".")[0] + ts_file = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + doc = { + "benchmark": "variant_vs_rawduck", + "timestamp": ts, + "git_commit": commit, + "git_branch": branch, + "duckdb_version": version, + "host": host, + "variant_note": ( + "DuckDB VARIANT as of v1.5.5. The v2.0 preview (shredded execution from " + "storage, extraction pushdown, Parquet shred, extra variant_* functions) " + "is not in this pin." + ), + "records": args.records, + "ingest_runs": args.runs, + "query_runs": args.query_runs, + "threads": int(os.environ["DUCKDB_THREADS"]) if os.environ.get("DUCKDB_THREADS") else None, + "source": str(cold_file), + "source_bytes": src_bytes, + "storage_version": "v1.5.0", + "ingest": ingest_out, + "queries": queries, + } + + out_path = ( + Path(args.output) + if args.output + else results_dir() / f"variant_{args.records}_{host_slug}_{ts_file}.json" + ) + out_path.parent.mkdir(parents=True, exist_ok=True) + out_path.write_text(json.dumps(doc, indent=2) + "\n") + print(f"Wrote {out_path}", file=sys.stderr) + print_summary(doc) + print( + f"\nSend back: {out_path}\n" + f" git={commit} branch={branch} duckdb={version}\n" + f" host={host.get('hostname')} cpu={host.get('cpu_brand') or host.get('machine')} " + f"cores={host.get('cpu_count')} ram={host.get('ram_gib', '?')} GiB", + file=sys.stderr, + ) + + shutil.rmtree(work, ignore_errors=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/benchmark/run_variant.sh b/scripts/benchmark/run_variant.sh new file mode 100755 index 0000000..8b0728d --- /dev/null +++ b/scripts/benchmark/run_variant.sh @@ -0,0 +1,45 @@ +#!/usr/bin/env bash +# VARIANT (DuckDB v1.5) vs RawDuck shredded tables. +# +# Same OTLP/JSON traces envelopes as run_otel.sh. Reports ingest, query, and +# on-disk size. VARIANT is measured as it exists in the pinned DuckDB (v1.5.5), +# not the v2.0 shredded-execution preview. +# +# Remote / new machine (post-build), from this branch: +# +# git clone --recurse-submodules … && git checkout feat/variant-benchmark +# GEN=ninja make release +# ./scripts/benchmark/run_variant.sh --quick +# ./scripts/benchmark/run_variant.sh --records 1000000 --runs 3 +# +# Send back the JSON under benchmark/results/variant_*.json (host, git, and +# DuckDB version are already inside the file). +# +# Examples: +# ./scripts/benchmark/run_variant.sh --quick +# ./scripts/benchmark/run_variant.sh --records 1000000 --runs 3 +set -euo pipefail + +export PATH="/bin:/usr/bin:/usr/local/bin:${HOME}/.pyenv/shims:${PATH:-}" +PYTHON="${PYTHON:-$(command -v python3 || true)}" +if [[ -z "${PYTHON}" || ! -x "${PYTHON}" ]]; then + echo "python3 not found (set PYTHON=...)" >&2 + exit 1 +fi + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +# shellcheck source=scripts/benchmark/lib.sh +source "${ROOT}/scripts/benchmark/lib.sh" + +bench_require_build + +export DUCKDB +export EXT +DUCKDB="$(bench_duckdb)" +EXT="$(bench_extension)" +export BENCH_DATA BENCH_WORK BENCH_RESULTS +BENCH_DATA="$(bench_data_dir)" +BENCH_WORK="$(bench_work_dir)" +BENCH_RESULTS="$(bench_results_dir)" + +exec "${PYTHON}" "${ROOT}/scripts/benchmark/run_variant.py" "$@" diff --git a/src/include/raw_functions.hpp b/src/include/raw_functions.hpp index 64b4fed..83fb761 100644 --- a/src/include/raw_functions.hpp +++ b/src/include/raw_functions.hpp @@ -38,6 +38,17 @@ struct RawIngestStats { }; RawIngestStats RawIngestPayload(ClientContext &context, const string &target, const string &payload, const RawParseOptions &options); +// 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 +// on every attempt. +RawIngestStats RawIngestParsedPayload(ClientContext &context, const string &target, shared_ptr parsed, + const string &payload, const RawParseOptions &options); +// HTTP/programmatic ingest entry point: manages conn's transaction itself +// and serializes a table's first-ever insert against concurrent racers +// (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); TableFunction GetRawServeFunction(); TableFunction GetRawServeStopFunction(); diff --git a/src/include/raw_json.hpp b/src/include/raw_json.hpp index 2d5c060..c27f515 100644 --- a/src/include/raw_json.hpp +++ b/src/include/raw_json.hpp @@ -35,7 +35,12 @@ struct RawNode { unordered_map child_lookup; unique_ptr element; - RawNode &GetOrCreateChild(const string &key); + RawNode &GetOrCreateChild(const char *key, idx_t key_len); + // Routing-only lookup (schema tree already built): linear-scans `children` + // for small fanout (the common case: no allocation, no hashing) and falls + // back to `child_lookup` once a node has enough children that scanning + // would lose. Returns nullptr if `key` isn't a known child. + const RawNode *FindChild(const char *key, idx_t key_len) const; }; // A flattened output column: `name` is the dotted path, `path` the segments to @@ -68,6 +73,17 @@ struct RawPayload { idx_t parse_errors = 0; // apply OTLP semantic normalization during Explode bool otlp_semantics = false; + // Backs every yyjson_read call made by Parse(): one malloc sized for the + // whole payload's worst-case memory need (yyjson_read_max_memory_usage) + // instead of one malloc/free pair per NDJSON line — every doc in `docs` + // then bump-allocates from this single buffer instead of glibc's arena. + // A vector, not a single pointer: small schema-stable batches get merged + // (MergeParsedPayloads moves `docs` from one RawPayload into another), + // and every merged-in doc's yyjson_alc still points at the pool buffer + // its own payload allocated — that buffer must move (not free) with it, + // or freeing the donor payload frees memory the surviving docs still use. + // Freed after all docs are (~RawPayload). + vector pool_buffers; RawPayload() = default; RawPayload(const RawPayload &) = delete; diff --git a/src/raw_api.cpp b/src/raw_api.cpp index 0fd3025..d03a6f9 100644 --- a/src/raw_api.cpp +++ b/src/raw_api.cpp @@ -259,12 +259,11 @@ void HandleIngest(const duckdb_httplib::Request &req, duckdb_httplib::Response & return; } } - conn.BeginTransaction(); try { auto options = otlp_signal.empty() ? RequestParseOptions(*conn.context, req, body) : ResolveTransform(*conn.context, "otlp-" + otlp_signal, ""); - auto stats = RawIngestPayload(*conn.context, table, body, options); - conn.Commit(); + auto parsed = RawParsedPayload::Process(body, options); + auto stats = RawIngestSerialized(conn, table, std::move(parsed), body, options); JsonDoc json; auto root = duckdb_yyjson::yyjson_mut_obj(json.doc); if (!otlp_signal.empty()) { @@ -322,10 +321,9 @@ void HandleOtlpProtobuf(const duckdb_httplib::Request &req, duckdb_httplib::Resp res.set_content(RawOtlpProtobufResponse(signal, 0, ""), "application/x-protobuf"); return; } - conn.BeginTransaction(); try { - auto stats = RawIngestPayload(*conn.context, table, payload, options); - conn.Commit(); + auto parsed = RawParsedPayload::Process(payload, options); + auto stats = RawIngestSerialized(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" : ""), diff --git a/src/raw_ingest.cpp b/src/raw_ingest.cpp index 56ea9bc..2d72271 100644 --- a/src/raw_ingest.cpp +++ b/src/raw_ingest.cpp @@ -9,6 +9,7 @@ #include "duckdb/catalog/catalog_entry/table_catalog_entry.hpp" #include "duckdb/common/case_insensitive_map.hpp" #include "duckdb/common/enums/database_modification_type.hpp" +#include "duckdb/common/exception/transaction_exception.hpp" #include "duckdb/common/file_system.hpp" #include "duckdb/common/vector_operations/vector_operations.hpp" #include "duckdb/function/table_function.hpp" @@ -68,6 +69,39 @@ class RawSchemaCache : public ObjectCacheEntry { unordered_map tables; }; +//===--------------------------------------------------------------------===// +// Table-creation serialization: concurrent HTTP/programmatic requests each +// open their own Connection/transaction. DuckDB's catalog allows only one +// of them to CREATE a given table name — the rest see a TransactionException +// ("write-write conflict") even though their payload is perfectly valid. +// `known_to_exist` is the steady-state fast path (checked, never locked, +// once a table has been seen): normal ingestion after a table's first-ever +// insert never touches `creation_lock` at all. `creation_lock` is only +// acquired for a table's *first* insert in this process, and held across +// the whole attempt through commit — a losing racer's catalog lookup can't +// see the winner's CREATE until that transaction actually commits, so +// releasing any earlier would just move the race, not remove it. One +// process-wide lock (not per-table): creating brand-new tables is rare and +// one-time per table, so serializing unrelated tables' first inserts against +// each other is an acceptable, simple tradeoff for never serializing +// steady-state appends to already-existing tables. +class RawTableCreationCache : public ObjectCacheEntry { +public: + static string ObjectType() { + return "rawduck_table_creation"; + } + string GetObjectType() override { + return ObjectType(); + } + optional_idx GetEstimatedCacheMemory() const override { + return optional_idx(); + } + + mutex lock; + case_insensitive_set_t known_to_exist; + mutex creation_lock; +}; + static uint64_t HashPayloadShape(const vector &columns) { uint64_t shape = 0xcbf29ce484222325ULL; for (auto &column : columns) { @@ -87,6 +121,13 @@ static void MergeParsedPayloads(RawParsedPayload &into, RawParsedPayload &&from) from.payload.docs.clear(); from.payload.rows.clear(); from.payload.parse_errors = 0; + // the moved docs' yyjson_alc still points at from's pool buffer(s): that + // ownership must move with them, or freeing `from` frees memory `into`'s + // docs still reference. + for (auto buffer : from.payload.pool_buffers) { + into.payload.pool_buffers.push_back(buffer); + } + from.payload.pool_buffers.clear(); } //===--------------------------------------------------------------------===// @@ -232,7 +273,7 @@ class RawAppendPool { worker.local_types = types; worker.local_slots = slots; worker.writer = make_uniq(context, storage); - auto collection = worker.writer->CreateCollection(storage, types); + auto collection = worker.writer->CreateCollection(storage, types, OptimisticWritePartialManagers::GLOBAL); collection->collection->InitializeEmpty(); worker.append_state = make_uniq(); collection->collection->InitializeAppend(*worker.append_state); @@ -289,9 +330,13 @@ class RawAppendPool { } idx_t total = 0; auto merge_threshold = storage.GetRowGroupSize() / 8; - // phase 1, parallel: finalize and flush each worker's collection. - // Writers capture the table's storage for compression metadata, so - // they are rebuilt against the CURRENT (possibly evolved) storage. + const bool flush_now = overlap_flush; + // phase 1, parallel: finalize each worker. Mid-append overlap_flush + // already wrote complete row groups; finish those to disk here. The + // default (in-memory until drain) skips that write: LocalMerge keeps + // uncompressed collections in local storage and CHECKPOINT packs them + // once — same GLOBAL partial-block path as DuckDB INSERT, without a + // second allocation that leaves free-list holes. vector flushers; for (auto &worker : workers) { // pad to the full published schema so layouts match the table @@ -309,10 +354,10 @@ class RawAppendPool { // created against it (no evolution) or rebuilt by ExtendWorker, and // it holds the partial blocks from incremental flushes, so reuse it // rather than dropping that state on the floor. - flushers.emplace_back([&worker, merge_threshold] { + flushers.emplace_back([&worker, merge_threshold, flush_now] { auto &collection = *worker.collection->collection; collection.FinalizeAppend(TransactionData(0, 0), *worker.append_state); - if (collection.GetTotalRows() >= merge_threshold) { + if (flush_now && collection.GetTotalRows() >= merge_threshold) { worker.writer->WriteUnflushedRowGroups(*worker.collection); worker.writer->FinalFlush(); } @@ -341,7 +386,9 @@ class RawAppendPool { storage.FinalizeLocalAppend(append_state); } else { storage.LocalMerge(context, *worker.collection); - storage.GetOptimisticWriter(context).Merge(*worker.writer); + if (flush_now) { + storage.GetOptimisticWriter(context).Merge(*worker.writer); + } } } workers.clear(); @@ -433,7 +480,11 @@ class RawAppendPool { } // pads this worker's collection with NULL-filled columns: metadata-only - // work through RowGroupCollection::AddColumn. The worker may have already + // work through RowGroupCollection::AddColumn. GLOBAL partial-block packing + // (same as DuckDB LocalStorage INSERT) leaves this vector empty: leftover + // space is shared across columns so CHECKPOINT can truncate. PER_COLUMN + // managers (one nearly-empty tail block per column) were the VARIANT-flat + // file-size gap — same codecs, twice the file. // flushed complete row groups to disk (incremental WriteNewRowGroup); // AddColumn extends those checkpointed row groups too (existing columns // stay on disk, the new column is materialized NULL in memory). The flush @@ -1184,8 +1235,18 @@ class RawIngestor { // the caller's active transaction. RawIngestStats RawIngestPayload(ClientContext &context, const string &target, const string &payload, const RawParseOptions &options) { + auto parsed = RawParsedPayload::Process(payload, options); + return RawIngestParsedPayload(context, target, std::move(parsed), payload, options); +} + +// Split out of RawIngestPayload so a retrying caller can reuse the same +// already-parsed payload across attempts instead of re-parsing and +// re-shredding the same bytes every time — parsing isn't what conflicts, +// only the catalog step is. +RawIngestStats RawIngestParsedPayload(ClientContext &context, const string &target, shared_ptr parsed, + const string &payload, const RawParseOptions &options) { RawIngestor ingestor(context, target, options); - ingestor.Ingest(payload); + ingestor.IngestParsed(std::move(parsed), payload); ingestor.Finish(); RawIngestStats stats; stats.created = ingestor.created; @@ -1196,6 +1257,52 @@ RawIngestStats RawIngestPayload(ClientContext &context, const string &target, co return stats; } +// Entry point for the HTTP/programmatic ingest path: manages `conn`'s +// transaction itself (a table's first insert must hold the creation lock +// across the whole attempt through commit; see RawTableCreationCache) and +// falls back to a single retry for the rarer conflict class the lock +// doesn't cover — concurrent ALTER on an already-existing table (e.g. two +// requests simultaneously discovering the same new column). +RawIngestStats RawIngestSerialized(Connection &conn, const string &target, shared_ptr parsed, + const string &payload, const RawParseOptions &options) { + auto &cache = *ObjectCache::GetObjectCache(*conn.context) + .GetOrCreate(RawTableCreationCache::ObjectType()); + auto known_to_exist = [&]() { + lock_guard guard(cache.lock); + return cache.known_to_exist.count(target) > 0; + }; + auto mark_known = [&]() { + lock_guard guard(cache.lock); + cache.known_to_exist.insert(target); + }; + + unique_lock creation_guard(cache.creation_lock, std::defer_lock); + if (!known_to_exist()) { + creation_guard.lock(); + if (known_to_exist()) { + // someone else created (and committed) it while we waited + creation_guard.unlock(); + } + } + + conn.BeginTransaction(); + try { + auto stats = RawIngestParsedPayload(*conn.context, target, parsed, payload, options); + conn.Commit(); + mark_known(); + return stats; + } catch (TransactionException &) { + conn.Rollback(); + } + // 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(); + auto stats = RawIngestParsedPayload(*conn.context, target, parsed, payload, options); + conn.Commit(); + mark_known(); + return stats; +} + // Streaming handle over RawIngestor for the INSERT-syntax path namespace { class RawStreamIngestorImpl : public RawStreamIngestor { @@ -1533,8 +1640,18 @@ static void RawIngestFileFunction(ClientContext &context, TableFunctionInput &da for (idx_t line = 0; line < take; line++) { split = pending.find('\n', split) + 1; } - emit(pending.substr(0, split)); - pending.erase(0, split); + // batches run several MB: copy the (small) unconsumed tail and move + // the (large) batch prefix out via resize-then-move, instead of + // copying the large prefix via substr on every batch boundary. + if (split >= pending.size()) { + emit(std::move(pending)); + pending.clear(); + } else { + string remainder(pending.data() + split, pending.size() - split); + pending.resize(split); + emit(std::move(pending)); + pending = std::move(remainder); + } pending_lines -= take; } } diff --git a/src/raw_json.cpp b/src/raw_json.cpp index 035b434..619a773 100644 --- a/src/raw_json.cpp +++ b/src/raw_json.cpp @@ -24,6 +24,13 @@ static constexpr idx_t RAW_MAX_COLUMNS = 10000; // of risking stack exhaustion on hostile inputs static constexpr idx_t RAW_MAX_NESTING = 128; static constexpr auto RAW_READ_FLAGS = duckdb_yyjson::YYJSON_READ_ALLOW_INF_AND_NAN; +// Below this fanout, a linear scan over `children` beats `child_lookup`: no +// temporary std::string, no hashing, and the scan stays within a couple of +// cache lines. Real-world objects (OTLP resource/span attrs, log records) +// overwhelmingly fall under this; wide flat schemas (GH Archive-style) still +// fall back to the hash map so a single object with hundreds of keys stays +// O(1) per lookup instead of O(n). +static constexpr idx_t RAW_CHILD_LINEAR_SCAN_LIMIT = 24; //===--------------------------------------------------------------------===// // Payload parsing @@ -33,6 +40,12 @@ RawPayload::~RawPayload() { for (auto doc : docs) { duckdb_yyjson::yyjson_doc_free(doc); } + // must run after the doc frees above: those write into their pool + // buffer's own free-list metadata (pool_free()), so a buffer can only be + // released once every doc allocated from it has been freed. + for (auto buffer : pool_buffers) { + free(buffer); + } } static void CollectRows(yyjson_val *root, vector &rows) { @@ -61,8 +74,51 @@ static void CheckRowUniformity(const vector &rows, bool &scalar_ro scalar_rows = !rows.empty() && object_rows == 0; } +// One malloc for the whole payload instead of one per yyjson_read call +// (the common NDJSON case: one call per line). Sized via yyjson's own +// worst-case estimator, which is linear in input length, so estimating once +// for the whole payload safely covers the sum of however it's split into +// documents. Returns nullptr (use the default allocator) on overflow or if +// the single malloc fails — pooling is purely opportunistic, never required. +static void *RawInitReadPool(const string &payload, duckdb_yyjson::yyjson_alc &pool_alc) { + if (payload.empty()) { + return nullptr; + } + auto pool_size = duckdb_yyjson::yyjson_read_max_memory_usage(payload.size(), RAW_READ_FLAGS); + if (pool_size == 0) { + return nullptr; + } + auto buffer = malloc(pool_size); + if (!buffer) { + return nullptr; + } + if (!duckdb_yyjson::yyjson_alc_pool_init(&pool_alc, buffer, pool_size)) { + free(buffer); + return nullptr; + } + return buffer; +} + +// Falls back to the default allocator if the pool is exhausted (a sizing +// edge case, not expected in practice) so pooling can never turn a payload +// that would otherwise parse successfully into a parse error. +static duckdb_yyjson::yyjson_doc *RawPoolRead(const char *data, size_t len, const duckdb_yyjson::yyjson_alc *alc) { + auto doc = duckdb_yyjson::yyjson_read_opts(const_cast(data), len, RAW_READ_FLAGS, alc, nullptr); + if (!doc && alc) { + doc = duckdb_yyjson::yyjson_read_opts(const_cast(data), len, RAW_READ_FLAGS, nullptr, nullptr); + } + return doc; +} + void RawPayload::Parse(const string &payload, const RawParseOptions &options) { - auto doc = duckdb_yyjson::yyjson_read(payload.c_str(), payload.size(), RAW_READ_FLAGS); + duckdb_yyjson::yyjson_alc pool_alc; + auto pool_buffer = RawInitReadPool(payload, pool_alc); + if (pool_buffer) { + pool_buffers.push_back(pool_buffer); + } + auto alc = pool_buffer ? &pool_alc : nullptr; + + auto doc = RawPoolRead(payload.c_str(), payload.size(), alc); if (doc) { docs.push_back(doc); CollectRows(duckdb_yyjson::yyjson_doc_get_root(doc), rows); @@ -90,7 +146,7 @@ void RawPayload::Parse(const string &payload, const RawParseOptions &options) { if (blank) { continue; } - auto line_doc = duckdb_yyjson::yyjson_read(line_begin, line_len, RAW_READ_FLAGS); + auto line_doc = RawPoolRead(line_begin, line_len, alc); if (!line_doc) { if (options.ignore_errors) { parse_errors++; @@ -329,17 +385,31 @@ static duckdb_yyjson::yyjson_mut_val *OtlpUnwrapAnyValue(duckdb_yyjson::yyjson_m return nullptr; } +// Objects rebuilt per row (merged span rows, nested "status" etc.) almost +// always fit this inline capacity: only a pathologically wide object spills +// to the heap. Avoids a malloc/free per object per row on the OTLP path. +static constexpr idx_t OTLP_NORMALIZE_INLINE_MEMBERS = 32; + static void OtlpNormalizeObject(duckdb_yyjson::yyjson_mut_doc *doc, duckdb_yyjson::yyjson_mut_val *object, idx_t depth) { + using MemberPair = pair; // rebuild members so attribute lists can spread into the parent - vector> members; + MemberPair inline_members[OTLP_NORMALIZE_INLINE_MEMBERS]; + idx_t inline_count = 0; + vector overflow_members; duckdb_yyjson::yyjson_mut_obj_iter iter; duckdb_yyjson::yyjson_mut_obj_iter_init(object, &iter); while (auto key = duckdb_yyjson::yyjson_mut_obj_iter_next(&iter)) { - members.emplace_back(key, duckdb_yyjson::yyjson_mut_obj_iter_get_val(key)); + auto value = duckdb_yyjson::yyjson_mut_obj_iter_get_val(key); + if (inline_count < OTLP_NORMALIZE_INLINE_MEMBERS) { + inline_members[inline_count++] = MemberPair(key, value); + } else { + overflow_members.emplace_back(key, value); + } } duckdb_yyjson::yyjson_mut_obj_clear(object); - for (auto &member : members) { + + auto process_member = [&](const MemberPair &member) { auto key_str = duckdb_yyjson::yyjson_mut_get_str(member.first); auto key_len = duckdb_yyjson::yyjson_mut_get_len(member.first); if (key_len == 10 && memcmp(key_str, "attributes", 10) == 0 && OtlpIsKeyValueArray(member.second)) { @@ -358,12 +428,18 @@ static void OtlpNormalizeObject(duckdb_yyjson::yyjson_mut_doc *doc, duckdb_yyjso duckdb_yyjson::yyjson_mut_obj_add(object, name, normalized); } } - continue; + return; } auto normalized = OtlpNormalizeValue(doc, member.second, key_str, key_len, depth + 1); if (!duckdb_yyjson::yyjson_mut_obj_getn(object, key_str, key_len)) { duckdb_yyjson::yyjson_mut_obj_add(object, member.first, normalized ? normalized : member.second); } + }; + for (idx_t i = 0; i < inline_count; i++) { + process_member(inline_members[i]); + } + for (auto &member : overflow_members) { + process_member(member); } } @@ -553,13 +629,29 @@ RawParseOptions RawExplodeOptions(const string &path) { // Inference //===--------------------------------------------------------------------===// -RawNode &RawNode::GetOrCreateChild(const string &key) { - auto entry = child_lookup.find(key); - if (entry != child_lookup.end()) { - return *children[entry->second].second; +const RawNode *RawNode::FindChild(const char *key, idx_t key_len) const { + if (children.size() <= RAW_CHILD_LINEAR_SCAN_LIMIT) { + for (auto &entry : children) { + auto &name = entry.first; + if (name.size() == key_len && memcmp(name.data(), key, key_len) == 0) { + return entry.second.get(); + } + } + return nullptr; } - child_lookup[key] = children.size(); - children.emplace_back(key, make_uniq()); + auto entry = child_lookup.find(string(key, key_len)); + return entry == child_lookup.end() ? nullptr : children[entry->second].second.get(); +} + +RawNode &RawNode::GetOrCreateChild(const char *key, idx_t key_len) { + if (auto existing = FindChild(key, key_len)) { + return *const_cast(existing); + } + string key_str(key, key_len); + // keep child_lookup populated from the start so it's already correct once + // fanout crosses the linear-scan threshold and FindChild starts using it. + child_lookup[key_str] = children.size(); + children.emplace_back(std::move(key_str), make_uniq()); return *children.back().second; } @@ -662,8 +754,9 @@ static void MergeValueInternal(RawNode &node, yyjson_val *val, idx_t depth) { duckdb_yyjson::yyjson_obj_iter_init(val, &iter); while (auto key = duckdb_yyjson::yyjson_obj_iter_next(&iter)) { auto child_val = duckdb_yyjson::yyjson_obj_iter_get_val(key); - auto key_str = string(duckdb_yyjson::yyjson_get_str(key), duckdb_yyjson::yyjson_get_len(key)); - MergeValueInternal(node.GetOrCreateChild(key_str), child_val, depth + 1); + MergeValueInternal( + node.GetOrCreateChild(duckdb_yyjson::yyjson_get_str(key), duckdb_yyjson::yyjson_get_len(key)), + child_val, depth + 1); } return; } @@ -851,12 +944,11 @@ void RawExtractor::Traverse(yyjson_val *val, const RawNode &node, idx_t row_idx) yyjson_obj_iter iter; duckdb_yyjson::yyjson_obj_iter_init(val, &iter); while (auto key = duckdb_yyjson::yyjson_obj_iter_next(&iter)) { - auto entry = - node.child_lookup.find(string(duckdb_yyjson::yyjson_get_str(key), duckdb_yyjson::yyjson_get_len(key))); - if (entry == node.child_lookup.end()) { + auto child = node.FindChild(duckdb_yyjson::yyjson_get_str(key), duckdb_yyjson::yyjson_get_len(key)); + if (!child) { continue; } - Traverse(duckdb_yyjson::yyjson_obj_iter_get_val(key), *node.children[entry->second].second, row_idx); + Traverse(duckdb_yyjson::yyjson_obj_iter_get_val(key), *child, row_idx); } } diff --git a/test/http/raw_api_compat.sh b/test/http/raw_api_compat.sh index 8b2a931..74eb8be 100755 --- a/test/http/raw_api_compat.sh +++ b/test/http/raw_api_compat.sh @@ -46,4 +46,29 @@ echo "$DESC" | python3 -c 'import json,sys; cols={c["name"] for c in json.load(s echo "== transform query param ==" curl -sf -X POST "${BASE}/v1/tables/traces2?transform=otlp-traces" "${AUTH[@]}" -d "$OTLP" | json_get "['inserted']" | grep -q 1 +echo "== concurrent create (first-insert serialization) ==" +# N clients racing to POST to a table that does not exist yet: each opens its +# own Connection/transaction, so only one would win the CREATE TABLE and the +# rest would see a DuckDB catalog conflict. RawIngestSerialized queues +# first-time creators behind an in-process lock instead of letting them race +# (and instead of retrying blind against an uncommitted winner), so every +# request lands cleanly. +CONCURRENT_TABLE="concurrent_create_test_$$" +PIDS=() +for i in $(seq 1 8); do + curl -sf -X POST "${BASE}/v1/tables/${CONCURRENT_TABLE}" "${AUTH[@]}" \ + -d "{\"n\": ${i}}" -o "/tmp/concurrent_create_${i}.json" & + PIDS+=($!) +done +for pid in "${PIDS[@]}"; do + wait "${pid}" +done +for i in $(seq 1 8); do + json_get "['inserted']" <"/tmp/concurrent_create_${i}.json" | grep -q 1 + rm -f "/tmp/concurrent_create_${i}.json" +done +COUNT=$(curl -sf -X POST "${BASE}/v1/query" "${AUTH[@]}" \ + -d "{\"sql\":\"SELECT count(*) AS c FROM ${CONCURRENT_TABLE}\"}") +echo "$COUNT" | json_get "['data'][0]['c']" | grep -q 8 + echo "OK: API compatibility smoke test passed"