feat(explorer): distinguish execution variants in identity, ranking and labels

CHANGELOG.md

@@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Documentation
 
+- The Explorer read model is at version 16 with execution-variant identity:
+  the pipeline records `variant_id`, `variant_label`, the requested execution
+  engine and gateway routing on every result, cohort and leaderboard row.
+  Two bundles that differ only in requested engine rank as separate variants
+  ("Polars (streaming)" beside "Polars"), and gateway-routed runs are
+  excluded from native ranking with `ranking_exclusion_reason=gateway_routed`.
+  Existing corpus values are unchanged: bundles without variant data keep
+  `variant_id` equal to the platform id.
+
 - The Polars version-matrix reproduction instructions now use
   `--execution-engine` for default, in-memory and streaming setups, and identify
   the earlier `engine` option name used by the recorded bundles.

_project/scripts/explorer_pipeline/contract.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 EXPLORER_BUILD_CONTRACT_VERSION = "6"
-EXPLORER_READ_MODEL_VERSION = 15
+EXPLORER_READ_MODEL_VERSION = 16
 EXPLORER_READ_MODEL_COMPATIBILITY = {
     "minimum_supported": EXPLORER_READ_MODEL_VERSION,
     "newer_policy": "warn-and-continue",

_project/scripts/explorer_pipeline/duckdb_builder.py

@@ -330,6 +330,11 @@ def _create_schema(self, con: Any) -> None:
                 scale_factor         DOUBLE   NOT NULL,
                 platform             VARCHAR  NOT NULL,
                 platform_id          VARCHAR  NOT NULL,
+                variant_id           VARCHAR  NOT NULL,
+                variant_label        VARCHAR  NOT NULL,
+                requested_execution_engine VARCHAR NOT NULL,
+                gateway_name         VARCHAR,
+                gateway_routed       BOOLEAN  NOT NULL,
                 driver_version       VARCHAR,
                 run_date             VARCHAR  NOT NULL,
                 power_score          DOUBLE,
@@ -403,6 +408,11 @@ def _create_schema(self, con: Any) -> None:
                 r.scale_factor,
                 r.platform,
                 r.platform_id,
+                r.variant_id,
+                r.variant_label,
+                r.requested_execution_engine,
+                r.gateway_name,
+                r.gateway_routed,
                 r.driver_version,
                 r.run_date,
                 r.power_score,
@@ -527,6 +537,8 @@ def _create_schema(self, con: Any) -> None:
                 phase                        VARCHAR  NOT NULL,
                 result_id                    VARCHAR  NOT NULL,
                 platform_id                  VARCHAR  NOT NULL,
+                variant_id                   VARCHAR  NOT NULL,
+                variant_label                VARCHAR  NOT NULL,
                 platform                     VARCHAR  NOT NULL,
                 short_id                     VARCHAR  NOT NULL,
                 trust_label                  VARCHAR  NOT NULL,
@@ -575,6 +587,11 @@ def _create_schema(self, con: Any) -> None:
                 scale_factor,
                 platform,
                 platform_id,
+                variant_id,
+                variant_label,
+                requested_execution_engine,
+                gateway_name,
+                gateway_routed,
                 driver_version,
                 run_date,
                 power_score,
@@ -629,6 +646,8 @@ def _create_schema(self, con: Any) -> None:
                 primary_metric   VARCHAR  NOT NULL,
                 primary_order    VARCHAR  NOT NULL,
                 platform_id      VARCHAR  NOT NULL,
+                variant_id       VARCHAR  NOT NULL,
+                variant_label    VARCHAR  NOT NULL,
                 platform         VARCHAR  NOT NULL,
                 result_id        VARCHAR  NOT NULL,
                 short_id         VARCHAR  NOT NULL,
@@ -653,7 +672,9 @@ def _create_schema(self, con: Any) -> None:
         con.execute("CREATE INDEX idx_cohort_metadata_platform ON cohort_metadata (cohort_key, platform_id)")
         con.execute("""
             CREATE TABLE meta_leaderboard (
-                platform_id  VARCHAR  PRIMARY KEY,
+                variant_id   VARCHAR  PRIMARY KEY,
+                variant_label VARCHAR NOT NULL,
+                platform_id  VARCHAR  NOT NULL,
                 platform     VARCHAR  NOT NULL,
                 avg_rank     DOUBLE,
                 n_cohorts    INTEGER  NOT NULL
@@ -774,6 +795,11 @@ def _populate_results(
                     entry.scale_factor,
                     entry.platform,
                     entry.platform_id,
+                    entry.variant_id,
+                    entry.variant_label,
+                    entry.requested_execution_engine,
+                    entry.gateway_name,
+                    entry.gateway_routed,
                     entry.driver_version,
                     entry.run_date,
                     entry.power_score,
@@ -971,6 +997,8 @@ def _populate_benchmark_rankings(
                         phase,
                         platform_row.result_id,
                         platform_row.platform_id,
+                        platform_row.variant_id or platform_row.platform_id,
+                        platform_row.variant_label or platform_row.platform,
                         platform_row.platform,
                         full_to_short.get(platform_row.result_id, ""),
                         platform_row.trust_label,
@@ -1094,6 +1122,8 @@ def _populate_cohort_metadata(
                         primary_metric,
                         primary_order,
                         p["platform_id"],
+                        p.get("variant_id") or p["platform_id"],
+                        p.get("variant_label") or p["platform"],
                         p["platform"],
                         p["result_id"],
                         p["short_id"],
@@ -1123,14 +1153,16 @@ def _populate_meta_leaderboard(self, con: Any, meta: dict[str, Any]) -> None:
         for p in meta.get("platforms", []):
             rows.append(
                 (
+                    p.get("variant_id") or p["platform_id"],
+                    p.get("variant_label") or p["platform"],
                     p["platform_id"],
                     p["platform"],
                     p.get("avg_rank"),
                     p.get("n_cohorts", 0),
                 )
             )
         if rows:
-            con.executemany("INSERT INTO meta_leaderboard VALUES (?, ?, ?, ?)", rows)
+            con.executemany("INSERT INTO meta_leaderboard VALUES (?, ?, ?, ?, ?, ?)", rows)
 
     def _populate_short_ids(self, con: Any, short_id_map: dict[str, str]) -> None:
         rows = list(short_id_map.items())

_project/scripts/explorer_pipeline/models.py

@@ -62,6 +62,23 @@ def _platform_id(raw: str) -> str:
     return stripped.lower().replace(" ", "-")
 
 
+def variant_label(
+    platform: str,
+    *,
+    variant_id: str,
+    requested_execution_engine: str = "default",
+    gateway_name: str | None = None,
+    gateway_routed: bool = False,
+) -> str:
+    if gateway_routed and gateway_name:
+        return f"{platform} via {str(gateway_name).capitalize()}"
+    if requested_execution_engine and requested_execution_engine != "default":
+        return f"{platform} ({requested_execution_engine})"
+    if variant_id and variant_id != _platform_id(platform):
+        return f"{platform} ({variant_id})"
+    return platform
+
+
 RANKING_ELIGIBLE_VISIBILITIES: frozenset[str] = frozenset(
     {
         "public-curated",
@@ -95,6 +112,11 @@ class ManifestEntry(BaseModel):
     scale_factor: float
     platform: str
     platform_id: str = ""
+    variant_id: str = ""
+    variant_label: str = ""
+    requested_execution_engine: str = "default"
+    gateway_name: str | None = None
+    gateway_routed: bool = False
     driver_version: str | None
     run_date: str
     power_score: float | None
@@ -147,6 +169,22 @@ class ManifestEntry(BaseModel):
     def _serialize_normalized_cost(self, cost: NormalizedCost) -> dict[str, Any]:
         return cost.to_dict()
 
+    @model_validator(mode="after")
+    def _default_variant_identity(self) -> ManifestEntry:
+        if not self.variant_id:
+            self.variant_id = self.platform_id
+        if not self.variant_label:
+            self.variant_label = variant_label(
+                self.platform,
+                variant_id=self.variant_id,
+                requested_execution_engine=self.requested_execution_engine,
+                gateway_name=self.gateway_name,
+                gateway_routed=self.gateway_routed,
+            )
+        if not self.requested_execution_engine:
+            self.requested_execution_engine = "default"
+        return self
+
     @model_validator(mode="after")
     def _default_logical_query_count(self) -> ManifestEntry:
         if self.logical_query_count == 0 and self.query_count > 0:
@@ -315,6 +353,11 @@ class DetailResult(BaseModel):
     scale_factor: float
     platform: str
     platform_id: str = ""
+    variant_id: str = ""
+    variant_label: str = ""
+    requested_execution_engine: str = "default"
+    gateway_name: str | None = None
+    gateway_routed: bool = False
     driver_version: str | None
     run_date: str
     total_duration_s: float
@@ -365,6 +408,22 @@ class DetailResult(BaseModel):
     physical_rendering_id: str | None = None
     basis_availability: BasisAvailability | None = None
 
+    @model_validator(mode="after")
+    def _default_variant_identity(self) -> DetailResult:
+        if not self.variant_id:
+            self.variant_id = self.platform_id
+        if not self.variant_label:
+            self.variant_label = variant_label(
+                self.platform,
+                variant_id=self.variant_id,
+                requested_execution_engine=self.requested_execution_engine,
+                gateway_name=self.gateway_name,
+                gateway_routed=self.gateway_routed,
+            )
+        if not self.requested_execution_engine:
+            self.requested_execution_engine = "default"
+        return self
+
     @field_serializer("normalized_cost")
     def _serialize_normalized_cost(self, cost: NormalizedCost) -> dict[str, Any]:
         return cost.to_dict()
@@ -408,6 +467,8 @@ def ranking_exclusion_reason(entry: ManifestEntry, primary_metric: str | None =
         return "tuning_not_applied"
     if entry.comparison_exclusion_reason is not None:
         return entry.comparison_exclusion_reason
+    if entry.gateway_routed:
+        return "gateway_routed"
 
     metric = primary_metric or get_ranking_config(entry.benchmark, canonical_phase(entry.test_type)).primary_metric
     value = primary_metric_value(entry, metric)
@@ -529,6 +590,8 @@ class PlatformRow(BaseModel):
     result_id: str
     short_id: str = ""
     platform_id: str
+    variant_id: str = ""
+    variant_label: str = ""
     platform: str
     platform_version: str | None
     tuning_mode: str | None
@@ -907,4 +970,5 @@ def _coerce_cost_block(cls, value: Any) -> Any:
     "timing_eligibility",
     "timing_exclusion_reason",
     "unavailable_normalized_cost_payload",
+    "variant_label",
 ]

_project/scripts/explorer_pipeline/pipeline.py

@@ -313,6 +313,8 @@ def _build_benchmark_summaries(
                 result_id=entry.result_id,
                 short_id=full_to_short.get(entry.result_id, ""),
                 platform_id=entry.platform_id,
+                variant_id=entry.variant_id,
+                variant_label=entry.variant_label,
                 platform=entry.platform,
                 platform_version=entry.platform_version,
                 tuning_mode=entry.tuning_mode,
@@ -412,6 +414,8 @@ def _rank_platforms_in_cohort(
         entries.append(
             {
                 "platform_id": row.platform_id,
+                "variant_id": row.variant_id,
+                "variant_label": row.variant_label,
                 "platform": row.platform,
                 "result_id": row.result_id,
                 "short_id": full_to_short.get(row.result_id, ""),
@@ -434,13 +438,15 @@ def _compute_average_ranks(platform_agg: dict[str, dict[str, Any]]) -> list[dict
         records.append(
             {
                 "platform_id": pdata["platform_id"],
+                "variant_id": pdata.get("variant_id") or pdata["platform_id"],
+                "variant_label": pdata.get("variant_label") or pdata["platform"],
                 "platform": pdata["platform"],
                 "ranks": ranks,
                 "avg_rank": avg_rank,
                 "n_cohorts": len(ranks),
             }
         )
-    records.sort(key=lambda p: (p["avg_rank"] is None, p["avg_rank"] or float("inf"), p["platform_id"]))
+    records.sort(key=lambda p: (p["avg_rank"] is None, p["avg_rank"] or float("inf"), p["variant_id"]))
     return records
 
 
@@ -473,9 +479,15 @@ def _build_meta_leaderboard(
             rank = entry["rank"]
             if rank is None:
                 continue
-            pid = entry["platform_id"]
+            pid = entry["variant_id"] or entry["platform_id"]
             if pid not in platform_agg:
-                platform_agg[pid] = {"platform_id": pid, "platform": entry["platform"], "ranks": {}}
+                platform_agg[pid] = {
+                    "platform_id": entry["platform_id"],
+                    "variant_id": entry["variant_id"],
+                    "variant_label": entry["variant_label"],
+                    "platform": entry["platform"],
+                    "ranks": {},
+                }
             existing = platform_agg[pid]["ranks"].get(cohort_key)
             if existing is None or rank < existing["rank"]:
                 platform_agg[pid]["ranks"][cohort_key] = {

_project/scripts/explorer_pipeline/transformer.py

@@ -8,6 +8,7 @@
 import math
 import re
 from collections import Counter
+from collections.abc import Mapping
 from decimal import Decimal, InvalidOperation
 from pathlib import Path
 from typing import Any, cast
@@ -32,6 +33,7 @@
     canonical_phase,
     ranking_exclusion_reason,
     timing_eligibility,
+    variant_label,
 )
 from benchbox.core.benchmark_registry import get_benchmark_support_status
 from benchbox.core.cost.models import CostScope, CostStatus, DeploymentMetadata, NormalizedCost
@@ -394,6 +396,36 @@ def _platform_version(bundle: BundleDocument) -> str | None:
     return str(val) if val and val != "unknown" else None
 
 
+def _variant_info(bundle: BundleDocument) -> tuple[str, str, str, str | None, bool]:
+    platform = bundle.platform
+    platform_name = str(platform.name)
+    pid = _platform_id(platform_name)
+    raw_variant = getattr(platform, "variant", None)
+    engine_raw = getattr(platform, "execution_engine", None)
+    engine = engine_raw if isinstance(engine_raw, Mapping) else {}
+    requested = engine.get("requested") or "default"
+    requested = str(requested)
+    gateway_raw = getattr(platform, "gateway", None)
+    gateway = gateway_raw if isinstance(gateway_raw, Mapping) else {}
+    routed = bool(gateway.get("routed", False))
+    gateway_name = gateway.get("name")
+    gateway_name = str(gateway_name) if gateway_name else None
+    if isinstance(raw_variant, str) and raw_variant.strip():
+        variant_id = raw_variant.strip()
+    elif requested != "default":
+        variant_id = f"{pid}+{requested}"
+    else:
+        variant_id = pid
+    label = variant_label(
+        platform_name,
+        variant_id=variant_id,
+        requested_execution_engine=requested,
+        gateway_name=gateway_name,
+        gateway_routed=routed,
+    )
+    return variant_id, label, requested, gateway_name, routed
+
+
 _EXECUTION_MODES = frozenset({"sql", "dataframe"})
 
 
@@ -1181,12 +1213,18 @@ def to_manifest_entry(
         )
         override = _override_display(bundle_path)
         test_type = _test_type(bundle)
+        variant = _variant_info(bundle)
         entry = ManifestEntry(
             result_id=rid,
             benchmark=benchmark,
             scale_factor=scale_factor,
             platform=platform,
             platform_id=_platform_id(platform),
+            variant_id=variant[0],
+            variant_label=variant[1],
+            requested_execution_engine=variant[2],
+            gateway_name=variant[3],
+            gateway_routed=variant[4],
             driver_version=_driver_version(bundle),
             run_date=run_date,
             power_score=_power_score(bundle),
@@ -1273,12 +1311,18 @@ def to_detail_result(
         timing_contract = timing_eligibility(display_timings, logical_query_count)
         normalized_cost = _normalized_cost(bundle)
         test_type = _test_type(bundle)
+        variant = _variant_info(bundle)
         detail = DetailResult(
             result_id=result_id,
             benchmark=benchmark,
             scale_factor=scale_factor,
             platform=platform,
             platform_id=_platform_id(platform),
+            variant_id=variant[0],
+            variant_label=variant[1],
+            requested_execution_engine=variant[2],
+            gateway_name=variant[3],
+            gateway_routed=variant[4],
             driver_version=_driver_version(bundle),
             run_date=run_date,
             total_duration_s=total_duration_s,