feat(perf): compare uncached and cached DuckDB

tests/mw-dev/scenario/perf/adapter_test.go

@@ -99,6 +99,48 @@ func TestExecutorRunsPerfStepAndWritesArtifacts(t *testing.T) {
 	}
 }
 
+func TestExecutorRecordsBothPGWireCacheVariantsWithEqualResources(t *testing.T) {
+	targets := []perfcore.Protocol{perfcore.ProtocolPGWireUncached, perfcore.ProtocolPGWireCached}
+	factory := &fakeDriverFactory{}
+	executor := NewExecutor(ExecutorConfig{
+		Connection: scenariosql.ConnectionConfig{DialHost: "127.0.0.1", SNISuffix: ".example.test", SSLMode: "require"},
+		OutputDir:  t.TempDir(), DriverFactory: factory,
+	})
+	err := executor.ExecuteStep(context.Background(), core.Step{
+		ID: "cache-comparison", Type: StepTypePerfQueries,
+		With: map[string]any{
+			"org_id": "scenario-org", "username": "root", "password": "test-password",
+			"catalog_file": writePerfCatalog(t, targets), "targets": []any{"pgwire_uncached", "pgwire_cached"},
+			"run_id": "cache-run", "worker_cpu": "3", "worker_memory": "12Gi",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	result, ok := executor.State().Result("cache-comparison")
+	if !ok || result.Summary.RunID != "cache-run" || result.Summary.TotalQueries != 2 || result.Summary.WarmupQueries != 2 {
+		t.Fatalf("unexpected result: %+v", result)
+	}
+	if len(factory.pgwireConnections) != 2 {
+		t.Fatalf("PGWire factories called for %d targets, want 2", len(factory.pgwireConnections))
+	}
+	for _, protocol := range targets {
+		connection, ok := factory.pgwireConnections[protocol]
+		if !ok || !strings.Contains(connection.DSN, "options='-c duckgres.worker_cpu=3 -c duckgres.worker_memory=12Gi'") {
+			t.Fatalf("%s connection missing standard resources: %+v", protocol, connection)
+		}
+	}
+	raw, err := os.ReadFile(filepath.Join(result.OutputDir, "query_results.csv"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, protocol := range targets {
+		if !strings.Contains(string(raw), "\nq1,i1,1,"+string(protocol)+",ok,") {
+			t.Fatalf("missing %s result for shared query/intent: %s", protocol, raw)
+		}
+	}
+}
+
 func TestExecutorRestrictsCatalogToStepTargets(t *testing.T) {
 	catalogPath := writePerfCatalog(t, []perfcore.Protocol{perfcore.ProtocolPGWire})
 	provisionState := provision.NewState()
@@ -487,20 +529,25 @@ func writePerfCatalog(t *testing.T, targets []perfcore.Protocol) string {
 }
 
 type fakeDriverFactory struct {
-	pgwireConnection scenariosql.PGWireConnection
-	pgwireErr        error
-	pgwireDriver     *fakeProtocolDriver
-	trinoConnection  trinodriver.ConnectionConfig
-	trinoContext     context.Context
-	trinoDriver      *fakeProtocolDriver
-	athenaConnection athenadriver.ConnectionConfig
-	athenaContext    context.Context
-	athenaDriver     *fakeProtocolDriver
+	pgwireConnections map[perfcore.Protocol]scenariosql.PGWireConnection
+	pgwireConnection  scenariosql.PGWireConnection
+	pgwireErr         error
+	pgwireDriver      *fakeProtocolDriver
+	trinoConnection   trinodriver.ConnectionConfig
+	trinoContext      context.Context
+	trinoDriver       *fakeProtocolDriver
+	athenaConnection  athenadriver.ConnectionConfig
+	athenaContext     context.Context
+	athenaDriver      *fakeProtocolDriver
 }
 
-func (f *fakeDriverFactory) NewPGWire(connection scenariosql.PGWireConnection) (perfcore.ProtocolDriver, error) {
+func (f *fakeDriverFactory) NewPGWire(connection scenariosql.PGWireConnection, protocol perfcore.Protocol) (perfcore.ProtocolDriver, error) {
+	if f.pgwireConnections == nil {
+		f.pgwireConnections = make(map[perfcore.Protocol]scenariosql.PGWireConnection)
+	}
+	f.pgwireConnections[protocol] = connection
 	f.pgwireConnection = connection
-	f.pgwireDriver = &fakeProtocolDriver{protocol: perfcore.ProtocolPGWire, err: f.pgwireErr}
+	f.pgwireDriver = &fakeProtocolDriver{protocol: protocol, err: f.pgwireErr}
 	return f.pgwireDriver, nil
 }
 

tests/mw-dev/scenario/perf/steps.go

@@ -21,7 +21,7 @@ import (
 const StepTypePerfQueries = "perf_queries"
 
 type DriverFactory interface {
-	NewPGWire(connection scenariosql.PGWireConnection) (perfcore.ProtocolDriver, error)
+	NewPGWire(connection scenariosql.PGWireConnection, protocol perfcore.Protocol) (perfcore.ProtocolDriver, error)
 	NewTrino(ctx context.Context, connection trinodriver.ConnectionConfig) (perfcore.ProtocolDriver, error)
 	NewAthena(ctx context.Context, connection athenadriver.ConnectionConfig) (perfcore.ProtocolDriver, error)
 }
@@ -312,7 +312,7 @@ func targetsFromWith(step core.Step) ([]perfcore.Protocol, error) {
 		}
 		target := perfcore.Protocol(value)
 		switch target {
-		case perfcore.ProtocolPGWire, perfcore.ProtocolTrino, perfcore.ProtocolAthena:
+		case perfcore.ProtocolPGWire, perfcore.ProtocolPGWireUncached, perfcore.ProtocolPGWireCached, perfcore.ProtocolTrino, perfcore.ProtocolAthena:
 		default:
 			return nil, classified(ErrorClassConfig, fmt.Errorf("step %s with.targets[%d] has unsupported perf protocol %q", step.ID, i, target))
 		}
@@ -356,12 +356,12 @@ func (e *Executor) driversForCatalog(ctx context.Context, catalog perfcore.Catal
 			continue
 		}
 		switch target {
-		case perfcore.ProtocolPGWire:
+		case perfcore.ProtocolPGWire, perfcore.ProtocolPGWireUncached, perfcore.ProtocolPGWireCached:
 			connection, err := e.pgwireConnection(spec)
 			if err != nil {
 				return nil, err
 			}
-			driver, err := e.driverFactory.NewPGWire(connection)
+			driver, err := e.driverFactory.NewPGWire(connection, target)
 			if err != nil {
 				return nil, classified(ErrorClassConfig, fmt.Errorf("create pgwire perf driver: %w", err))
 			}
@@ -483,12 +483,17 @@ func closeDrivers(drivers map[perfcore.Protocol]perfcore.ProtocolDriver) {
 	}
 }
 
-func (defaultDriverFactory) NewPGWire(connection scenariosql.PGWireConnection) (perfcore.ProtocolDriver, error) {
+func (defaultDriverFactory) NewPGWire(connection scenariosql.PGWireConnection, protocol perfcore.Protocol) (perfcore.ProtocolDriver, error) {
 	db, err := connection.OpenDB()
 	if err != nil {
 		return nil, err
 	}
-	return pgdriver.NewWithDB(db), nil
+	driver, err := pgdriver.NewWithDBAndProtocol(db, protocol)
+	if err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+	return driver, nil
 }
 
 func (defaultDriverFactory) NewTrino(ctx context.Context, connection trinodriver.ConnectionConfig) (perfcore.ProtocolDriver, error) {

tests/mw-dev/scenario/runner_test.go

@@ -745,8 +745,8 @@ func assertPerfTargetsOnlyPGWire(t *testing.T, step core.Step) {
 func assertPerfTargetsPGWireTrinoAndAthena(t *testing.T, step core.Step) {
 	t.Helper()
 	targets, ok := step.With["targets"].([]any)
-	if !ok || len(targets) != 3 || targets[0] != "pgwire" || targets[1] != "trino" || targets[2] != "athena" {
-		t.Fatalf("perf step %s targets = %#v, want [pgwire trino athena]", step.ID, step.With["targets"])
+	if !ok || len(targets) != 4 || targets[0] != "pgwire_uncached" || targets[1] != "pgwire_cached" || targets[2] != "trino" || targets[3] != "athena" {
+		t.Fatalf("perf step %s targets = %#v, want [pgwire_uncached pgwire_cached trino athena]", step.ID, step.With["targets"])
 	}
 }
 

tests/mw-dev/scenario/scenarios/posthog_frozen_perf.yaml

@@ -80,7 +80,7 @@ steps:
       org_id: ${env:DUCKGRES_SCENARIO_ORG_ID}
       catalog: ducklake
       catalog_file: ../../../perf/queries/ducklake_posthog_tables.yaml
-      targets: [pgwire, trino, athena]
+      targets: [pgwire_uncached, pgwire_cached, trino, athena]
       trino_ca_cert_file: ${env:DUCKGRES_SCENARIO_TRINO_CA_CERT}
       trino_startup_timeout: 2m
       trino_startup_poll_interval: 2s

tests/perf/README.md

@@ -4,7 +4,8 @@ This package contains the golden-query performance harness.
 
 ## Protocol Drivers
 
-Catalogs may target `pgwire`, `trino`, `athena`, or any supported combination. All drivers execute the same
+Catalogs may target `pgwire`, `pgwire_uncached`, `pgwire_cached`, `trino`,
+`athena`, or any supported combination. All drivers execute the same
 rendered statement stored in the existing `pgwire_sql` catalog field; the
 legacy field name is retained for catalog compatibility and must not be used
 to create a second, protocol-specific query definition. Keep shared benchmark
@@ -32,9 +33,48 @@ readiness, Kubernetes Secret projection, and file-authenticator refresh.
 When a catalog targets multiple protocols, the runner completes all warmup and
 measured iterations for one protocol before starting the next protocol in the
 catalog's declared target order. This keeps each protocol's connection and
-worker cache active throughout its measurements and prevents slow queries in
+worker context active throughout its measurements and prevents slow queries in
 one protocol from changing another protocol's cache context.
 
+### DuckDB cache comparison
+
+`posthog_frozen_perf` runs four separately labeled targets in one result set,
+in order: **`pgwire_uncached` (baseline)**, `pgwire_cached`, `trino`, `athena`.
+Both DuckDB variants use identical queries, worker resource requests, warmup
+counts, and measured iterations. Each variant finishes before the next begins;
+Trino and Athena run once, not once per cache mode. Legacy `pgwire` catalogs
+keep their existing behavior and are not relabeled as uncached history.
+
+Only the perf driver changes cache settings. It pins its PGWire connection and
+applies `SET GLOBAL` before that variant's first warmup and outside query timing:
+
+| Setting | Uncached baseline | Cached |
+| --- | --- | --- |
+| `enable_external_file_cache` | `false` | `true` |
+| `parquet_metadata_cache` | `false` | `false` |
+| `enable_http_metadata_cache` | `false` | `false` |
+
+The cached variant matches DuckDB's default remote-file caching policy rather
+than enabling metadata caches that ordinary workers leave off. Setup is lazy,
+so the cached driver's construction cannot enable caching during the uncached
+phase. A lost pinned connection fails the query instead of silently connecting
+to an unconfigured worker. Worker startup and production configuration are
+unchanged.
+
+Use the isolated scenario workflow for this comparison. Local
+`just scenario-frozen-perf` runs require a dedicated test warehouse: these
+settings affect the underlying worker globally, and the two variants must not
+run concurrently against a shared worker. The scenario's sequential phases
+also work when connections reuse the same worker. If cache setup fails, inspect
+the artifact errors and recreate the test stack; do not publish fallback runs
+under either explicit cache-mode label. Teardown removes the test workers.
+
+Warmup does not imply that the entire dataset fits in memory. Query-local
+buffering, prefetching, DuckLake catalog caching, and separate cache
+proxies/extensions are unaffected, so "uncached" here is not a fully cold
+end-to-end read path. The paired query and intent IDs use `balanced_v4` to
+separate this methodology from `balanced_v3`; the dataset is unchanged.
+
 ## Paired Query Catalogs
 
 Existing catalogs continue to use `queries:` unchanged. A catalog may contain

tests/perf/core/catalog.go

@@ -389,7 +389,7 @@ func validateCatalog(c Catalog) error {
 	seenTargets := map[Protocol]struct{}{}
 	for _, target := range c.Targets {
 		switch target {
-		case ProtocolPGWire, ProtocolTrino, ProtocolAthena:
+		case ProtocolPGWire, ProtocolPGWireUncached, ProtocolPGWireCached, ProtocolTrino, ProtocolAthena:
 		default:
 			return fmt.Errorf("unsupported target protocol %q", target)
 		}

tests/perf/core/catalog_test.go

@@ -26,7 +26,7 @@ func TestCheckedInCatalogsLoad(t *testing.T) {
 			}
 			wantTargets := []Protocol{ProtocolPGWire}
 			if filepath.Base(path) == "ducklake_posthog_tables.yaml" {
-				wantTargets = []Protocol{ProtocolPGWire, ProtocolTrino, ProtocolAthena}
+				wantTargets = []Protocol{ProtocolPGWireUncached, ProtocolPGWireCached, ProtocolTrino, ProtocolAthena}
 			}
 			if !reflect.DeepEqual(catalog.Targets, wantTargets) {
 				t.Fatalf("catalog targets = %v, want %v", catalog.Targets, wantTargets)
@@ -49,27 +49,27 @@ func TestCheckedInPostHogCatalogPublishesCompleteStablePairs(t *testing.T) {
 		t.Fatalf("LoadCatalog: %v", err)
 	}
 	want := []string{
-		"q_events_total_balanced_v3__raw_view",
-		"q_events_total_balanced_v3__ducklake_table",
-		"q_events_total_balanced_v3__athena_external",
-		"q_events_count_one_day_balanced_v3__raw_view",
-		"q_events_count_one_day_balanced_v3__ducklake_table",
-		"q_events_count_one_day_balanced_v3__athena_external",
-		"q_events_by_name_march_2026_balanced_v3__raw_view",
-		"q_events_by_name_march_2026_balanced_v3__ducklake_table",
-		"q_events_by_name_march_2026_balanced_v3__athena_external",
-		"q_events_distinct_persons_balanced_v3__raw_view",
-		"q_events_distinct_persons_balanced_v3__ducklake_table",
-		"q_events_distinct_persons_balanced_v3__athena_external",
-		"q_persons_total_balanced_v3__raw_view",
-		"q_persons_total_balanced_v3__ducklake_table",
-		"q_persons_total_balanced_v3__athena_external",
-		"q_persons_daily_april_2026_balanced_v3__raw_view",
-		"q_persons_daily_april_2026_balanced_v3__ducklake_table",
-		"q_persons_daily_april_2026_balanced_v3__athena_external",
-		"q_events_daily_march_2026_balanced_v3__raw_view",
-		"q_events_daily_march_2026_balanced_v3__ducklake_table",
-		"q_events_daily_march_2026_balanced_v3__athena_external",
+		"q_events_total_balanced_v4__raw_view",
+		"q_events_total_balanced_v4__ducklake_table",
+		"q_events_total_balanced_v4__athena_external",
+		"q_events_count_one_day_balanced_v4__raw_view",
+		"q_events_count_one_day_balanced_v4__ducklake_table",
+		"q_events_count_one_day_balanced_v4__athena_external",
+		"q_events_by_name_march_2026_balanced_v4__raw_view",
+		"q_events_by_name_march_2026_balanced_v4__ducklake_table",
+		"q_events_by_name_march_2026_balanced_v4__athena_external",
+		"q_events_distinct_persons_balanced_v4__raw_view",
+		"q_events_distinct_persons_balanced_v4__ducklake_table",
+		"q_events_distinct_persons_balanced_v4__athena_external",
+		"q_persons_total_balanced_v4__raw_view",
+		"q_persons_total_balanced_v4__ducklake_table",
+		"q_persons_total_balanced_v4__athena_external",
+		"q_persons_daily_april_2026_balanced_v4__raw_view",
+		"q_persons_daily_april_2026_balanced_v4__ducklake_table",
+		"q_persons_daily_april_2026_balanced_v4__athena_external",
+		"q_events_daily_march_2026_balanced_v4__raw_view",
+		"q_events_daily_march_2026_balanced_v4__ducklake_table",
+		"q_events_daily_march_2026_balanced_v4__athena_external",
 	}
 	if got := queryIDs(catalog); !reflect.DeepEqual(got, want) {
 		t.Fatalf("checked-in PostHog query IDs changed: got %v want %v", got, want)
@@ -78,7 +78,7 @@ func TestCheckedInPostHogCatalogPublishesCompleteStablePairs(t *testing.T) {
 		t.Fatalf("checked-in PostHog measure iterations = %d, want 4 for balanced target order", catalog.MeasureIterations)
 	}
 	for _, query := range catalog.Queries {
-		if !strings.HasSuffix(query.IntentID, "_balanced_v3") {
+		if !strings.HasSuffix(query.IntentID, "_balanced_v4") {
 			t.Fatalf("checked-in PostHog query %s has unversioned methodology intent %q", query.QueryID, query.IntentID)
 		}
 		if strings.Contains(query.PGWireSQL, "TIMESTAMPTZ '") {

tests/perf/core/intent_matcher.go

@@ -13,7 +13,7 @@ func NewIntentMatcher() *IntentMatcher {
 
 func (m *IntentMatcher) SQLFor(query Query, protocol Protocol) (string, error) {
 	switch protocol {
-	case ProtocolPGWire, ProtocolTrino:
+	case ProtocolPGWire, ProtocolPGWireUncached, ProtocolPGWireCached, ProtocolTrino:
 		if strings.TrimSpace(query.CanonicalSQL()) == "" {
 			return "", fmt.Errorf("query %s missing canonical SQL", query.QueryID)
 		}

tests/perf/core/intent_matcher_test.go

@@ -9,7 +9,7 @@ func TestIntentMatcherReturnsCanonicalSQLForBothProtocols(t *testing.T) {
 		IntentID:  "i1",
 		PGWireSQL: "SELECT 1",
 	}
-	for _, protocol := range []Protocol{ProtocolPGWire, ProtocolTrino} {
+	for _, protocol := range []Protocol{ProtocolPGWire, ProtocolPGWireUncached, ProtocolPGWireCached, ProtocolTrino} {
 		if got, err := m.SQLFor(q, protocol); err != nil || got != "SELECT 1" {
 			t.Fatalf("unexpected %s SQL result: sql=%q err=%v", protocol, got, err)
 		}

tests/perf/core/runner.go

@@ -168,9 +168,9 @@ func querySupportsProtocol(query Query, protocol Protocol) bool {
 	case "":
 		return true
 	case StorageTargetRawView:
-		return protocol == ProtocolPGWire
+		return protocol == ProtocolPGWire || protocol == ProtocolPGWireUncached || protocol == ProtocolPGWireCached
 	case StorageTargetDuckLakeTable:
-		return protocol == ProtocolPGWire || protocol == ProtocolTrino
+		return protocol == ProtocolPGWire || protocol == ProtocolPGWireUncached || protocol == ProtocolPGWireCached || protocol == ProtocolTrino
 	case StorageTargetAthenaExternal:
 		return protocol == ProtocolAthena
 	default:

tests/perf/core/runner_test.go

@@ -305,6 +305,56 @@ func TestRunnerRoutesEachStorageVariantOnlyToItsComparableProtocol(t *testing.T)
 	}
 }
 
+func TestRunnerRunsUncachedAndCachedPGWireAsDistinctComparableResults(t *testing.T) {
+	uncached, cached := ProtocolPGWireUncached, ProtocolPGWireCached
+	var events []string
+	uncachedDriver := &testDriver{protocol: uncached, events: &events}
+	cachedDriver := &testDriver{protocol: cached, events: &events}
+	sink := &inMemorySink{}
+	runner := NewQueryRunner(RunnerConfig{
+		RunID: "cache-comparison",
+		Catalog: Catalog{
+			Targets: []Protocol{uncached, cached}, WarmupIterations: 1, MeasureIterations: 1,
+			Queries: []Query{
+				{QueryID: "q__raw_view", IntentID: "intent", StorageTarget: StorageTargetRawView},
+				{QueryID: "q__ducklake_table", IntentID: "intent", StorageTarget: StorageTargetDuckLakeTable},
+				{QueryID: "q__athena_external", IntentID: "intent", StorageTarget: StorageTargetAthenaExternal},
+			},
+		},
+		Drivers: map[Protocol]ProtocolDriver{uncached: uncachedDriver, cached: cachedDriver}, Sink: sink,
+	})
+	summary, err := runner.Run(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if summary.RunID != "cache-comparison" || summary.TotalQueries != 4 || summary.WarmupQueries != 4 {
+		t.Fatalf("unexpected summary: %+v", summary)
+	}
+	for _, driver := range []*testDriver{uncachedDriver, cachedDriver} {
+		if want := []string{"q__raw_view", "q__ducklake_table", "q__raw_view", "q__ducklake_table"}; !reflect.DeepEqual(driver.queryIDs, want) {
+			t.Fatalf("%s routed queries = %v, want %v", driver.protocol, driver.queryIDs, want)
+		}
+	}
+	wantEvents := []string{
+		"pgwire_uncached/q__raw_view", "pgwire_uncached/q__ducklake_table",
+		"pgwire_uncached/q__raw_view", "pgwire_uncached/q__ducklake_table",
+		"pgwire_cached/q__raw_view", "pgwire_cached/q__ducklake_table",
+		"pgwire_cached/q__raw_view", "pgwire_cached/q__ducklake_table",
+	}
+	if !reflect.DeepEqual(events, wantEvents) {
+		t.Fatalf("execution order = %v, want %v", events, wantEvents)
+	}
+	for index, result := range sink.results {
+		wantProtocol := uncached
+		if index >= 2 {
+			wantProtocol = cached
+		}
+		if result.Protocol != wantProtocol || result.IntentID != "intent" || result.MeasureIteration != 1 {
+			t.Fatalf("result %d = %+v", index, result)
+		}
+	}
+}
+
 func (d *testDriver) Close() error { return nil }
 
 type inMemorySink struct {

tests/perf/core/types.go

@@ -5,9 +5,11 @@ import "time"
 type Protocol string
 
 const (
-	ProtocolPGWire Protocol = "pgwire"
-	ProtocolTrino  Protocol = "trino"
-	ProtocolAthena Protocol = "athena"
+	ProtocolPGWire         Protocol = "pgwire"
+	ProtocolPGWireUncached Protocol = "pgwire_uncached"
+	ProtocolPGWireCached   Protocol = "pgwire_cached"
+	ProtocolTrino          Protocol = "trino"
+	ProtocolAthena         Protocol = "athena"
 )
 
 // StorageTarget identifies the physical relation family selected for a paired