Isolate perf runs by protocol

tests/perf/README.md

@@ -17,6 +17,12 @@ authenticated `SELECT 1` outside query timing, retrying for up to 2 minutes at
 2-second intervals by default. This absorbs the bounded delay between Trino
 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
+one protocol from changing another protocol's cache context.
+
 ## Paired Query Catalogs
 
 Existing catalogs continue to use `queries:` unchanged. A catalog may contain

tests/perf/core/catalog_test.go

@@ -49,20 +49,20 @@ func TestCheckedInPostHogCatalogPublishesCompleteStablePairs(t *testing.T) {
 		t.Fatalf("LoadCatalog: %v", err)
 	}
 	want := []string{
-		"q_events_total_balanced_v2__raw_view",
-		"q_events_total_balanced_v2__ducklake_table",
-		"q_events_count_one_day_balanced_v2__raw_view",
-		"q_events_count_one_day_balanced_v2__ducklake_table",
-		"q_events_by_name_march_2026_balanced_v2__raw_view",
-		"q_events_by_name_march_2026_balanced_v2__ducklake_table",
-		"q_events_distinct_persons_balanced_v2__raw_view",
-		"q_events_distinct_persons_balanced_v2__ducklake_table",
-		"q_persons_total_balanced_v2__raw_view",
-		"q_persons_total_balanced_v2__ducklake_table",
-		"q_persons_daily_april_2026_balanced_v2__raw_view",
-		"q_persons_daily_april_2026_balanced_v2__ducklake_table",
-		"q_events_daily_march_2026_balanced_v2__raw_view",
-		"q_events_daily_march_2026_balanced_v2__ducklake_table",
+		"q_events_total_balanced_v3__raw_view",
+		"q_events_total_balanced_v3__ducklake_table",
+		"q_events_count_one_day_balanced_v3__raw_view",
+		"q_events_count_one_day_balanced_v3__ducklake_table",
+		"q_events_by_name_march_2026_balanced_v3__raw_view",
+		"q_events_by_name_march_2026_balanced_v3__ducklake_table",
+		"q_events_distinct_persons_balanced_v3__raw_view",
+		"q_events_distinct_persons_balanced_v3__ducklake_table",
+		"q_persons_total_balanced_v3__raw_view",
+		"q_persons_total_balanced_v3__ducklake_table",
+		"q_persons_daily_april_2026_balanced_v3__raw_view",
+		"q_persons_daily_april_2026_balanced_v3__ducklake_table",
+		"q_events_daily_march_2026_balanced_v3__raw_view",
+		"q_events_daily_march_2026_balanced_v3__ducklake_table",
 	}
 	if got := queryIDs(catalog); !reflect.DeepEqual(got, want) {
 		t.Fatalf("checked-in PostHog query IDs changed: got %v want %v", got, want)
@@ -71,7 +71,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_v2") {
+		if !strings.HasSuffix(query.IntentID, "_balanced_v3") {
 			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/runner.go

@@ -78,16 +78,18 @@ func (r *QueryRunner) Run(ctx context.Context) (RunSummary, error) {
 		}
 	}
 
-	warmupIterations := r.cfg.Catalog.WarmupIterations
-	for i := 0; i < warmupIterations; i++ {
-		if err := r.executeIteration(ctx, false, 0, &summary); err != nil {
-			return summary, err
+	for _, protocol := range r.cfg.Catalog.Targets {
+		warmupIterations := r.cfg.Catalog.WarmupIterations
+		for i := 0; i < warmupIterations; i++ {
+			if err := r.executeIteration(ctx, protocol, false, 0, &summary); err != nil {
+				return summary, err
+			}
 		}
-	}
-	measureIterations := r.cfg.Catalog.MeasureIterations
-	for i := 0; i < measureIterations; i++ {
-		if err := r.executeIteration(ctx, true, i+1, &summary); err != nil {
-			return summary, err
+		measureIterations := r.cfg.Catalog.MeasureIterations
+		for i := 0; i < measureIterations; i++ {
+			if err := r.executeIteration(ctx, protocol, true, i+1, &summary); err != nil {
+				return summary, err
+			}
 		}
 	}
 
@@ -108,51 +110,49 @@ func (r *QueryRunner) MetricsGatherer() prometheus.Gatherer {
 	return r.metrics.Gatherer()
 }
 
-func (r *QueryRunner) executeIteration(ctx context.Context, measure bool, measureIteration int, summary *RunSummary) error {
+func (r *QueryRunner) executeIteration(ctx context.Context, protocol Protocol, measure bool, measureIteration int, summary *RunSummary) error {
 	for _, query := range queriesForIteration(r.cfg.Catalog.Queries, measureIteration) {
+		if !querySupportsProtocol(query, protocol) {
+			continue
+		}
 		args := orderedParamValues(query.Params)
-		for _, protocol := range r.cfg.Catalog.Targets {
-			if !querySupportsProtocol(query, protocol) {
-				continue
-			}
-			driver := r.cfg.Drivers[protocol]
-			started := r.cfg.Now()
-			result := QueryResult{
-				QueryID:          query.QueryID,
-				IntentID:         query.IntentID,
-				MeasureIteration: measureIteration,
-				Protocol:         protocol,
-				StartedAt:        started,
-			}
+		driver := r.cfg.Drivers[protocol]
+		started := r.cfg.Now()
+		result := QueryResult{
+			QueryID:          query.QueryID,
+			IntentID:         query.IntentID,
+			MeasureIteration: measureIteration,
+			Protocol:         protocol,
+			StartedAt:        started,
+		}
 
-			execResult, err := driver.Execute(ctx, query, args)
-			if execResult.Duration <= 0 {
-				execResult.Duration = time.Since(started)
-			}
-			result.Duration = execResult.Duration
-			result.Rows = execResult.Rows
-			if err != nil {
-				result.Status = "error"
-				result.Error = err.Error()
-				result.ErrorClass = "execution_error"
-			} else {
-				result.Status = "ok"
-			}
-			r.metrics.Observe(result)
+		execResult, err := driver.Execute(ctx, query, args)
+		if execResult.Duration <= 0 {
+			execResult.Duration = time.Since(started)
+		}
+		result.Duration = execResult.Duration
+		result.Rows = execResult.Rows
+		if err != nil {
+			result.Status = "error"
+			result.Error = err.Error()
+			result.ErrorClass = "execution_error"
+		} else {
+			result.Status = "ok"
+		}
+		r.metrics.Observe(result)
 
-			if measure {
-				summary.TotalQueries++
-				if result.Status == "error" {
-					summary.TotalErrors++
-				}
-				if r.cfg.Sink != nil {
-					if err := r.cfg.Sink.Record(result); err != nil {
-						return fmt.Errorf("sink record (%s/%s): %w", protocol, query.QueryID, err)
-					}
+		if measure {
+			summary.TotalQueries++
+			if result.Status == "error" {
+				summary.TotalErrors++
+			}
+			if r.cfg.Sink != nil {
+				if err := r.cfg.Sink.Record(result); err != nil {
+					return fmt.Errorf("sink record (%s/%s): %w", protocol, query.QueryID, err)
 				}
-			} else {
-				summary.WarmupQueries++
 			}
+		} else {
+			summary.WarmupQueries++
 		}
 	}
 	return nil

tests/perf/core/runner_test.go

@@ -2,6 +2,7 @@ package core
 
 import (
 	"context"
+	"fmt"
 	"reflect"
 	"testing"
 	"time"
@@ -11,16 +12,76 @@ type testDriver struct {
 	protocol Protocol
 	calls    int
 	queryIDs []string
+	events   *[]string
 }
 
 func (d *testDriver) Protocol() Protocol { return d.protocol }
 
 func (d *testDriver) Execute(_ context.Context, query Query, _ []any) (ExecutionResult, error) {
 	d.calls++
 	d.queryIDs = append(d.queryIDs, query.QueryID)
+	if d.events != nil {
+		*d.events = append(*d.events, string(d.protocol)+"/"+query.QueryID)
+	}
 	return ExecutionResult{Rows: 1}, nil
 }
 
+func TestRunnerCompletesWarmupAndMeasurementsForEachProtocolBeforeStartingNext(t *testing.T) {
+	const trino Protocol = "trino"
+	events := []string{}
+	pg := &testDriver{protocol: ProtocolPGWire, events: &events}
+	trinoDriver := &testDriver{protocol: trino, events: &events}
+	sink := &inMemorySink{}
+	runner := NewQueryRunner(RunnerConfig{
+		Catalog: Catalog{
+			Name:              "protocol-isolation",
+			WarmupIterations:  1,
+			MeasureIterations: 2,
+			Targets:           []Protocol{ProtocolPGWire, trino},
+			Queries: []Query{
+				{
+					QueryID:   "q1",
+					IntentID:  "i1",
+					PGWireSQL: "SELECT 1",
+				},
+			},
+		},
+		Drivers: map[Protocol]ProtocolDriver{
+			ProtocolPGWire: pg,
+			trino:          trinoDriver,
+		},
+		Sink: sink,
+		Now:  func() time.Time { return time.Unix(1700000000, 0) },
+	})
+
+	summary, err := runner.Run(context.Background())
+	if err != nil {
+		t.Fatalf("Run returned error: %v", err)
+	}
+	want := []string{
+		"pgwire/q1", "pgwire/q1", "pgwire/q1",
+		"trino/q1", "trino/q1", "trino/q1",
+	}
+	if !reflect.DeepEqual(events, want) {
+		t.Fatalf("execution order: got %v want %v", events, want)
+	}
+	if got, want := summaryProtocolIterations(sink), []string{"pgwire/1", "pgwire/2", "trino/1", "trino/2"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("unexpected sink records: got %v want %v", got, want)
+	}
+	if summary.WarmupQueries != 2 || summary.TotalQueries != 4 {
+		t.Fatalf("unexpected summary: %+v", summary)
+	}
+}
+
+func summaryProtocolIterations(sink *inMemorySink) []string {
+	results := sink.results
+	got := make([]string, 0, len(results))
+	for _, result := range results {
+		got = append(got, fmt.Sprintf("%s/%d", result.Protocol, result.MeasureIteration))
+	}
+	return got
+}
+
 func TestRunnerExecutesPairedQueriesThroughExistingRuntimeContract(t *testing.T) {
 	catalog, err := ParseCatalog([]byte(pairedCatalogYAML(`
 paired_queries:

tests/perf/queries/ducklake_posthog_tables.yaml

@@ -1,5 +1,5 @@
-name: posthog-frozen-ducklake-golden-v2
-description: Identical query shapes with balanced execution order over frozen raw Parquet views and production-shaped DuckLake tables.
+name: posthog-frozen-ducklake-golden-v3
+description: Identical query shapes with protocol-isolated, balanced execution order over frozen raw Parquet views and production-shaped DuckLake tables.
 seed: 42
 dataset_scale: 1
 targets:
@@ -17,14 +17,14 @@ relation_variants:
     persons: posthog.persons
 
 paired_queries:
-  - query_id_base: q_events_total_balanced_v2
-    intent_id: intent_events_total_balanced_v2
+  - query_id_base: q_events_total_balanced_v3
+    intent_id: intent_events_total_balanced_v3
     tags: [nightly, frozen, posthog, events, aggregate, paired]
     params: {}
     sql_template: SELECT COUNT(*) AS events FROM {{ relation "events" }}
 
-  - query_id_base: q_events_count_one_day_balanced_v2
-    intent_id: intent_events_count_one_day_balanced_v2
+  - query_id_base: q_events_count_one_day_balanced_v3
+    intent_id: intent_events_count_one_day_balanced_v3
     tags: [nightly, frozen, posthog, events, aggregate, one-day, paired]
     params: {}
     sql_template: >
@@ -33,32 +33,32 @@ paired_queries:
       WHERE "timestamp" >= CAST('2026-03-01 00:00:00+00:00' AS TIMESTAMP WITH TIME ZONE)
         AND "timestamp" < CAST('2026-03-02 00:00:00+00:00' AS TIMESTAMP WITH TIME ZONE)
 
-  - query_id_base: q_events_by_name_march_2026_balanced_v2
-    intent_id: intent_events_by_name_march_2026_balanced_v2
+  - query_id_base: q_events_by_name_march_2026_balanced_v3
+    intent_id: intent_events_by_name_march_2026_balanced_v3
     tags: [nightly, frozen, posthog, events, aggregate, analytics, paired]
     params: {}
     sql_template: SELECT event, COUNT(*) AS events FROM {{ relation "events" }} WHERE "timestamp" >= CAST('2026-03-01 00:00:00+00:00' AS TIMESTAMP WITH TIME ZONE) AND "timestamp" < CAST('2026-03-18 00:00:00+00:00' AS TIMESTAMP WITH TIME ZONE) GROUP BY event ORDER BY events DESC, event LIMIT 20
 
-  - query_id_base: q_events_distinct_persons_balanced_v2
-    intent_id: intent_events_distinct_persons_balanced_v2
+  - query_id_base: q_events_distinct_persons_balanced_v3
+    intent_id: intent_events_distinct_persons_balanced_v3
     tags: [nightly, frozen, posthog, events, aggregate, distinct, paired]
     params: {}
     sql_template: SELECT COUNT(DISTINCT person_id) AS distinct_persons FROM {{ relation "events" }} WHERE person_id IS NOT NULL
 
-  - query_id_base: q_persons_total_balanced_v2
-    intent_id: intent_persons_total_balanced_v2
+  - query_id_base: q_persons_total_balanced_v3
+    intent_id: intent_persons_total_balanced_v3
     tags: [nightly, frozen, posthog, persons, aggregate, paired]
     params: {}
     sql_template: SELECT COUNT(*) AS persons FROM {{ relation "persons" }}
 
-  - query_id_base: q_persons_daily_april_2026_balanced_v2
-    intent_id: intent_persons_daily_april_2026_balanced_v2
+  - query_id_base: q_persons_daily_april_2026_balanced_v3
+    intent_id: intent_persons_daily_april_2026_balanced_v3
     tags: [nightly, frozen, posthog, persons, aggregate, time-series, paired]
     params: {}
     sql_template: SELECT date_trunc('day', _timestamp) AS day, COUNT(*) AS persons FROM {{ relation "persons" }} WHERE _timestamp >= CAST('2026-04-01 00:00:00+00:00' AS TIMESTAMP WITH TIME ZONE) AND _timestamp < CAST('2026-05-01 00:00:00+00:00' AS TIMESTAMP WITH TIME ZONE) GROUP BY 1 ORDER BY 1
 
-  - query_id_base: q_events_daily_march_2026_balanced_v2
-    intent_id: intent_events_daily_march_2026_balanced_v2
+  - query_id_base: q_events_daily_march_2026_balanced_v3
+    intent_id: intent_events_daily_march_2026_balanced_v3
     tags: [nightly, frozen, posthog, events, aggregate, time-series, paired]
     params: {}
     sql_template: SELECT date_trunc('day', "timestamp") AS day, COUNT(*) AS events FROM {{ relation "events" }} WHERE "timestamp" >= CAST('2026-03-01 00:00:00+00:00' AS TIMESTAMP WITH TIME ZONE) AND "timestamp" < CAST('2026-03-18 00:00:00+00:00' AS TIMESTAMP WITH TIME ZONE) GROUP BY 1 ORDER BY 1