Generalize cuVS Bench plugin CLI

python/cuvs_bench/cuvs_bench/backends/base.py

@@ -11,7 +11,7 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import Dict, List, Any, Optional, TYPE_CHECKING
+from typing import Dict, List, Any, Optional, Sequence, TYPE_CHECKING
 import numpy as np
 
 if TYPE_CHECKING:
@@ -368,6 +368,10 @@ class BenchmarkBackend(ABC):
         - requires_network : bool - Whether backend requires network (default: False)
     """
 
+    # Registration exposes this optional class-level default to the generic
+    # CLI.
+    default_algorithm: Optional[str] = None
+
     def __init__(self, config: Dict[str, Any]):
         """Initialize backend with configuration."""
         self.config = config
@@ -441,7 +445,7 @@ def search(
         batch_size : int, optional
             Number of queries to process at once (default: 10000)
         mode : str, optional
-            "latency" (measure individual query latency with percentiles) or
+            "latency" (measure batch latency with percentiles) or
             "throughput" (measure overall QPS) (default: "latency")
         force : bool, optional
             Whether to force the execution regardless of existing results (default: False)
@@ -495,6 +499,13 @@ def cleanup(self) -> None:
         """
         pass
 
+    @classmethod
+    def result_failure_message(
+        cls, results: Sequence[BuildResult | SearchResult]
+    ) -> Optional[str]:
+        """Return a fatal result summary when the backend requires one."""
+        return None
+
     def _check_gpu_available(self) -> bool:
         """
         Check if GPU is available (same logic as run.py rmm_present()).

python/cuvs_bench/cuvs_bench/backends/cpp_gbench.py

@@ -51,6 +51,8 @@ class CppGoogleBenchmarkBackend(BenchmarkBackend):
     >>> print(backend.algo)  # "cuvs_ivf_flat" (from config)
     """
 
+    default_algorithm = "cuvs_cagra"
+
     def __init__(self, config: Dict[str, Any]):
         """Initialize C++ benchmark backend."""
         super().__init__(config)

python/cuvs_bench/cuvs_bench/orchestrator/orchestrator.py

@@ -87,6 +87,12 @@ def __init__(self, backend_type: str = "cpp_gbench"):
         # Instantiate config loader
         self.config_loader = loader_class()
 
