CHORE: Improve PR performance sensitivity and measurement reliability

eng/profiler_benchmarks/README.md

@@ -30,7 +30,23 @@ improvement thresholds are unchanged.
 
 CI uses the PR merge's first parent as the exact base. It reuses the
 profiling-enabled candidate build after pytest and builds the base separately.
-Fresh processes run five measured pairs after one warmup pair in alternating order.
+Fresh processes run five measured pairs after one discarded warmup pair in alternating order.
+Each worker also warms every steady-state task once before measuring it in that
+same process. Connection opening remains a single cold invocation.
+
+Short read/query tasks use fixed blocks shared by both revisions: SELECT and
+256 KiB LOB fetching repeat 100 times, row-by-row fetching 10 times, CTE queries
+20 times, and positional/named query workloads 3 times. On the observed hosted
+Unix controls these blocks target roughly 100 ms or more; they are not adaptive
+duration gates. Other workloads remain single-pass. Setup and cleanup keep their
+existing timing boundaries.
+
+The reported wall time is the mean time per original workload within each block,
+not the total block duration. Raw inner timings and aggregate counters are retained
+in the artifacts; phase diagnostics and call counts are normalized per repetition.
+Repetition and in-process warmup counts must match across all base/candidate pairs.
+Retained single-pass reports remain readable. Repetitions do not increase the number
+of independent paired comparisons, and the 1 ms floor remains per original workload.
 
 Workers have six minutes each. CI allows 90 minutes inside a 100-minute step and
 160-minute job; local runs receive 105 minutes because they build both revisions.
@@ -50,8 +66,18 @@ is published. A failed aggregate build can still publish usable profiler artifac
 Exact-head reports may finalize after merge; stale heads are ignored. Missing,
 malformed, canceled, incomplete, or invalid data remains unavailable.
 
-The report highlights consistent slowdowns and improvements using the same 20%
-median change, 1 ms absolute change, and 80% pair-agreement requirements.
+The report highlights consistent slowdowns and improvements in two advisory tiers:
+
+- Tentative small impact: more than 10% median paired movement.
+- Large impact: more than 20% median paired movement.
+
+Both tiers require at least 1 ms between median per-workload runtimes and at least
+80% of pairs exceeding that tier's relative threshold in the same direction.
+Large impact is evaluated first. A median above 20% with agreement only at 10%
+can therefore remain tentative small impact. Inconsistent movement produces no
+headline, badge, or signal fingerprint; its percentage and agreement remain in
+the expanded timing table as no signal. Missing environments remain unavailable
+and a clean partial assessment is labeled incomplete.
 
 The publisher waits up to 220 minutes inside a 230-minute workflow. The first main
 comparison after introduction may be incomplete because its parent lacks this

eng/profiler_benchmarks/controller.py

@@ -11,6 +11,7 @@
 import platform
 import re
 import signal
+import statistics
 import subprocess
 import sys
 import tarfile
@@ -145,6 +146,49 @@ def load_suite():
     return core, workloads
 
 
+def measure_scenario(profiler, name):
+    repetitions = workloads.MEASUREMENT_REPEATS.get(name, 1)
+    warmups = 0 if name == "connect" else 1
+    results = []
+    identity = None
+    for index in range(warmups + repetitions):
+        with contextlib.redirect_stdout(io.StringIO()):
+            result = profiler.run(name)[0]
+        if result["cpp"] is None or result["py"] is None:
+            raise RuntimeError(f"Scenario {name} was skipped")
+        if not result["cpp"]:
+            raise RuntimeError(f"Scenario {name} has no native samples")
+        work = result.get("detail", "Connection: 1").split(" (")[0]
+        if identity is not None and work != identity:
+            raise RuntimeError(f"Workload changed within scenario {name}")
+        identity = work
+        if index >= warmups:
+            results.append(result)
+
+    times = [result["wall_ms"] for result in results]
+    output = dict(
+        wall_ms=statistics.mean(times),
+        wall_samples_ms=times,
+        repetitions=repetitions,
+        warmups=warmups,
+        work=identity,
+    )
+    for layer in ("cpp", "py"):
+        stats = {}
+        for result in results:
+            for label, counter in result[layer].items():
+                if label not in stats:
+                    stats[label] = dict(counter)
+                else:
+                    total = stats[label]
+                    total["calls"] += counter["calls"]
+                    total["total_us"] += counter["total_us"]
+                    total["min_us"] = min(total["min_us"], counter["min_us"])
+                    total["max_us"] = max(total["max_us"], counter["max_us"])
+        output[layer] = stats
+    return output
+
+
 def worker(args):
     # Import the chosen driver FIRST, then the SAME workload/controller for both
     # revisions. Never mix two native extensions into one interpreter.
@@ -165,22 +209,14 @@ def worker(args):
                 json.dumps(dict(status="running", active_scenario=name, scenarios=output)),
                 encoding="utf-8",
             )
-            # Keep phase tables out of logs, but never hide which workload stalled.
-            with contextlib.redirect_stdout(io.StringIO()):
-                result = profiler.run(name)[0]
-            if result["cpp"] is None or result["py"] is None:
-                raise RuntimeError(f"Scenario {name} was skipped")
-            if not result["cpp"]:
-                raise RuntimeError(f"Scenario {name} has no native samples")
-            output[name] = {key: result[key] for key in ("wall_ms", "cpp", "py")}
-            output[name]["work"] = result.get("detail", "Connection: 1").split(" (")[0]
+            output[name] = measure_scenario(profiler, name)
             args.output.write_text(
                 json.dumps(
                     dict(status="running", active_scenario=None, scenarios=output), allow_nan=False
                 ),
                 encoding="utf-8",
             )
-            print(f"Completed scenario: {name} ({result['wall_ms']:.3f} ms)", flush=True)
+            print(f"Completed scenario: {name} ({output[name]['wall_ms']:.3f} ms)", flush=True)
         print("Collecting server metadata", flush=True)
         profiler._ensure_connection()
         with profiler._conn.cursor() as cursor: