Ensure frozen perf uses standard worker size

docs/runbooks/scenario-runner.md

@@ -48,14 +48,24 @@ Frozen dataset scenarios additionally require:
 export DUCKGRES_SCENARIO_FROZEN_S3_URI="s3://<dev-managed-bucket>/frozen_v1/"
 ```
 
-The full suite, fast suite, and targeted frozen perf scenarios exercise PGWire
-only. Frozen perf records per-query success and failure rows in
-`query_results.csv`.
+The targeted frozen perf scenario also requires the standard Duckgres worker
+shape deployed with the scenario stack:
+
+```bash
+export DUCKGRES_K8S_WORKER_CPU_REQUEST="3"
+export DUCKGRES_K8S_WORKER_MEMORY_REQUEST="12Gi"
+```
+
+The full and fast suites exercise PGWire only. The targeted frozen perf
+scenario compares PGWire and Trino. It records per-query success and failure
+rows in `query_results.csv`.
 Measured query errors fail the perf DAG step after its artifacts are written;
 independent sibling steps continue to run.
 
 A `perf_queries` step can set `with.targets` to a non-empty subset of the
-catalog's targets. Duckgres perf catalogs and scenarios are pgwire-only.
+catalog's targets. Optional `with.worker_cpu` and `with.worker_memory` values
+are sent as PGWire startup options. Both default to empty, which leaves worker
+selection to the server; set both for resource-controlled comparisons.
 
 Do not commit concrete dev endpoints, secrets, org IDs, or private bucket names.
 
@@ -149,7 +159,7 @@ The frozen dbt scenario uses:
 
 dbt artifacts are written under `artifacts/scenario/<run_id>/dbt/`, including per-command stdout/stderr logs, `target/` artifacts, and dbt logs. Install `dbt-postgres` locally or set `DUCKGRES_SCENARIO_DBT_BIN` to the dbt executable to use.
 
-The frozen dbt workload requests a 2 CPU, 4Gi worker through the dbt connection's `duckgres.worker_cpu` and `duckgres.worker_memory` startup options. It also sets `with.connect_timeout: 360`, long enough for the control plane's five-minute worker queue to provision a cold Karpenter node. Other scenario workloads use the isolated control plane's default worker size; `scenario-dev` sets that default to 2 CPU and 8Gi to add process headroom for repeated frozen pgwire aggregates. A `dbt_run` step can opt into a different size or connection window with `with.worker_cpu`, `with.worker_memory`, and `with.connect_timeout`.
+The frozen dbt workload requests a 2 CPU, 4Gi worker through the dbt connection's `duckgres.worker_cpu` and `duckgres.worker_memory` startup options. It also sets `with.connect_timeout: 360`, long enough for the control plane's five-minute worker queue to provision a cold Karpenter node. Frozen perf requests the scenario stack's standard worker shape through the same startup GUCs; `scenario-dev` currently configures 3 CPU and 12Gi, matching the aggregate execution resources of its three Trino workers. This explicit sizing bypasses the exploratory worker tier so every measured PGWire query uses the comparison resource budget. A `dbt_run` or `perf_queries` step can opt into a different worker shape with `with.worker_cpu` and `with.worker_memory`.
 
 `perf_queries` defaults `with.fail_on_query_errors` to `true`. A measured query
 error therefore marks that DAG step failed and appears in the scenario result,

tests/mw-dev/README.md

@@ -97,6 +97,8 @@ verified HTTPS and never disables certificate verification.
 
 Optional perf-step settings are:
 
+- `worker_cpu` (default empty, leaving PGWire worker selection to the server)
+- `worker_memory` (default empty, leaving PGWire worker selection to the server)
 - `trino_schema` (default `posthog`)
 - `trino_ca_cert_file` (default empty, using system roots)
 - `trino_startup_timeout` (default `2m`)
@@ -110,7 +112,10 @@ before warmup or measured statements run. For the isolated mw-dev cell, use
 scenario Job mounts the per-run CA from `duckgres-trino-tls` and passes that
 path through `DUCKGRES_SCENARIO_TRINO_CA_CERT`; the perf adapter verifies the
 coordinator certificate and retries its first authenticated query for the
-bounded Secret-projection window. The paired catalog remains the single SQL
+bounded Secret-projection window. It also passes the deployed
+`DUCKGRES_K8S_WORKER_CPU_REQUEST` and `DUCKGRES_K8S_WORKER_MEMORY_REQUEST` into
+the perf step, which requests that exact shape through PGWire startup options
+and therefore bypasses the exploratory worker tier. The paired catalog remains the single SQL
 source: direct-Parquet `raw_view` members run only through PGWire, while each
 production-shaped `ducklake_table` member runs through both PGWire and Trino.
 To reproduce the scheduled run, deploy and test with `E2E_SUITE=trino` and the

tests/mw-dev/run.sh

@@ -536,6 +536,8 @@ spec:
             - { name: DUCKGRES_SCENARIO_FROZEN_S3_URI, value: "$FROZEN_S3_URI" }
             - { name: DUCKGRES_SCENARIO_TRINO_CA_CERT, value: "/trino-ca/ca.crt" }
             - { name: DUCKGRES_SCENARIO_DBT_BIN, value: "dbt" }
+            - { name: DUCKGRES_K8S_WORKER_CPU_REQUEST, value: "$DUCKGRES_K8S_WORKER_CPU_REQUEST" }
+            - { name: DUCKGRES_K8S_WORKER_MEMORY_REQUEST, value: "$DUCKGRES_K8S_WORKER_MEMORY_REQUEST" }
             # The Crossplane composition grants this isolated service account
             # exact-name access to only the matching CNPG credential Secret.
             - { name: DUCKGRES_SCENARIO_ORG_ID, value: "ci-pr-${PR_NUMBER}-cnpg" }

tests/mw-dev/run_sh_test.go

@@ -1156,6 +1156,13 @@ func TestScenarioPodIsProtectedFromKarpenterDisruption(t *testing.T) {
 	if got := annotations["karpenter.sh/do-not-disrupt"]; got != "true" {
 		t.Fatalf("scenario Pod karpenter.sh/do-not-disrupt = %v, want true", got)
 	}
+	env := deploymentContainerEnv(manifest, "scenario")
+	if got := env["DUCKGRES_K8S_WORKER_CPU_REQUEST"]; got != "$DUCKGRES_K8S_WORKER_CPU_REQUEST" {
+		t.Fatalf("scenario worker CPU env = %q, want deployed standard worker CPU", got)
+	}
+	if got := env["DUCKGRES_K8S_WORKER_MEMORY_REQUEST"]; got != "$DUCKGRES_K8S_WORKER_MEMORY_REQUEST" {
+		t.Fatalf("scenario worker memory env = %q, want deployed standard worker memory", got)
+	}
 }
 
 func TestControlPlaneServiceDoesNotExposeFlight(t *testing.T) {

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

@@ -50,6 +50,8 @@ func TestExecutorRunsPerfStepAndWritesArtifacts(t *testing.T) {
 			"catalog_file":    catalogPath,
 			"run_id":          "scenario-run-1",
 			"dataset_version": "posthog-file-views-v1",
+			"worker_cpu":      "3",
+			"worker_memory":   "12Gi",
 		},
 	})
 	if err != nil {
@@ -70,6 +72,9 @@ func TestExecutorRunsPerfStepAndWritesArtifacts(t *testing.T) {
 	if strings.Contains(pgwireDSN, "hostaddr=") {
 		t.Fatalf("pgwire dsn = %q, should not use unsupported lib/pq hostaddr", pgwireDSN)
 	}
+	if !strings.Contains(pgwireDSN, "options='-c duckgres.worker_cpu=3 -c duckgres.worker_memory=12Gi'") {
+		t.Fatalf("pgwire dsn = %q, want explicit standard worker sizing options", pgwireDSN)
+	}
 	if factory.pgwireConnection.DialAddress != "10.0.0.10:5432" {
 		t.Fatalf("pgwire direct address = %q, want 10.0.0.10:5432", factory.pgwireConnection.DialAddress)
 	}

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

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -64,6 +65,8 @@ type stepSpec struct {
 	OutputSubdir      string
 	ReadOnly          bool
 	FailOnQueryErrors bool
+	WorkerCPU         string
+	WorkerMemory      string
 	TrinoSchema       string
 	TrinoCACertFile   string
 	TrinoStartup      trinodriver.StartupOptions
@@ -255,6 +258,8 @@ func (e *Executor) parseStep(step core.Step) (stepSpec, error) {
 		OutputSubdir:      stringFromWith(step, "output_subdir", "perf"),
 		ReadOnly:          boolFromWith(step, "read_only", true),
 		FailOnQueryErrors: boolFromWith(step, "fail_on_query_errors", true),
+		WorkerCPU:         stringFromWith(step, "worker_cpu", ""),
+		WorkerMemory:      stringFromWith(step, "worker_memory", ""),
 		TrinoSchema:       stringFromWith(step, "trino_schema", "posthog"),
 		TrinoCACertFile:   stringFromWith(step, "trino_ca_cert_file", ""),
 		TrinoStartup: trinodriver.StartupOptions{
@@ -392,6 +397,19 @@ func (e *Executor) pgwireConnection(spec stepSpec) (scenariosql.PGWireConnection
 	cfg.Database = spec.Database
 	cfg.Username = spec.Username
 	cfg.Password = spec.Password
+	workerOptions := make([]string, 0, 4)
+	if spec.WorkerCPU != "" {
+		workerOptions = append(workerOptions, "-c", "duckgres.worker_cpu="+spec.WorkerCPU)
+	}
+	if spec.WorkerMemory != "" {
+		workerOptions = append(workerOptions, "-c", "duckgres.worker_memory="+spec.WorkerMemory)
+	}
+	if len(workerOptions) > 0 {
+		if cfg.StartupOptions != "" {
+			cfg.StartupOptions += " "
+		}
+		cfg.StartupOptions += strings.Join(workerOptions, " ")
+	}
 	connection, err := cfg.PGWire()
 	if err != nil {
 		return scenariosql.PGWireConnection{}, classified(ErrorClassConfig, err)

tests/mw-dev/scenario/runner_test.go

@@ -164,6 +164,8 @@ func TestProvisionSmokeScenarioUsesIsolatedStackWarehouseIdentityAndSupportedSte
 func TestFrozenSuccessScenariosUseIsolatedStackWarehouseIdentity(t *testing.T) {
 	const scenarioOrgID = "ci-pr-123-cnpg"
 	t.Setenv("DUCKGRES_SCENARIO_TRINO_CA_CERT", "/tmp/test-trino-ca.crt")
+	t.Setenv("DUCKGRES_K8S_WORKER_CPU_REQUEST", "3")
+	t.Setenv("DUCKGRES_K8S_WORKER_MEMORY_REQUEST", "12Gi")
 
 	for _, scenarioFile := range []string{
 		"posthog_frozen_metadata.yaml",
@@ -450,6 +452,8 @@ func TestFrozenPerfScenarioUsesSupportedStepsAndRelativeCatalog(t *testing.T) {
 	t.Setenv("DUCKGRES_SCENARIO_FROZEN_S3_URI", "s3://example-frozen/frozen_v1/")
 	t.Setenv("DUCKGRES_SCENARIO_ORG_ID", "ci-pr-123-cnpg")
 	t.Setenv("DUCKGRES_SCENARIO_TRINO_CA_CERT", "/tmp/test-trino-ca.crt")
+	t.Setenv("DUCKGRES_K8S_WORKER_CPU_REQUEST", "3")
+	t.Setenv("DUCKGRES_K8S_WORKER_MEMORY_REQUEST", "12Gi")
 
 	scenario, _, err := loadScenarioForRun(filepath.Join("scenarios", "posthog_frozen_perf.yaml"))
 	if err != nil {
@@ -490,6 +494,12 @@ func TestFrozenPerfScenarioUsesSupportedStepsAndRelativeCatalog(t *testing.T) {
 		if got, _ := step.With["trino_ca_cert_file"].(string); got != "/tmp/test-trino-ca.crt" {
 			t.Fatalf("perf Trino CA file = %q, want resolved environment path", got)
 		}
+		if got, _ := step.With["worker_cpu"].(string); got != "3" {
+			t.Fatalf("perf worker_cpu = %q, want standard worker CPU", got)
+		}
+		if got, _ := step.With["worker_memory"].(string); got != "12Gi" {
+			t.Fatalf("perf worker_memory = %q, want standard worker memory", got)
+		}
 	}
 	if !foundPerf {
 		t.Fatal("expected frozen perf scenario to include a perf_queries step")
@@ -500,6 +510,8 @@ func TestFrozenPerfScenarioBuildsAndValidatesPostHogTablesBeforePerf(t *testing.
 	t.Setenv("DUCKGRES_SCENARIO_FROZEN_S3_URI", "s3://example-frozen/frozen_v1/")
 	t.Setenv("DUCKGRES_SCENARIO_ORG_ID", "ci-pr-123-cnpg")
 	t.Setenv("DUCKGRES_SCENARIO_TRINO_CA_CERT", "/tmp/test-trino-ca.crt")
+	t.Setenv("DUCKGRES_K8S_WORKER_CPU_REQUEST", "3")
+	t.Setenv("DUCKGRES_K8S_WORKER_MEMORY_REQUEST", "12Gi")
 
 	scenario, _, err := loadScenarioForRun(filepath.Join("scenarios", "posthog_frozen_perf.yaml"))
 	if err != nil {

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

@@ -4,6 +4,8 @@ required_env:
   - DUCKGRES_SCENARIO_ORG_ID
   - DUCKGRES_SCENARIO_FROZEN_S3_URI
   - DUCKGRES_SCENARIO_TRINO_CA_CERT
+  - DUCKGRES_K8S_WORKER_CPU_REQUEST
+  - DUCKGRES_K8S_WORKER_MEMORY_REQUEST
 steps:
   - id: provision
     type: provision_warehouse
@@ -81,6 +83,8 @@ steps:
       run_id: ${run_id}
       dataset_version: posthog-file-views-v1
       fail_on_query_errors: true
+      worker_cpu: ${env:DUCKGRES_K8S_WORKER_CPU_REQUEST}
+      worker_memory: ${env:DUCKGRES_K8S_WORKER_MEMORY_REQUEST}
 
   - id: deprovision
     type: deprovision_warehouse

tests/mw-dev/scenario/script_test.go

@@ -51,6 +51,8 @@ func TestScenarioRunScriptCheckEnvIncludesScenarioRequiredEnv(t *testing.T) {
 		"DUCKGRES_SCENARIO_ORG_ID",
 		"DUCKGRES_SCENARIO_FROZEN_S3_URI",
 		"DUCKGRES_SCENARIO_TRINO_CA_CERT",
+		"DUCKGRES_K8S_WORKER_CPU_REQUEST",
+		"DUCKGRES_K8S_WORKER_MEMORY_REQUEST",
 	} {
 		if !strings.Contains(text, name) {
 			t.Fatalf("script output %q missing %s", text, name)

tests/mw-dev/scenario/sql/connection.go

@@ -23,6 +23,7 @@ type ConnectionConfig struct {
 	SSLMode         string
 	ConnectTimeout  int
 	ApplicationName string
+	StartupOptions  string
 }
 
 // PGWireConnection separates the hostname used for managed-warehouse TLS
@@ -75,6 +76,9 @@ func (c ConnectionConfig) PGWire() (PGWireConnection, error) {
 	if c.ApplicationName != "" {
 		values = append(values, [2]string{"application_name", c.ApplicationName})
 	}
+	if c.StartupOptions != "" {
+		values = append(values, [2]string{"options", c.StartupOptions})
+	}
 
 	parts := make([]string, 0, len(values))
 	for _, kv := range values {