From ec25139ae9a6749717fe01196892ac39845c0727 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 14 Oct 2026 17:51:23 -0700 Subject: [PATCH 1/4] build(duckdb): pin DuckDB v2.0.0-alpha46057 Move the embedded engine from v2.0.0-alpha43763 (96063b9e39) to v2.0.0-alpha46057 (b15358c371), the newest completed DuckDB 2 build. The duckdb-v2.0.0-alpha46057 release mirrors DuckDB's staging archives byte for byte, and the wrapper pins their SHA-256 values. The engine adds one built-in, read_single_parquet_file. It reads files, so the telemetry SQL boundary blocks it like read_parquet, and the function snapshot is regenerated. No built-in was removed or retyped, and no code change was needed. SELECT version() reports v2.0.0-alpha46057. --- CONTRIBUTING.md | 6 +++--- THIRD_PARTY_NOTICES | 8 ++++---- internal/duckdb/native.go | 4 ++-- internal/query/sql_boundary.go | 4 ++-- internal/query/sql_boundary_test.go | 2 ++ internal/query/testdata/duckdb-functions.txt | 3 ++- scripts/with-duckdb.sh | 10 +++++----- site/src/content/docs/explanation/storage-model.mdx | 2 +- third_party/notices/duckdb-NOTICE.txt | 4 ++-- 9 files changed, 23 insertions(+), 20 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 5f793e6b..2db2e832 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -29,7 +29,7 @@ Keep this file out of Git and Docker inputs. See [runtime secret delivery](docs/operations.md#runtime-secret-delivery). The DuckDB wrapper downloads the pinned, checksum-verified engine from -[Fanout's dependency artifact mirror](https://github.com/labstack/fanout/releases/tag/duckdb-v2.0.0-alpha43763) +[Fanout's dependency artifact mirror](https://github.com/labstack/fanout/releases/tag/duckdb-v2.0.0-alpha46057) on the first build. Linux needs the C++ runtime (`libstdc++`); macOS needs Xcode Command Line Tools and `libc++`. Cross-compilation is unsupported. Use `just` recipes or prefix Go commands with the wrapper: @@ -46,8 +46,8 @@ Alternatively configure your editor to launch `gopls` through `bash scripts/with-duckdb.sh gopls serve`. An already-running editor process must be restarted to inherit these variables. -The dependency mirror preserves the original DuckDB `v2.0.0-alpha43763` -archives at upstream commit `96063b9e39` byte for byte. Its `duckdb-` tag +The dependency mirror preserves the original DuckDB `v2.0.0-alpha46057` +archives at upstream commit `b15358c371` byte for byte. Its `duckdb-` tag namespace is separate from Fanout's product CalVer releases; new engine bytes require a new artifact tag and checksum pins. It never supplies a second engine or downloads extensions at runtime. diff --git a/THIRD_PARTY_NOTICES b/THIRD_PARTY_NOTICES index 0230def8..c5365208 100644 --- a/THIRD_PARTY_NOTICES +++ b/THIRD_PARTY_NOTICES @@ -73,7 +73,7 @@ COMPONENT INVENTORY - Go: modernc.org/mathutil v1.7.1 - Go: modernc.org/memory v1.12.1 - Go: modernc.org/sqlite v1.60.1 -- Native: DuckDB v2.0.0-alpha43763 (96063b9e39749cc0f087bb11501d7edbee527957) +- Native: DuckDB v2.0.0-alpha46057 (b15358c371) - npm: @ag-ui/client 1.0.2 (declared license: MIT) - npm: @ag-ui/core 1.0.2 (declared license: MIT) - npm: @ag-ui/encoder 1.0.2 (declared license: MIT) @@ -5122,11 +5122,11 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. --- Applies to ------------------------------------------------------------- -- Native: DuckDB v2.0.0-alpha43763 (96063b9e39749cc0f087bb11501d7edbee527957) / duckdb-NOTICE +- Native: DuckDB v2.0.0-alpha46057 (b15358c371) / duckdb-NOTICE ---------------------------------------------------------------------------- -DuckDB 96063b9e39749cc0f087bb11501d7edbee527957 — upstream license and bundled third-party notices. -Source: https://github.com/duckdb/duckdb/tree/96063b9e39749cc0f087bb11501d7edbee527957 +DuckDB b15358c371 — upstream license and bundled third-party notices. +Source: https://github.com/duckdb/duckdb/tree/b15358c371 --- LICENSE --- diff --git a/internal/duckdb/native.go b/internal/duckdb/native.go index 2634533b..01492361 100644 --- a/internal/duckdb/native.go +++ b/internal/duckdb/native.go @@ -18,8 +18,8 @@ static int register_extensions(void) { */ import "C" -const Version = "v2.0.0-alpha43763" -const SourceCommit = "96063b9e39749cc0f087bb11501d7edbee527957" +const Version = "v2.0.0-alpha46057" +const SourceCommit = "b15358c371" func init() { if C.GoString(C.duckdb_library_version()) != Version { diff --git a/internal/query/sql_boundary.go b/internal/query/sql_boundary.go index ac4d80d5..bd07d65b 100644 --- a/internal/query/sql_boundary.go +++ b/internal/query/sql_boundary.go @@ -24,8 +24,8 @@ var blockedSQLFunctions = map[string]bool{ "query": true, "query_table": true, "json_execute_serialized_sql": true, "current_setting": true, "getvariable": true, "getenv": true, "read_csv": true, "read_csv_auto": true, "read_json": true, "read_json_auto": true, - "read_text": true, "read_blob": true, "read_parquet": true, "glob": true, - "http_get": true, "write_file": true, "write_log": true, + "read_text": true, "read_blob": true, "read_parquet": true, "read_single_parquet_file": true, + "glob": true, "http_get": true, "write_file": true, "write_log": true, "current_query": true, "current_query_id": true, "pg_get_viewdef": true, "pg_get_constraintdef": true, "format_type": true, "get_block_size": true, } diff --git a/internal/query/sql_boundary_test.go b/internal/query/sql_boundary_test.go index 31c67373..56f3ac37 100644 --- a/internal/query/sql_boundary_test.go +++ b/internal/query/sql_boundary_test.go @@ -24,6 +24,8 @@ func TestSQLBoundaryRejectsHiddenReadsAndMutations(t *testing.T) { "SELECT * FROM (WITH rollup_state AS (SELECT 1) SELECT * FROM rollup_state) x, rollup_state", "SELECT * FROM query('SELECT * FROM rollup_state')", "SELECT * FROM read_parquet('/etc/passwd')", + "SELECT * FROM read_single_parquet_file('/etc/passwd')", + "SELECT read_single_parquet_file('/etc/passwd')", "SELECT * FROM (SHOW ALL TABLES)", "SELECT * FROM read_batches", "SELECT * FROM read_span_events", diff --git a/internal/query/testdata/duckdb-functions.txt b/internal/query/testdata/duckdb-functions.txt index 6fb83c69..c5df5f5c 100644 --- a/internal/query/testdata/duckdb-functions.txt +++ b/internal/query/testdata/duckdb-functions.txt @@ -1,4 +1,4 @@ -# engine v2.0.0-alpha43763 +# engine v2.0.0-alpha46057 != scalar !__postfix scalar !~~ scalar @@ -961,6 +961,7 @@ read_ndjson_objects table read_parquet table read_single_csv_file table read_single_json_file table +read_single_parquet_file table read_text table reduce scalar regexp_escape scalar diff --git a/scripts/with-duckdb.sh b/scripts/with-duckdb.sh index 8175e277..0e4ffa3f 100644 --- a/scripts/with-duckdb.sh +++ b/scripts/with-duckdb.sh @@ -3,26 +3,26 @@ set -euo pipefail root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -version=v2.0.0-alpha43763 +version=v2.0.0-alpha46057 case "$(uname -s)/$(uname -m)" in Darwin/arm64) platform=osx-arm64 - checksum=42320cd312d1a55f2fbe2e527b4f95aed6e0b9ee77e64092fa2a5b0879024206 + checksum=e78e6a3ab5c35c1451650cdd234e9f42c8c9485412aaf063dfaae995f0dcda6a system_libs='-lc++ -Wl,-no_warn_duplicate_libraries' ;; Darwin/x86_64) platform=osx-amd64 - checksum=c13adabde6b7aac99281d745da2bdfad0c482b306e1b3d3d7391ef806e967368 + checksum=a1dd2b83754484dcd0980ed1c46d5726ec2a2fd26090d43341dc74f1060abff3 system_libs='-lc++ -Wl,-no_warn_duplicate_libraries' ;; Linux/x86_64) platform=linux-amd64 - checksum=5d62c5bca5474722e172d19c51ba4c7b3711444b70f6478b3b38b94bfb5681b2 + checksum=4cb2df2e767f2a08fb38a0c39b8dadd1ad0eb07fd5cd6431092e709be41a323e system_libs='-lstdc++ -lm -ldl -pthread' ;; Linux/aarch64) platform=linux-arm64 - checksum=af0ed15848a76a013764419b4f47020ae0aa7e71396f2365aa61a95ecba456ee + checksum=81cf834f18a7004d412862d25bbbd334c1e335432c94a2cbcb6c0f560ebbc3eb system_libs='-lstdc++ -lm -ldl -pthread' ;; *) diff --git a/site/src/content/docs/explanation/storage-model.mdx b/site/src/content/docs/explanation/storage-model.mdx index e01155de..36239b9c 100644 --- a/site/src/content/docs/explanation/storage-model.mdx +++ b/site/src/content/docs/explanation/storage-model.mdx @@ -98,7 +98,7 @@ string encoding when querying them for a JSON client. ## Engine and format are pinned -Fanout embeds DuckDB `v2.0.0-alpha43763`, a preview engine, with its extensions +Fanout embeds DuckDB `v2.0.0-alpha46057`, a preview engine, with its extensions compiled into the binary. It downloads no extensions at runtime. Native source builds use `scripts/with-duckdb.sh`, which checks the platform archive and uses the matching engine headers; the ordinary `just` build and test recipes invoke diff --git a/third_party/notices/duckdb-NOTICE.txt b/third_party/notices/duckdb-NOTICE.txt index 256d6756..c0888af8 100644 --- a/third_party/notices/duckdb-NOTICE.txt +++ b/third_party/notices/duckdb-NOTICE.txt @@ -1,5 +1,5 @@ -DuckDB 96063b9e39749cc0f087bb11501d7edbee527957 — upstream license and bundled third-party notices. -Source: https://github.com/duckdb/duckdb/tree/96063b9e39749cc0f087bb11501d7edbee527957 +DuckDB b15358c371 — upstream license and bundled third-party notices. +Source: https://github.com/duckdb/duckdb/tree/b15358c371 --- LICENSE --- From f724fbbf776116775c82e5febb83a5acfd29b98a Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 14 Oct 2026 15:47:10 -0700 Subject: [PATCH 2/4] chore(ci): drop renovate[bot] from the CLA allowlist Fanout uses Dependabot for dependency updates; Renovate is not configured, so its bot no longer needs a CLA exemption. --- .github/workflows/cla.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/cla.yml b/.github/workflows/cla.yml index cbf92d69..27ed0cb2 100644 --- a/.github/workflows/cla.yml +++ b/.github/workflows/cla.yml @@ -36,4 +36,4 @@ jobs: # a bot commit into the release history and trigger a CI run plus an # image publish every time somebody signs. branch: cla-signatures - allowlist: vishr,dependabot[bot],renovate[bot] + allowlist: vishr,dependabot[bot] From 686a6adb7bfd3b9cb75c6eb61de1c745994f723d Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 14 Oct 2026 21:40:41 -0700 Subject: [PATCH 3/4] perf(telemetry): cut commit allocation and dashboard scan work Every new zstd encoder at the library's fastest level allocated 8 MiB of match history, and the garbage collector empties the encoder pool, so that history was most of what a commit allocated. Parquet pages are compressed with a 1 MiB window instead, which covers a whole page. A 50,000-span commit now allocates 116-121 MB instead of 369-383 MB and takes 293-322 ms instead of 472-509 ms; the allocation ratchet drops from 7.0x to 3.0x. - Buffer trace-index entries instead of one write call per trace. - Measure a request's bytes once at submission rather than on admission, grouping and release. - Build each telemetry Parquet schema once per row type. - Dictionary-encode span kind and status, which come from fixed OTLP enums; client-supplied names stay plain. Bound every dictionary column at 1 MiB, after which it continues in plain pages. - Explain an empty panel with one scan that counts every omitted-filter rerun, falling back to separate counts when a predicate fails. --- go.mod | 2 +- internal/ingest/server.go | 14 ++- internal/panel/compile.go | 29 +++-- internal/panel/diagnose.go | 84 +++++++++++-- internal/panel/diagnose_regression_test.go | 31 ++++- internal/telemetry/commit_memory_test.go | 18 +-- internal/telemetry/parquet.go | 12 +- internal/telemetry/parquet_codec.go | 52 ++++++++ internal/telemetry/parquet_encoding_test.go | 124 +++++++++++++++++++ internal/telemetry/parquet_rows.go | 25 ++-- internal/telemetry/parquet_schema.go | 20 ++- internal/telemetry/store/admission_test.go | 14 +-- internal/telemetry/store/batch_bytes_test.go | 14 +-- internal/telemetry/store/writer.go | 30 +++-- internal/telemetry/trace_index.go | 27 ++-- 15 files changed, 407 insertions(+), 89 deletions(-) create mode 100644 internal/telemetry/parquet_codec.go create mode 100644 internal/telemetry/parquet_encoding_test.go diff --git a/go.mod b/go.mod index 1a85b188..71291640 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/golang-jwt/jwt/v5 v5.3.1 github.com/google/jsonschema-go v0.4.3 github.com/google/uuid v1.6.0 + github.com/klauspost/compress v1.20.1 github.com/labstack/echo/v5 v5.4.0 github.com/modelcontextprotocol/go-sdk v1.8.0 github.com/parquet-go/parquet-go v0.32.0 @@ -50,7 +51,6 @@ require ( github.com/goccy/go-json v0.11.2 // indirect github.com/google/flatbuffers v25.12.19+incompatible // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.31.0 // indirect - github.com/klauspost/compress v1.20.1 // indirect github.com/klauspost/cpuid/v2 v2.4.0 // indirect github.com/kylelemons/godebug v1.1.0 // indirect github.com/mattn/go-isatty v0.0.24 // indirect diff --git a/internal/ingest/server.go b/internal/ingest/server.go index c40d5960..c2740ccf 100644 --- a/internal/ingest/server.go +++ b/internal/ingest/server.go @@ -3,11 +3,13 @@ package ingest import ( "bytes" "context" + "encoding/hex" "encoding/json" "errors" "fmt" "log/slog" "math" + "strconv" "strings" "time" @@ -90,8 +92,8 @@ func (s *Server) exportTraces(ctx context.Context, req *collectortrace.ExportTra } row := telemetry.Span{ Namespace: namespace, - TraceID: fmt.Sprintf("%x", sp.TraceId), - SpanID: fmt.Sprintf("%x", sp.SpanId), + TraceID: hex.EncodeToString(sp.TraceId), + SpanID: hex.EncodeToString(sp.SpanId), ParentSpanID: hexOrEmpty(sp.ParentSpanId), ServiceName: svc, Name: sp.Name, @@ -575,7 +577,7 @@ func spanAttrInt(attrs []*common.KeyValue, keys ...string) string { // Handle integer values (common for status codes) if kv.Value != nil { if iv, ok := kv.Value.Value.(*common.AnyValue_IntValue); ok { - return fmt.Sprintf("%d", iv.IntValue) + return strconv.FormatInt(iv.IntValue, 10) } } } @@ -731,7 +733,7 @@ func hexOrEmpty(b []byte) string { if len(b) == 0 { return "" } - return fmt.Sprintf("%x", b) + return hex.EncodeToString(b) } func eventsToJSON(events []*tracepb.Span_Event) string { @@ -778,8 +780,8 @@ func linksToJSON(links []*tracepb.Span_Link) string { out := make([]link, 0, len(links)) for _, l := range links { ln := link{ - TraceID: fmt.Sprintf("%x", l.TraceId), - SpanID: fmt.Sprintf("%x", l.SpanId), + TraceID: hex.EncodeToString(l.TraceId), + SpanID: hex.EncodeToString(l.SpanId), TraceState: l.TraceState, } if len(l.Attributes) > 0 { diff --git a/internal/panel/compile.go b/internal/panel/compile.go index afd35d38..742a25b7 100644 --- a/internal/panel/compile.go +++ b/internal/panel/compile.go @@ -114,26 +114,31 @@ func buildWhere(sig *signal, filters []Filter, scope Scope) (string, []any, erro if scope.dropped(f) { continue } - empty := false - for _, name := range f.Params { - if v, ok := scope.Vars[name]; ok && !v.All && len(v.Values) == 0 { - empty = true - } - } - if empty { - clauses = append(clauses, "FALSE") - continue - } - text, _, values, err := bindParams(f.Expr, scope.lookup(f)) + clause, values, err := filterClause(f, scope) if err != nil { return "", nil, err } - clauses = append(clauses, "("+text+")") + clauses = append(clauses, clause) args = append(args, values...) } return strings.Join(clauses, " AND "), args, nil } +// filterClause renders one filter that the scope keeps. A variable bound to +// no values matches nothing. +func filterClause(f Filter, scope Scope) (string, []any, error) { + for _, name := range f.Params { + if v, ok := scope.Vars[name]; ok && !v.All && len(v.Values) == 0 { + return "FALSE", nil, nil + } + } + text, _, values, err := bindParams(f.Expr, scope.lookup(f)) + if err != nil { + return "", nil, err + } + return "(" + text + ")", values, nil +} + // compileQuery turns a structured query into one statement. The bucket is // selected as "_t" and named "time" in the frame, so it never collides with a // signal's own time column. diff --git a/internal/panel/diagnose.go b/internal/panel/diagnose.go index b42edce8..e1e306c4 100644 --- a/internal/panel/diagnose.go +++ b/internal/panel/diagnose.go @@ -55,7 +55,13 @@ func (e *Executor) diagnose(ctx context.Context, p *Panel, filters []Filter, sco return "The selected metric points have no defined values for these measures in this time range." } } - total, err := e.count(ctx, sig, nil, scope) + total, without, err := e.filterCounts(ctx, sig, active, scope) + if err != nil { + // One statement fails when any predicate fails on rows another filter + // normally excludes, such as a cast; counted separately, a failing + // rerun hides only its own count. + total, without, err = e.rerunCounts(ctx, sig, active, scope) + } if err != nil { return "" } @@ -63,13 +69,10 @@ func (e *Executor) diagnose(ctx context.Context, p *Panel, filters []Filter, sco return fmt.Sprintf("No %s were recorded in this time range.", sig.name) } best, recovered := -1, int64(0) - for i := range min(len(active), maxDiagnosedFilters) { - others := append(append([]Filter{}, active[:i]...), active[i+1:]...) - n, err := e.count(ctx, sig, others, scope) - if err != nil || n <= recovered { - continue + for i, n := range without { + if n > recovered { + best, recovered = i, n } - best, recovered = i, n } if best >= 0 { f := active[best] @@ -104,6 +107,73 @@ func (e *Executor) count(ctx context.Context, sig *signal, filters []Filter, sco return n, rows.Err() } +// filterCounts scans the signal once for everything the diagnosis compares: +// the rows in the window, and for each of the first maxDiagnosedFilters +// active filters the rows that satisfy every other active filter. Each FILTER +// clause is the predicate the rerun count would have carried, so one +// statement replaces one scan per filter without changing a single count. +func (e *Executor) filterCounts(ctx context.Context, sig *signal, active []Filter, scope Scope) (total int64, without []int64, err error) { + window, windowArgs, err := buildWhere(sig, nil, scope) + if err != nil { + return 0, nil, err + } + clauses := make([]string, len(active)) + clauseArgs := make([][]any, len(active)) + for i, f := range active { + if clauses[i], clauseArgs[i], err = filterClause(f, scope); err != nil { + return 0, nil, err + } + } + // Positional parameters bind in textual order: the FILTER clauses precede + // the window predicate in the statement, so their arguments come first. + without = make([]int64, min(len(active), maxDiagnosedFilters)) + selects := []string{"count(*)"} + dest := []any{&total} + var args []any + for i := range without { + var others []string + for j := range active { + if j != i { + others = append(others, clauses[j]) + args = append(args, clauseArgs[j]...) + } + } + if len(others) == 0 { + selects = append(selects, "count(*)") + } else { + selects = append(selects, "count(*) FILTER (WHERE "+strings.Join(others, " AND ")+")") + } + dest = append(dest, &without[i]) + } + args = append(args, windowArgs...) + rows, err := e.engine.QueryContext(ctx, "SELECT "+strings.Join(selects, ", ")+" FROM "+structuredSource(sig.name)+" WHERE "+window, args...) + if err != nil { + return 0, nil, err + } + defer rows.Close() + if rows.Next() { + if err := rows.Scan(dest...); err != nil { + return 0, nil, err + } + } + return total, without, rows.Err() +} + +// rerunCounts is filterCounts one statement at a time. A rerun that fails +// counts no rows. +func (e *Executor) rerunCounts(ctx context.Context, sig *signal, active []Filter, scope Scope) (int64, []int64, error) { + total, err := e.count(ctx, sig, nil, scope) + if err != nil { + return 0, nil, err + } + without := make([]int64, min(len(active), maxDiagnosedFilters)) + for i := range without { + others := append(append([]Filter{}, active[:i]...), active[i+1:]...) + without[i], _ = e.count(ctx, sig, others, scope) + } + return total, without, nil +} + func (e *Executor) topValues(ctx context.Context, sig *signal, fieldText string, filters []Filter, scope Scope) string { ref, err := sig.field(fieldText) if err != nil { diff --git a/internal/panel/diagnose_regression_test.go b/internal/panel/diagnose_regression_test.go index 905fed4d..39a5449b 100644 --- a/internal/panel/diagnose_regression_test.go +++ b/internal/panel/diagnose_regression_test.go @@ -88,6 +88,30 @@ func TestDiagnosisChoosesMostSelectiveQualifyingFilter(t *testing.T) { } } +func TestDiagnosisSkipsFailingFilterRerun(t *testing.T) { + engine, repo := newTestEngine(t) + n := fixtureStart.UnixNano() + commit(t, repo, []telemetry.Span{ + {ServiceName: "checkout", HTTPStatusCode: "200", HTTPRoute: "/ok", StartUnixNanos: n, EndUnixNanos: n + 1, IngestedAt: n}, + {ServiceName: "frontend", HTTPStatusCode: "", HTTPRoute: "/ok", StartUnixNanos: n, EndUnixNanos: n + 1, IngestedAt: n}, + }, nil) + e := NewExecutor(engine, 30) + e.now = func() time.Time { return fixtureStart.Add(time.Hour) } + d := Dashboard{Name: "Diagnosis", Time: Time{Range: "1h"}, Panels: []Panel{{ + ID: "empty", Title: "Empty", Viz: "table", Query: &Query{ + From: "spans", Measures: []string{"count()"}, By: []string{"service"}, + Where: []string{"service = 'checkout'", "CAST(http_status_code AS BIGINT) > 100", "http_route = '/missing'"}, + }, + }}} + results, err := e.Run(t.Context(), RunRequest{Dashboard: d}) + if err != nil { + t.Fatal(err) + } + if r := results[0]; r.Status != StatusEmpty || !strings.Contains(r.Diagnosis, "match http_route = '/missing'. Without that filter, 1 do.") { + t.Fatalf("result = %+v", r) + } +} + // countingEngine counts the statements a diagnosis issues. type countingEngine struct { Engine @@ -119,8 +143,9 @@ func TestDiagnosisBoundsReruns(t *testing.T) { } scope := Scope{Start: fixtureStart, End: fixtureStart.Add(time.Hour), Vars: map[string]Value{"service": {Values: []string{"checkout"}}}} _ = e.diagnose(t.Context(), &d.Panels[0], checked.Filters["empty"], scope) - // One total count, at most maxDiagnosedFilters reruns, and one value lookup. - if counting.queries > 2+maxDiagnosedFilters { - t.Fatalf("diagnosis ran %d queries, want at most %d", counting.queries, 2+maxDiagnosedFilters) + // One scan counts the window and the first maxDiagnosedFilters reruns + // together; a value lookup for the blamed equality is the only other read. + if counting.queries > 2 { + t.Fatalf("diagnosis ran %d queries, want at most 2", counting.queries) } } diff --git a/internal/telemetry/commit_memory_test.go b/internal/telemetry/commit_memory_test.go index 43210d46..96fa6e37 100644 --- a/internal/telemetry/commit_memory_test.go +++ b/internal/telemetry/commit_memory_test.go @@ -29,21 +29,21 @@ import ( // The assertion is on total bytes allocated rather than peak live heap. // Peak heap carries whatever slack the collector happened to be holding when // the sampler looked, which varies with GOMAXPROCS and machine speed: the same -// code measured 2.79x locally and 3.83x on a CI runner, and the two encodings' -// peak-heap distributions overlap. Total allocation does not overlap and barely -// varies -- it answers "how many copies does this make", which is the question. -// Peak heap is still reported, because it is what the kernel kills on. +// code measured 2.79x locally and 3.83x on a CI runner. Total allocation +// answers "how many copies does this make", which is the question. Peak heap +// is still reported, because it is what the kernel kills on. // -// This is a ratchet, not a target, and it is placed between two measured -// distributions rather than guessed. Dictionary-encoded near-unique JSON -// columns allocate 10.68x-10.90x; writing them plain allocates 4.40x-5.69x. -// The limit sits in that gap. +// This is a ratchet, not a target, and it is placed above a measured +// distribution rather than guessed. With parquetCodec's 1 MiB zstd window a +// commit allocates 1.1x-2.0x its payload. The library's default 4 MiB window +// measured 3x-9x: encoders rebuilt after each garbage collection allocated +// 8 MiB of match history apiece, and that history outweighed every copy. // // What remains above 1x is a conversion copy of every JSON column // (makeSpanParquetRow) plus row-struct headers and the writer's own buffers. // Lower this constant as each is addressed; see the ordered plan in the // ingest-memory issue. -const commitAllocRatioLimit = 7.0 +const commitAllocRatioLimit = 3.0 // syntheticSpan builds a span whose JSON columns look like production: a // resource block shared across the batch, and attributes/events/links that are diff --git a/internal/telemetry/parquet.go b/internal/telemetry/parquet.go index b73a90e2..08426cd2 100644 --- a/internal/telemetry/parquet.go +++ b/internal/telemetry/parquet.go @@ -17,7 +17,6 @@ import ( "time" "github.com/parquet-go/parquet-go" - "github.com/parquet-go/parquet-go/compress/zstd" "github.com/zeebo/xxh3" "golang.org/x/sync/errgroup" ) @@ -27,8 +26,12 @@ const ( SchemaBatch = "_schema" + BatchSuffix batchMetadataVersion = 4 parquetPageSize = 64 << 10 - parquetRowGroupRows = 50_000 - maxTraceQueryResults = 500 + // A column whose distinct values outgrow this switches to plain pages for + // the rest of its row group, so client-supplied text cannot grow a + // dictionary without bound. + parquetDictionaryMaxBytes = 1 << 20 + parquetRowGroupRows = 50_000 + maxTraceQueryResults = 500 ) // commitPublishWait caps how long one ingest commit waits for the publish @@ -1274,9 +1277,10 @@ func spanSortingColumns() parquet.SortingOption { func parquetWriterOptions(pageSize int, extra ...parquet.WriterOption) []parquet.WriterOption { options := []parquet.WriterOption{ - parquet.Compression(&zstd.Codec{Level: zstd.SpeedFastest, Concurrency: 1}), + parquet.Compression(parquetCodec), parquet.MaxRowsPerRowGroup(parquetRowGroupRows), parquet.PageBufferSize(pageSize), + parquet.DictionaryMaxBytes(parquetDictionaryMaxBytes), } return append(options, extra...) } diff --git a/internal/telemetry/parquet_codec.go b/internal/telemetry/parquet_codec.go new file mode 100644 index 00000000..eb6d6960 --- /dev/null +++ b/internal/telemetry/parquet_codec.go @@ -0,0 +1,52 @@ +package telemetry + +import ( + "sync" + + "github.com/klauspost/compress/zstd" + "github.com/parquet-go/parquet-go/compress" + parquetzstd "github.com/parquet-go/parquet-go/compress/zstd" + "github.com/parquet-go/parquet-go/format" +) + +// parquetCodec compresses every Parquet page Fanout writes. Each page is an +// independent zstd frame of about parquetPageSize bytes, so a 1 MiB window +// covers a whole data page. The library's fastest level defaults to a 4 MiB +// window, and each new encoder then allocates 8 MiB of match history. Pooled +// encoders are dropped at every garbage collection, so ingest rebuilt them +// often enough that their history was most of what a commit allocated. +var parquetCodec compress.Codec = &pageCodec{} + +const pageCodecWindow = 1 << 20 + +type pageCodec struct { + encoders sync.Pool // *zstd.Encoder + decoder parquetzstd.Codec +} + +func (c *pageCodec) String() string { return "ZSTD" } +func (c *pageCodec) CompressionCodec() format.CompressionCodec { return format.Zstd } + +func (c *pageCodec) Encode(dst, src []byte) ([]byte, error) { + e, _ := c.encoders.Get().(*zstd.Encoder) + if e == nil { + var err error + e, err = zstd.NewWriter(nil, + zstd.WithEncoderConcurrency(1), + zstd.WithEncoderLevel(zstd.SpeedFastest), + zstd.WithWindowSize(pageCodecWindow), + zstd.WithLowerEncoderMem(true), + zstd.WithZeroFrames(true), + zstd.WithEncoderCRC(false), + ) + if err != nil { + return nil, err + } + } + defer c.encoders.Put(e) + return e.EncodeAll(src, dst[:0]), nil +} + +func (c *pageCodec) Decode(dst, src []byte) ([]byte, error) { + return c.decoder.Decode(dst, src) +} diff --git a/internal/telemetry/parquet_encoding_test.go b/internal/telemetry/parquet_encoding_test.go new file mode 100644 index 00000000..c94ce1d4 --- /dev/null +++ b/internal/telemetry/parquet_encoding_test.go @@ -0,0 +1,124 @@ +package telemetry + +import ( + "errors" + "fmt" + "io" + "os" + "path/filepath" + "slices" + "strings" + "testing" + + "github.com/parquet-go/parquet-go" + "github.com/parquet-go/parquet-go/format" +) + +// Dashboard panels filter on span kind and status; a dictionary lets the +// engine evaluate those predicates once per distinct value instead of decoding +// one string per row. Identifiers stay plain: a dictionary over a near-unique +// column is a second copy of it. Namespaces, services and operation names stay +// plain because clients bound neither their length nor their count. +func TestSpanParquetDictionaryEncodesLowCardinalityColumns(t *testing.T) { + dir := t.TempDir() + store, err := OpenParquetStore(dir) + if err != nil { + t.Fatal(err) + } + spans := make([]Span, 300) + for i := range spans { + spans[i] = Span{Namespace: "prod", TraceID: strings.Repeat("a", 30) + string(rune('a'+i%26)) + string(rune('a'+i/26)), SpanID: strings.Repeat("b", 14) + string(rune('a'+i%26)) + string(rune('a'+i/26)), + ServiceName: "checkout", Name: "POST /orders", Kind: "SPAN_KIND_SERVER", StatusCode: "STATUS_CODE_UNSET", StartUnixNanos: int64(i + 1), EndUnixNanos: int64(i + 2), IngestedAt: 999} + } + if err := store.CommitBatch(t.Context(), BatchMetadata{ID: "encodings"}, spans, nil, nil); err != nil { + t.Fatal(err) + } + f, err := os.Open(filepath.Join(store.BatchPath("encodings"), "spans.parquet")) + if err != nil { + t.Fatal(err) + } + defer f.Close() + info, _ := f.Stat() + file, err := parquet.OpenFile(f, info.Size()) + if err != nil { + t.Fatal(err) + } + encodings := map[string][]format.Encoding{} + for _, chunk := range file.Metadata().RowGroups[0].Columns { + if len(chunk.MetaData.PathInSchema) == 1 { + encodings[chunk.MetaData.PathInSchema[0]] = chunk.MetaData.Encoding + } + } + for _, column := range []string{"kind", "status"} { + if !slices.Contains(encodings[column], format.RLEDictionary) { + t.Errorf("%s is not dictionary encoded: %v", column, encodings[column]) + } + } + for _, column := range []string{"namespace", "service", "operation", "trace_id", "span_id", "parent_span_id", "events_json", "links_json"} { + if slices.Contains(encodings[column], format.RLEDictionary) { + t.Errorf("%s is dictionary encoded: %v", column, encodings[column]) + } + } + // The encoding is physical layout, not schema: the fixed format-4 schema + // still matches, so files written either way remain readable together. + if !parquet.EqualNodes(file.Schema(), telemetrySchema[spanParquetRow]()) { + t.Fatal("dictionary encoding changed the physical schema") + } + rows, totals, err := store.Trace(t.Context(), TraceQuery{TraceID: spans[7].TraceID, StartNanos: 0, EndNanos: 1000, Limit: 10}) + if err != nil || len(rows) != 1 || totals.Spans != 1 || rows[0].Kind != "SPAN_KIND_SERVER" || rows[0].StatusCode != "STATUS_CODE_UNSET" { + t.Fatalf("round trip: %#v %#v %v", rows, totals, err) + } +} + +// HTTP routes are client-supplied text. Past parquetDictionaryMaxBytes the +// column continues in plain pages, so distinct routes stop growing its +// dictionary once a page flushes, and every row still reads back. +func TestSpanDictionaryFallsBackToPlainPastItsBound(t *testing.T) { + store, err := OpenParquetStore(t.TempDir()) + if err != nil { + t.Fatal(err) + } + spans := make([]Span, 20_000) + for i := range spans { + spans[i] = Span{Namespace: "prod", TraceID: fmt.Sprintf("%032x", i), SpanID: fmt.Sprintf("%016x", i), + ServiceName: "checkout", HTTPRoute: fmt.Sprintf("/%0100d", i), Name: "op", Kind: "SPAN_KIND_SERVER", StatusCode: "STATUS_CODE_OK", + StartUnixNanos: int64(i + 1), EndUnixNanos: int64(i + 2), IngestedAt: 999} + } + if err := store.CommitBatch(t.Context(), BatchMetadata{ID: "bounded"}, spans, nil, nil); err != nil { + t.Fatal(err) + } + f, err := os.Open(filepath.Join(store.BatchPath("bounded"), "spans.parquet")) + if err != nil { + t.Fatal(err) + } + defer f.Close() + info, _ := f.Stat() + file, err := parquet.OpenFile(f, info.Size()) + if err != nil { + t.Fatal(err) + } + var stats []format.PageEncodingStats + for _, chunk := range file.Metadata().RowGroups[0].Columns { + if chunk.MetaData.PathInSchema[0] == "http_route" { + stats = chunk.MetaData.EncodingStats + } + } + plainPages := false + for _, s := range stats { + plainPages = plainPages || (s.PageType != format.DictionaryPage && s.Encoding == format.Plain) + } + if !plainPages { + t.Fatalf("http_route stayed dictionary encoded past its bound: %+v", stats) + } + rows := make([]spanParquetRow, len(spans)) + if n, err := parquet.NewGenericReader[spanParquetRow](file).Read(rows); n != len(spans) || (err != nil && !errors.Is(err, io.EOF)) { + t.Fatalf("read %d rows: %v", n, err) + } + routes := map[string]bool{} + for _, row := range rows { + routes[row.HTTPRoute] = true + } + if len(routes) != len(spans) || !routes[spans[len(spans)-1].HTTPRoute] { + t.Fatalf("read back %d distinct routes, want %d", len(routes), len(spans)) + } +} diff --git a/internal/telemetry/parquet_rows.go b/internal/telemetry/parquet_rows.go index 9d0c4a4d..08f66cc2 100644 --- a/internal/telemetry/parquet_rows.go +++ b/internal/telemetry/parquet_rows.go @@ -3,14 +3,19 @@ package telemetry // Dictionary encoding is applied per column chunk and holds every distinct // value for the whole row group, which ingest flushes only at // parquetRowGroupRows. That pays for itself on a column whose values repeat — -// service names, HTTP methods, scope identifiers, the resource block shared by -// every span in a ResourceSpans — and costs a full second copy of the payload -// on a column whose values do not. Per-row events, links and exemplars are -// near-unique by construction and are written plain. Attributes and resources -// use shredded VARIANT columns with a bounded shared metadata dictionary. -// Dictionary-encoding whole attribute payloads made one 50k-span commit -// hold several times the batch it was handed, which is what -// TestCommitBatchPeakHeapStaysNearPayloadSize pins. +// span kinds and statuses, HTTP methods, scope identifiers, the resource +// block shared by every span in a ResourceSpans — and costs a full second copy +// of the payload on a column whose values do not. Per-row IDs, events, links +// and exemplars are near-unique by construction and are written plain. Span +// namespaces, services and operation names stay plain too: clients bound +// neither their length nor their count, and parquetDictionaryMaxBytes is only +// checked when a page flushes. Attributes and resources use shredded VARIANT +// columns with a bounded shared metadata dictionary. +// Dictionary-encoding whole attribute payloads made one 50k-span commit hold +// several times the batch it was handed; commit_memory_test.go bounds the +// allocation ratio. Dashboard scans filter and group on the low-cardinality +// span columns, and the engine evaluates those predicates against a +// dictionary instead of decoding one string per row. type spanParquetRow struct { Namespace string `parquet:"namespace"` @@ -20,13 +25,13 @@ type spanParquetRow struct { ParentSpanID string `parquet:"parent_span_id"` Service string `parquet:"service"` Operation string `parquet:"operation"` - Kind string `parquet:"kind"` + Kind string `parquet:"kind,dict"` StartTime int64 `parquet:"start_time,timestamp(nanosecond)"` EndTime int64 `parquet:"end_time,timestamp(nanosecond)"` StartUnixNano int64 `parquet:"start_unix_nano"` EndUnixNano int64 `parquet:"end_unix_nano"` DurationMS float64 `parquet:"duration_ms"` - Status string `parquet:"status"` + Status string `parquet:"status,dict"` StatusMessage string `parquet:"status_message"` Resource any `parquet:"resource,variant"` Attributes any `parquet:"attributes,variant"` diff --git a/internal/telemetry/parquet_schema.go b/internal/telemetry/parquet_schema.go index ae3f1a40..88d267ff 100644 --- a/internal/telemetry/parquet_schema.go +++ b/internal/telemetry/parquet_schema.go @@ -1,10 +1,28 @@ package telemetry -import "github.com/parquet-go/parquet-go" +import ( + "reflect" + "sync" + + "github.com/parquet-go/parquet-go" +) + +// Schemas are immutable once built, so one per row type serves every commit, +// compaction and validation instead of reflecting over the struct each time. +var telemetrySchemas sync.Map // reflect.Type -> *parquet.Schema // Every writer and full-row reader uses this physical schema, including empty // schema batches and compaction. Dotted OTel names remain literal object keys. func telemetrySchema[T any]() *parquet.Schema { + key := reflect.TypeFor[T]() + if cached, ok := telemetrySchemas.Load(key); ok { + return cached.(*parquet.Schema) + } + schema, _ := telemetrySchemas.LoadOrStore(key, buildTelemetrySchema[T]()) + return schema.(*parquet.Schema) +} + +func buildTelemetrySchema[T any]() *parquet.Schema { base := parquet.SchemaOf(new(T)) fields := parquet.Group{} for _, field := range base.Fields() { diff --git a/internal/telemetry/store/admission_test.go b/internal/telemetry/store/admission_test.go index 9309423d..c63613a2 100644 --- a/internal/telemetry/store/admission_test.go +++ b/internal/telemetry/store/admission_test.go @@ -24,20 +24,20 @@ func TestSubmitRefusesWorkBeyondTheInFlightBudget(t *testing.T) { w.inFlightBudget = 4096 big := Batch{Spans: []telemetry.Span{{Attributes: map[string]any{"payload": strings.Repeat("x", 8192)}}}} - if got := w.reserve(big); got != nil { + if got := w.reserve(batchBytes(big)); got != nil { t.Fatalf("a lone request must be admitted however large, got %v", got) } // Now over budget on a single oversized reservation. small := Batch{Spans: []telemetry.Span{{Attributes: map[string]any{"payload": strings.Repeat("x", 16)}}}} - if err := w.reserve(small); !errors.Is(err, ErrIngestOverBudget) { + if err := w.reserve(batchBytes(small)); !errors.Is(err, ErrIngestOverBudget) { t.Errorf("reserve while over budget = %v, want ErrIngestOverBudget", err) } - w.release(big) - if err := w.reserve(small); err != nil { + w.release(batchBytes(big)) + if err := w.reserve(batchBytes(small)); err != nil { t.Errorf("reserve after the budget was released = %v, want nil", err) } - w.release(small) + w.release(batchBytes(small)) if left := w.inFlightBytes.Load(); left != 0 { t.Errorf("in-flight bytes = %d after every release, want 0", left) } @@ -50,7 +50,7 @@ func TestSubmitBudgetOfZeroAdmitsEverything(t *testing.T) { w.inFlightBudget = 0 huge := Batch{Spans: []telemetry.Span{{Attributes: map[string]any{"payload": strings.Repeat("x", 1<<20)}}}} for range 64 { - if err := w.reserve(huge); err != nil { + if err := w.reserve(batchBytes(huge)); err != nil { t.Fatalf("unbounded budget refused a request: %v", err) } } @@ -63,7 +63,7 @@ func TestSubmitOverBudgetIsReportedToTheCaller(t *testing.T) { w := NewWriter(nil, 1000) w.inFlightBudget = 1 batch := Batch{Spans: []telemetry.Span{{Attributes: map[string]any{"payload": strings.Repeat("x", 4096)}}}} - if err := w.reserve(batch); err != nil { + if err := w.reserve(batchBytes(batch)); err != nil { t.Fatalf("first reservation must be admitted: %v", err) } ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) diff --git a/internal/telemetry/store/batch_bytes_test.go b/internal/telemetry/store/batch_bytes_test.go index 087e0e7b..f2f18ca2 100644 --- a/internal/telemetry/store/batch_bytes_test.go +++ b/internal/telemetry/store/batch_bytes_test.go @@ -43,21 +43,21 @@ func TestRichSharedResourcesReachRowTargetAndUseActualAdmissionBytes(t *testing. return batch } first, second := newExport(), newExport() - if !groupBatchFits(batchBytes(first), second, 1) { + if !groupBatchFits(batchBytes(first), batchBytes(second), 1) { t.Fatal("24-key shared resources prematurely cut a 50k-row group") } w := NewWriter(nil, maxGroupBatchRows) w.SetInFlightBudget(maxGroupBatchBytes) for _, batch := range []Batch{first, second} { - if err := w.reserve(batch); err != nil { + if err := w.reserve(batchBytes(batch)); err != nil { t.Fatalf("actual retained exports fit the budget: %v", err) } } - if err := w.reserve(newExport()); err == nil { + if err := w.reserve(batchBytes(newExport())); err == nil { t.Fatal("distinct requests must each charge their own resource and row payloads") } - w.release(first) - w.release(second) + w.release(batchBytes(first)) + w.release(batchBytes(second)) if got := w.inFlightBytes.Load(); got != 0 { t.Fatalf("reservation leak: %d", got) } @@ -98,10 +98,10 @@ func TestGroupBatchBytesAdmitsALoneOversizedRequest(t *testing.T) { if got := batchBytes(huge); got <= maxGroupBatchBytes { t.Fatalf("fixture is wrong: batchBytes = %d, want more than the %d ceiling", got, maxGroupBatchBytes) } - if !groupBatchFits(0, huge, 0) { + if !groupBatchFits(0, batchBytes(huge), 0) { t.Error("an empty group must admit any single request, however large") } - if groupBatchFits(maxGroupBatchBytes, huge, 1) { + if groupBatchFits(maxGroupBatchBytes, batchBytes(huge), 1) { t.Error("a non-empty group must not accept a request that takes it past the ceiling") } } diff --git a/internal/telemetry/store/writer.go b/internal/telemetry/store/writer.go index c1decb15..860a831c 100644 --- a/internal/telemetry/store/writer.go +++ b/internal/telemetry/store/writer.go @@ -48,6 +48,10 @@ type Writer struct { type submission struct { batch Batch + // bytes is batchBytes(batch), measured once at submission: the estimate + // walks every attribute map, and admission, grouping and release all + // need the same figure. + bytes int ack chan error } @@ -62,18 +66,19 @@ type commitJob struct { // point, and a dropped request would defeat it. var ErrIngestOverBudget = errors.New("telemetry ingest is over its in-flight byte budget") -// reserve charges a batch against the in-flight budget, or refuses it. +// reserve charges a batch's measured bytes against the in-flight budget, +// or refuses it. // // An empty ledger admits anything, however large. One request is one atomic // batch and refusing the only thing in flight would shed load the process is // not actually short of -- the budget exists to stop requests accumulating, // not to reject a single large one. A budget of zero is unbounded, so an // unconfigured deployment behaves exactly as it did before. -func (w *Writer) reserve(batch Batch) error { +func (w *Writer) reserve(bytes int) error { if w.inFlightBudget <= 0 { return nil } - cost := int64(batchBytes(batch)) + cost := int64(bytes) for { current := w.inFlightBytes.Load() if current > 0 && current > int64(w.inFlightBudget)-cost { @@ -86,11 +91,11 @@ func (w *Writer) reserve(batch Batch) error { } } -func (w *Writer) release(batch Batch) { +func (w *Writer) release(bytes int) { if w.inFlightBudget <= 0 { return } - w.inFlightBytes.Add(-int64(batchBytes(batch))) + w.inFlightBytes.Add(-int64(bytes)) } func NewWriter(repository *Repository, batchSize int) *Writer { @@ -117,11 +122,12 @@ func (w *Writer) Submit(ctx context.Context, batch Batch) error { // Charged here rather than at the transport because both transports funnel // through this call, and it already blocks until durable acknowledgement -- // so the ledger measures exactly the bytes the process is holding. - if err := w.reserve(batch); err != nil { + bytes := batchBytes(batch) + if err := w.reserve(bytes); err != nil { return err } - defer w.release(batch) - request := submission{batch: batch, ack: make(chan error, 1)} + defer w.release(bytes) + request := submission{batch: batch, bytes: bytes, ack: make(chan error, 1)} select { case w.submissions <- request: metrics.UpdateQueueDepth("batch", len(w.submissions)) @@ -251,13 +257,13 @@ drained: if len(group) > 0 && rows+nextRows > limit { break } - if !groupBatchFits(groupBytes, next.batch, len(group)) { + if !groupBatchFits(groupBytes, next.bytes, len(group)) { break } requests = requests[1:] group = append(group, next) rows += nextRows - groupBytes += batchBytes(next.batch) + groupBytes += next.bytes batch.Spans = append(batch.Spans, next.batch.Spans...) batch.Logs = append(batch.Logs, next.batch.Logs...) batch.Metrics = append(batch.Metrics, next.batch.Metrics...) @@ -400,11 +406,11 @@ func batchBytes(batch Batch) int { // atomic directory, and refusing it would drop data rather than batch it more // carefully -- the same reason the row ceiling lets a lone oversized request // through. -func groupBatchFits(groupBytes int, next Batch, groupLen int) bool { +func groupBatchFits(groupBytes, nextBytes, groupLen int) bool { if groupLen == 0 { return true } - return groupBytes <= maxGroupBatchBytes-batchBytes(next) + return groupBytes <= maxGroupBatchBytes-nextBytes } func batchRows(batch Batch) int { diff --git a/internal/telemetry/trace_index.go b/internal/telemetry/trace_index.go index 5fb8fddb..efbc499f 100644 --- a/internal/telemetry/trace_index.go +++ b/internal/telemetry/trace_index.go @@ -31,13 +31,17 @@ type traceIndex struct { } type traceIndexWriter struct { - file *os.File - path string - last traceRange - rows uint64 - entries uint64 - have bool - closed bool + file *os.File + // Entries are 24 bytes and there is one per trace; buffering them keeps + // a 50,000-span batch from costing tens of thousands of write calls. + buffered *bufio.Writer + encoded [traceIndexEntrySize]byte + path string + last traceRange + rows uint64 + entries uint64 + have bool + closed bool } func newTraceIndexWriter(path string) (*traceIndexWriter, error) { @@ -50,7 +54,7 @@ func newTraceIndexWriter(path string) (*traceIndexWriter, error) { _ = os.Remove(path) return nil, err } - return &traceIndexWriter{file: file, path: path}, nil + return &traceIndexWriter{file: file, buffered: bufio.NewWriterSize(file, 64<<10), path: path}, nil } // Append adds one Parquet row hash. Input must be ordered by hash, which lets @@ -91,6 +95,9 @@ func (w *traceIndexWriter) Close() error { return w.fail(err) } } + if err := w.buffered.Flush(); err != nil { + return w.fail(err) + } var header [traceIndexHeaderSize]byte copy(header[:], traceIndexMagic) binary.LittleEndian.PutUint64(header[8:], w.entries) @@ -117,11 +124,11 @@ func (w *traceIndexWriter) Abort() { } func (w *traceIndexWriter) flush() error { - var encoded [traceIndexEntrySize]byte + encoded := w.encoded[:] binary.LittleEndian.PutUint64(encoded[0:8], w.last.hash) binary.LittleEndian.PutUint64(encoded[8:16], w.last.row) binary.LittleEndian.PutUint64(encoded[16:24], w.last.count) - if _, err := w.file.Write(encoded[:]); err != nil { + if _, err := w.buffered.Write(encoded[:]); err != nil { return err } w.entries++ From 7361cd03987a1f79344d7477211c4d705d69074c Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 14 Oct 2026 22:20:51 -0700 Subject: [PATCH 4/4] fix(query): keep cancelled reads from poisoning pooled connections A read transaction begun with the request context was rolled back by database/sql on its own goroutine when that context was cancelled, as a dashboard refresh does to the requests it supersedes. DuckDB could deliver the cancelled statement's interrupt to that ROLLBACK, so the transaction stayed open and the pooled connection failed every later read with "Current transaction is aborted" or "cannot start a transaction within a transaction". Cancelling reads in a loop poisoned the pool in 4 of 7 runs on DuckDB v2.0.0-alpha46057 and 1 of 7 on alpha43763. Read transactions now begin with a context that cannot be cancelled and are rolled back by their owner; statements inside still run under the request context. A connection whose BEGIN or ROLLBACK fails is discarded instead of returned to the pool. The same loop then passed 6 of 6 runs. --- internal/query/duck.go | 57 ++++++++++++++++++++++++++++--------- internal/query/duck_test.go | 46 ++++++++++++++++++++++++++++++ 2 files changed, 89 insertions(+), 14 deletions(-) diff --git a/internal/query/duck.go b/internal/query/duck.go index 7e8458ce..e2f2c72f 100644 --- a/internal/query/duck.go +++ b/internal/query/duck.go @@ -1661,16 +1661,13 @@ func (d *Duck) QueryContext(ctx context.Context, query string, args ...any) (que var tx *sql.Tx var db snapshotSQL = conn if window, ok := queryrows.ReadWindow(ctx); ok && d.repository != nil { - tx, err = conn.BeginTx(ctx, nil) + tx, err = beginRead(ctx, conn) if err == nil { db = tx query, err = d.bindSnapshot(ctx, tx, query, window) } if err != nil { - if tx != nil { - _ = tx.Rollback() - } - _ = conn.Close() + endRead(conn, tx) d.parquetMu.RUnlock() return nil, err } @@ -1679,17 +1676,11 @@ func (d *Duck) QueryContext(ctx context.Context, query string, args ...any) (que rows, err := db.QueryContext(ctx, query, args...) metrics.DuckDBStatement.Observe(time.Since(started).Seconds()) if err != nil { - if tx != nil { - _ = tx.Rollback() - } - _ = conn.Close() + endRead(conn, tx) d.parquetMu.RUnlock() return nil, err } release := func() { - if tx != nil { - _ = tx.Rollback() - } // The connection goes back to the pool only once its rows are done // with it, which is what closing the rows first guarantees. // @@ -1698,12 +1689,41 @@ func (d *Duck) QueryContext(ctx context.Context, query string, args ...any) (que // return the connection when the context died, and an explicitly held // one is the caller's until it says otherwise. Every call site closes // on every path; this is what it costs if one stops. - _ = conn.Close() + endRead(conn, tx) d.parquetMu.RUnlock() } return &lockedRows{Rows: rows, unlock: release}, nil } +// beginRead opens a read transaction that only its owner ends. A transaction +// begun with the request context is rolled back by database/sql from another +// goroutine when that context is cancelled, while DuckDB may still deliver the +// cancelled statement's interrupt to the ROLLBACK. The rollback then fails, +// the transaction stays open, and the pooled connection fails every later read. +// Statements inside still run under ctx, so cancellation still stops them. +func beginRead(ctx context.Context, conn *sql.Conn) (*sql.Tx, error) { + tx, err := conn.BeginTx(context.WithoutCancel(ctx), nil) + if err != nil { + discardConn(conn) + } + return tx, err +} + +// endRead rolls back tx, if any, and returns conn to the pool. A connection +// whose rollback failed may still hold the transaction, so it is discarded +// instead of being handed to the next reader. +func endRead(conn *sql.Conn, tx *sql.Tx) { + if tx != nil && tx.Rollback() != nil { + discardConn(conn) + } + _ = conn.Close() +} + +// discardConn closes conn's driver connection rather than pooling it. +func discardConn(conn *sql.Conn) { + _ = conn.Raw(func(any) error { return driver.ErrBadConn }) +} + type lockedRows struct { *sql.Rows unlockOnce sync.Once @@ -2047,5 +2067,14 @@ LIMIT %d; // WithReadTransaction keeps derived annotation rows and completeness flags in // one read snapshot. Its callback only receives the query surface. func (d *Duck) WithReadTransaction(ctx context.Context, read func(queryrows.Queryer) error) error { - return (queryrows.SQLAdapter{DB: d.DB}).WithReadTransaction(ctx, read) + conn, err := d.acquireRead(ctx) + if err != nil { + return err + } + tx, err := beginRead(ctx, conn) + defer endRead(conn, tx) + if err != nil { + return err + } + return read(queryrows.SQLAdapter{DB: tx}) } diff --git a/internal/query/duck_test.go b/internal/query/duck_test.go index 3095731f..46313fa6 100644 --- a/internal/query/duck_test.go +++ b/internal/query/duck_test.go @@ -222,6 +222,52 @@ WHERE bucket >= ? AND bucket < ?`, eventTime.Add(-time.Minute), eventTime.Add(ti } } +// A cancelled statement's interrupt can reach the ROLLBACK that ends its read +// transaction, leaving the transaction open. The read must roll back itself +// after cancellation, and a connection whose rollback failed must leave the +// pool instead of failing every later read. +func TestReadTransactionRollbackFailureDiscardsTheConnection(t *testing.T) { + for _, rollbackErr := range []error{nil, errors.New("INTERRUPT Error: Interrupted!")} { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatal(err) + } + conn, err := db.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + mock.ExpectBegin() + tx, err := beginRead(ctx, conn) + if err != nil { + t.Fatal(err) + } + rollback := mock.ExpectRollback() + if rollbackErr != nil { + rollback.WillReturnError(rollbackErr) + } + cancel() + // database/sql rolls back a transaction begun with a cancelled context + // on its own goroutine, where a failed rollback cannot be seen. + time.Sleep(20 * time.Millisecond) + if mock.ExpectationsWereMet() == nil { + t.Fatal("cancellation rolled the read transaction back before its owner ended it") + } + endRead(conn, tx) + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } + want := 1 + if rollbackErr != nil { + want = 0 + } + if open := db.Stats().OpenConnections; open != want { + t.Errorf("rollback error %v: %d pooled connections, want %d", rollbackErr, open, want) + } + _ = db.Close() + } +} + func TestQueryContextHoldsParquetLockUntilRowsFinish(t *testing.T) { db, mock, err := sqlmock.New() if err != nil {