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]
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/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/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/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/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 {
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/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++
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 ---