+    def result_failure_message(
+        self, results: List[Union[BuildResult, SearchResult]]
+    ) -> Optional[str]:
+        """Return the selected backend's fatal result summary, if any."""
+        return self.backend_class.result_failure_message(results)
+
     def run_benchmark(
         self,
         mode: str = "sweep",

python/cuvs_bench/cuvs_bench/run/__main__.py

@@ -6,17 +6,64 @@
 import json
 import os
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional
 
 import click
 import yaml
 
+from ..backends.registry import get_backend_class
+from ..orchestrator import BenchmarkOrchestrator
 from .data_export import (
     convert_json_to_csv_build,
     convert_json_to_csv_search,
     write_results_to_csv,
 )
-from ..orchestrator import BenchmarkOrchestrator
+
+
+_DEFAULT_BACKEND = "cpp_gbench"
+_BACKEND_CONFIG_CACHE_KEY = "cuvs_bench.backend_config"
+
+
+def _read_backend_config(path: str) -> dict[str, Any]:
+    with open(path, "r") as stream:
+        config = yaml.safe_load(stream)
+    if not isinstance(config, dict):
+        raise ValueError(
+            "--backend-config must parse to a mapping, "
+            f"got {type(config).__name__}"
+        )
+    if "backend" not in config:
+        raise ValueError("--backend-config must include a 'backend' field")
+    return dict(config)
+
+
+def _backend_config_snapshot(path: str) -> dict[str, Any]:
+    """Return one backend configuration snapshot per CLI invocation."""
+    context = click.get_current_context()
+    cached_config = context.meta.get(_BACKEND_CONFIG_CACHE_KEY)
+    if cached_config is None:
+        cached_config = _read_backend_config(path)
+        context.meta[_BACKEND_CONFIG_CACHE_KEY] = cached_config
+    return dict(cached_config)
+
+
+def _default_algorithm() -> str:
+    """Choose the displayed prompt default from the selected backend."""
+    context = click.get_current_context(silent=True)
+    backend = _DEFAULT_BACKEND
+    if context is not None and (
+        backend_config := context.params.get("backend_config")
+    ):
+        backend = _backend_config_snapshot(backend_config)["backend"]
+    backend_class = get_backend_class(str(backend))
+    if backend_class.default_algorithm is not None:
+        return backend_class.default_algorithm
+    default_backend_class = get_backend_class(_DEFAULT_BACKEND)
+    if default_backend_class.default_algorithm is None:
+        raise RuntimeError(
+            "The default backend must define a default algorithm"
+        )
+    return default_backend_class.default_algorithm
 
 
 @click.command()
@@ -88,12 +135,13 @@
 @click.option("--search", is_flag=True, help="Perform the search")
 @click.option(
     "--algorithms",
-    default="cuvs_cagra",
+    default=_default_algorithm,
     show_default=True,
     prompt="Enter the comma separated list of named algorithms to run",
     help="Run only comma separated list of named algorithms. If parameters "
     "`groups` and `algo-groups` are both undefined, then group `base` "
-    "is run by default.",
+    "is run by default. A backend may provide its own prompt default; "
+    "otherwise the CLI default is retained.",
 )
 @click.option(
     "--groups",
@@ -251,10 +299,10 @@ def main(
     n_trials : Optional[int]
         Number of Optuna trials for tune mode.
     backend_config : Optional[str]
-        Path to YAML config for non-C++ backends. If not provided,
-        defaults to the C++ Google Benchmark backend. The YAML file
-        must contain a 'backend' field (e.g., 'opensearch', 'elastic')
-        and any backend-specific connection parameters (host, port, etc.).
+        Path to YAML config for non-C++ backends. If not provided, defaults
+        to the C++ Google Benchmark backend. The YAML file must contain a
+        'backend' field (e.g., 'opensearch', 'elastic') and any
+        backend-specific connection parameters (host, port, etc.).
 
     """
     if data_export:
@@ -271,18 +319,10 @@ def main(
     if not build and not search:
         build = search = True
 
-    backend_type = "cpp_gbench"
+    backend_type = _DEFAULT_BACKEND
     backend_kwargs = {}
     if backend_config:
-        with open(backend_config, "r") as f:
-            cfg = yaml.safe_load(f)
-        if not isinstance(cfg, dict):
-            raise ValueError(
-                f"--backend-config must parse to a mapping, "
-                f"got {type(cfg).__name__}"
-            )
-        if "backend" not in cfg:
-            raise ValueError("--backend-config must include a 'backend' field")
+        cfg = _backend_config_snapshot(backend_config)
         backend_type = cfg.pop("backend")
         backend_kwargs = cfg
 
@@ -321,6 +361,8 @@ def main(
             convert_json_to_csv_search(dataset, dataset_path)
     else:
         write_results_to_csv(results, dataset, dataset_path, count, batch_size)
+        if failure_message := orchestrator.result_failure_message(results):
+            raise click.ClickException(failure_message)
 
 
 if __name__ == "__main__":

python/cuvs_bench/cuvs_bench/run/data_export.py

@@ -52,11 +52,15 @@
     },
 }
 
+_SEARCH_PARAMS_IDENTITY_COLUMN = "_search_params_identity"
+
 
 def write_results_to_csv(results, dataset, dataset_path, count, batch_size):
     """Write Python-backend results using the existing plotting CSV schema."""
     grouped = defaultdict(list)
     for result in results:
+        if result.metadata.get("dry_run"):
+            continue
         group = result.metadata.get("group")
         index_name = result.metadata.get("index_name")
         if group is None or (
@@ -84,32 +88,43 @@ def write_results_to_csv(results, dataset, dataset_path, count, batch_size):
 
 
 def _write_build_results(results, algorithm, group, dataset, dataset_path):
-    output_dir = os.path.join(dataset_path, dataset, "result", "build")
-    os.makedirs(output_dir, exist_ok=True)
     algo_name = algorithm if group == "base" else f"{algorithm}_{group}"
 
     rows = []
     for result in results:
-        if not result.success or result.metadata.get("skipped"):
+        if (
+            not result.success
+            or result.metadata.get("skipped")
+            or result.metadata.get("dry_run")
+        ):
             continue
         metadata = _scalar_metadata(result.metadata)
         rows.append(
             {
                 **result.build_params,
                 **metadata,
                 "algo_name": algo_name,
-                "index_name": result.index_path,
+                "index_name": result.metadata.get(
+                    "index_name", result.index_path
+                ),
                 "time": result.build_time_seconds,
             }
         )
 
+    if not rows:
+        # A failed, skipped, or dry run must not replace prior measurements.
+        return
+
+    output_dir = os.path.join(dataset_path, dataset, "result", "build")
+    os.makedirs(output_dir, exist_ok=True)
     columns = ["algo_name", "index_name", "time"]
     dataframe = pd.DataFrame(rows)
     build_file = os.path.join(output_dir, f"{algorithm},{group}.csv")
 
     complete_run = all(
         result.success and not result.metadata.get("skipped")
         for result in results
+        if not result.metadata.get("dry_run")
     )
     if not complete_run and os.path.exists(build_file):
         dataframe = pd.concat(
@@ -118,11 +133,6 @@ def _write_build_results(results, algorithm, group, dataset, dataset_path):
             sort=False,
         )
 
-    if dataframe.empty:
-        # Do not replace an existing measurement with a skipped or failed
-        # build, and do not create an empty result file.
-        return
-
     dataframe = dataframe.drop_duplicates(subset=["index_name"], keep="last")
     dataframe = dataframe[
         columns + [name for name in dataframe if name not in columns]
@@ -133,13 +143,15 @@ def _write_build_results(results, algorithm, group, dataset, dataset_path):
 def _write_search_results(
     results, algorithm, group, dataset, dataset_path, count, batch_size
 ):
-    output_dir = os.path.join(dataset_path, dataset, "result", "search")
-    os.makedirs(output_dir, exist_ok=True)
     algo_name = algorithm if group == "base" else f"{algorithm}_{group}"
 
     rows = []
     for result in results:
-        if not result.success:
+        if (
+            not result.success
+            or result.metadata.get("skipped")
+            or result.metadata.get("dry_run")
+        ):
             continue
         metadata = _scalar_metadata(result.metadata)
         search_params = (
@@ -156,9 +168,18 @@ def _write_search_results(
                 "latency": result.metadata.get(
                     "latency_seconds", result.search_time_ms / 1000.0
                 ),
+                _SEARCH_PARAMS_IDENTITY_COLUMN: _search_params_identity(
+                    result.search_params
+                ),
             }
         )
 
+    if not rows:
+        # Preserve prior evidence when every attempted search failed or was dry.
+        return
+
+    output_dir = os.path.join(dataset_path, dataset, "result", "search")
+    os.makedirs(output_dir, exist_ok=True)
     columns = [
         "algo_name",
         "index_name",
@@ -167,12 +188,9 @@ def _write_search_results(
         "latency",
     ]
     dataframe = pd.DataFrame(rows)
-    if dataframe.empty:
-        dataframe = pd.DataFrame(columns=columns)
-    else:
-        dataframe = dataframe[
-            columns + [name for name in dataframe if name not in columns]
-        ]
+    dataframe = dataframe[
+        columns + [name for name in dataframe if name not in columns]
+    ]
 
     build_file = os.path.join(
         dataset_path,
@@ -196,10 +214,53 @@ def _write_search_results(
 
     stem = f"{algorithm},{group},k{count},bs{batch_size}"
     raw_file = os.path.join(output_dir, f"{stem},raw.csv")
+    complete_run = all(
+        result.success and not result.metadata.get("skipped")
+        for result in results
+    )
+    if not complete_run and os.path.exists(raw_file):
+        existing = _with_legacy_search_param_identities(
+            pd.read_csv(raw_file)
+        )
+        identity_columns = ["index_name", _SEARCH_PARAMS_IDENTITY_COLUMN]
+        dataframe = dataframe.drop_duplicates(
+            subset=identity_columns, keep="last"
+        )
+        existing_identities = pd.MultiIndex.from_frame(
+            existing[identity_columns]
+        )
+        replacement_identities = pd.MultiIndex.from_frame(
+            dataframe[identity_columns]
+        )
+        existing = existing.loc[
+            ~existing_identities.isin(replacement_identities)
+        ]
+        dataframe = pd.concat(
+            [existing, dataframe], ignore_index=True, sort=False
+        )
     dataframe.to_csv(raw_file, index=False)
     frontier_file = os.path.join(output_dir, f"{stem}.json")
-    write_frontier(frontier_file, dataframe, "throughput")
-    write_frontier(frontier_file, dataframe, "latency")
+    frontier_data = dataframe.drop(columns=[_SEARCH_PARAMS_IDENTITY_COLUMN])
+    write_frontier(frontier_file, frontier_data, "throughput")
+    write_frontier(frontier_file, frontier_data, "latency")
+
+
+def _search_params_identity(search_params):
+    """Return a stable key for the parameter plans represented by one row."""
+    return json.dumps(search_params, sort_keys=True, separators=(",", ":"))
+
+
+def _with_legacy_search_param_identities(dataframe):
+    """Give legacy rows opaque keys so a partial run cannot erase them."""
+    if _SEARCH_PARAMS_IDENTITY_COLUMN not in dataframe:
+        dataframe[_SEARCH_PARAMS_IDENTITY_COLUMN] = pd.NA
+
+    missing_identity = dataframe[_SEARCH_PARAMS_IDENTITY_COLUMN].isna()
+    dataframe.loc[missing_identity, _SEARCH_PARAMS_IDENTITY_COLUMN] = [
+        f"legacy:{row_number}"
+        for row_number in dataframe.index[missing_identity]
+    ]
+    return dataframe
 
 
 def _scalar_metadata(metadata):