From 3aaf4fd6e9b355237dbbbc2f7ad959e278c66a70 Mon Sep 17 00:00:00 2001 From: Bill Guowei Yang Date: Thu, 3 Sep 2026 09:18:58 -0400 Subject: [PATCH] Isolate perf runs by protocol --- tests/perf/README.md | 6 ++ tests/perf/core/catalog_test.go | 30 +++--- tests/perf/core/runner.go | 96 +++++++++---------- tests/perf/core/runner_test.go | 61 ++++++++++++ .../perf/queries/ducklake_posthog_tables.yaml | 32 +++---- 5 files changed, 146 insertions(+), 79 deletions(-) diff --git a/tests/perf/README.md b/tests/perf/README.md index 46aa24b7..96dc3244 100644 --- a/tests/perf/README.md +++ b/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 diff --git a/tests/perf/core/catalog_test.go b/tests/perf/core/catalog_test.go index b7e8bfc8..4ba05606 100644 --- a/tests/perf/core/catalog_test.go +++ b/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 '") { diff --git a/tests/perf/core/runner.go b/tests/perf/core/runner.go index fdf02037..cb57405f 100644 --- a/tests/perf/core/runner.go +++ b/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 diff --git a/tests/perf/core/runner_test.go b/tests/perf/core/runner_test.go index 27e5c34e..9b8195b3 100644 --- a/tests/perf/core/runner_test.go +++ b/tests/perf/core/runner_test.go @@ -2,6 +2,7 @@ package core import ( "context" + "fmt" "reflect" "testing" "time" @@ -11,6 +12,7 @@ type testDriver struct { protocol Protocol calls int queryIDs []string + events *[]string } func (d *testDriver) Protocol() Protocol { return d.protocol } @@ -18,9 +20,68 @@ 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: diff --git a/tests/perf/queries/ducklake_posthog_tables.yaml b/tests/perf/queries/ducklake_posthog_tables.yaml index 5cb4784f..edb6721a 100644 --- a/tests/perf/queries/ducklake_posthog_tables.yaml +++ b/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