diff --git a/mlpstorage_py/benchmarks/vdb_summary.py b/mlpstorage_py/benchmarks/vdb_summary.py new file mode 100644 index 00000000..7decb855 --- /dev/null +++ b/mlpstorage_py/benchmarks/vdb_summary.py @@ -0,0 +1,110 @@ +"""Canonical reportgen ``summary.json`` for VectorDB runs. + +The VDB benchmark writes its native metrics into ``statistics.json`` — +single-node at the run root, MPI under ``vectordb//`` — but never a +canonical ``summary.json`` at the run root, which is where reportgen +(``report_generator.py::_load_workload_summary``) reads. Without a bridge +every ``vdb_*`` metric column comes out blank. + +This module is that bridge, as two pure functions that take a directory +path (NOT a benchmark instance, so reportgen can call them without +constructing a full ``VectorDBBenchmark``): + + * :func:`build_vdb_summary` — locate the native QUERY-phase stats file, + project it, coerce a dict ``recall`` to its scalar ``mean_recall``. + Returns the dict, or ``None`` when no native file exists / is readable. + * :func:`write_vdb_summary` — build + persist ``/summary.json``. + Producer-only; returns the written path or ``None``. + +The producer calls :func:`write_vdb_summary` after a run so fresh +submissions ship ``summary.json``. reportgen calls :func:`build_vdb_summary` +IN MEMORY for legacy packages that predate the producer write — it never +writes into the submission tree (packages may be read-only or reviewed by +someone who does not own them). +""" + +import json +import os +from typing import Any, Dict, Optional + +# Native query-phase stats files, in priority order. The load phase +# (``load_statistics.json`` / ``vectordb/load/``) is intentionally excluded +# — the final table wants QUERY metrics, never datagen/load metrics. +# +# * single-node simple AND enhanced both write ``statistics.json`` at the +# run root (see vdbbench/simple_bench.py + enhanced_bench.py). +# * MPI writes into ``vectordb//`` (see +# benchmarks/vectordbbench.py::_write_summary_from_mpi_output / +# _base_output_dir). +_QUERY_STATS_CANDIDATES = ( + "statistics.json", + os.path.join("vectordb", "simple", "statistics.json"), + os.path.join("vectordb", "enhanced", "enhanced_statistics.json"), + os.path.join("vectordb", "simple", "vdb_multi_node_summary.json"), + os.path.join("vectordb", "enhanced", "vdb_multi_node_summary.json"), +) + +_SUMMARY_FILENAME = "summary.json" + + +def _scalar_recall(recall: Any) -> Any: + """Reduce a dict recall block to its scalar ``mean_recall``. + + The producer emits ``recall`` as a dict (``{mean_recall, median_recall, + ...}``); reportgen and the final table want a single scalar. A recall + that is already a scalar (or absent) is returned unchanged. + """ + if isinstance(recall, dict): + return recall.get("mean_recall") + return recall + + +def _find_native_stats_file(result_dir: str) -> Optional[str]: + """Return the path to the query-phase native stats file, or ``None``.""" + for rel in _QUERY_STATS_CANDIDATES: + candidate = os.path.join(result_dir, rel) + if os.path.isfile(candidate): + return candidate + return None + + +def build_vdb_summary(result_dir: str) -> Optional[Dict[str, Any]]: + """Build the canonical reportgen summary dict from native VDB output. + + Copies the native stats dict whole (harmless extra keys are + future-proof if reportgen reads more later) and overrides ``recall`` + with its scalar form. Returns ``None`` when no native query-phase stats + file exists under ``result_dir`` or it cannot be parsed. + """ + if not result_dir: + return None + stats_path = _find_native_stats_file(result_dir) + if stats_path is None: + return None + try: + with open(stats_path, "r", encoding="utf-8") as fd: + stats = json.load(fd) + except (OSError, ValueError): + return None + if not isinstance(stats, dict): + return None + + summary = dict(stats) + if "recall" in summary: + summary["recall"] = _scalar_recall(summary["recall"]) + return summary + + +def write_vdb_summary(result_dir: str) -> Optional[str]: + """Persist ``/summary.json`` from native VDB output. + + Producer-only. Returns the written path, or ``None`` when there was no + native query-phase stats file to project. + """ + summary = build_vdb_summary(result_dir) + if summary is None: + return None + out_path = os.path.join(result_dir, _SUMMARY_FILENAME) + with open(out_path, "w", encoding="utf-8") as fd: + json.dump(summary, fd, indent=2, sort_keys=True) + return out_path diff --git a/mlpstorage_py/benchmarks/vectordbbench.py b/mlpstorage_py/benchmarks/vectordbbench.py index 7dd527f6..bf5d14a9 100644 --- a/mlpstorage_py/benchmarks/vectordbbench.py +++ b/mlpstorage_py/benchmarks/vectordbbench.py @@ -136,9 +136,35 @@ def __init__(self, args, **kwargs): if not getattr(args, "what_if", False) and self.command != "datasize": self._validate_vdb_dependencies() + # Surface the VDB reportgen identity keys into metadata['parameters'] + # (base.get_metadata sources that block from self.combined_params). + # reportgen reads run.parameters for the final-table identity columns + # Vector Count / Vector Dimension / Index Type / Engine. + self.combined_params = self._reportgen_params() + self.verify_benchmark() self.logger.status("Instantiated the VectorDB Benchmark...") + def _reportgen_params(self) -> Dict[str, Any]: + """Identity keys reportgen reads from ``metadata['parameters']``. + + Keys match exactly what ``report_generator._aggregate_vdb`` looks up + (``engine`` / ``index_type`` / ``num_vectors`` / ``dimension``). A + value not present on ``self.args`` (e.g. ``num_vectors`` on a bare + ``run`` invocation) stays ``None`` — reportgen emits the column + present-but-blank rather than dropping it. + """ + index_type = ( + getattr(self.args, "index_type", None) + or getattr(self.args, "vdb_index", None) + ) + return { + "engine": getattr(self.args, "vdb_engine", None), + "index_type": index_type, + "num_vectors": getattr(self.args, "num_vectors", None), + "dimension": getattr(self.args, "dimension", None), + } + def _resolve_storage_args(self): """Record the VDB storage location on ``self.args`` (storage#802). @@ -920,9 +946,31 @@ def execute_run(self) -> int: enhanced_bench via enhanced-bench """ if self._is_distributed(): - return self._execute_run_distributed() + rc = self._execute_run_distributed() + else: + rc = self._execute_run_single_node() - return self._execute_run_single_node() + # Emit the canonical reportgen summary.json at the run root so fresh + # submissions ship it (reportgen backfills legacy packages in-memory). + # Best-effort: a missing native stats file or write error must not + # fail an otherwise-successful run. + if not getattr(self.args, "what_if", False): + try: + from mlpstorage_py.benchmarks.vdb_summary import write_vdb_summary + + written = write_vdb_summary(self.run_result_output) + if written: + self.logger.status(f"Wrote reportgen summary: {written}") + else: + self.logger.verbose( + "No native VDB query stats found; summary.json not written." + ) + except Exception as exc: # noqa: BLE001 — best-effort, never fatal + self.logger.warning( + f"Could not write reportgen summary.json: {exc}" + ) + + return rc def _execute_run_single_node(self) -> int: """Execute existing single-node VectorDB run path.""" diff --git a/mlpstorage_py/report_generator.py b/mlpstorage_py/report_generator.py index 077a5b47..78e5b685 100755 --- a/mlpstorage_py/report_generator.py +++ b/mlpstorage_py/report_generator.py @@ -20,14 +20,14 @@ from typing import List, Dict, Any, Optional, Set from mlpstorage_py.mlps_logging import setup_logging, apply_logging_options -from mlpstorage_py.config import MLPS_DEBUG, BENCHMARK_TYPES, EXIT_CODE, PARAM_VALIDATION, LLM_MODELS, MODELS, ACCELERATORS +from mlpstorage_py.config import MLPS_DEBUG, BENCHMARK_TYPES, EXIT_CODE, PARAM_VALIDATION, LLM_MODELS, LLM_ALLOWED_VALUES, MODELS, ACCELERATORS from mlpstorage_py.rules import get_runs_files, BenchmarkVerifier, BenchmarkRun, Issue, RunID from mlpstorage_py.rules.datagen_hierarchy import ( validate_datagen_leaf, validate_supported_model, ) from mlpstorage_py.errors import ConfigurationError -from mlpstorage_py.utils import flatten_nested_dict, remove_nan_values +from mlpstorage_py.utils import flatten_nested_dict, remove_nan_values, MLPSJsonEncoder from mlpstorage_py.reporting import ( ResultsDirectoryValidator, ValidationMessageFormatter, @@ -52,6 +52,36 @@ ) +class Hyperlink: + """A results-table cell that is a (visible-text, href) pair. + + Renders differently per output format (Task F, v3.0 results table): + + - CSV: ``str(link)`` yields an HTML anchor ``text`` + (``csv.DictWriter`` str-coerces non-string cell values, and + ``flatten_nested_dict`` treats this non-dict object as a leaf). + - JSON: ``MLPSJsonEncoder`` serializes it via its ``__dict__`` branch + to ``{"text": ..., "href": ...}``. + + ``href`` is a repo-root-relative URL (the aggregated-submissions repo + root == ````); a future change prepends the full repo URL. + """ + + def __init__(self, text: str, href: str) -> None: + self.text = text + self.href = href + + def __str__(self) -> str: + return f'{self.text}' + + def __repr__(self) -> str: + return f"Hyperlink(text={self.text!r}, href={self.href!r})" + + def __eq__(self, other: Any) -> bool: + return (isinstance(other, Hyperlink) + and self.text == other.text and self.href == other.href) + + @dataclass class Result: """Container for a single benchmark run result.""" @@ -506,6 +536,11 @@ def _workload_result_to_row( 'model': model_val, 'accelerator': accelerator_val, } + # Task F: shared System-Under-Test block (from system-description.yaml), + # inserted after the D-10 prefix and before metric groups so the + # prefix stays first / issues stays last. + row.update(self._sut_columns( + category_val, orgname_key, systemname_key, first_run)) # Aggregated metric columns (D-11 grouped body). for metric_key, metric_val in (workload_result.metrics or {}).items(): row[metric_key] = metric_val @@ -521,6 +556,122 @@ def _workload_result_to_row( return row + # Task F: shared System-Under-Test column order (v3.0 results table). + # Group prefix `sut_`; placed after the D-10 prefix by _ordered_fieldnames. + _SUT_COLUMNS = [ + 'sut_public_id', + 'sut_organization', + 'sut_name', + 'sut_description', + 'sut_type', + 'sut_access_protocol', + 'sut_availability', + 'sut_rus', + 'sut_integrated_client_storage', + 'sut_usable_capacity_tib', + 'sut_code', + 'sut_logs', + ] + + def _sut_columns( + self, + category: str, + orgname: str, + systemname: str, + first_run: Any, + ) -> Dict[str, Any]: + """Build the shared SUT block for a workload row (Task F). + + AUTO-filled: ``sut_organization`` (path), ``sut_name`` / + ``sut_description`` (Hyperlinks to the system-description + ``.yaml`` / ``.pdf``), ``sut_rus`` (``total_rack_units``). The + remaining six cells are present-but-blank placeholders the + submitter fills manually. Hrefs are repo-root-relative (Option A). + """ + cols: Dict[str, Any] = {key: "" for key in self._SUT_COLUMNS} + cols['sut_organization'] = orgname or "" + + if category and orgname and systemname: + base = f"{category}/{orgname}/systems/{systemname}" + submission_name, rack_units = self._read_system_description( + first_run, systemname) + name_text = submission_name or systemname + cols['sut_name'] = Hyperlink(name_text, f"{base}.yaml") + cols['sut_description'] = Hyperlink(name_text, f"{base}.pdf") + if rack_units is not None: + cols['sut_rus'] = rack_units + + # Task F-b: Code/Logs anchors -> the run's content-addressed + # code-image pool dir (repo-root-relative). Logs is a + # placeholder pointing at the same dir for now (future change + # will retarget it). Blank when the pointer is absent/malformed. + code_href = self._code_image_href(category, orgname, first_run) + if code_href: + cols['sut_code'] = Hyperlink("code", code_href) + cols['sut_logs'] = Hyperlink("logs", code_href) + return cols + + def _code_image_href( + self, category: str, orgname: str, first_run: Any, + ) -> Optional[str]: + """Repo-root-relative URL of the run's ``code-`` pool dir. + + Resolves the run leaf's ``.mlps-code-image`` pointer via the + canonical code_image helpers (D-61 pointer parse, D-62 pool-dir + name). Returns ``None`` when the pointer is absent or unreadable — + the caller then leaves Code/Logs blank. + """ + result_dir = getattr(first_run, 'result_dir', None) + if not result_dir: + return None + from pathlib import Path + from mlpstorage_py.submission_checker.tools.code_image import ( + _read_pointer, _pool_dir_name, CodeImageError, + ) + try: + _alg, full_hash = _read_pointer(Path(result_dir), self.logger) + except (FileNotFoundError, CodeImageError, OSError) as e: + self.logger.warning( + "reportgen: no/invalid code-image pointer at %s: %s; " + "Code/Logs will be blank.", result_dir, e) + return None + return f"{category}/{orgname}/{_pool_dir_name(full_hash)}/" + + def _read_system_description(self, first_run: Any, systemname: str): + """Locate + parse ``systems/.yaml`` for a workload. + + Walks up from the run's ``result_dir`` to the first ancestor that + contains ``systems/.yaml`` (the ```` level of the + submission tree). Returns ``(submission_name, total_rack_units)``, + each ``None`` when the file is absent/unreadable — the caller then + falls back to the systemname text and a blank RU cell. + """ + result_dir = getattr(first_run, 'result_dir', None) + if not result_dir: + return None, None + import yaml # local import: pyyaml is a runtime dep, keep top clean + from pathlib import Path + start = Path(result_dir) + for ancestor in [start, *start.parents]: + candidate = ancestor / "systems" / f"{systemname}.yaml" + if candidate.is_file(): + try: + with open(candidate, "r") as fh: + data = yaml.safe_load(fh) or {} + sut = data.get('system_under_test') or {} + solution = sut.get('solution') or {} + return (solution.get('submission_name'), + sut.get('total_rack_units')) + except (OSError, yaml.YAMLError) as e: + self.logger.warning( + "reportgen: could not read system description %s: %s", + candidate, e) + return None, None + self.logger.warning( + "reportgen: no systems/%s.yaml found above %s; SUT block will " + "fall back to systemname + blank RUs.", systemname, result_dir) + return None, None + def _model_group_folder_for_workload( self, workload_result: 'Result' ) -> Optional[str]: @@ -1002,6 +1153,45 @@ def _aggregate_training( if basename.startswith("train_"): basename = basename[len("train_"):] out[f"train_mean_of_{basename}"] = outer + + # Slice-Training: v3.0 final-table columns the list-valued metric + # aggregation above does not carry. num_hosts / num_accelerators + # live at summary.json top level; Read B/W is DLIO's scalar + # ``train_io_mean_MB_per_second`` (MiB/s despite the ``MB`` label — + # DLIO's statscounter.py computes ``samples/s * record_size / 1024 + # / 1024`` and logs it as "MiB/second"). All three are dropped by + # the metadata-complete parse path (system_info is never + # reconstructed; scalar metrics are filtered out), so read them + # straight from each run's summary.json — consistent with the + # vdb / kvcache branches. + summaries = [self._load_workload_summary(inv) for inv in non_warmup] + + # Identity columns — run configuration, identical across the + # measured invocations; take the first invocation that reports each. + def _first_present(field: str) -> Any: + for s in summaries: + value = s.get(field) + if value is not None: + return value + return None + + out["train_num_client_nodes"] = _first_present("num_hosts") + out["train_num_simulated_accelerators"] = _first_present( + "num_accelerators" + ) + + # Read B/W (GiB/s): mean of DLIO's per-run MiB/s scalar across all + # non-warmup invocations, MiB -> GiB (binary, /1024). Present-but- + # blank when any invocation omits the scalar (blank only when + # truly absent — no silent partial-mean). + io_mibps = [ + (s.get("metric") or {}).get("train_io_mean_MB_per_second") + for s in summaries + ] + if io_mibps and all(isinstance(v, (int, float)) for v in io_mibps): + out["train_read_bw_gibps"] = fmean(io_mibps) / 1024.0 + else: + out["train_read_bw_gibps"] = None return out def _aggregate_checkpointing( @@ -1055,6 +1245,81 @@ def _aggregate_checkpointing( if basename.startswith("checkpoint_"): basename = basename[len("checkpoint_"):] out[f"checkpoint_mean_of_{basename}"] = outer + + # Slice-Checkpointing: v3.0 final-table columns. Write/Read B/W and + # durations come from DLIO's per-run SCALAR keys + # (save_/load_checkpoint_io_mean_GB_per_second, _duration_mean_ + # seconds) — already GiB/s BINARY (checkpoint_size is in GiB and + # throughput = size / seconds; DLIO logs it "GiB/second"), so NO + # conversion. These scalars are dropped by the metadata-complete + # parse path (list-only metric filter) and are absent from the + # fabricated fixtures, so read them straight from each run's + # summary.json — consistent with the training / vdb / kvcache + # branches. num_hosts is summary top-level; Checkpoint Mode is a + # param; DP Instances is a per-model constant. + summaries = [self._load_workload_summary(r) for r in runs] + + def _first_present(field: str) -> Any: + for s in summaries: + value = s.get(field) + if value is not None: + return value + return None + + out["checkpoint_num_client_nodes"] = _first_present("num_hosts") + + # Checkpoint Mode (Full / Subset). The benchmark writes + # ``checkpoint.mode = "subset"`` only for subset runs + # (benchmarks/dlio.py:add_checkpoint_params); a "default" / absent + # mode is a full-model run. Handle both the nested params dict and + # a flattened ``checkpoint.mode`` key defensively. + params = runs[0].parameters or {} + mode_raw = None + ckpt_params = params.get("checkpoint") + if isinstance(ckpt_params, dict): + mode_raw = ckpt_params.get("mode") + if mode_raw is None: + mode_raw = params.get("checkpoint.mode") + out["checkpoint_mode"] = "Subset" if mode_raw == "subset" else "Full" + + # DP Instances — configured data parallelism int(ClosedGPUs / + # GPUpDP) from LLM_ALLOWED_VALUES (the same value + # benchmarks/dlio.py:add_checkpoint_params computes). Per-model + # constant; blank when the model is unknown. + allowed = LLM_ALLOWED_VALUES.get(runs[0].model) + if allowed: + _min_procs, _zero_level, gpu_per_dp, closed_gpus = allowed + out["checkpoint_dp_instances"] = ( + int(closed_gpus / gpu_per_dp) if gpu_per_dp else None + ) + else: + out["checkpoint_dp_instances"] = None + + # B/W (verbatim GiB/s) + durations — mean of DLIO's per-run scalar + # across invocations (Rules.md §2.1.23 permits 1-2). Present-but- + # blank when any invocation omits a field (blank only when truly + # absent — no silent partial-mean). + def _scalar_mean(field: str) -> Optional[float]: + values = [(s.get("metric") or {}).get(field) for s in summaries] + if values and all( + isinstance(v, (int, float)) and not isinstance(v, bool) + for v in values + ): + return fmean(values) + return None + + out["checkpoint_write_bw_gibps"] = _scalar_mean( + "save_checkpoint_io_mean_GB_per_second" + ) + out["checkpoint_write_duration_secs"] = _scalar_mean( + "save_checkpoint_duration_mean_seconds" + ) + out["checkpoint_read_bw_gibps"] = _scalar_mean( + "load_checkpoint_io_mean_GB_per_second" + ) + out["checkpoint_read_duration_secs"] = _scalar_mean( + "load_checkpoint_duration_mean_seconds" + ) return out def _aggregate_vdb( @@ -1085,7 +1350,7 @@ def _aggregate_vdb( for CLI/YAML args on disk). """ run = runs[0] - summary = self._load_workload_summary(run) + summary = self._load_vdb_summary(run) _VDB_METRIC_FIELDS = ( "throughput_qps", @@ -1099,8 +1364,12 @@ def _aggregate_vdb( out[f"vdb_{field_name}"] = summary.get(field_name) # Recall (D-21) — first summary.json, then recall_stats.json fallback - # (per submission_checker/checks/vdb_checks.py:462-469). + # (per submission_checker/checks/vdb_checks.py:462-469). A dict recall + # block is reduced to its scalar ``mean_recall``; the value is emitted + # as a PERCENTAGE (0-100) to match the "Recall Percentage" column. recall = summary.get("recall") + if isinstance(recall, dict): + recall = recall.get("mean_recall") if recall is None and run.result_dir: recall_stats_path = os.path.join(run.result_dir, "recall_stats.json") if os.path.isfile(recall_stats_path): @@ -1108,19 +1377,92 @@ def _aggregate_vdb( with open(recall_stats_path, "r") as f: recall_stats = json.load(f) recall = recall_stats.get("recall") + if isinstance(recall, dict): + recall = recall.get("mean_recall") except (OSError, ValueError) as e: self.logger.warning( f"vdb: could not read recall_stats.json at " f"{recall_stats_path}: {e}" ) - out["vdb_recall"] = recall + out["vdb_recall"] = (recall * 100) if isinstance(recall, (int, float)) else None + + # Read B/W + # Client Nodes derive from the ``disk_io`` sub-block. + # Read B/W is converted bytes/sec -> GiB/s (binary, /1024**3). + disk_io = summary.get("disk_io") + read_bps = disk_io.get("total_bytes_read_per_sec") if isinstance(disk_io, dict) else None + out["vdb_read_bw_gibps"] = ( + read_bps / (1024 ** 3) if isinstance(read_bps, (int, float)) else None + ) + out["vdb_num_client_nodes"] = ( + disk_io.get("host_count") if isinstance(disk_io, dict) else None + ) + # Storage IOPs is not measured (disk_io carries bytes, not op counts) — + # emit the column present-but-blank so the table has it (submitter or a + # future instrumentation task fills it). + out["vdb_storage_iops"] = None + + # Identity columns (D-15). Prefer the structured ``parameters`` block + # (the combined_params fix); fall back to the persisted per-run + # ``metadata['args']`` snapshot for legacy packages whose parameters + # block is empty. Blank only when truly absent from both. + out["vdb_num_vectors"] = self._vdb_identity(run, "num_vectors", ("num_vectors",)) + out["vdb_dimension"] = self._vdb_identity(run, "dimension", ("dimension",)) + out["vdb_engine"] = self._vdb_identity(run, "engine", ("vdb_engine", "engine")) + out["vdb_index_type"] = self._vdb_identity( + run, "index_type", ("index_type", "vdb_index") + ) + return out + + @staticmethod + def _vdb_identity(run: BenchmarkRun, param_key: str, arg_keys: tuple) -> Any: + """Resolve a VDB identity value: parameters first, then persisted args. - # Identity columns (D-15). ``BenchmarkRun`` exposes CLI/YAML args - # via ``.parameters``; use ``.get`` so missing keys emit ``None``. + ``param_key`` is looked up in ``run.parameters`` (the combined_params + block). If absent/None, ``arg_keys`` are tried in order against + ``run.run_args`` (the persisted ``metadata['args']`` snapshot). + Returns ``None`` when the value is in neither. + """ params = run.parameters or {} - out["vdb_engine"] = params.get("engine") - out["vdb_index_type"] = params.get("index_type") - return out + value = params.get(param_key) + if value is not None: + return value + args = getattr(run, "run_args", None) or {} + for ak in arg_keys: + candidate = args.get(ak) + if candidate is not None: + return candidate + return None + + def _load_vdb_summary(self, run: BenchmarkRun) -> Dict[str, Any]: + """Load a VDB run's ``summary.json``, backfilling legacy packages. + + Fresh submissions ship ``summary.json`` (written by the producer). + Legacy packages predate that write — they carry only the native + ``statistics.json``. For those, derive the summary IN MEMORY via + ``vdb_summary.build_vdb_summary``; NEVER write into the submission + package (it may be read-only or owned by another submitter). + """ + summary = self._load_workload_summary(run) + if summary: + return summary + if run.result_dir: + try: + from mlpstorage_py.benchmarks.vdb_summary import build_vdb_summary + + built = build_vdb_summary(run.result_dir) + except Exception as e: # noqa: BLE001 — backfill is best-effort + self.logger.warning( + f"vdb: could not derive summary from native stats at " + f"{run.result_dir!r}: {e}" + ) + built = None + if built: + self.logger.verbose( + f"vdb: derived summary in-memory from native stats at " + f"{run.result_dir!r} (legacy package, no summary.json)." + ) + return built + return {} def _aggregate_kvcache( self, @@ -1163,6 +1505,12 @@ def _aggregate_kvcache( params = run.parameters or {} out["kvcache_performance_profile"] = params.get("performance_profile") + # Slice-KVCache: shared ``# Client Nodes`` column for the v3.0 + # final table. Source is the top-level ``host_count`` field that + # kvcache's ``_write_run_summary`` persists (benchmarks/kvcache.py). + # This is a run-wide count, not per-option, so it is emitted once. + out["kvcache_num_client_nodes"] = summary.get("host_count") + # Top-level per-run aggregates (D-14). Copy verbatim from source; # Zero sentinels from kvcache's "fmean-or-zero" idiom are # preserved (D-22 boundary — do NOT re-interpret). @@ -1815,7 +2163,7 @@ def write_json_file(self, results, target_dir: Optional[str] = None): json_file = os.path.join(out_dir, 'results.json') self.logger.info(f'Writing results to {json_file}') with open(json_file, 'w') as f: - json.dump(results, f, indent=2) + json.dump(results, f, indent=2, cls=MLPSJsonEncoder) def write_csv_file(self, results, target_dir: Optional[str] = None): out_dir = target_dir if target_dir is not None else self.results_dir @@ -1867,12 +2215,16 @@ def _ordered_fieldnames(self, rows: List[dict]) -> List[str]: trailing_set = set(trailing) remaining = all_keys - prefix_set - trailing_set + # Task F: shared SUT block leads the body (right after the D-10 + # prefix, before the metric groups). Ordered by _SUT_COLUMNS, not + # alphabetically, so the block reads in spreadsheet order. + sut_cols = [k for k in self._SUT_COLUMNS if k in remaining] train_cols = sorted(k for k in remaining if k.startswith('train_')) checkpoint_cols = sorted(k for k in remaining if k.startswith('checkpoint_')) vdb_cols = sorted(k for k in remaining if k.startswith('vdb_')) kvcache_cols = sorted(k for k in remaining if k.startswith('kvcache_')) - grouped = train_cols + checkpoint_cols + vdb_cols + kvcache_cols + grouped = sut_cols + train_cols + checkpoint_cols + vdb_cols + kvcache_cols other = sorted(k for k in remaining if k not in set(grouped)) # `other` catches any un-prefixed column not covered by the diff --git a/mlpstorage_py/rules/models.py b/mlpstorage_py/rules/models.py index 2712701d..cba80028 100755 --- a/mlpstorage_py/rules/models.py +++ b/mlpstorage_py/rules/models.py @@ -84,6 +84,10 @@ class BenchmarkRunData: result_dir: Optional[str] = None accelerator: Optional[str] = None end_datetime: str = "" + # Persisted per-run ``metadata['args']`` snapshot (vars(self.args) at run + # time). reportgen uses it as a fallback source for identity columns when + # the structured ``parameters`` block is empty (legacy packages). + run_args: Dict[str, Any] = field(default_factory=dict) @dataclass @@ -758,6 +762,7 @@ def extract(benchmark) -> BenchmarkRunData: override_parameters=override_parameters, system_info=system_info, accelerator=getattr(benchmark.args, 'accelerator_type', None), + run_args=vars(benchmark.args) if hasattr(benchmark, 'args') else {}, result_dir=benchmark.run_result_output if hasattr(benchmark, 'run_result_output') else None, metrics=None, ) @@ -841,6 +846,7 @@ def parse(self, result_dir: str, metadata: Optional[Dict] = None) -> BenchmarkRu metrics=summary.get("metric"), result_dir=result_dir, accelerator=accelerator, + run_args=(metadata or {}).get('args', {}) or {}, ) def _load_summary(self, result_dir: str) -> Optional[Dict]: @@ -962,6 +968,7 @@ def _from_metadata(self, metadata: Dict, result_dir: str) -> BenchmarkRunData: metrics=run_metrics, result_dir=result_dir, accelerator=metadata.get('accelerator'), + run_args=metadata.get('args', {}) or {}, ) def _load_summary(self, result_dir: str) -> Optional[Dict]: @@ -1077,6 +1084,16 @@ def parameters(self): def override_parameters(self): return self._data.override_parameters + @property + def run_args(self): + """Persisted per-run ``metadata['args']`` snapshot (or ``{}``). + + reportgen uses this as a fallback source for VDB identity columns + when the structured ``parameters`` block is empty (legacy packages + that predate the ``combined_params`` fix). + """ + return self._data.run_args + @property def system_info(self): return self._data.system_info diff --git a/tests/unit/test_aggregation.py b/tests/unit/test_aggregation.py index bb82161c..2d4e4290 100644 --- a/tests/unit/test_aggregation.py +++ b/tests/unit/test_aggregation.py @@ -419,13 +419,19 @@ def test_training_5run_mean_excludes_warmup(self, tmp_path): ) def test_training_output_keys_use_train_mean_of_prefix(self, tmp_path): - """Every emitted training key carries the ``train_mean_of_`` prefix (D-13/D-14). + """Metric-mean keys carry ``train_mean_of_``; all keys carry ``train_`` (D-13/D-14). D-13 rule: ``_?``. Training source metric keys are ``train_au_percentage``, ``train_throughput_samples_per_second``, ``train_io_throughput_MB_per_second``; the emitted output keys strip the redundant ``train_`` and insert ``mean_of_``. + + Slice-Training additionally emits v3.0 final-table columns + (``train_num_client_nodes``, ``train_num_simulated_accelerators``, + ``train_read_bw_gibps``) that are NOT metric means — they keep the + ``train_`` group prefix (so they sort into the training column + group) but do not carry ``mean_of_``. """ gen = _make_bare_generator(tmp_path) runs_root = tmp_path / "workload" @@ -443,12 +449,267 @@ def test_training_output_keys_use_train_mean_of_prefix(self, tmp_path): assert expected_keys <= set(result), ( f"Expected {expected_keys} <= keys, got {set(result)}" ) - # Every emitted training column starts with train_mean_of_. + # Slice-Training final-table columns are new non-mean training cols. + final_table_cols = { + "train_num_client_nodes", + "train_num_simulated_accelerators", + "train_read_bw_gibps", + } + # Every training column keeps the ``train_`` group prefix (column + # ordering invariant); metric-mean columns additionally carry + # ``mean_of_``. for k in result: - assert k.startswith("train_mean_of_"), ( + assert k.startswith("train_"), ( f"Training result key {k!r} does not carry the " - f"``train_mean_of_`` prefix (D-13/D-14)." + f"``train_`` group prefix (column-ordering invariant)." + ) + if k not in final_table_cols: + assert k.startswith("train_mean_of_"), ( + f"Training metric key {k!r} does not carry the " + f"``train_mean_of_`` prefix (D-13/D-14)." + ) + + +# --------------------------------------------------------------------------- # +# TestTrainingFinalTableColumns — Slice-Training (v3.0 final results tables) # +# --------------------------------------------------------------------------- # + + +def _write_training_result_dir( + parent: pathlib.Path, + ts: str, + *, + io_mean_mibps: Optional[float], + num_hosts: int = 2, + num_accelerators: int = 8, +) -> str: + """Materialize a realistic training run dir and return its path. + + The written ``summary.json`` mirrors real DLIO output: top-level + ``num_hosts`` / ``num_accelerators`` and a ``metric`` block carrying + list-valued series PLUS the scalar ``train_io_mean_MB_per_second`` + (DLIO's per-run I/O bandwidth, MiB/s despite the ``MB`` label — see + ``dlio_benchmark/utils/statscounter.py``, logged as "MiB/second"). + ``io_mean_mibps=None`` omits the scalar (legacy / partial run). + """ + run_dir = parent / ts + run_dir.mkdir(parents=True, exist_ok=True) + metric: Dict[str, Any] = { + "train_au_percentage": [95.0, 95.5, 95.2], + "train_throughput_samples_per_second": [12500.0, 12480.0, 12510.0], + } + if io_mean_mibps is not None: + metric["train_io_mean_MB_per_second"] = io_mean_mibps + summary = { + "start": ts, + "end": ts, + "num_accelerators": num_accelerators, + "num_hosts": num_hosts, + "metric": metric, + } + (run_dir / "summary.json").write_text(json.dumps(summary)) + return str(run_dir) + + +class TestTrainingFinalTableColumns: + """Slice-Training: v3.0 final-table columns (issue #823). + + ``# Client Nodes`` (num_hosts), ``# Simulated Accelerators`` + (num_accelerators) and ``Read B/W (GiB/s)`` all come from summary.json + top-level / the DLIO scalar ``train_io_mean_MB_per_second`` — fields + the metadata-complete parse path drops (system_info unreconstructed; + scalar metrics filtered out). reportgen reads them straight from + summary.json, like the vdb / kvcache branches. + """ + + def _five_runs(self, tmp_path, io_values): + """Build 5 training BenchmarkRuns; run.metrics is list-filtered + (as production's ``_from_metadata`` delivers it) while each run's + on-disk summary.json carries the full realistic payload.""" + runs_root = tmp_path / "workload" + runs: List[BenchmarkRun] = [] + for i, io in enumerate(io_values): + ts = f"202607010{i}1500" + result_dir = _write_training_result_dir( + runs_root, ts, io_mean_mibps=io + ) + # Production delivers only list-valued metric keys to + # _aggregate_training; the scalar io_mean lives in summary.json. + runs.append( + _make_run( + benchmark_type=BENCHMARK_TYPES.training, + model="unet3d", + result_dir=result_dir, + metrics={ + "train_au_percentage": [95.0, 95.5, 95.2], + "train_throughput_samples_per_second": [ + 12500.0, 12480.0, 12510.0 + ], + }, + accelerator="h100", + run_datetime=ts, + ) ) + return runs + + def test_read_bw_client_nodes_and_accelerators(self, tmp_path): + """Read B/W (MiB/s scalar mean ÷1024), # Client Nodes, # Sim Accelerators.""" + gen = _make_bare_generator(tmp_path) + io_values = [4096.0, 4096.0, 2048.0, 2048.0, 3072.0] # MiB/s per run + runs = self._five_runs(tmp_path, io_values) + + result = gen._aggregate_workload_metrics(runs, warmup_set=set()) + + # MiB/s mean -> GiB/s (binary, /1024). + expected_gibps = (sum(io_values) / len(io_values)) / 1024.0 + assert result["train_read_bw_gibps"] == pytest.approx(expected_gibps) + assert result["train_num_client_nodes"] == 2 + assert result["train_num_simulated_accelerators"] == 8 + + def test_read_bw_blank_when_scalar_absent(self, tmp_path): + """A run missing ``train_io_mean_MB_per_second`` -> present-but-blank B/W.""" + gen = _make_bare_generator(tmp_path) + io_values = [4096.0, None, 2048.0, 2048.0, 3072.0] + runs = self._five_runs(tmp_path, io_values) + + result = gen._aggregate_workload_metrics(runs, warmup_set=set()) + + assert result["train_read_bw_gibps"] is None + # Identity columns still resolve from the runs that do report them. + assert result["train_num_client_nodes"] == 2 + assert result["train_num_simulated_accelerators"] == 8 + + +# --------------------------------------------------------------------------- # +# TestCheckpointingFinalTableColumns — Slice-Checkpointing (v3.0 tables) # +# --------------------------------------------------------------------------- # + + +def _write_checkpointing_result_dir( + parent: pathlib.Path, + ts: str, + *, + write_bw: Optional[float], + write_dur: Optional[float], + read_bw: Optional[float], + read_dur: Optional[float], + num_hosts: int = 4, + num_accelerators: int = 64, +) -> str: + """Materialize a realistic checkpointing run dir; return its path. + + The written ``summary.json`` mirrors REAL DLIO checkpointing output: + top-level ``num_hosts`` / ``num_accelerators`` and a ``metric`` block + of SCALARS — ``save_/load_checkpoint_io_mean_GB_per_second`` (already + GiB/s binary: checkpoint_size in GiB / seconds, DLIO logs "GiB/second") + and ``save_/load_checkpoint_duration_mean_seconds``. NOT the fabricated + list keys the shared fixtures use. A ``None`` omits that scalar. + """ + run_dir = parent / ts + run_dir.mkdir(parents=True, exist_ok=True) + metric: Dict[str, Any] = {} + if write_bw is not None: + metric["save_checkpoint_io_mean_GB_per_second"] = write_bw + if write_dur is not None: + metric["save_checkpoint_duration_mean_seconds"] = write_dur + if read_bw is not None: + metric["load_checkpoint_io_mean_GB_per_second"] = read_bw + if read_dur is not None: + metric["load_checkpoint_duration_mean_seconds"] = read_dur + summary = { + "start": ts, + "end": ts, + "num_accelerators": num_accelerators, + "num_hosts": num_hosts, + "metric": metric, + } + (run_dir / "summary.json").write_text(json.dumps(summary)) + return str(run_dir) + + +class TestCheckpointingFinalTableColumns: + """Slice-Checkpointing: v3.0 final-table columns (issue #823). + + Write/Read B/W (already GiB/s, no conversion) and Write/Read Duration + come from DLIO's per-run SCALAR keys read straight from summary.json + (dropped by the list-only metric filter; absent from the fabricated + fixtures). # Client Nodes = num_hosts; Checkpoint Mode = param; + DP Instances = int(ClosedGPUs / GPUpDP) per model. + """ + + def _run(self, tmp_path, *, model, parameters, ts="20260703_100000", + write_bw=45.0, write_dur=12.0, read_bw=52.0, read_dur=9.0): + runs_root = tmp_path / "workload" + result_dir = _write_checkpointing_result_dir( + runs_root, ts, + write_bw=write_bw, write_dur=write_dur, + read_bw=read_bw, read_dur=read_dur, + num_hosts=4, num_accelerators=64, + ) + # Production delivers empty run.metrics for checkpointing (all DLIO + # keys are scalars, filtered out by _from_metadata's list filter). + return _make_run( + benchmark_type=BENCHMARK_TYPES.checkpointing, + model=model, + result_dir=result_dir, + metrics={}, + parameters=parameters, + accelerator=None, + run_datetime=ts, + ) + + def test_bw_durations_client_nodes_and_dp(self, tmp_path): + """B/W (verbatim GiB/s), durations, # Client Nodes, DP Instances.""" + gen = _make_bare_generator(tmp_path) + # llama3-70b -> ClosedGPUs=64, GPUpDP=8 -> DP = 8. + run = self._run( + tmp_path, model="llama3-70b", + parameters={"checkpoint": {"mode": "default"}}, + ) + + result = gen._aggregate_workload_metrics([run], warmup_set=set()) + + # B/W already GiB/s binary — verbatim, no conversion. + assert result["checkpoint_write_bw_gibps"] == pytest.approx(45.0) + assert result["checkpoint_read_bw_gibps"] == pytest.approx(52.0) + assert result["checkpoint_write_duration_secs"] == pytest.approx(12.0) + assert result["checkpoint_read_duration_secs"] == pytest.approx(9.0) + assert result["checkpoint_num_client_nodes"] == 4 + assert result["checkpoint_dp_instances"] == 8 + # Non-subset / default mode -> Full. + assert result["checkpoint_mode"] == "Full" + + def test_subset_mode_and_dp_for_1t(self, tmp_path): + """checkpoint.mode='subset' -> 'Subset'; 1250B (llama3-1t) DP = 2.""" + gen = _make_bare_generator(tmp_path) + # llama3-1t -> ClosedGPUs=1024, GPUpDP=512 -> DP = 2. + run = self._run( + tmp_path, model="llama3-1t", + parameters={"checkpoint": {"mode": "subset"}}, + ) + + result = gen._aggregate_workload_metrics([run], warmup_set=set()) + + assert result["checkpoint_mode"] == "Subset" + assert result["checkpoint_dp_instances"] == 2 + + def test_bw_blank_when_scalar_absent(self, tmp_path): + """A run missing the DLIO scalar -> present-but-blank B/W/duration.""" + gen = _make_bare_generator(tmp_path) + run = self._run( + tmp_path, model="llama3-8b", + parameters={"checkpoint": {"mode": "default"}}, + write_bw=None, read_dur=None, + ) + + result = gen._aggregate_workload_metrics([run], warmup_set=set()) + + assert result["checkpoint_write_bw_gibps"] is None + assert result["checkpoint_read_duration_secs"] is None + # Fields that ARE present still resolve. + assert result["checkpoint_read_bw_gibps"] == pytest.approx(52.0) + assert result["checkpoint_num_client_nodes"] == 4 + assert result["checkpoint_dp_instances"] == 1 # llama3-8b # --------------------------------------------------------------------------- # @@ -565,7 +826,9 @@ def test_vdb_in_summary_recall_path(self, tmp_path): assert result["vdb_p95_latency_ms"] == summary["p95_latency_ms"] assert result["vdb_p99_latency_ms"] == summary["p99_latency_ms"] assert result["vdb_p999_latency_ms"] == summary["p999_latency_ms"] - assert result["vdb_recall"] == summary["recall"] + # Recall is emitted as a PERCENTAGE (0-100), not the source fraction + # — the final-table column header is "Recall Percentage". + assert result["vdb_recall"] == pytest.approx(summary["recall"] * 100) # D-15 identity columns. assert result["vdb_engine"] == "milvus" assert result["vdb_index_type"] == "hnsw" @@ -599,7 +862,214 @@ def test_vdb_recall_fallback_via_recall_stats_json(self, tmp_path): assert result["vdb_recall"] is not None, ( "vdb_recall must be populated via the recall_stats.json fallback" ) - assert result["vdb_recall"] == recall_stats["recall"] + # Percentage form (0-100), consistent with the in-summary recall path. + assert result["vdb_recall"] == pytest.approx(recall_stats["recall"] * 100) + + +def _write_vdb_result_dir( + parent: pathlib.Path, + ts: str, + *, + summary: Optional[Dict[str, Any]] = None, + statistics: Optional[Dict[str, Any]] = None, + args: Optional[Dict[str, Any]] = None, + parameters: Optional[Dict[str, Any]] = None, +) -> pathlib.Path: + """Materialize a VDB run dir on disk for from_result_dir-based tests. + + Writes a ``__metadata.json`` so ``BenchmarkRun.from_result_dir`` + routes through ``ResultFilesExtractor._from_metadata``. Optionally writes + a canonical ``summary.json`` and/or a native ``statistics.json``. + """ + run_dir = parent / ts + run_dir.mkdir(parents=True, exist_ok=True) + metadata = { + "benchmark_type": "vector_database", + "run_datetime": ts, + "num_processes": 4, + "parameters": parameters if parameters is not None else {}, + "model": "milvus_HNSW", + "command": "run", + } + if args is not None: + metadata["args"] = args + (run_dir / f"vector_database_{ts}_metadata.json").write_text(json.dumps(metadata)) + if summary is not None: + (run_dir / "summary.json").write_text(json.dumps(summary)) + if statistics is not None: + (run_dir / "statistics.json").write_text(json.dumps(statistics)) + return run_dir + + +class TestVdbFinalTableColumns: + """Slice-VDB: the v3.0 final-table metric/identity columns (issue #823).""" + + def test_read_bw_and_client_nodes_from_disk_io(self, tmp_path): + """Read B/W (GiB/s) and # Client Nodes derive from ``disk_io``. + + ``disk_io.total_bytes_read_per_sec`` -> ``vdb_read_bw_gibps`` divided + by 1024**3 (GiB, binary). ``disk_io.host_count`` -> #Client Nodes. + Storage IOPs is not measured -> present-but-blank. + """ + gen = _make_bare_generator(tmp_path) + runs_root = tmp_path / "workload" + summary = { + "throughput_qps": 900.0, + "mean_latency_ms": 3.0, + "p95_latency_ms": 5.0, + "p99_latency_ms": 7.0, + "p999_latency_ms": 9.0, + "recall": 0.95, + "disk_io": { + "total_bytes_read_per_sec": 2 * 1024 ** 3, # 2 GiB/s + "host_count": 4, + }, + } + run_dir = _write_vdb_result_dir(runs_root, "20260704_130000", summary=summary) + run = BenchmarkRun.from_result_dir(str(run_dir)) + + result = gen._aggregate_workload_metrics([run], warmup_set=set()) + + assert result["vdb_read_bw_gibps"] == pytest.approx(2.0) + assert result["vdb_num_client_nodes"] == 4 + assert result["vdb_storage_iops"] is None + + def test_recall_percentage(self, tmp_path): + """Scalar recall in summary.json is emitted as a percentage.""" + gen = _make_bare_generator(tmp_path) + runs_root = tmp_path / "workload" + summary = {"throughput_qps": 1.0, "recall": 0.9421} + run_dir = _write_vdb_result_dir(runs_root, "20260704_131000", summary=summary) + run = BenchmarkRun.from_result_dir(str(run_dir)) + + result = gen._aggregate_workload_metrics([run], warmup_set=set()) + + assert result["vdb_recall"] == pytest.approx(94.21) + + def test_dict_recall_coerced_then_percentage(self, tmp_path): + """A dict recall in summary.json is reduced to mean_recall then ×100.""" + gen = _make_bare_generator(tmp_path) + runs_root = tmp_path / "workload" + summary = {"throughput_qps": 1.0, "recall": {"mean_recall": 0.88, "k": 10}} + run_dir = _write_vdb_result_dir(runs_root, "20260704_132000", summary=summary) + run = BenchmarkRun.from_result_dir(str(run_dir)) + + result = gen._aggregate_workload_metrics([run], warmup_set=set()) + + assert result["vdb_recall"] == pytest.approx(88.0) + + def test_legacy_backfill_from_statistics_json_in_memory(self, tmp_path): + """A legacy package with native statistics.json but NO summary.json. + + reportgen must backfill the metrics IN MEMORY (via build_vdb_summary) + and must NOT write summary.json into the submission package. + """ + gen = _make_bare_generator(tmp_path) + runs_root = tmp_path / "workload" + statistics = { + "throughput_qps": 321.0, + "p99_latency_ms": 11.0, + "recall": {"mean_recall": 0.97}, + "disk_io": {"total_bytes_read_per_sec": 1024 ** 3, "host_count": 2}, + } + run_dir = _write_vdb_result_dir( + runs_root, "20260704_133000", statistics=statistics + ) + # Fixture invariant: no canonical summary.json exists yet. + assert not (run_dir / "summary.json").exists() + run = BenchmarkRun.from_result_dir(str(run_dir)) + + result = gen._aggregate_workload_metrics([run], warmup_set=set()) + + assert result["vdb_throughput_qps"] == 321.0 + assert result["vdb_p99_latency_ms"] == 11.0 + assert result["vdb_recall"] == pytest.approx(97.0) + assert result["vdb_read_bw_gibps"] == pytest.approx(1.0) + assert result["vdb_num_client_nodes"] == 2 + # reportgen must NOT mutate the submission package. + assert not (run_dir / "summary.json").exists(), ( + "reportgen must derive summary in-memory, never write into the package" + ) + + def test_identity_columns_fall_back_to_persisted_args(self, tmp_path): + """Legacy packages have empty parameters; identity comes from metadata['args']. + + engine/index_type/num_vectors/dimension fall back to the persisted + per-run ``metadata['args']`` snapshot (keys ``vdb_engine`` / + ``vdb_index`` / ``num_vectors`` / ``dimension``) when the + ``parameters`` block lacks them. + """ + gen = _make_bare_generator(tmp_path) + runs_root = tmp_path / "workload" + summary = {"throughput_qps": 1.0} + run_dir = _write_vdb_result_dir( + runs_root, + "20260704_134000", + summary=summary, + parameters={}, # legacy: no reportgen params block + args={ + "vdb_engine": "milvus", + "vdb_index": "HNSW", + "num_vectors": 5000000, + "dimension": 768, + }, + ) + run = BenchmarkRun.from_result_dir(str(run_dir)) + + result = gen._aggregate_workload_metrics([run], warmup_set=set()) + + assert result["vdb_engine"] == "milvus" + assert result["vdb_index_type"] == "HNSW" + assert result["vdb_num_vectors"] == 5000000 + assert result["vdb_dimension"] == 768 + + def test_parameters_win_over_args_for_identity(self, tmp_path): + """When present, the parameters block wins over the args fallback.""" + gen = _make_bare_generator(tmp_path) + runs_root = tmp_path / "workload" + summary = {"throughput_qps": 1.0} + run_dir = _write_vdb_result_dir( + runs_root, + "20260704_135000", + summary=summary, + parameters={"engine": "milvus", "index_type": "DISKANN", + "num_vectors": 10, "dimension": 128}, + args={"vdb_engine": "WRONG", "vdb_index": "WRONG", + "num_vectors": 999, "dimension": 999}, + ) + run = BenchmarkRun.from_result_dir(str(run_dir)) + + result = gen._aggregate_workload_metrics([run], warmup_set=set()) + + assert result["vdb_engine"] == "milvus" + assert result["vdb_index_type"] == "DISKANN" + assert result["vdb_num_vectors"] == 10 + assert result["vdb_dimension"] == 128 + + +class TestBenchmarkRunArgs: + """``BenchmarkRun.run_args`` surfaces the persisted metadata['args'].""" + + def test_run_args_populated_from_metadata(self, tmp_path): + run_dir = _write_vdb_result_dir( + tmp_path / "workload", + "20260704_140000", + summary={"throughput_qps": 1.0}, + args={"vdb_engine": "milvus", "num_vectors": 42}, + ) + run = BenchmarkRun.from_result_dir(str(run_dir)) + + assert run.run_args == {"vdb_engine": "milvus", "num_vectors": 42} + + def test_run_args_defaults_empty_when_absent(self, tmp_path): + run_dir = _write_vdb_result_dir( + tmp_path / "workload", + "20260704_141000", + summary={"throughput_qps": 1.0}, + ) + run = BenchmarkRun.from_result_dir(str(run_dir)) + + assert run.run_args == {} # --------------------------------------------------------------------------- # @@ -714,6 +1184,33 @@ def test_kvcache_zero_value_passthrough_verbatim(self, tmp_path): assert result[col] is not None assert isinstance(result[col], float) + def test_kvcache_num_client_nodes_from_host_count(self, tmp_path): + """Slice-KVCache: ``# Client Nodes`` derives from ``summary['host_count']``. + + The v3.0 kvcache final table carries a shared ``# Client Nodes`` + column (not per-option). Its source is the top-level ``host_count`` + field kvcache's ``_write_run_summary`` already persists + (``benchmarks/kvcache.py:926``). reportgen must surface it as + ``kvcache_num_client_nodes``. + + Note on B/W units (resolved 2026-07-21): the per-option + ``aggregated_read/write_bandwidth_gbps`` values are ALREADY GiB/s + binary — kvcache's ``cache.py:990-991`` computes them as + ``(bytes / 1024**3) / duration`` despite the ``_gbps`` name. So the + final table's "Read/Write B/W (GiB/s)" columns flow verbatim from + the existing per-option columns with NO conversion; only + ``# Client Nodes`` was missing. + """ + gen = _make_bare_generator(tmp_path) + runs_root = tmp_path / "workload" + runs_root.mkdir() + run = _kvcache_run(runs_root, "20260704_140000") + + result = gen._aggregate_workload_metrics([run], warmup_set=set()) + + summary = _load_summary(runs_root / "20260704_140000" / "summary.json") + assert result["kvcache_num_client_nodes"] == summary["host_count"] + # --------------------------------------------------------------------------- # # TestInvalidRulesStrict — D-23 / D-24 / D-26 / D-27 / D-29 # diff --git a/tests/unit/test_benchmarks_vectordb.py b/tests/unit/test_benchmarks_vectordb.py index 7fb3b172..c67f4792 100755 --- a/tests/unit/test_benchmarks_vectordb.py +++ b/tests/unit/test_benchmarks_vectordb.py @@ -262,6 +262,35 @@ def test_metadata_datagen_command_fields(self, datagen_args, tmp_path): assert meta['index_type'] == 'HNSW' assert meta['model'] == 'milvus_HNSW' + def test_metadata_parameters_block_populated_for_reportgen( + self, datagen_args, tmp_path + ): + """metadata['parameters'] must carry the VDB reportgen identity keys. + + reportgen reads ``run.parameters`` (= ``metadata['parameters']``, + sourced from ``self.combined_params``) for the VDB final-table + identity columns Vector Count / Vector Dimension / Index Type / + Engine. VectorDBBenchmark historically never set + ``combined_params`` so this block was ``{}`` — leaving all four + columns blank. Pin the exact keys reportgen consumes. + """ + with patch('mlpstorage_py.benchmarks.base.generate_output_location') as mock_gen, \ + patch('mlpstorage_py.benchmarks.vectordbbench.read_config_from_file', return_value={}), \ + patch('mlpstorage_py.benchmarks.vectordbbench.VectorDBBenchmark.verify_benchmark'), \ + patch('mlpstorage_py.benchmarks.vectordbbench.VectorDBBenchmark._validate_vdb_dependencies'): + output_dir = str(tmp_path / "output") + mock_gen.return_value = output_dir + + from mlpstorage_py.benchmarks.vectordbbench import VectorDBBenchmark + bm = VectorDBBenchmark(datagen_args) + meta = bm.metadata + + params = meta['parameters'] + assert params.get('engine') == 'milvus' + assert params.get('index_type') == 'HNSW' + assert params.get('num_vectors') == 5000000 + assert params.get('dimension') == 768 + def test_metadata_connection_info(self, run_args, tmp_path): """Verify host/port connection info in metadata.""" run_args.host = '10.0.0.50' diff --git a/tests/unit/test_reportgen_sut_block.py b/tests/unit/test_reportgen_sut_block.py new file mode 100644 index 00000000..49092ca2 --- /dev/null +++ b/tests/unit/test_reportgen_sut_block.py @@ -0,0 +1,182 @@ +"""Task F — SUT (System Under Test) block population in reportgen output. + +reportgen must emit the shared System-Under-Test columns (v3.0 results +table, `Results Table Structure.xlsx`) into results.{csv,json} for every +workload row, sourced from the row's `system-description.yaml`: + +- ``sut_organization`` ← orgname (path) +- ``sut_name`` ← Hyperlink(text=solution.submission_name, + href=//systems/.yaml) +- ``sut_description`` ← Hyperlink(text=solution.submission_name, + href=//systems/.pdf) +- ``sut_rus`` ← system_under_test.total_rack_units +- blank placeholders (manual fill post-reportgen): ``sut_public_id``, + ``sut_type``, ``sut_access_protocol``, ``sut_availability``, + ``sut_integrated_client_storage``, ``sut_usable_capacity_tib`` + +Encoding (user-confirmed): +- ``.csv`` → hyperlink cell is an HTML anchor string + ``text`` +- ``.json`` → hyperlink cell is a ``{"text": ..., "href": ...}`` object + +Href base = repo-root-relative (Option A): ```` is the +aggregated-submissions repo root, so ``closed//systems/.yaml`` +is a valid relative URL. + +Regression guard: the D-10 6-column prefix stays first and D-12 ``issues`` +stays last — the ``sut_*`` columns are a body group between them. +""" + +from __future__ import annotations + +import csv +import json +import pathlib +import shutil +from argparse import Namespace + +import pytest + +from mlpstorage_py.report_generator import ReportGenerator + +_REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent.parent +_FIXTURES_ROOT = _REPO_ROOT / "tests" / "fixtures" / "sample_results" + +_SUBMISSION_NAME = "AcmeSysA" +_RACK_UNITS = 14 + +_ACME_YAML = f"""\ +system_under_test: + solution: + submission_name: {_SUBMISSION_NAME} + total_rack_units: {_RACK_UNITS} +""" + + +def _prepare_tree(tmp_path: pathlib.Path) -> pathlib.Path: + """Copy the multi_orgname fixture and inject acme's system-description.yaml.""" + fixture_src = _FIXTURES_ROOT / "multi_orgname" + assert fixture_src.is_dir(), f"expected fixture at {fixture_src}" + dest = tmp_path / "repo_root" + shutil.copytree(fixture_src, dest) + systems_dir = dest / "closed" / "acme" / "systems" + systems_dir.mkdir(parents=True, exist_ok=True) + (systems_dir / "system-a.yaml").write_text(_ACME_YAML) + return dest + + +def _run_reportgen(root: pathlib.Path) -> None: + gen = ReportGenerator(str(root), args=Namespace(debug=False), + validate_structure=False) + rc = gen.generate_reports() + assert rc == 0, f"generate_reports returned {rc}" + + +def _acme_per_model_dir(root: pathlib.Path) -> pathlib.Path: + return (root / "closed" / "acme" / "results" / "system-a" + / "training" / "unet3d" / "run") + + +class TestSutBlockJson: + def test_json_sut_columns_populated(self, tmp_path): + root = _prepare_tree(tmp_path) + _run_reportgen(root) + rows = json.loads((_acme_per_model_dir(root) / "results.json").read_text()) + assert len(rows) == 1 + row = rows[0] + + assert row["sut_organization"] == "acme" + assert row["sut_rus"] == _RACK_UNITS + # Hyperlinks serialize as {"text","href"} objects in JSON. + assert row["sut_name"] == { + "text": _SUBMISSION_NAME, + "href": "closed/acme/systems/system-a.yaml", + } + assert row["sut_description"] == { + "text": _SUBMISSION_NAME, + "href": "closed/acme/systems/system-a.pdf", + } + # Blank placeholders present-but-empty (manual fill). + for blank in ("sut_public_id", "sut_type", "sut_access_protocol", + "sut_availability", "sut_integrated_client_storage", + "sut_usable_capacity_tib"): + assert row[blank] == "", f"{blank} should be blank, got {row[blank]!r}" + + +class TestSutBlockCsv: + def test_csv_hyperlinks_render_as_html_anchor(self, tmp_path): + root = _prepare_tree(tmp_path) + _run_reportgen(root) + with open(_acme_per_model_dir(root) / "results.csv", newline="") as fh: + rows = list(csv.DictReader(fh)) + assert len(rows) == 1 + row = rows[0] + assert row["sut_name"] == ( + f'{_SUBMISSION_NAME}' + ) + assert row["sut_description"] == ( + f'{_SUBMISSION_NAME}' + ) + assert row["sut_organization"] == "acme" + assert row["sut_rus"] == str(_RACK_UNITS) + + +_CODE_HASH = "0123456789abcdef0123456789abcdef" # md5-tree-v2 32-hex +_CODE_DIR = "code-01234567" # code- +_ACME_LEAF = ("closed/acme/results/system-a/training/unet3d/run/20260706_100000") + + +def _inject_code_pointer(root: pathlib.Path) -> None: + leaf = root / _ACME_LEAF + (leaf / ".mlps-code-image").write_text(f"md5-tree-v2:{_CODE_HASH}\n") + + +class TestSutCodeLogsLinks: + """Task F-b: Code/Logs anchors resolved from the run's .mlps-code-image.""" + + def test_json_code_logs_hyperlinks(self, tmp_path): + root = _prepare_tree(tmp_path) + _inject_code_pointer(root) + _run_reportgen(root) + rows = json.loads((_acme_per_model_dir(root) / "results.json").read_text()) + row = rows[0] + expected_href = f"closed/acme/{_CODE_DIR}/" + assert row["sut_code"] == {"text": "code", "href": expected_href} + # Logs is a placeholder pointing at the same code-image dir for now. + assert row["sut_logs"] == {"text": "logs", "href": expected_href} + + def test_csv_code_logs_anchor(self, tmp_path): + root = _prepare_tree(tmp_path) + _inject_code_pointer(root) + _run_reportgen(root) + with open(_acme_per_model_dir(root) / "results.csv", newline="") as fh: + row = next(csv.DictReader(fh)) + href = f"closed/acme/{_CODE_DIR}/" + assert row["sut_code"] == f'code' + assert row["sut_logs"] == f'logs' + + def test_missing_pointer_leaves_code_logs_blank(self, tmp_path): + # No pointer injected → graceful blank, no crash. + root = _prepare_tree(tmp_path) + _run_reportgen(root) + rows = json.loads((_acme_per_model_dir(root) / "results.json").read_text()) + row = rows[0] + assert row["sut_code"] == "" + assert row["sut_logs"] == "" + + +class TestSutBlockDoesNotBreakColumnInvariants: + def test_prefix_first_issues_last_sut_in_between(self, tmp_path): + root = _prepare_tree(tmp_path) + _run_reportgen(root) + with open(_acme_per_model_dir(root) / "results.csv", newline="") as fh: + header = next(csv.reader(fh)) + assert header[:6] == [ + "category", "orgname", "systemname", + "benchmark_type", "model", "accelerator", + ], f"D-10 prefix broken: {header[:6]}" + assert header[-1] == "issues", f"D-12 trailing issues broken: {header[-1]}" + # sut_ columns exist and sit strictly between prefix and issues. + sut_idxs = [i for i, c in enumerate(header) if c.startswith("sut_")] + assert sut_idxs, f"no sut_ columns in header: {header}" + assert min(sut_idxs) >= 6 and max(sut_idxs) < len(header) - 1 diff --git a/tests/unit/test_vdb_summary.py b/tests/unit/test_vdb_summary.py new file mode 100644 index 00000000..79bd4951 --- /dev/null +++ b/tests/unit/test_vdb_summary.py @@ -0,0 +1,141 @@ +"""Tests for ``mlpstorage_py.benchmarks.vdb_summary``. + +The VDB producer writes its native metrics into ``statistics.json`` +(single-node at the run root, MPI under ``vectordb//``) — never a +canonical ``summary.json`` at the run root, which is where reportgen +reads. ``vdb_summary`` bridges that gap with two pure functions: + + * ``build_vdb_summary(result_dir)`` — locate the native QUERY-phase + stats file, project it, coerce a dict ``recall`` to its scalar + ``mean_recall``. Returns the dict, or ``None`` when no native file + exists. + * ``write_vdb_summary(result_dir)`` — build + persist + ``/summary.json`` (producer-only). Returns the path, or + ``None`` when there was nothing to write. + +reportgen calls ``build_vdb_summary`` IN MEMORY for legacy packages that +predate the producer write — it never writes into the submission tree. +""" + +import json +import os + +import pytest + +from mlpstorage_py.benchmarks import vdb_summary + + +def _write_json(path, data): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fd: + json.dump(data, fd) + + +_FULL_STATS = { + "throughput_qps": 1250.5, + "mean_latency_ms": 12.3, + "p95_latency_ms": 18.7, + "p99_latency_ms": 25.4, + "p999_latency_ms": 41.2, + "recall": {"k": 10, "mean_recall": 0.982, "median_recall": 0.99}, + "disk_io": { + "applicable": True, + "total_bytes_read_per_sec": 1073741824.0, # exactly 1 GiB/s + "host_count": 3, + }, +} + + +class TestBuildVdbSummary: + def test_single_node_statistics_json_at_root(self, tmp_path): + """Single-node writes ``statistics.json`` at the run root.""" + _write_json(str(tmp_path / "statistics.json"), _FULL_STATS) + + out = vdb_summary.build_vdb_summary(str(tmp_path)) + + assert out is not None + assert out["throughput_qps"] == 1250.5 + assert out["p99_latency_ms"] == 25.4 + # dict recall coerced to its scalar mean_recall. + assert out["recall"] == 0.982 + # disk_io carried through untouched (reportgen converts units). + assert out["disk_io"]["total_bytes_read_per_sec"] == 1073741824.0 + assert out["disk_io"]["host_count"] == 3 + + def test_mpi_simple_subdir(self, tmp_path): + """MPI simple phase lands under ``vectordb/simple/statistics.json``.""" + _write_json( + str(tmp_path / "vectordb" / "simple" / "statistics.json"), + {"throughput_qps": 50.0, "recall": {"mean_recall": 0.9}}, + ) + + out = vdb_summary.build_vdb_summary(str(tmp_path)) + + assert out is not None + assert out["throughput_qps"] == 50.0 + assert out["recall"] == 0.9 + + def test_mpi_enhanced_subdir(self, tmp_path): + """MPI enhanced phase lands under ``vectordb/enhanced/enhanced_statistics.json``.""" + _write_json( + str(tmp_path / "vectordb" / "enhanced" / "enhanced_statistics.json"), + {"throughput_qps": 77.0, "recall": {"mean_recall": 0.85}}, + ) + + out = vdb_summary.build_vdb_summary(str(tmp_path)) + + assert out is not None + assert out["throughput_qps"] == 77.0 + assert out["recall"] == 0.85 + + def test_scalar_recall_passes_through(self, tmp_path): + """A recall that is already a scalar is preserved verbatim.""" + _write_json( + str(tmp_path / "statistics.json"), + {"throughput_qps": 10.0, "recall": 0.75}, + ) + + out = vdb_summary.build_vdb_summary(str(tmp_path)) + + assert out["recall"] == 0.75 + + def test_missing_native_file_returns_none(self, tmp_path): + """No native stats anywhere -> ``None`` (reportgen leaves columns blank).""" + assert vdb_summary.build_vdb_summary(str(tmp_path)) is None + + def test_load_statistics_is_not_treated_as_query(self, tmp_path): + """``load_statistics.json`` is the datagen phase, not a query result. + + The load phase must never be mistaken for the query summary — the + table wants QUERY metrics. + """ + _write_json( + str(tmp_path / "vectordb" / "load" / "load_statistics.json"), + {"throughput_qps": 999.0}, + ) + + assert vdb_summary.build_vdb_summary(str(tmp_path)) is None + + def test_malformed_json_returns_none(self, tmp_path): + (tmp_path / "statistics.json").write_text("{ not json", encoding="utf-8") + assert vdb_summary.build_vdb_summary(str(tmp_path)) is None + + +class TestWriteVdbSummary: + def test_writes_summary_json_at_root(self, tmp_path): + _write_json(str(tmp_path / "statistics.json"), _FULL_STATS) + + path = vdb_summary.write_vdb_summary(str(tmp_path)) + + assert path == str(tmp_path / "summary.json") + assert os.path.isfile(path) + with open(path, encoding="utf-8") as fd: + written = json.load(fd) + assert written["throughput_qps"] == 1250.5 + assert written["recall"] == 0.982 # scalar, matching build output + + def test_no_native_file_writes_nothing(self, tmp_path): + path = vdb_summary.write_vdb_summary(str(tmp_path)) + + assert path is None + assert not (tmp_path / "summary.json").exists()