diff --git a/.github/workflows/scenario-dev.yml b/.github/workflows/scenario-dev.yml index 6db5b211..745468d8 100644 --- a/.github/workflows/scenario-dev.yml +++ b/.github/workflows/scenario-dev.yml @@ -1,4 +1,5 @@ name: scenario-dev +run-name: scenario-dev / ${{ inputs.scenario || 'posthog_frozen_perf' }} / Trino ${{ inputs.trino_perf_shape || 'baseline' }} on: workflow_dispatch: @@ -13,6 +14,17 @@ on: required: false default: "" type: string + trino_perf_shape: + description: Trino execution shape (experiments require posthog_frozen_perf) + required: false + default: baseline + type: choice + options: + - baseline + - large + - scaleout + - large-scaleout + - all schedule: - cron: "17 8 * * *" @@ -25,7 +37,24 @@ concurrency: cancel-in-progress: false jobs: + plan: + runs-on: ubuntu-24.04 + timeout-minutes: 5 + outputs: + matrix: ${{ steps.shapes.outputs.matrix }} + env: + SCENARIO_NAME: ${{ github.event_name == 'schedule' && 'posthog_frozen_perf' || inputs.scenario }} + TRINO_PERF_SHAPE: ${{ inputs.trino_perf_shape || 'baseline' }} + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + - name: Select Trino shapes + id: shapes + run: | + matrix="$(bash scripts/scenario_trino_shapes.sh)" + echo "matrix=$matrix" >> "$GITHUB_OUTPUT" + scenario-runner-image: + needs: [plan] uses: ./.github/workflows/_image-build.yml with: dockerfile: tests/mw-dev/scenario/Dockerfile @@ -37,6 +66,7 @@ jobs: ecr-role: ${{ vars.AWS_ECR_PRS_PUBLISH_IAM_ROLE }} duckgres-image: + needs: [plan] if: ${{ github.event_name != 'workflow_dispatch' || inputs.duckgres_image == '' }} uses: ./.github/workflows/_image-build.yml with: @@ -57,12 +87,20 @@ jobs: ecr-role: ${{ vars.AWS_ECR_PRS_PUBLISH_IAM_ROLE }} scenario: - needs: [scenario-runner-image, duckgres-image] - if: ${{ always() && needs.scenario-runner-image.result == 'success' && (needs.duckgres-image.result == 'success' || (github.event_name == 'workflow_dispatch' && inputs.duckgres_image != '')) }} + name: scenario (${{ matrix.shape }}) + needs: [plan, scenario-runner-image, duckgres-image] + if: ${{ always() && needs.plan.result == 'success' && needs.scenario-runner-image.result == 'success' && (needs.duckgres-image.result == 'success' || (github.event_name == 'workflow_dispatch' && inputs.duckgres_image != '')) }} + strategy: + # Each job includes teardown. A fresh runner also refreshes credentials + # and the timeout budget before admitting the next shape. + max-parallel: 1 + fail-fast: false + matrix: ${{ fromJSON(needs.plan.outputs.matrix) }} runs-on: ubuntu-24.04 timeout-minutes: 270 env: SCENARIO_NAME: ${{ github.event_name == 'schedule' && 'posthog_frozen_perf' || inputs.scenario }} + TRINO_PERF_SHAPE: ${{ matrix.shape }} SCENARIO_RUNNER_IMAGE: ${{ needs.scenario-runner-image.outputs.image }} WORKER_IMAGE: ${{ (github.event_name == 'workflow_dispatch' && inputs.duckgres_image) || needs.duckgres-image.outputs.image }} CONTROLPLANE_IMAGE: ${{ (github.event_name == 'workflow_dispatch' && inputs.duckgres_image) || needs.duckgres-image.outputs.image }} @@ -76,8 +114,8 @@ jobs: TRINO_POD_IDENTITY_ROLE: ${{ secrets.MW_DEV_TRINO_POD_IDENTITY_ROLE }} TRINO_IMAGE: ghcr.io/posthog/trino:4505364c570d6b51edecd299b603fca4b6693d86@sha256:ac80c275fd18a439d25da5652ab5cd3c80bcbdd2d88d64c9722dc3e8bb68ba07 E2E_SUITE: ${{ (github.event_name == 'schedule' || inputs.scenario == 'posthog_frozen_perf') && 'trino' || 'neutral' }} - PR_NUMBER: ${{ github.run_id }} - NAMESPACE: duckgres-ci-pr-${{ github.run_id }} + PR_NUMBER: ${{ github.run_id }}${{ matrix.suffix }} + NAMESPACE: duckgres-ci-pr-${{ github.run_id }}${{ matrix.suffix }} DUCKGRES_SCENARIO_MAX_RUNTIME: 4h DUCKGRES_SCENARIO_GO_TEST_TIMEOUT: 4h15m # Add process headroom for repeated full-dataset pgwire aggregates. @@ -95,7 +133,7 @@ jobs: go-version-file: go.mod - name: Test scenario workflow scripts - run: go test -count=1 ./tests/mw-dev/scenario ./tests/mw-dev ./tests/perf/publishercli + run: go test -count=1 ./tests/mw-dev/scenario ./tests/mw-dev ./tests/perf/publishercli ./tests/perf/shapecompare ./cmd/duckgres-perf-shape-summary - name: Configure AWS credentials (OIDC) uses: aws-actions/configure-aws-credentials@517a711dbcd0e402f90c77e7e2f81e849156e31d # v6.2.2 @@ -120,19 +158,32 @@ jobs: run: aws eks update-kubeconfig --name "$CLUSTER_NAME" --region "$AWS_REGION" --alias "$KUBE_CONTEXT" - name: Load Athena perf configuration - if: env.SCENARIO_NAME == 'posthog_frozen_perf' + if: env.SCENARIO_NAME == 'posthog_frozen_perf' && matrix.shape == 'baseline' run: bash scripts/scenario_athena_config.sh >> "$GITHUB_ENV" - name: Deploy isolated Duckgres stack + id: deploy run: tests/mw-dev/run.sh deploy - name: Run selected scenario + id: benchmark run: tests/mw-dev/run.sh test-scenario - name: Publish scenario summary if: always() run: | set -euo pipefail + shape_file=artifacts/scenario-dev/trino-perf-shape.json + if [ -f "$shape_file" ]; then + { + echo '## Trino execution shape' + echo + echo '```json' + cat "$shape_file" + echo '```' + echo + } >> "$GITHUB_STEP_SUMMARY" + fi summary_file="$(find artifacts/scenario-dev -type f -name scenario_summary.md -print -quit 2>/dev/null || true)" if [ -n "$summary_file" ]; then cat "$summary_file" >> "$GITHUB_STEP_SUMMARY" @@ -149,20 +200,36 @@ jobs: run: tests/mw-dev/run.sh diagnostics - name: Teardown + id: teardown if: always() run: tests/mw-dev/run.sh teardown + - name: Record shape outcome + if: always() + env: + DEPLOY_OUTCOME: ${{ steps.deploy.outcome }} + SCENARIO_OUTCOME: ${{ steps.benchmark.outcome }} + TEARDOWN_OUTCOME: ${{ steps.teardown.outcome }} + run: | + mkdir -p artifacts/scenario-dev + jq -n --arg shape "$TRINO_PERF_SHAPE" \ + --arg deploy "$DEPLOY_OUTCOME" --arg scenario "$SCENARIO_OUTCOME" \ + --arg teardown "$TEARDOWN_OUTCOME" \ + '$ARGS.named' > artifacts/scenario-dev/shape-result.json + - name: Upload scenario artifacts if: always() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: - name: scenario-dev-${{ github.run_id }}-${{ github.run_attempt }} + name: scenario-dev-${{ github.run_id }}-${{ github.run_attempt }}-${{ matrix.shape }} path: artifacts/scenario-dev/ if-no-files-found: warn retention-days: 14 - name: Publish scenario perf results - if: ${{ always() && github.ref == 'refs/heads/main' }} + # Shape experiments remain downloadable artifacts; publishing them + # would mix different resource budgets into the daily baseline. + if: ${{ always() && github.ref == 'refs/heads/main' && env.TRINO_PERF_SHAPE == 'baseline' && inputs.trino_perf_shape != 'all' }} timeout-minutes: 10 env: MW_DEV_SCENARIO_PERF_SECRET_ID: ${{ vars.MW_DEV_SCENARIO_PERF_SECRET_ID }} @@ -195,3 +262,40 @@ jobs: fi done exit "$publish_failed" + + compare-shapes: + needs: [plan, scenario] + if: ${{ always() && needs.plan.result == 'success' && inputs.trino_perf_shape == 'all' }} + runs-on: ubuntu-24.04 + timeout-minutes: 10 + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6.5.0 + with: + go-version-file: go.mod + - name: Download shape artifacts + # Missing uploads are reported as missing shapes by the comparison. + continue-on-error: true + uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0 + with: + pattern: scenario-dev-${{ github.run_id }}-${{ github.run_attempt }}-* + path: artifacts/trino-shapes + merge-multiple: false + - name: Compare Trino shapes + run: go run ./cmd/duckgres-perf-shape-summary --artifacts-dir artifacts/trino-shapes > trino-shape-comparison.md + - name: Publish Trino shape comparison + if: always() + run: | + if [ -s trino-shape-comparison.md ]; then + cat trino-shape-comparison.md >> "$GITHUB_STEP_SUMMARY" + else + echo 'Trino shape comparison unavailable. Inspect the shape jobs and their artifacts.' >> "$GITHUB_STEP_SUMMARY" + fi + - name: Upload Trino shape comparison + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: trino-shape-comparison-${{ github.run_id }}-${{ github.run_attempt }} + path: trino-shape-comparison.md + if-no-files-found: warn + retention-days: 14 diff --git a/cmd/duckgres-perf-shape-summary/main.go b/cmd/duckgres-perf-shape-summary/main.go new file mode 100644 index 00000000..e3b904ae --- /dev/null +++ b/cmd/duckgres-perf-shape-summary/main.go @@ -0,0 +1,33 @@ +package main + +import ( + "flag" + "fmt" + "io" + "os" + + "github.com/posthog/duckgres/tests/perf/shapecompare" +) + +func main() { + if err := run(os.Args[1:], os.Stdout); err != nil { + _, _ = fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run(args []string, stdout io.Writer) error { + flags := flag.NewFlagSet("duckgres-perf-shape-summary", flag.ContinueOnError) + dir := flags.String("artifacts-dir", "", "required directory containing one artifact directory per Trino shape") + if err := flags.Parse(args); err != nil { + return err + } + if *dir == "" || flags.NArg() != 0 { + return fmt.Errorf("usage: duckgres-perf-shape-summary --artifacts-dir ") + } + report, err := shapecompare.Generate(*dir) + if _, writeErr := io.WriteString(stdout, report); writeErr != nil { + return writeErr + } + return err +} diff --git a/cmd/duckgres-perf-shape-summary/main_test.go b/cmd/duckgres-perf-shape-summary/main_test.go new file mode 100644 index 00000000..54bcc22c --- /dev/null +++ b/cmd/duckgres-perf-shape-summary/main_test.go @@ -0,0 +1,33 @@ +package main + +import ( + "bytes" + "errors" + "strings" + "testing" +) + +func TestRunRequiresArtifactsDirectory(t *testing.T) { + for _, args := range [][]string{nil, {"--artifacts-dir", "test", "extra"}} { + if err := run(args, &bytes.Buffer{}); err == nil { + t.Fatal("expected usage error") + } + } +} + +func TestRunPublishesIncompleteReportBeforeFailing(t *testing.T) { + var out bytes.Buffer + err := run([]string{"--artifacts-dir", t.TempDir()}, &out) + if err == nil || !strings.Contains(out.String(), "Comparison incomplete") || !strings.Contains(out.String(), "missing artifact") { + t.Fatalf("report=%s, error=%v", out.String(), err) + } +} + +type failedWriter struct{} + +func (failedWriter) Write([]byte) (int, error) { return 0, errors.New("write failed") } +func TestRunReturnsWriteFailure(t *testing.T) { + if err := run([]string{"--artifacts-dir", t.TempDir()}, failedWriter{}); err == nil || err.Error() != "write failed" { + t.Fatalf("got %v", err) + } +} diff --git a/justfile b/justfile index ba08dc73..f7768cdf 100644 --- a/justfile +++ b/justfile @@ -301,6 +301,7 @@ test: test-unit: go test -v -p 1 . ./configresolve/... ./duckdbservice/... ./server/... ./transpiler/... ./internal/... ./tests/manifests/... go test -v -count=1 ./tests/mw-dev/... + go test -v -count=1 ./tests/perf/shapecompare ./cmd/duckgres-perf-shape-summary # Run scenario runner unit tests [group('test')] diff --git a/scripts/scenario_run.sh b/scripts/scenario_run.sh index c8a07cbd..37519688 100755 --- a/scripts/scenario_run.sh +++ b/scripts/scenario_run.sh @@ -21,6 +21,7 @@ Optional environment: DUCKGRES_SCENARIO_DBT_BIN DUCKGRES_SCENARIO_MAX_RUNTIME DUCKGRES_SCENARIO_GO_TEST_TIMEOUT + DUCKGRES_SCENARIO_PERF_MODE (default: full; trino-only for posthog-frozen-perf) Scenario-specific required environment: DUCKGRES_SCENARIO_ORG_ID (required by successful provisioning scenarios) @@ -83,6 +84,29 @@ root_relative_path() { scenario_file="$(root_relative_path "$scenario_file")" output_base="$(root_relative_path "$output_base")" +# Keep preflight aligned with the Go scenario loader: Trino-only removes the +# Athena target and its environment requirements before template resolution. +perf_mode="${DUCKGRES_SCENARIO_PERF_MODE:-full}" +case "$perf_mode" in + full) ;; + trino-only) + scenario_name="$(awk '/^name:[[:space:]]*/ { + sub(/^name:[[:space:]]*/, "") + sub(/[[:space:]]*$/, "") + gsub(/^["'\'']|["'\'']$/, "") + print; exit + }' "$scenario_file")" + if [ "$scenario_name" != posthog-frozen-perf ]; then + echo "DUCKGRES_SCENARIO_PERF_MODE=trino-only requires the posthog-frozen-perf scenario." >&2 + exit 2 + fi + ;; + *) + echo "DUCKGRES_SCENARIO_PERF_MODE must be full or trino-only." >&2 + exit 2 + ;; +esac + scenario_required_env() { awk ' /^[^[:space:]]/ { in_required = 0 } @@ -104,6 +128,11 @@ required=( ) if [ -f "$scenario_file" ]; then while IFS= read -r name; do + if [ "$perf_mode" = trino-only ]; then + case "$name" in + DUCKGRES_SCENARIO_ATHENA_REGION|DUCKGRES_SCENARIO_ATHENA_WORKGROUP|DUCKGRES_SCENARIO_ATHENA_DATABASE|DUCKGRES_SCENARIO_ATHENA_RESULTS_S3_URI) continue ;; + esac + fi required+=("$name") done < <(scenario_required_env "$scenario_file") elif [ "$check_env_only" -eq 1 ]; then diff --git a/scripts/scenario_trino_shapes.sh b/scripts/scenario_trino_shapes.sh new file mode 100644 index 00000000..0225c1ec --- /dev/null +++ b/scripts/scenario_trino_shapes.sh @@ -0,0 +1,24 @@ +#!/usr/bin/env bash +# Print the scenario workflow matrix. Single-shape runs retain the original run +# ID; all-shape runs append distinct numeric suffixes for harness isolation. +set -euo pipefail + +shape="${TRINO_PERF_SHAPE:-baseline}" +case "$shape" in + baseline|large|scaleout|large-scaleout|all) ;; + *) + printf '%s\n' 'Invalid TRINO_PERF_SHAPE: expected baseline, large, scaleout, large-scaleout, or all' >&2 + exit 1 + ;; +esac + +if [[ "$shape" != baseline && "${SCENARIO_NAME:-}" != posthog_frozen_perf ]]; then + printf '%s\n' 'Trino shape experiments require posthog_frozen_perf' >&2 + exit 1 +fi + +if [[ "$shape" == all ]]; then + printf '%s\n' '{"include":[{"shape":"baseline","suffix":"1"},{"shape":"large","suffix":"2"},{"shape":"scaleout","suffix":"3"},{"shape":"large-scaleout","suffix":"4"}]}' +else + printf '{"include":[{"shape":"%s","suffix":""}]}\n' "$shape" +fi diff --git a/tests/mw-dev/README.md b/tests/mw-dev/README.md index d5c770e0..0b128a04 100644 --- a/tests/mw-dev/README.md +++ b/tests/mw-dev/README.md @@ -57,6 +57,20 @@ The isolated control plane's default worker request is configurable through `scenario-dev.yml` explicitly overrides them to 3 CPU and 12Gi for the frozen perf workload. Direct `run.sh` callers can make the same explicit override. +Manual frozen-perf runs can select `TRINO_PERF_SHAPE=baseline|large|scaleout|large-scaleout` +(`trino_perf_shape` in workflow dispatch). The default remains `baseline` for +scheduled runs and the Trino E2E lane. Workflow dispatch also accepts `all` to +run all four shapes sequentially with shared image builds and one comparison +summary. Each shape receives a separate temporary stack and cleanup. Baseline +runs the full cross-engine benchmark; nonbaseline shapes measure only Trino, +with identical SQL, warmup, and measured iterations, while retaining setup and +validation. The harness derives `DUCKGRES_SCENARIO_PERF_MODE=full` or +`trino-only` from the shape and records it in provenance; an inherited value +cannot override it. Nonbaseline shapes need no Athena configuration or scenario +Pod Identity role. See the +[Trino experiment runbook](../perf/README.md#trino-worker-shape-experiments) +for resource budgets, result provenance, sequential comparison, and recovery. + ### Scenario Trino readiness Scenarios that opt an org into Trino in their `provision_warehouse` request can diff --git a/tests/mw-dev/manifests.trino.tmpl.yaml b/tests/mw-dev/manifests.trino.tmpl.yaml index fcde21bc..b3c1efa0 100644 --- a/tests/mw-dev/manifests.trino.tmpl.yaml +++ b/tests/mw-dev/manifests.trino.tmpl.yaml @@ -52,9 +52,9 @@ data: http-server.https.keystore.path=/etc/trino/tls/keystore.p12 http-server.https.keystore.key=${ENV:TRINO_TLS_KEYSTORE_PASSWORD} discovery.uri=http://duckgres-trino.${NAMESPACE}.svc:8080 - query.max-memory=6GB + query.max-memory=${TRINO_QUERY_MEMORY} # The coordinator does not execute tasks and retains headroom in its 2G - # heap; execution workers carry the 2GB per-node allowance below. + # heap; execution workers carry the selected per-node allowance below. query.max-memory-per-node=1GB catalog.management=dynamic catalog.store=posthog @@ -94,7 +94,7 @@ data: node.data-dir=/data/trino jvm.config: | -server - -Xmx3G + -Xmx${TRINO_WORKER_HEAP} -XX:+UseG1GC -XX:+ExitOnOutOfMemoryError -Dfile.encoding=UTF-8 @@ -102,8 +102,8 @@ data: coordinator=false http-server.http.port=8080 discovery.uri=http://duckgres-trino.${NAMESPACE}.svc:8080 - query.max-memory=6GB - query.max-memory-per-node=2GB + query.max-memory=${TRINO_QUERY_MEMORY} + query.max-memory-per-node=${TRINO_QUERY_MEMORY_PER_NODE} shutdown.grace-period=10s catalog.management=dynamic internal-communication.shared-secret=${ENV:TRINO_INTERNAL_COMMUNICATION_SHARED_SECRET} @@ -261,11 +261,9 @@ spec: initialDelaySeconds: 20 periodSeconds: 5 resources: - requests: { cpu: "1", memory: 4Gi } - # Three replicas provide the same aggregate 3-CPU/12Gi execution - # budget as the frozen-perf Duckgres worker. Keep requests equal - # to limits so node contention cannot change the comparison. - limits: { cpu: "1", memory: 4Gi } + requests: { cpu: "${TRINO_WORKER_CPU}", memory: ${TRINO_WORKER_MEMORY} } + # Keep requests equal to limits for every experimental shape. + limits: { cpu: "${TRINO_WORKER_CPU}", memory: ${TRINO_WORKER_MEMORY} } volumeMounts: - { name: config, mountPath: /etc/trino/node.properties, subPath: node.properties } - { name: config, mountPath: /etc/trino/jvm.config, subPath: jvm.config } diff --git a/tests/mw-dev/run.sh b/tests/mw-dev/run.sh index bfee09d1..8bf9f921 100755 --- a/tests/mw-dev/run.sh +++ b/tests/mw-dev/run.sh @@ -36,6 +36,75 @@ case "$E2E_SUITE" in esac TRINO_IMAGE="${TRINO_IMAGE:-ghcr.io/posthog/trino:4505364c570d6b51edecd299b603fca4b6693d86@sha256:ac80c275fd18a439d25da5652ab5cd3c80bcbdd2d88d64c9722dc3e8bb68ba07}" TRINO_TLS_PASSWORD="${TRINO_TLS_PASSWORD:-duckgres-e2e-keystore}" +TRINO_PERF_SHAPE="${TRINO_PERF_SHAPE:-baseline}" + +# Fixed experimental shapes keep worker sizing and aggregate memory limits in +# sync. Do not accept independent resource overrides that would change the A/B. +configure_trino_perf_shape() { + case "$TRINO_PERF_SHAPE" in + baseline) TRINO_WORKER_REPLICAS=3; TRINO_WORKER_CPU=1 ;; + large) TRINO_WORKER_REPLICAS=1; TRINO_WORKER_CPU=3 ;; + scaleout) TRINO_WORKER_REPLICAS=6; TRINO_WORKER_CPU=1 ;; + large-scaleout) TRINO_WORKER_REPLICAS=2; TRINO_WORKER_CPU=3 ;; + *) echo "TRINO_PERF_SHAPE must be baseline, large, scaleout, or large-scaleout (got $TRINO_PERF_SHAPE)" >&2; return 2 ;; + esac + if [ "$TRINO_PERF_SHAPE" != baseline ] && { [ "$SCENARIO_NAME" != posthog_frozen_perf ] || [ "$E2E_SUITE" != trino ]; }; then + echo "Nonbaseline TRINO_PERF_SHAPE requires SCENARIO_NAME=posthog_frozen_perf and E2E_SUITE=trino." >&2 + return 2 + fi + # Benchmark selection follows the validated shape, not inherited environment: + # keep cross-engine reference measurements once, then vary only Trino. + DUCKGRES_SCENARIO_PERF_MODE=full + if [ "$TRINO_PERF_SHAPE" != baseline ]; then + DUCKGRES_SCENARIO_PERF_MODE=trino-only + fi + TRINO_WORKER_MEMORY="$((TRINO_WORKER_CPU * 4))Gi" + TRINO_WORKER_HEAP="$((TRINO_WORKER_CPU * 3))G" + TRINO_QUERY_MEMORY_PER_NODE="$((TRINO_WORKER_CPU * 2))GB" + TRINO_QUERY_MEMORY="$((TRINO_WORKER_REPLICAS * TRINO_WORKER_CPU * 2))GB" +} + +trino_perf_shape_json() { + jq -n \ + --arg shape "$TRINO_PERF_SHAPE" \ + --arg perf_mode "$DUCKGRES_SCENARIO_PERF_MODE" \ + --argjson worker_replicas "$TRINO_WORKER_REPLICAS" \ + --arg worker_cpu "$TRINO_WORKER_CPU" --arg worker_memory "$TRINO_WORKER_MEMORY" \ + --arg worker_heap "$TRINO_WORKER_HEAP" \ + --arg query_memory_per_node "$TRINO_QUERY_MEMORY_PER_NODE" --arg query_memory "$TRINO_QUERY_MEMORY" \ + --argjson total_worker_cpu "$((TRINO_WORKER_REPLICAS * TRINO_WORKER_CPU))" \ + --argjson total_worker_memory_gib "$((TRINO_WORKER_REPLICAS * TRINO_WORKER_CPU * 4))" \ + --arg trino_image "$TRINO_IMAGE" --arg duckgres_worker_image "${WORKER_IMAGE:-}" \ + --arg controlplane_image "${CONTROLPLANE_IMAGE:-}" --arg scenario_runner_image "${SCENARIO_RUNNER_IMAGE:-}" \ + --arg duckgres_worker_cpu "$DUCKGRES_K8S_WORKER_CPU_REQUEST" \ + --arg duckgres_worker_memory "$DUCKGRES_K8S_WORKER_MEMORY_REQUEST" \ + --arg git_sha "${GITHUB_SHA:-}" --arg run_id "${GITHUB_RUN_ID:-}" \ + '$ARGS.named + {coordinator_cpu_request: "1", coordinator_cpu_limit: "2", coordinator_memory: "3Gi", coordinator_heap: "2G", coordinator_query_memory_per_node: "1GB", architecture: "arm64"}' +} + +record_trino_perf_shape() { + [ "$SCENARIO_NAME" = posthog_frozen_perf ] && [ "$E2E_SUITE" = trino ] || return 0 + mkdir -p "$SCENARIO_ARTIFACTS_DIR" + trino_perf_shape_json > "$SCENARIO_ARTIFACTS_DIR/trino-perf-shape.json" + echo "Trino perf shape: $TRINO_PERF_SHAPE; $TRINO_WORKER_REPLICAS workers x $TRINO_WORKER_CPU CPU/$TRINO_WORKER_MEMORY; heap=$TRINO_WORKER_HEAP; query memory=$TRINO_QUERY_MEMORY_PER_NODE per worker, $TRINO_QUERY_MEMORY cluster" +} + +verify_trino_perf_shape() { + [ "$SCENARIO_NAME" = posthog_frozen_perf ] && [ "$E2E_SUITE" = trino ] || return 0 + if [ -f "$SCENARIO_ARTIFACTS_DIR/trino-perf-shape.json" ]; then + # A separate test-scenario invocation must not silently relabel a stack + # deployed with another shape, image, resource budget, or workflow run. + # The runner image may be supplied only at test time; it does not size the + # deployed execution workers. + if ! jq -e --argjson expected "$(trino_perf_shape_json)" \ + 'del(.scenario_runner_image) == ($expected | del(.scenario_runner_image))' \ + "$SCENARIO_ARTIFACTS_DIR/trino-perf-shape.json" >/dev/null; then + echo "TRINO_PERF_SHAPE configuration does not match recorded deployment; restore the deployment environment or redeploy with the requested shape." >&2 + return 2 + fi + fi + record_trino_perf_shape +} # Internal secret for the per-PR control plane. Random per run; never reused. # Stamped into the rendered manifests and handed to the in-cluster harness. @@ -74,6 +143,7 @@ require_pr_identity() { } render() { + configure_trino_perf_shape : "${WORKER_IMAGE:?}" "${CONTROLPLANE_IMAGE:?}" "${PR_NUMBER:?}" ensure_secret_dir [ -f "$internal_secret_file" ] || (umask 077; openssl rand -hex 16 > "$internal_secret_file") @@ -94,8 +164,11 @@ render() { TRINO_CA_CERT_B64="$(base64 < "$trino_ca_cert_file" | tr -d '\n')" \ TRINO_SERVER_P12_B64="$(base64 < "$trino_server_p12_file" | tr -d '\n')" \ TRINO_IMAGE="$TRINO_IMAGE" TRINO_TLS_PASSWORD="$TRINO_TLS_PASSWORD" \ + TRINO_WORKER_CPU="$TRINO_WORKER_CPU" TRINO_WORKER_MEMORY="$TRINO_WORKER_MEMORY" \ + TRINO_WORKER_HEAP="$TRINO_WORKER_HEAP" TRINO_QUERY_MEMORY="$TRINO_QUERY_MEMORY" \ + TRINO_QUERY_MEMORY_PER_NODE="$TRINO_QUERY_MEMORY_PER_NODE" \ NAMESPACE="$NS" PR_NUMBER="$PR_NUMBER" \ - envsubst '$NAMESPACE $PR_NUMBER $TRINO_IMAGE $TRINO_TLS_PASSWORD $TRINO_CA_CERT_B64 $TRINO_SERVER_P12_B64' \ + envsubst '$NAMESPACE $PR_NUMBER $TRINO_IMAGE $TRINO_TLS_PASSWORD $TRINO_CA_CERT_B64 $TRINO_SERVER_P12_B64 $TRINO_WORKER_CPU $TRINO_WORKER_MEMORY $TRINO_WORKER_HEAP $TRINO_QUERY_MEMORY $TRINO_QUERY_MEMORY_PER_NODE' \ < "$HERE/manifests.trino.tmpl.yaml" fi } @@ -324,6 +397,8 @@ reset_pr_stack() { } cmd_deploy() { + configure_trino_perf_shape + record_trino_perf_shape reset_pr_stack echo "::group::Apply manifests ($NS)" @@ -337,7 +412,7 @@ cmd_deploy() { ensure_pod_identity restart_cp_with_identity - if [ "$SCENARIO_NAME" = "posthog_frozen_perf" ]; then + if [ "$SCENARIO_NAME" = "posthog_frozen_perf" ] && [ "$DUCKGRES_SCENARIO_PERF_MODE" = full ]; then ensure_scenario_pod_identity fi @@ -359,13 +434,11 @@ cmd_deploy() { # Patch the Deployment resources directly. The CI deployer intentionally # cannot patch the deployments/scale subresource, while it already needs # narrowly scoped Deployment patch access for the control-plane config. - # Three 1-CPU/4Gi workers match the frozen-perf Duckgres worker's - # aggregate 3-CPU/12Gi execution budget while exercising Trino's - # distributed path. + # Admit exactly the selected worker count after identity propagation. "${KUBECTL[@]}" -n "$NS" patch deployment duckgres-trino-coordinator \ --type=merge -p '{"spec":{"replicas":1}}' "${KUBECTL[@]}" -n "$NS" patch deployment duckgres-trino-worker \ - --type=merge -p '{"spec":{"replicas":3}}' + --type=merge -p "{\"spec\":{\"replicas\":$TRINO_WORKER_REPLICAS}}" "${KUBECTL[@]}" -n "$NS" rollout status deploy/duckgres-trino-coordinator --timeout=300s "${KUBECTL[@]}" -n "$NS" rollout status deploy/duckgres-trino-worker --timeout=300s fi @@ -488,6 +561,8 @@ scenario_job_name() { } cmd_test_scenario() { + configure_trino_perf_shape + verify_trino_perf_shape local scenario_file scenario_name : "${SCENARIO_RUNNER_IMAGE:?SCENARIO_RUNNER_IMAGE is required}" @@ -549,6 +624,7 @@ spec: - { name: DUCKGRES_SCENARIO_SNI_SUFFIX, value: "$suffix" } - { name: DUCKGRES_SCENARIO_FROZEN_S3_URI, value: "$FROZEN_S3_URI" } - { name: DUCKGRES_SCENARIO_TRINO_CA_CERT, value: "/trino-ca/ca.crt" } + - { name: DUCKGRES_SCENARIO_PERF_MODE, value: "$DUCKGRES_SCENARIO_PERF_MODE" } - { name: DUCKGRES_SCENARIO_ATHENA_REGION, value: "$AWS_REGION" } - { name: DUCKGRES_SCENARIO_ATHENA_WORKGROUP, value: "${DUCKGRES_SCENARIO_ATHENA_WORKGROUP:-}" } - { name: DUCKGRES_SCENARIO_ATHENA_DATABASE, value: "${DUCKGRES_SCENARIO_ATHENA_DATABASE:-}" } @@ -717,6 +793,13 @@ copy_scenario_artifacts() { failure_reason="${failure_reason}${failure_reason:+; }missing required scenario artifact $artifact" fi done + if [ "$copy_failed" -eq 0 ]; then + if [ "$SCENARIO_NAME" = posthog_frozen_perf ] && [ "$E2E_SUITE" = trino ] && \ + ! cp "$SCENARIO_ARTIFACTS_DIR/trino-perf-shape.json" "$staging/trino-perf-shape.json"; then + copy_failed=1 + failure_reason="failed to copy Trino shape provenance" + fi + fi if [ "$copy_failed" -eq 0 ]; then if mv "$staging" "$dest"; then echo "Copied scenario artifacts to $dest." diff --git a/tests/mw-dev/run_sh_test.go b/tests/mw-dev/run_sh_test.go index 65297d97..7acbddbe 100644 --- a/tests/mw-dev/run_sh_test.go +++ b/tests/mw-dev/run_sh_test.go @@ -2,6 +2,7 @@ package e2emwdev_test import ( "bytes" + "encoding/json" "io" "os" "os/exec" @@ -156,6 +157,19 @@ func TestDeployCreatesDedicatedScenarioPodIdentityForAthenaPerf(t *testing.T) { } } +func TestBaselinePerfStillRequiresAthenaPodIdentity(t *testing.T) { + fakes := newRunSHFakes(t) + out, err := runSHCommand(t, fakes.binDir, "deploy", + "SCENARIO_DEV_ALLOW_DUCKLING_DELETE=1", + "SCENARIO_NAME=posthog_frozen_perf", + "SCENARIO_POD_IDENTITY_ROLE=", + "DUCKGRES_SCENARIO_PERF_MODE=trino-only", + ).CombinedOutput() + if err == nil || !strings.Contains(string(out), "SCENARIO_POD_IDENTITY_ROLE is required for the Athena perf scenario") { + t.Fatalf("baseline must require Athena identity despite inherited Trino-only mode: %v\n%s", err, out) + } +} + func TestTrinoWorkersMatchDuckgresAggregateCompute(t *testing.T) { raw, err := os.ReadFile("manifests.trino.tmpl.yaml") if err != nil { @@ -168,6 +182,11 @@ func TestTrinoWorkersMatchDuckgresAggregateCompute(t *testing.T) { "${TRINO_TLS_PASSWORD}", "test-password", "${TRINO_CA_CERT_B64}", "dGVzdA==", "${TRINO_SERVER_P12_B64}", "dGVzdA==", + "${TRINO_WORKER_CPU}", "1", + "${TRINO_WORKER_MEMORY}", "4Gi", + "${TRINO_WORKER_HEAP}", "3G", + "${TRINO_QUERY_MEMORY_PER_NODE}", "2GB", + "${TRINO_QUERY_MEMORY}", "6GB", ).Replace(string(raw)) decoder := utilyaml.NewYAMLOrJSONDecoder(strings.NewReader(rendered), 4096) @@ -229,6 +248,197 @@ func TestTrinoWorkersMatchDuckgresAggregateCompute(t *testing.T) { } } +func TestTrinoPerfShapesDeployAndRecordResources(t *testing.T) { + for _, tc := range []struct { + shape, cpu, memory, heap, perNode, cluster string + replicas, totalCPU, totalMemoryGiB int + }{ + {"baseline", "1", "4Gi", "3G", "2GB", "6GB", 3, 3, 12}, + {"large", "3", "12Gi", "9G", "6GB", "6GB", 1, 3, 12}, + {"scaleout", "1", "4Gi", "3G", "2GB", "12GB", 6, 6, 24}, + {"large-scaleout", "3", "12Gi", "9G", "6GB", "12GB", 2, 6, 24}, + } { + t.Run(tc.shape, func(t *testing.T) { + fakes := newRunSHFakes(t) + perfMode, scenarioRole := "trino-only", "" + if tc.shape == "baseline" { + perfMode, scenarioRole = "full", "arn:aws:iam::123456789012:role/test-scenario" + } + for _, name := range []string{"duckgres-ci-trino-ca.crt", "duckgres-ci-trino-server.p12"} { + if err := os.WriteFile(filepath.Join(filepath.Dir(fakes.binDir), "secrets", name), []byte("tls-test-secret"), 0o600); err != nil { + t.Fatal(err) + } + } + out, err := runSHCommand(t, fakes.binDir, "deploy", + "SCENARIO_DEV_ALLOW_DUCKLING_DELETE=1", "E2E_SUITE=trino", "SCENARIO_NAME=posthog_frozen_perf", + "TRINO_PERF_SHAPE="+tc.shape, "TRINO_POD_IDENTITY_ROLE=arn:aws:iam::123456789012:role/test-trino", + "SCENARIO_POD_IDENTITY_ROLE="+scenarioRole, + "DUCKGRES_SCENARIO_PERF_MODE=inherited-invalid-mode", + "TRINO_IMAGE=example.invalid/trino:experiment", "TRINO_TLS_PASSWORD=never-record-this-secret", + ).CombinedOutput() + if err != nil { + t.Fatalf("deploy failed: %v\n%s", err, out) + } + calls := fakes.calls(t) + if got := strings.Contains(calls, "--service-account duckgres-scenario --role-arn"); got != (tc.shape == "baseline") { + t.Errorf("scenario Pod Identity present = %v, want %v", got, tc.shape == "baseline") + } + for _, want := range []string{ + `patch deployment duckgres-trino-worker --type=merge -p {"spec":{"replicas":` + strconv.Itoa(tc.replicas) + `}}`, + `requests: { cpu: "` + tc.cpu + `", memory: ` + tc.memory + ` }`, + `limits: { cpu: "` + tc.cpu + `", memory: ` + tc.memory + ` }`, + "-Xmx" + tc.heap, "query.max-memory-per-node=" + tc.perNode, "query.max-memory=" + tc.cluster, + "-Xmx2G", "query.max-memory-per-node=1GB", "${ENV:TRINO_INTERNAL_COMMUNICATION_SHARED_SECRET}", + } { + if !strings.Contains(calls, want) { + t.Errorf("rendered deploy missing %q", want) + } + } + raw, err := os.ReadFile(filepath.Join(filepath.Dir(fakes.binDir), "scenario-artifacts", "trino-perf-shape.json")) + if err != nil { + t.Fatal(err) + } + var metadata map[string]any + if err := json.Unmarshal(raw, &metadata); err != nil { + t.Fatal(err) + } + for key, want := range map[string]any{ + "shape": tc.shape, "worker_replicas": float64(tc.replicas), "worker_cpu": tc.cpu, + "worker_memory": tc.memory, "worker_heap": tc.heap, "query_memory_per_node": tc.perNode, + "query_memory": tc.cluster, "total_worker_cpu": float64(tc.totalCPU), "total_worker_memory_gib": float64(tc.totalMemoryGiB), + "trino_image": "example.invalid/trino:experiment", "perf_mode": perfMode, + } { + if metadata[key] != want { + t.Errorf("metadata[%s] = %v, want %v", key, metadata[key], want) + } + } + if strings.Contains(string(raw), "secret") { + t.Errorf("metadata contains a secret: %s", raw) + } + if !strings.Contains(string(out), "Trino perf shape: "+tc.shape) { + t.Errorf("missing shape summary: %s", out) + } + }) + } +} + +func TestTrinoPerfShapeDerivesRunnerMode(t *testing.T) { + for _, shape := range []string{"baseline", "large", "scaleout", "large-scaleout"} { + t.Run(shape, func(t *testing.T) { + fakes := newRunSHFakes(t) + out, err := runSHCommand(t, fakes.binDir, "test-scenario", + "SCENARIO_RUNNER_IMAGE=example.invalid/scenario:test", "TRINO_PERF_SHAPE="+shape, + "E2E_SUITE=trino", "SCENARIO_NAME=posthog_frozen_perf", + "DUCKGRES_SCENARIO_PERF_MODE=inherited-invalid-mode", + ).CombinedOutput() + if err != nil { + t.Fatalf("scenario failed: %v\n%s", err, out) + } + mode := "trino-only" + if shape == "baseline" { + mode = "full" + } + want := `name: DUCKGRES_SCENARIO_PERF_MODE, value: "` + mode + `"` + if calls := fakes.calls(t); !strings.Contains(calls, want) || strings.Contains(calls, "inherited-invalid-mode") { + t.Fatalf("runner must receive derived mode %s, never inherited override; calls:\n%s", mode, calls) + } + }) + } +} + +func TestTrinoPerfShapeRejectsInvalidSelectionBeforeCloudCalls(t *testing.T) { + for _, env := range [][]string{ + {"TRINO_PERF_SHAPE=unknown", "E2E_SUITE=trino", "SCENARIO_NAME=posthog_frozen_perf"}, + {"TRINO_PERF_SHAPE=large", "E2E_SUITE=neutral", "SCENARIO_NAME=posthog_frozen_perf"}, + {"TRINO_PERF_SHAPE=scaleout", "E2E_SUITE=trino", "SCENARIO_NAME=full-suite"}, + } { + for _, subcommand := range []string{"deploy", "test-scenario"} { + fakes := newRunSHFakes(t) + out, err := runSHCommand(t, fakes.binDir, subcommand, env...).CombinedOutput() + if err == nil || !strings.Contains(string(out), "TRINO_PERF_SHAPE") { + t.Errorf("invalid shape accepted: %s", out) + } + if calls := fakes.calls(t); calls != "" { + t.Errorf("invalid shape performed cloud calls: %s", calls) + } + } + } +} + +func TestTrinoPerfShapeProvenanceSurvivesDeployFailure(t *testing.T) { + fakes := newRunSHFakes(t) + out, err := runSHCommand(t, fakes.binDir, "deploy", "TRINO_PERF_SHAPE=large", "E2E_SUITE=trino", "SCENARIO_NAME=posthog_frozen_perf").CombinedOutput() + if err == nil || !strings.Contains(string(out), "refusing to reuse") { + t.Fatalf("expected failed stack reset: %v\n%s", err, out) + } + if _, err := os.Stat(filepath.Join(filepath.Dir(fakes.binDir), "scenario-artifacts", "trino-perf-shape.json")); err != nil { + t.Fatalf("failure lost provenance: %v", err) + } +} + +func TestTrinoPerfShapeIncludedWithScenarioArtifacts(t *testing.T) { + fakes := newRunSHFakes(t) + out, err := runSHCommand(t, fakes.binDir, "test-scenario", + "SCENARIO_RUNNER_IMAGE=example.invalid/scenario:test", "TRINO_PERF_SHAPE=large-scaleout", + "E2E_SUITE=trino", "SCENARIO_NAME=posthog_frozen_perf", + ).CombinedOutput() + if err != nil { + t.Fatalf("scenario failed: %v\n%s", err, out) + } + root := filepath.Join(filepath.Dir(fakes.binDir), "scenario-artifacts") + entries, err := os.ReadDir(root) + if err != nil { + t.Fatal(err) + } + for _, entry := range entries { + if entry.IsDir() { + if _, err := os.Stat(filepath.Join(root, entry.Name(), "trino-perf-shape.json")); err != nil { + t.Fatalf("scenario lost provenance: %v", err) + } + return + } + } + t.Fatal("missing scenario result directory") +} + +func TestTrinoPerfShapeDoesNotBlockTeardown(t *testing.T) { + fakes := newRunSHFakes(t) + out, err := runSHCommand(t, fakes.binDir, "teardown", "TRINO_PERF_SHAPE=invalid", "SCENARIO_DEV_ALLOW_DUCKLING_DELETE=1").CombinedOutput() + if err != nil { + t.Fatalf("invalid shape blocked teardown: %v\n%s", err, out) + } + if !strings.Contains(fakes.calls(t), "delete namespace duckgres-ci-pr-123") { + t.Fatal("teardown did not delete namespace") + } +} + +func TestTrinoPerfShapeRejectsScenarioProvenanceMismatch(t *testing.T) { + fakes := newRunSHFakes(t) + // Even a failed deployment leaves the shape selected for that attempt. + _, _ = runSHCommand(t, fakes.binDir, "deploy", "TRINO_PERF_SHAPE=large", "E2E_SUITE=trino", "SCENARIO_NAME=posthog_frozen_perf").CombinedOutput() + priorCalls := fakes.calls(t) + out, err := runSHCommand(t, fakes.binDir, "test-scenario", + "SCENARIO_RUNNER_IMAGE=example.invalid/scenario:test", "E2E_SUITE=trino", "SCENARIO_NAME=posthog_frozen_perf", + ).CombinedOutput() + if err == nil || !strings.Contains(string(out), "does not match recorded deployment") { + t.Fatalf("scenario accepted different deployment shape: %v\n%s", err, out) + } + if calls := fakes.calls(t); calls != priorCalls { + t.Fatal("mismatched scenario made cloud calls") + } + raw, err := os.ReadFile(filepath.Join(filepath.Dir(fakes.binDir), "scenario-artifacts", "trino-perf-shape.json")) + if err != nil { + t.Fatal(err) + } + var metadata map[string]any + if err := json.Unmarshal(raw, &metadata); err != nil { + t.Fatal(err) + } + if metadata["shape"] != "large" { + t.Fatalf("overwrote deployed shape: %s", raw) + } +} + func TestScheduledCleanupKeepsGoingWhenDucklingsDoNotDelete(t *testing.T) { fakes := newRunSHFakes(t) @@ -1987,7 +2197,12 @@ printf 'test-secret\n' writeFake(t, binDir, "envsubst", `#!/usr/bin/env bash printf 'envsubst %s\n' "$*" >> "$RUN_SH_TEST_CALLS" -cat +template="$(cat)" +for token in $1; do + name="${token#\$}" + template="${template//\$\{$name\}/${!name}}" +done +printf '%s\n' "$template" `) writeFake(t, binDir, "curl", `#!/usr/bin/env bash @@ -2046,6 +2261,7 @@ func runSHCommand(t *testing.T, binDir, subcommand string, extraEnv ...string) * "EKS_CLUSTER_NAME=test-cluster", "AWS_REGION=us-east-1", "E2E_SUITE=neutral", + "TRINO_PERF_SHAPE=baseline", "SCENARIO_NAME=full-suite", "SCENARIO_POD_IDENTITY_ROLE=", "SCENARIO_ARTIFACTS_DIR="+filepath.Join(filepath.Dir(binDir), "scenario-artifacts"), diff --git a/tests/mw-dev/scenario/perf/adapter_test.go b/tests/mw-dev/scenario/perf/adapter_test.go index 33821b38..c42041c3 100644 --- a/tests/mw-dev/scenario/perf/adapter_test.go +++ b/tests/mw-dev/scenario/perf/adapter_test.go @@ -264,6 +264,43 @@ func TestExecutorBuildsTrinoDriverFromReadinessState(t *testing.T) { } } +func TestExecutorRunsFrozenCatalogTrinoOnly(t *testing.T) { + state := provision.NewState() + state.StoreProvisionResponse("scenario-org", provision.ProvisionResponse{Password: "test-password"}) + state.StoreTrinoStatus("scenario-org", provision.TrinoStatus{ + Cell: provision.TrinoCell{ID: "test-cell", CoordinatorURL: "https://trino.example.test:8443"}, + Enabled: true, Available: true, + Status: &provision.TrinoOrgStatus{ + Org: "scenario-org", Cell: "test-cell", Principal: "test-principal", Catalog: "test_catalog", State: provision.WarehouseStateReady, + }, + }) + factory := &fakeDriverFactory{} + executor := NewExecutor(ExecutorConfig{ + ProvisionState: state, OutputDir: t.TempDir(), DriverFactory: factory, + }) + err := executor.ExecuteStep(context.Background(), core.Step{ + ID: "perf_queries", Type: StepTypePerfQueries, + With: map[string]any{ + "org_id": "scenario-org", "run_id": "test-run", + "catalog_file": filepath.Join("..", "..", "..", "perf", "queries", "ducklake_posthog_tables.yaml"), + "targets": []any{"trino"}, "trino_ca_cert_file": "/tmp/test-ca.crt", + }, + }) + if err != nil { + t.Fatal(err) + } + if factory.pgwireDriver != nil || factory.athenaDriver != nil || factory.trinoDriver == nil { + t.Fatal("Trino-only must not initialize non-Trino drivers") + } + result, ok := executor.State().Result("perf_queries") + if !ok || result.Summary.WarmupQueries != 7 || result.Summary.TotalQueries != 28 || result.Summary.TotalErrors != 0 { + t.Fatalf("expected seven warmups and four measurements per Trino query: %+v", result) + } + if !factory.trinoDriver.closed { + t.Fatal("Trino driver was not closed") + } +} + func TestExecutorBuildsAthenaDriverFromExplicitOnDemandConfig(t *testing.T) { catalogPath := writePerfCatalog(t, []perfcore.Protocol{perfcore.ProtocolAthena}) provisionState := provision.NewState() diff --git a/tests/mw-dev/scenario/perf_mode_test.go b/tests/mw-dev/scenario/perf_mode_test.go new file mode 100644 index 00000000..f38eacff --- /dev/null +++ b/tests/mw-dev/scenario/perf_mode_test.go @@ -0,0 +1,93 @@ +package scenario + +import ( + "path/filepath" + "reflect" + "strings" + "testing" + + "github.com/posthog/duckgres/tests/mw-dev/scenario/core" +) + +func TestLoadScenarioForRunTrinoOnly(t *testing.T) { + t.Setenv("DUCKGRES_SCENARIO_PERF_MODE", "trino-only") + path := filepath.Join("scenarios", "posthog_frozen_perf.yaml") + original, err := core.LoadScenario(path) + if err != nil { + t.Fatal(err) + } + for _, key := range original.RequiredEnv { + value := "test-value" + if strings.HasPrefix(key, "DUCKGRES_SCENARIO_ATHENA_") { + value = "" + } + t.Setenv(key, value) + } + got, absPath, err := loadScenarioForRun(path) + if err != nil { + t.Fatal(err) + } + if missing := missingRequiredEnv(got); len(missing) != 0 { + t.Fatalf("Trino-only requires unrelated environment: %v", missing) + } + if _, err := resolveRunTemplates(got, "test-run"); err != nil { + t.Fatalf("Trino-only must resolve without Athena configuration: %v", err) + } + want := resolveScenarioFilePaths(original, filepath.Dir(absPath)) + if len(got.Steps) != len(want.Steps) { + t.Fatal("Trino-only must retain provisioning, table setup, validation and cleanup") + } + for i, step := range got.Steps { + if step.Type != "perf_queries" { + if !reflect.DeepEqual(step, want.Steps[i]) { + t.Fatalf("non-perf step %s changed", step.ID) + } + continue + } + if !reflect.DeepEqual(step.With["targets"], []any{"trino"}) { + t.Fatalf("targets=%v, want only trino", step.With["targets"]) + } + for key, value := range want.Steps[i].With { + if strings.HasPrefix(key, "athena_") { + if _, exists := step.With[key]; exists { + t.Errorf("unused Athena field %s retained", key) + } + } else if key != "targets" && !reflect.DeepEqual(step.With[key], value) { + t.Errorf("shared perf field %s changed", key) + } + } + } + for _, key := range got.RequiredEnv { + if strings.HasPrefix(key, "DUCKGRES_SCENARIO_ATHENA_") { + t.Errorf("unused required environment retained: %s", key) + } + } +} + +func TestLoadScenarioForRunFullModeUnchanged(t *testing.T) { + for _, file := range []string{"posthog_frozen_perf.yaml", "full-suite.yaml"} { + path := filepath.Join("scenarios", file) + t.Setenv("DUCKGRES_SCENARIO_PERF_MODE", "") + want, _, err := loadScenarioForRun(path) + if err != nil { + t.Fatal(err) + } + t.Setenv("DUCKGRES_SCENARIO_PERF_MODE", "full") + got, _, err := loadScenarioForRun(path) + if err != nil || !reflect.DeepEqual(got, want) { + t.Fatalf("full mode changed %s: %v", file, err) + } + } +} + +func TestLoadScenarioForRunRejectsInvalidPerfMode(t *testing.T) { + for _, tc := range []struct{ file, mode string }{ + {"posthog_frozen_perf.yaml", "typo"}, + {"full-suite.yaml", "trino-only"}, + } { + t.Setenv("DUCKGRES_SCENARIO_PERF_MODE", tc.mode) + if _, _, err := loadScenarioForRun(filepath.Join("scenarios", tc.file)); err == nil { + t.Fatalf("accepted mode %q for %s", tc.mode, tc.file) + } + } +} diff --git a/tests/mw-dev/scenario/runner_test.go b/tests/mw-dev/scenario/runner_test.go index 5b395cdd..1e6913d6 100644 --- a/tests/mw-dev/scenario/runner_test.go +++ b/tests/mw-dev/scenario/runner_test.go @@ -1012,9 +1012,47 @@ func loadScenarioForRun(path string) (core.Scenario, string, error) { if err != nil { return core.Scenario{}, "", err } + scenario, err = selectPerfMode(scenario, os.Getenv("DUCKGRES_SCENARIO_PERF_MODE")) + if err != nil { + return core.Scenario{}, "", err + } return resolveScenarioFilePaths(scenario, filepath.Dir(absPath)), absPath, nil } +// Select targets before required-env validation and template resolution, so a +// Trino-only experiment does not need credentials for engines it never runs. +// The shared catalog and all provisioning/validation/cleanup steps stay intact. +func selectPerfMode(s core.Scenario, mode string) (core.Scenario, error) { + if mode == "" || mode == "full" { + return s, nil + } + if mode != "trino-only" || s.Name != "posthog-frozen-perf" { + return core.Scenario{}, fmt.Errorf("DUCKGRES_SCENARIO_PERF_MODE must be full, or trino-only for posthog-frozen-perf") + } + out := s + out.RequiredEnv = make([]string, 0, len(s.RequiredEnv)) + for _, key := range s.RequiredEnv { + if !strings.HasPrefix(key, "DUCKGRES_SCENARIO_ATHENA_") { + out.RequiredEnv = append(out.RequiredEnv, key) + } + } + out.Steps = make([]core.Step, len(s.Steps)) + for i, step := range s.Steps { + if step.Type == scenarioperf.StepTypePerfQueries { + with := make(map[string]any, len(step.With)) + for key, value := range step.With { + if !strings.HasPrefix(key, "athena_") { + with[key] = value + } + } + with["targets"] = []any{"trino"} + step.With = with + } + out.Steps[i] = step + } + return out, nil +} + func resolveScenarioFilePaths(s core.Scenario, baseDir string) core.Scenario { out := s out.Steps = make([]core.Step, len(s.Steps)) diff --git a/tests/mw-dev/scenario/script_test.go b/tests/mw-dev/scenario/script_test.go index f018125d..ef74407c 100644 --- a/tests/mw-dev/scenario/script_test.go +++ b/tests/mw-dev/scenario/script_test.go @@ -67,6 +67,72 @@ func TestScenarioRunScriptCheckEnvIncludesScenarioRequiredEnv(t *testing.T) { } } +func TestScenarioRunScriptPerfModePreflight(t *testing.T) { + for _, tc := range []struct { + name, mode, scenario, missing, want string + execute bool + }{ + {name: "trino-only needs no Athena", mode: "trino-only"}, + {name: "Trino-only reaches Go without Athena", mode: "trino-only", execute: true}, + {name: "full still requires Athena", mode: "full", want: "DUCKGRES_SCENARIO_ATHENA_WORKGROUP"}, + {name: "default still requires Athena", want: "DUCKGRES_SCENARIO_ATHENA_WORKGROUP"}, + {name: "Trino CA still required", mode: "trino-only", missing: "DUCKGRES_SCENARIO_TRINO_CA_CERT", want: "DUCKGRES_SCENARIO_TRINO_CA_CERT"}, + {name: "invalid mode", mode: "other", want: "DUCKGRES_SCENARIO_PERF_MODE"}, + {name: "invalid mode does not reach Go", mode: "other", want: "DUCKGRES_SCENARIO_PERF_MODE", execute: true}, + {name: "wrong scenario", mode: "trino-only", scenario: "provision_smoke", want: "trino-only requires"}, + } { + t.Run(tc.name, func(t *testing.T) { + scenario := tc.scenario + if scenario == "" { + scenario = "posthog_frozen_perf" + } + binDir := t.TempDir() + marker := filepath.Join(binDir, "external-call") + for _, bin := range []string{"go", "aws", "kubectl"} { + if err := os.WriteFile(filepath.Join(binDir, bin), []byte("#!/bin/sh\ntouch \"$TEST_EXTERNAL_CALL\"\nexit 99\n"), 0o700); err != nil { + t.Fatal(err) + } + } + if err := os.WriteFile(filepath.Join(binDir, "go"), []byte("#!/bin/sh\nprintf '%s\\n' \"$DUCKGRES_SCENARIO_PERF_MODE\" \"$*\" > \"$TEST_EXTERNAL_CALL\"\n"), 0o700); err != nil { + t.Fatal(err) + } + args := []string{filepath.Join("..", "..", "..", "scripts", "scenario_run.sh")} + if !tc.execute { + args = append(args, "--check-env") + } + args = append(args, "tests/mw-dev/scenario/scenarios/"+scenario+".yaml") + cmd := exec.Command("bash", args...) + cmd.Env = []string{"PATH=" + binDir + string(os.PathListSeparator) + os.Getenv("PATH"), "TEST_EXTERNAL_CALL=" + marker, "DUCKGRES_SCENARIO_PERF_MODE=" + tc.mode} + for key, value := range map[string]string{ + "DUCKGRES_SCENARIO_API_BASE": "http://127.0.0.1", "DUCKGRES_SCENARIO_INTERNAL_SECRET": "test-secret", + "DUCKGRES_SCENARIO_PG_HOST": "127.0.0.1", "DUCKGRES_SCENARIO_SNI_SUFFIX": ".dev.example", + "DUCKGRES_SCENARIO_ORG_ID": "test-org", "DUCKGRES_SCENARIO_FROZEN_S3_URI": "s3://example/frozen/", + "DUCKGRES_SCENARIO_TRINO_CA_CERT": "/tmp/test-ca.crt", "DUCKGRES_K8S_WORKER_CPU_REQUEST": "3", + "DUCKGRES_K8S_WORKER_MEMORY_REQUEST": "12Gi", + } { + if key != tc.missing { + cmd.Env = append(cmd.Env, key+"="+value) + } + } + out, err := cmd.CombinedOutput() + if tc.want == "" && err != nil { + t.Fatalf("preflight should succeed: %v\n%s", err, out) + } + if tc.want != "" && (err == nil || !strings.Contains(string(out), tc.want)) { + t.Fatalf("preflight should reject with %q: %v\n%s", tc.want, err, out) + } + if tc.execute && tc.want == "" { + call, err := os.ReadFile(marker) + if err != nil || !strings.Contains(string(call), "trino-only\ntest -count=1 ./tests/mw-dev/scenario") { + t.Fatalf("entrypoint did not invoke Go with inherited Trino-only mode: %v\n%s", err, call) + } + } else if _, err := os.Stat(marker); !os.IsNotExist(err) { + t.Fatalf("preflight made an external call: %v", err) + } + }) + } +} + func TestDevScenarioWorkflowUsesUnifiedMwDevHarness(t *testing.T) { workflowPath := filepath.Join("..", "..", "..", ".github", "workflows", "scenario-dev.yml") raw, err := os.ReadFile(workflowPath) @@ -142,7 +208,6 @@ func TestDevScenarioWorkflowUsesUnifiedMwDevHarness(t *testing.T) { "DUCKGRES_SCENARIO_CONFIG_SECRET", "DUCKGRES_SCENARIO_INTERNAL_SECRET_NAME", "DUCKGRES_SCENARIO_INTERNAL_SECRET_KEY", - "matrix:", "trino.trino.svc", "DUCKGRES_SCENARIO_API_BASE: ${{ secrets.", "DUCKGRES_SCENARIO_INTERNAL_SECRET: ${{ secrets.", diff --git a/tests/mw-dev/scenario/trino_shape_workflow_test.go b/tests/mw-dev/scenario/trino_shape_workflow_test.go new file mode 100644 index 00000000..6808cb2b --- /dev/null +++ b/tests/mw-dev/scenario/trino_shape_workflow_test.go @@ -0,0 +1,130 @@ +package scenario + +import ( + "os" + "path/filepath" + "reflect" + "strings" + "testing" + + "gopkg.in/yaml.v3" +) + +func TestScenarioWorkflowTrinoShapeExperiments(t *testing.T) { + raw, err := os.ReadFile(filepath.Join("..", "..", "..", ".github", "workflows", "scenario-dev.yml")) + if err != nil { + t.Fatal(err) + } + var workflow struct { + On struct { + Dispatch struct { + Inputs map[string]struct { + Type string `yaml:"type"` + Default string `yaml:"default"` + Options []string `yaml:"options"` + } `yaml:"inputs"` + } `yaml:"workflow_dispatch"` + } `yaml:"on"` + Jobs map[string]struct { + Needs []string `yaml:"needs"` + If string `yaml:"if"` + Strategy struct { + MaxParallel int `yaml:"max-parallel"` + FailFast *bool `yaml:"fail-fast"` + Matrix string `yaml:"matrix"` + } `yaml:"strategy"` + Env map[string]string `yaml:"env"` + Steps []struct { + Name string `yaml:"name"` + If string `yaml:"if"` + Run string `yaml:"run"` + With map[string]any `yaml:"with"` + } `yaml:"steps"` + } `yaml:"jobs"` + } + if err := yaml.Unmarshal(raw, &workflow); err != nil { + t.Fatal(err) + } + input, ok := workflow.On.Dispatch.Inputs["trino_perf_shape"] + if !ok || input.Type != "choice" || input.Default != "baseline" || !reflect.DeepEqual(input.Options, []string{"baseline", "large", "scaleout", "large-scaleout", "all"}) { + t.Fatalf("manual dispatch must offer baseline, three experiments, and all; got %+v", input) + } + job := workflow.Jobs["scenario"] + if got := job.Env["TRINO_PERF_SHAPE"]; got != "${{ matrix.shape }}" { + t.Fatalf("scenario must receive each concrete shape, never all: %q", got) + } + if job.Strategy.MaxParallel != 1 || job.Strategy.FailFast == nil || *job.Strategy.FailFast || job.Strategy.Matrix != "${{ fromJSON(needs.plan.outputs.matrix) }}" { + t.Fatalf("all shapes must run sequentially and retain remaining results on failure: %+v", job.Strategy) + } + if job.Env["PR_NUMBER"] != "${{ github.run_id }}${{ matrix.suffix }}" || job.Env["NAMESPACE"] != "duckgres-ci-pr-${{ github.run_id }}${{ matrix.suffix }}" { + t.Fatal("each shape must have a distinct cleanup and warehouse identity") + } + if !reflect.DeepEqual(job.Needs, []string{"plan", "scenario-runner-image", "duckgres-image"}) { + t.Fatal("shapes must reuse the same pair of image builds") + } + plan := workflow.Jobs["plan"] + if plan.Env["TRINO_PERF_SHAPE"] != "${{ inputs.trino_perf_shape || 'baseline' }}" { + t.Fatal("scheduled/default runs must plan a single baseline shape") + } + var foundPublish, foundSummary, foundAthena bool + var teardownIndex, outcomeIndex, uploadIndex = -1, -1, -1 + for i, step := range job.Steps { + if step.Name == "Load Athena perf configuration" { + foundAthena = true + if step.If != "env.SCENARIO_NAME == 'posthog_frozen_perf' && matrix.shape == 'baseline'" { + t.Fatalf("only baseline needs Athena configuration: %s", step.If) + } + } + if step.Name == "Publish scenario perf results" { + foundPublish = true + if step.If != "${{ always() && github.ref == 'refs/heads/main' && env.TRINO_PERF_SHAPE == 'baseline' && inputs.trino_perf_shape != 'all' }}" { + t.Fatalf("experiment measurements must not enter baseline historical results: %s", step.If) + } + } + if step.Name == "Publish scenario summary" { + foundSummary = true + if !strings.Contains(step.Run, "trino-perf-shape.json") { + t.Fatal("workflow summary must include the shape artifact so results are attributable") + } + } + switch step.Name { + case "Teardown": + teardownIndex = i + if step.If != "always()" { + t.Fatal("each matrix member must teardown on failure") + } + case "Record shape outcome": + outcomeIndex = i + if step.If != "always()" || !strings.Contains(step.Run, "shape-result.json") { + t.Fatal("each shape must record failure/cleanup outcomes") + } + case "Upload scenario artifacts": + uploadIndex = i + if step.If != "always()" || step.With["name"] != "scenario-dev-${{ github.run_id }}-${{ github.run_attempt }}-${{ matrix.shape }}" { + t.Fatal("each shape must upload uniquely named artifacts even on failure") + } + } + } + if !foundPublish || !foundSummary || !foundAthena { + t.Fatal("missing summary or performance publication step") + } + if teardownIndex < 0 || outcomeIndex <= teardownIndex || uploadIndex <= outcomeIndex { + t.Fatal("outcome and artifact publication must follow teardown") + } + comparison, ok := workflow.Jobs["compare-shapes"] + if !ok || !reflect.DeepEqual(comparison.Needs, []string{"plan", "scenario"}) || !strings.Contains(comparison.If, "always()") || !strings.Contains(comparison.If, "inputs.trino_perf_shape == 'all'") { + t.Fatal("all mode must summarize all matrix outcomes, including failed ones") + } + var foundCompare, foundComparisonPublish bool + for _, step := range comparison.Steps { + if strings.Contains(step.Run, "go run ./cmd/duckgres-perf-shape-summary") { + foundCompare = true + } + if strings.Contains(step.Run, "$GITHUB_STEP_SUMMARY") && step.If == "always()" { + foundComparisonPublish = true + } + } + if !foundCompare || !foundComparisonPublish { + t.Fatal("comparison must render results and publish incomplete summaries on failure") + } +} diff --git a/tests/mw-dev/scenario/trino_shapes_script_test.go b/tests/mw-dev/scenario/trino_shapes_script_test.go new file mode 100644 index 00000000..10ee058a --- /dev/null +++ b/tests/mw-dev/scenario/trino_shapes_script_test.go @@ -0,0 +1,109 @@ +package scenario + +import ( + "bytes" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "reflect" + "strconv" + "strings" + "testing" +) + +func TestTrinoShapesScriptPlansIsolatedRuns(t *testing.T) { + type entry struct { + Shape string `json:"shape"` + Suffix string `json:"suffix"` + } + for _, tc := range []struct { + name string + scenario string + shape string + want []entry + }{ + {name: "default", scenario: "full-suite", want: []entry{{Shape: "baseline"}}}, + {name: "baseline", scenario: "posthog_frozen_perf", shape: "baseline", want: []entry{{Shape: "baseline"}}}, + {name: "large", scenario: "posthog_frozen_perf", shape: "large", want: []entry{{Shape: "large"}}}, + {name: "scaleout", scenario: "posthog_frozen_perf", shape: "scaleout", want: []entry{{Shape: "scaleout"}}}, + {name: "large-scaleout", scenario: "posthog_frozen_perf", shape: "large-scaleout", want: []entry{{Shape: "large-scaleout"}}}, + {name: "all", scenario: "posthog_frozen_perf", shape: "all", want: []entry{ + {Shape: "baseline", Suffix: "1"}, + {Shape: "large", Suffix: "2"}, + {Shape: "scaleout", Suffix: "3"}, + {Shape: "large-scaleout", Suffix: "4"}, + }}, + } { + t.Run(tc.name, func(t *testing.T) { + stdout, stderr, err := runTrinoShapesScript(tc.scenario, tc.shape) + if err != nil { + t.Fatalf("script failed: %v: %s", err, stderr) + } + if stderr != "" { + t.Fatalf("unexpected stderr: %q", stderr) + } + var matrix struct { + Include []entry `json:"include"` + } + if err := json.Unmarshal([]byte(stdout), &matrix); err != nil { + t.Fatalf("expected JSON-only output, got %q: %v", stdout, err) + } + if !reflect.DeepEqual(matrix.Include, tc.want) { + t.Fatalf("matrix=%+v, want %+v", matrix.Include, tc.want) + } + // The harness derives namespaces and identifiers from PR_NUMBER, which + // the workflow constructs by appending each suffix to the run ID. + seen := make(map[string]bool) + for _, item := range matrix.Include { + id := "12345678901" + item.Suffix + if _, err := strconv.ParseUint(id, 10, 64); err != nil { + t.Fatalf("run identifier is not numeric: %q", id) + } + if seen[id] { + t.Fatalf("duplicate run identifier: %q", id) + } + seen[id] = true + } + }) + } +} + +func TestTrinoShapesScriptRejectsInvalidInputBeforeOutput(t *testing.T) { + for _, tc := range []struct { + name string + scenario string + shape string + want string + }{ + {name: "unknown shape", scenario: "posthog_frozen_perf", shape: "huge", want: "Invalid TRINO_PERF_SHAPE"}, + {name: "JSON injection", scenario: "posthog_frozen_perf", shape: "large\"}\n", want: "Invalid TRINO_PERF_SHAPE"}, + {name: "large other scenario", scenario: "full-suite", shape: "large", want: "require posthog_frozen_perf"}, + {name: "scaleout other scenario", scenario: "posthog_frozen_metadata", shape: "scaleout", want: "require posthog_frozen_perf"}, + {name: "large-scaleout other scenario", scenario: "full-suite", shape: "large-scaleout", want: "require posthog_frozen_perf"}, + {name: "all other scenario", scenario: "full-suite", shape: "all", want: "require posthog_frozen_perf"}, + {name: "all missing scenario", shape: "all", want: "require posthog_frozen_perf"}, + } { + t.Run(tc.name, func(t *testing.T) { + stdout, stderr, err := runTrinoShapesScript(tc.scenario, tc.shape) + if err == nil || !strings.Contains(stderr, tc.want) { + t.Fatalf("err=%v stderr=%q, want %q", err, stderr, tc.want) + } + if stdout != "" { + t.Fatalf("failed validation emitted matrix output: %q", stdout) + } + }) + } +} + +func runTrinoShapesScript(scenario, shape string) (string, string, error) { + cmd := exec.Command("bash", filepath.Join("..", "..", "..", "scripts", "scenario_trino_shapes.sh")) + cmd.Env = []string{"PATH=" + os.Getenv("PATH"), "SCENARIO_NAME=" + scenario} + if shape != "" { + cmd.Env = append(cmd.Env, "TRINO_PERF_SHAPE="+shape) + } + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + err := cmd.Run() + return stdout.String(), stderr.String(), err +} diff --git a/tests/perf/README.md b/tests/perf/README.md index aef39dd2..6ded26ba 100644 --- a/tests/perf/README.md +++ b/tests/perf/README.md @@ -75,6 +75,111 @@ 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. +### Trino worker shape experiments + +The `scenario-dev` workflow accepts a manual `trino_perf_shape` choice for +`posthog_frozen_perf`. Scheduled runs and callers that omit the choice retain +`baseline`. The three experiments vary worker size and total execution +resources independently: + +| Shape | Workers | CPU / memory per worker | Total CPU / memory | Heap per worker | Query memory per worker / cluster | +| --- | ---: | --- | --- | --- | --- | +| `baseline` | 3 | 1 / 4 GiB | 3 / 12 GiB | 3 GiB | 2 / 6 GiB | +| `large` | 1 | 3 / 12 GiB | 3 / 12 GiB | 9 GiB | 6 / 6 GiB | +| `scaleout` | 6 | 1 / 4 GiB | 6 / 24 GiB | 3 GiB | 2 / 12 GiB | +| `large-scaleout` | 2 | 3 / 12 GiB | 6 / 24 GiB | 9 GiB | 6 / 12 GiB | + +Worker CPU and memory requests equal their limits. The coordinator keeps its +existing resources, 2 GiB heap, and 1 GiB per-node query limit; its cluster +query-memory setting follows the selected budget. JVM heap headroom retains +Trino's default of 30% of the heap. Duckgres remains at 3 CPU / 12 GiB in the +workflow, so only `baseline` and `large` match its execution resource budget. +Coordinator and supporting-service resources are additional to the table. + +Run the entire comparison with one dispatch on the PR branch (or `main` after +merging): + +```bash +gh workflow run scenario-dev.yml --ref codex/trino-perf-shape-experiments \ + -f scenario=posthog_frozen_perf \ + -f trino_perf_shape=all +``` + +`all` builds the runner and Duckgres images once, then runs the four shapes as +a sequential matrix (`max-parallel: 1`). Each shape gets one temporary stack, +the existing warmup and four measured iterations, and teardown before the next +job starts. Numeric suffixes `1` through `4` on the workflow run ID give each +shape a separate namespace and warehouse identity. Each job has its own +270-minute timeout and fresh credentials. A failed shape does not cancel the +remaining jobs. This is one measurement round, with no repeated deployments +per shape. The Trino image remains pinned by the workflow. + +The baseline runs the full cross-engine benchmark (uncached and cached +Duckgres, Trino, and Athena). The three nonbaseline shapes run only Trino +measurements, avoiding redundant Duckgres and Athena work. Each still performs +the same isolated warehouse provisioning, frozen-data setup, and table/view +validation. Trino SQL, query order, warmup, and four measured iterations are +unchanged. Nonbaseline shapes do not load Athena configuration or create the +scenario runner's Athena Pod Identity association. + +The final `compare-shapes` job presents Trino medians, baseline-relative +speedups, and CPU-budget efficiency in one workflow summary and a downloadable +`trino-shape-comparison--` artifact. Missing or failed results and +teardown failures are marked incomplete and fail the comparison job. Each +shape's raw artifact is named `scenario-dev---` and includes +`shape-result.json` with deployment, scenario, and teardown outcomes. + +Individual choices remain available: replace `all` with `baseline`, `large`, +`scaleout`, or `large-scaleout`. Default and scheduled invocations execute just +`baseline`, with the full workload. Individual nonbaseline choices also run +Trino-only measurements. Compare identical query IDs and protocol labels from the current +`balanced_v4` catalog; do not mix them with older methodology. + +The workflow title and summary identify the shape. The downloadable scenario +artifact includes `trino-perf-shape.json` with configured resource and image +provenance and `perf_mode` (`full` or `trino-only`), including on deployment +failure. A copy accompanies the collected +scenario results. Nonbaseline experiments and the entire `all` comparison +(including its baseline member) are artifact-only and do not publish to the +daily baseline's historical tables. Trino SQL, cache settings, warmup count, +and measured iteration count are identical across shapes; other protocols are +measured only in the baseline. + +Use per-query median latency and allocated CPU-seconds (total worker CPU times +elapsed seconds) to compare speed and resource efficiency. The distinct-person +query is the primary diagnostic case. The harness excludes warmups from +`query_results.csv`; these results are measured iterations, not a separately +instrumented cold-cache benchmark. A topology speedup alone does not distinguish +GC, throttling, exchange traffic, scan throughput, or memory pressure: correlate +with execution telemetry before attributing its cause. + +For local harness development, set `TRINO_PERF_SHAPE` to one concrete shape +alongside `SCENARIO_NAME=posthog_frozen_perf`, `E2E_SUITE=trino`, and the usual +isolated-stack environment, and use that same environment for `run.sh deploy` +and `run.sh test-scenario`. The test invocation checks any saved deployment +provenance, including the measurement mode derived from the selected shape. +`DUCKGRES_SCENARIO_PERF_MODE` is set by the harness and cannot override that +selection. The harness rejects a different shape, resource budget, or deployment +image; restore the deployment environment or redeploy before continuing. Use a separate +`SCENARIO_ARTIFACTS_DIR` for each local stack. `all` is a workflow selection, +not a shape accepted by `run.sh`. Install `jq` on the machine running `run.sh` to +write the JSON provenance (the GitHub runner already includes it). Explicitly +set the Duckgres worker variables to +`3` and `12Gi` to reproduce the workflow. The generic harness retains its +smaller Duckgres defaults. `just scenario-frozen-perf` uses an existing +warehouse and does not resize Trino. Unknown shapes and nonbaseline shapes +outside the isolated frozen-perf scenario are rejected before cloud mutations. + +If deployment or testing fails, inspect the selected shape artifact and pod +events, then use the normal `run.sh teardown` with the same namespace and run +identity. Teardown remains available even with a malformed shape selection. +For an `all` run, use the suffixed `PR_NUMBER` and namespace shown in the failed +shape job, not the unsuffixed workflow run ID. Rerun the whole workflow for a +complete combined report; rerunning only failed jobs produces a new attempt +with missing shape artifacts, which the comparison intentionally rejects. +Rerun in a fresh isolated stack; a partially started worker pool is not a valid +measurement. Restore `baseline` to return to the daily configuration. + ## Paired Query Catalogs Existing catalogs continue to use `queries:` unchanged. A catalog may contain diff --git a/tests/perf/shapecompare/compare.go b/tests/perf/shapecompare/compare.go new file mode 100644 index 00000000..7ca4fccf --- /dev/null +++ b/tests/perf/shapecompare/compare.go @@ -0,0 +1,313 @@ +// Package shapecompare summarizes the four-shape Trino experiment without +// publishing SQL, connection details, or raw artifact error messages. +package shapecompare + +import ( + "encoding/csv" + "encoding/json" + "errors" + "fmt" + "io/fs" + "math" + "os" + "path/filepath" + "regexp" + "slices" + "strconv" + "strings" +) + +type shape struct { + name string + replicas, cpu int +} + +var shapes = []shape{{"baseline", 3, 1}, {"large", 1, 3}, {"scaleout", 6, 1}, {"large-scaleout", 2, 3}} +var safeQueryID = regexp.MustCompile(`^q_[a-z0-9_]{1,160}__ducklake_table$`) + +type result struct { + status string + medians map[string]float64 + dataset string + image string +} + +// Generate always returns a Markdown report, including when artifacts are +// missing or malformed. A non-nil error means the comparison is incomplete. +func Generate(dir string) (string, error) { + results := make(map[string]result) + candidates := make(map[string][]string) + entries, readErr := os.ReadDir(dir) + unknown := false + for _, entry := range entries { + if !entry.IsDir() { + continue + } + root := filepath.Join(dir, entry.Name()) + var metadata struct { + Shape string `json:"shape"` + } + _ = readJSON(filepath.Join(root, "trino-perf-shape.json"), &metadata) + name := metadata.Shape + if !knownShape(name) { + // Artifact names still identify a shape if deployment failed before + // provenance was written. Check the longest suffix first. + for _, suffix := range []string{"large-scaleout", "baseline", "scaleout", "large"} { + if strings.HasSuffix(entry.Name(), "-"+suffix) { + name = suffix + break + } + } + } + if !knownShape(name) { + unknown = true + continue + } + candidates[name] = append(candidates[name], root) + } + complete := readErr == nil && !unknown + for _, s := range shapes { + paths := candidates[s.name] + var r result + switch len(paths) { + case 0: + r.status = "missing artifact" + case 1: + r = load(paths[0], s) + default: + r.status = "duplicate artifacts" + } + results[s.name] = r + } + base := results["baseline"] + if base.status == "complete" { + for _, s := range shapes[1:] { + r := results[s.name] + if r.status == "complete" && (r.dataset != base.dataset || !sameQueries(r.medians, base.medians)) { + r.status = "dataset or query set differs from baseline" + results[s.name] = r + } + if r.status == "complete" && r.image != base.image { + r.status = "image differs from baseline" + results[s.name] = r + } + } + } + var out strings.Builder + out.WriteString("## Trino worker-shape comparison\n\n") + out.WriteString("Only Trino measurements contribute to latency and speedup comparisons; baseline artifacts may also include other engines.\n\n") + out.WriteString("| Shape | Execution workers | Status |\n|---|---|---|\n") + for _, s := range shapes { + r := results[s.name] + fmt.Fprintf(&out, "| %s | %d × %d CPU / %d GiB | %s |\n", s.name, s.replicas, s.cpu, s.cpu*4, r.status) + if r.status != "complete" { + complete = false + } + } + if unknown { + out.WriteString("\nUnrecognized artifact directories were found.\n") + } + if complete { + out.WriteString("\nComparison complete.\n") + } else { + out.WriteString("\nComparison incomplete. Failed or partial shapes are excluded from latency comparisons.\n") + } + querySet := make(map[string]bool) + for _, r := range results { + if r.status == "complete" { + for q := range r.medians { + querySet[q] = true + } + } + } + queries := make([]string, 0, len(querySet)) + for q := range querySet { + queries = append(queries, q) + } + slices.Sort(queries) + if len(queries) > 0 { + out.WriteString("\nTrino measured-query median latency in seconds (four measured iterations per query):\n\n| Query | baseline | large | scaleout | large-scaleout |\n|---|---:|---:|---:|---:|\n") + for _, q := range queries { + fmt.Fprintf(&out, "| %s |", q) + for _, s := range shapes { + r := results[s.name] + value, ok := r.medians[q] + if r.status == "complete" && ok { + fmt.Fprintf(&out, " %.3f |", value) + } else { + out.WriteString(" — |") + } + } + out.WriteByte('\n') + } + } + if base.status == "complete" { + out.WriteString("\nWorkload speedup uses the sum of query medians, relative to baseline. Allocated CPU-budget efficiency is speedup divided by the execution CPU increase (3 CPU baseline; 6 CPU for scaleout shapes). This does not measure CPU utilization. Coordinator resources are excluded.\n\n| Shape | Workload speedup | Allocated CPU-budget efficiency |\n|---|---:|---:|\n") + for _, s := range shapes { + r := results[s.name] + if r.status == "complete" { + speedup := sum(base.medians) / sum(r.medians) + fmt.Fprintf(&out, "| %s | %.2f× | %.1f%% |\n", s.name, speedup, speedup*3/float64(s.replicas*s.cpu)*100) + } else { + fmt.Fprintf(&out, "| %s | — | — |\n", s.name) + } + } + } + if !complete { + return out.String(), errors.New("trino shape comparison incomplete; see the Markdown status table") + } + return out.String(), nil +} + +func knownShape(name string) bool { + for _, s := range shapes { + if s.name == name { + return true + } + } + return false +} + +func load(root string, s shape) result { + fail := func(status string) result { return result{status: status} } + var job struct { + Shape string `json:"shape"` + Deploy string `json:"deploy"` + Scenario string `json:"scenario"` + Teardown string `json:"teardown"` + } + if readJSON(filepath.Join(root, "shape-result.json"), &job) != nil || job.Shape != s.name { + return fail("missing or malformed job result") + } + if job.Deploy != "success" || job.Scenario != "success" || job.Teardown != "success" { + return fail("job did not complete successfully") + } + var provenance struct { + Shape string `json:"shape"` + Replicas int `json:"worker_replicas"` + CPU string `json:"worker_cpu"` + TotalCPU int `json:"total_worker_cpu"` + TotalMemory int `json:"total_worker_memory_gib"` + Image string `json:"trino_image"` + PerfMode string `json:"perf_mode"` + } + if readJSON(filepath.Join(root, "trino-perf-shape.json"), &provenance) != nil || provenance.Shape != s.name || provenance.Replicas != s.replicas || provenance.CPU != strconv.Itoa(s.cpu) || provenance.TotalCPU != s.replicas*s.cpu || provenance.TotalMemory != s.replicas*s.cpu*4 || provenance.Image == "" { + return fail("missing or inconsistent resource provenance") + } + // Older experiment artifacts predate perf_mode and remain comparable. + if provenance.PerfMode != "" && provenance.PerfMode != "full" && provenance.PerfMode != "trino-only" { + return fail("missing or inconsistent resource provenance") + } + var scenarioFiles []string + err := filepath.WalkDir(root, func(path string, d fs.DirEntry, e error) error { + if e != nil { + return e + } + if !d.IsDir() && d.Name() == "scenario_summary.json" { + scenarioFiles = append(scenarioFiles, path) + } + return nil + }) + if err != nil || len(scenarioFiles) != 1 { + return fail("missing or duplicate scenario summary") + } + var scenario struct { + Status string `json:"status"` + FailedSteps int `json:"failed_steps"` + } + if readJSON(scenarioFiles[0], &scenario) != nil || (scenario.Status != "success" && scenario.Status != "success_with_retries") || scenario.FailedSteps != 0 { + return fail("scenario failed or incomplete") + } + perfDir := filepath.Join(filepath.Dir(scenarioFiles[0]), "perf") + var summary struct { + TotalQueries int `json:"total_queries"` + TotalErrors int `json:"total_errors"` + Dataset string `json:"dataset_version"` + } + if readJSON(filepath.Join(perfDir, "summary.json"), &summary) != nil || summary.TotalQueries <= 0 || summary.TotalErrors != 0 || summary.Dataset == "" { + return fail("perf summary failed or incomplete") + } + rows, e := readCSV(filepath.Join(perfDir, "query_results.csv")) + if e != nil || len(rows) < 2 { + return fail("missing or malformed query results") + } + header := []string{"query_id", "intent_id", "measure_iteration", "protocol", "status", "error", "error_class", "rows", "duration_ms", "started_at"} + if !slices.Equal(rows[0], header) { + return fail("missing or malformed query results") + } + if len(rows)-1 != summary.TotalQueries { + return fail("query count does not match summary") + } + seen := make(map[string]bool) + samples := make(map[string][]float64) + for _, row := range rows[1:] { + if len(row) != len(header) { + return fail("failed or invalid query result") + } + if provenance.PerfMode == "trino-only" && row[3] != "trino" { + return fail("query protocol does not match declared perf mode") + } + ms, msErr := strconv.ParseFloat(row[8], 64) + iteration, iterationErr := strconv.Atoi(row[2]) + count, countErr := strconv.ParseInt(row[7], 10, 64) + if row[4] != "ok" || row[5] != "" || row[6] != "" || msErr != nil || math.IsNaN(ms) || math.IsInf(ms, 0) || ms <= 0 || iterationErr != nil || iteration < 1 || iteration > 4 || countErr != nil || count < 0 { + return fail("failed or invalid query result") + } + key := row[0] + "\x00" + row[3] + "\x00" + row[2] + if seen[key] { + return fail("duplicate query iteration") + } + seen[key] = true + if row[3] == "trino" { + if !safeQueryID.MatchString(row[0]) { + return fail("failed or invalid query result") + } + samples[row[0]] = append(samples[row[0]], ms/1000) + } + } + if len(samples) == 0 { + return fail("no Trino measurements") + } + medians := make(map[string]float64) + for q, values := range samples { + if len(values) != 4 { + return fail("incomplete measured iterations") + } + slices.Sort(values) + medians[q] = (values[1] + values[2]) / 2 + } + return result{status: "complete", medians: medians, dataset: summary.Dataset, image: provenance.Image} +} + +func readJSON(path string, dst any) error { + b, e := os.ReadFile(path) + if e != nil { + return e + } + return json.Unmarshal(b, dst) +} +func readCSV(path string) ([][]string, error) { + b, e := os.ReadFile(path) + if e != nil { + return nil, e + } + return csv.NewReader(strings.NewReader(string(b))).ReadAll() +} +func sameQueries(a, b map[string]float64) bool { + if len(a) != len(b) { + return false + } + for q := range a { + if _, ok := b[q]; !ok { + return false + } + } + return true +} +func sum(values map[string]float64) float64 { + var total float64 + for _, v := range values { + total += v + } + return total +} diff --git a/tests/perf/shapecompare/compare_test.go b/tests/perf/shapecompare/compare_test.go new file mode 100644 index 00000000..e06dda2a --- /dev/null +++ b/tests/perf/shapecompare/compare_test.go @@ -0,0 +1,339 @@ +package shapecompare + +import ( + "encoding/csv" + "encoding/json" + "os" + "path/filepath" + "slices" + "strconv" + "strings" + "testing" +) + +func TestGenerateComparison(t *testing.T) { + dir := completeFixture(t) + got, err := Generate(dir) + if err != nil { + t.Fatal(err) + } + for _, want := range []string{"Comparison complete", "baseline | 3 × 1 CPU / 4 GiB | complete", "large-scaleout | 2 × 3 CPU / 12 GiB | complete", "| q_events_total_balanced_v4__ducklake_table | 10.000 | 5.000 | 5.000 | 2.500 |", "| large | 2.00× | 200.0% |", "| scaleout | 2.00× | 100.0% |", "| large-scaleout | 4.00× | 200.0% |"} { + if !strings.Contains(got, want) { + t.Errorf("missing %q in:\n%s", want, got) + } + } +} + +func TestGenerateUsesMedianAndIgnoresOtherProtocolLatency(t *testing.T) { + dir := completeFixture(t) + mutateCSV(t, dir, func(rows [][]string) [][]string { + for i, value := range []string{"1000", "3000", "7000", "99000"} { + rows[i+1][8] = value + } + for i := 1; i <= 4; i++ { + row := slices.Clone(rows[i]) + row[3] = "pgwire_cached" + row[8] = "1" + rows = append(rows, row) + } + return rows + }) + writeJSON(t, filepath.Join(dir, "artifact-large", "scenario", "perf", "summary.json"), map[string]any{"total_queries": 8, "total_errors": 0, "dataset_version": "fixture"}) + // The nested provenance copy belongs to this artifact, not another run. + writeJSON(t, filepath.Join(dir, "artifact-large", "scenario", "trino-perf-shape.json"), map[string]string{"shape": "large"}) + got, err := Generate(dir) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(got, "| q_events_total_balanced_v4__ducklake_table | 10.000 | 5.000 | 5.000 | 2.500 |") { + t.Fatal(got) + } +} + +func TestGenerateFullBaselineAndTrinoOnlyExperiments(t *testing.T) { + dir := mixedProtocolFixture(t) + got, err := Generate(dir) + if err != nil { + t.Fatal(err) + } + for _, want := range []string{ + "Comparison complete", + "Only Trino measurements contribute to latency and speedup comparisons; baseline artifacts may also include other engines.", + "| large | 2.00× | 200.0% |", + "| large-scaleout | 4.00× | 200.0% |", + } { + if !strings.Contains(got, want) { + t.Errorf("missing %q in:\n%s", want, got) + } + } + if strings.Count(got, "__ducklake_table |") != 7 || strings.Contains(got, "__raw_view") { + t.Fatalf("comparison must contain exactly the seven Trino table queries:\n%s", got) + } +} + +func TestGenerateMixedProtocolsStillRejectsIncompleteMeasurements(t *testing.T) { + dir := mixedProtocolFixture(t) + mutateCSV(t, dir, func(rows [][]string) [][]string { return rows[:len(rows)-1] }) + writeJSON(t, filepath.Join(dir, "artifact-large", "scenario", "perf", "summary.json"), map[string]any{ + "total_queries": 27, "warmup_queries": 7, "total_errors": 0, "dataset_version": "fixture", + }) + got, err := Generate(dir) + if err == nil || !strings.Contains(got, "incomplete measured iterations") { + t.Fatalf("missing fourth Trino iteration must fail comparison: err=%v\n%s", err, got) + } +} + +func TestGenerateValidatesDeclaredPerfMode(t *testing.T) { + for _, tc := range []struct { + name, mode, want string + }{ + {"unknown mode", "unknown", "missing or inconsistent resource provenance"}, + {"Trino-only with other protocols", "trino-only", "query protocol does not match declared perf mode"}, + } { + t.Run(tc.name, func(t *testing.T) { + dir := mixedProtocolFixture(t) + setPerfMode(t, dir, "baseline", tc.mode) + got, err := Generate(dir) + if err == nil || !strings.Contains(got, tc.want) { + t.Fatalf("invalid declared mode must fail: err=%v\n%s", err, got) + } + }) + } +} + +func setPerfMode(t *testing.T, dir, shape, mode string) { + t.Helper() + path := filepath.Join(dir, "artifact-"+shape, "trino-perf-shape.json") + var metadata map[string]any + if err := readJSON(path, &metadata); err != nil { + t.Fatal(err) + } + metadata["perf_mode"] = mode + writeJSON(t, path, metadata) +} + +func mixedProtocolFixture(t *testing.T) string { + t.Helper() + dir := completeFixture(t) + queries := []string{ + "events_total", "events_count_one_day", "events_by_name_march_2026", + "events_distinct_persons", "persons_total", "persons_daily_april_2026", "events_daily_march_2026", + } + type protocolVariant struct{ protocol, representation string } + for _, s := range shapes { + mode := "trino-only" + variants := []protocolVariant{{"trino", "ducklake_table"}} + if s.name == "baseline" { + mode = "full" + variants = append(variants, + protocolVariant{"pgwire_uncached", "raw_view"}, + protocolVariant{"pgwire_uncached", "ducklake_table"}, + protocolVariant{"pgwire_cached", "raw_view"}, + protocolVariant{"pgwire_cached", "ducklake_table"}, + protocolVariant{"athena", "athena_external"}, + ) + } + setPerfMode(t, dir, s.name, mode) + rows := [][]string{{"query_id", "intent_id", "measure_iteration", "protocol", "status", "error", "error_class", "rows", "duration_ms", "started_at"}} + for _, variant := range variants { + ms := "1" // Non-Trino timings must not affect shape speedups. + if variant.protocol == "trino" { + ms = map[string]string{"baseline": "10000", "large": "5000", "scaleout": "5000", "large-scaleout": "2500"}[s.name] + } + for _, query := range queries { + for iteration := 1; iteration <= 4; iteration++ { + rows = append(rows, []string{"q_" + query + "_balanced_v4__" + variant.representation, "intent_" + query, strconv.Itoa(iteration), variant.protocol, "ok", "", "", "1", ms, "2026-01-01T00:00:00Z"}) + } + } + } + perf := filepath.Join(dir, "artifact-"+s.name, "scenario", "perf") + // The runner counts warmups in summary.json but emits only measured + // executions to query_results.csv: baseline 42/168; experiments 7/28. + writeJSON(t, filepath.Join(perf, "summary.json"), map[string]any{ + "total_queries": len(rows) - 1, "warmup_queries": len(variants) * len(queries), "total_errors": 0, "dataset_version": "fixture", + }) + writeCSV(t, filepath.Join(perf, "query_results.csv"), rows) + } + return dir +} + +func TestGenerateIncompleteAndUnsafeArtifacts(t *testing.T) { + for _, tc := range []struct { + name string + mutate func(*testing.T, string) + want string + }{ + {"missing directory", func(t *testing.T, d string) { + if err := os.RemoveAll(d); err != nil { + t.Fatal(err) + } + }, "missing artifact"}, + {"missing result", func(t *testing.T, d string) { remove(t, filepath.Join(d, "artifact-large", "shape-result.json")) }, "missing or malformed job result"}, + {"missing provenance", func(t *testing.T, d string) { remove(t, filepath.Join(d, "artifact-large", "trino-perf-shape.json")) }, "missing or inconsistent resource provenance"}, + {"different image", func(t *testing.T, d string) { + p := filepath.Join(d, "artifact-large", "trino-perf-shape.json") + var metadata map[string]any + if err := readJSON(p, &metadata); err != nil { + t.Fatal(err) + } + metadata["trino_image"] = "other-image" + writeJSON(t, p, metadata) + }, "image differs from baseline"}, + {"different dataset", func(t *testing.T, d string) { + writeJSON(t, filepath.Join(d, "artifact-large", "scenario", "perf", "summary.json"), map[string]any{"total_queries": 4, "total_errors": 0, "dataset_version": "different"}) + }, "dataset or query set differs from baseline"}, + {"different queries", func(t *testing.T, d string) { + mutateCSV(t, d, func(rows [][]string) [][]string { + for _, row := range rows[1:] { + row[0] = "q_persons_total_balanced_v4__ducklake_table" + } + return rows + }) + }, "dataset or query set differs from baseline"}, + {"teardown failed", func(t *testing.T, d string) { + writeJSON(t, filepath.Join(d, "artifact-large", "shape-result.json"), map[string]string{"shape": "large", "deploy": "success", "scenario": "success", "teardown": "failure"}) + }, "job did not complete successfully"}, + {"scenario failed", func(t *testing.T, d string) { + writeJSON(t, filepath.Join(d, "artifact-large", "scenario", "scenario_summary.json"), map[string]any{"status": "failed", "error": "SECRET"}) + }, "scenario failed or incomplete"}, + {"missing csv", func(t *testing.T, d string) { + remove(t, filepath.Join(d, "artifact-large", "scenario", "perf", "query_results.csv")) + }, "missing or malformed query results"}, + {"failed row", func(t *testing.T, d string) { + mutateCSV(t, d, func(rows [][]string) [][]string { + rows[1][4] = "error" + rows[1][5] = "SECRET https://private.example" + rows[1][8] = "0" + return rows + }) + }, "failed or invalid query result"}, + {"NaN duration", func(t *testing.T, d string) { + mutateCSV(t, d, func(rows [][]string) [][]string { rows[1][8] = "NaN"; return rows }) + }, "failed or invalid query result"}, + {"partial rows", func(t *testing.T, d string) { + mutateCSV(t, d, func(rows [][]string) [][]string { return rows[:len(rows)-1] }) + }, "query count does not match summary"}, + {"duplicate iteration", func(t *testing.T, d string) { + mutateCSV(t, d, func(rows [][]string) [][]string { rows[2] = rows[1]; return rows }) + }, "duplicate query iteration"}, + {"unsafe query ID", func(t *testing.T, d string) { + mutateCSV(t, d, func(rows [][]string) [][]string { rows[1][0] = "SECRET |