perf(telemetry): cut commit allocation; fix(query): cancelled reads poisoning the pool; pin DuckDB v2.0.0-alpha46057

.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]

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.

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 ---
 

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

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 {

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 {

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.

internal/panel/diagnose.go

@@ -55,21 +55,24 @@ 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 ""
 	}
 	if total == 0 {
 		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 {

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)
 	}
 }