From b7e4cd4b38eea2ed5d67914ea5d846fe171891e3 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Mon, 20 Jul 2026 22:13:05 -0700 Subject: [PATCH 01/14] =?UTF-8?q?test(reportgen):=20RED=20=E2=80=94=20SUT?= =?UTF-8?q?=20block=20columns=20in=20results.{csv,json}=20(Task=20F-a)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Assert reportgen emits the shared System-Under-Test block for each workload row, sourced from system-description.yaml: sut_organization, sut_name/sut_description as hyperlinks (HTML anchor in CSV, {text,href} in JSON), sut_rus from total_rack_units, and blank manual-fill placeholders. Guards the D-10 6-col prefix / D-12 trailing-issues invariants. Fails until F-a is implemented. --- tests/unit/test_reportgen_sut_block.py | 138 +++++++++++++++++++++++++ 1 file changed, 138 insertions(+) create mode 100644 tests/unit/test_reportgen_sut_block.py diff --git a/tests/unit/test_reportgen_sut_block.py b/tests/unit/test_reportgen_sut_block.py new file mode 100644 index 00000000..0d4e2da2 --- /dev/null +++ b/tests/unit/test_reportgen_sut_block.py @@ -0,0 +1,138 @@ +"""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) + + +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 From 425ed218c6f36edc0c3850baef78512d76107b22 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Mon, 20 Jul 2026 22:17:21 -0700 Subject: [PATCH 02/14] feat(reportgen): populate shared System-Under-Test block (Task F-a) reportgen now emits the v3.0 results-table SUT columns for every workload row, sourced from the row's system-description.yaml: - sut_organization (path), sut_rus (total_rack_units) - sut_name / sut_description as Hyperlink cells -> the system-description .yaml / .pdf (repo-root-relative href, Option A) - sut_public_id / sut_type / sut_access_protocol / sut_availability / sut_integrated_client_storage / sut_usable_capacity_tib as present-but-blank placeholders (submitter fills manually) New Hyperlink value type renders as an HTML anchor in results.csv (via __str__) and a {text,href} object in results.json (via MLPSJsonEncoder; write_json_file now passes cls=MLPSJsonEncoder). The sut_* group is ordered right after the D-10 6-col prefix and before metric groups, so the prefix stays first and 'issues' stays last. Green: new test_reportgen_sut_block.py + all 4 CI suites pass. --- mlpstorage_py/report_generator.py | 124 +++++++++++++++++++++++++++++- 1 file changed, 121 insertions(+), 3 deletions(-) diff --git a/mlpstorage_py/report_generator.py b/mlpstorage_py/report_generator.py index 077a5b47..1280b3a3 100755 --- a/mlpstorage_py/report_generator.py +++ b/mlpstorage_py/report_generator.py @@ -27,7 +27,7 @@ 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,85 @@ 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', + ] + + 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 + return cols + + 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]: @@ -1815,7 +1929,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 +1981,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 From add3749fa4d1ce706ebd478352095055b22dc335 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Mon, 20 Jul 2026 22:20:50 -0700 Subject: [PATCH 03/14] =?UTF-8?q?test(reportgen):=20RED=20=E2=80=94=20Code?= =?UTF-8?q?/Logs=20code-image=20links=20(Task=20F-b)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Assert reportgen resolves each run's .mlps-code-image pointer to code- and emits sut_code/sut_logs anchors -> closed//code-/ (Logs a placeholder to the same dir). Missing pointer -> blank, no crash. --- tests/unit/test_reportgen_sut_block.py | 44 ++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/tests/unit/test_reportgen_sut_block.py b/tests/unit/test_reportgen_sut_block.py index 0d4e2da2..49092ca2 100644 --- a/tests/unit/test_reportgen_sut_block.py +++ b/tests/unit/test_reportgen_sut_block.py @@ -121,6 +121,50 @@ def test_csv_hyperlinks_render_as_html_anchor(self, tmp_path): 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) From 30d06dc0aff21b68cb826cb38bf22344dd6e7ee0 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Mon, 20 Jul 2026 22:26:29 -0700 Subject: [PATCH 04/14] feat(reportgen): resolve Code/Logs code-image links (Task F-b) Adds sut_code / sut_logs to the SUT block. Each is resolved from the run leaf's .mlps-code-image pointer via the canonical code_image helpers (_read_pointer D-61, _pool_dir_name D-62) to a repo-root-relative anchor -> //code-/. Logs is a placeholder pointing at the same code-image dir for now (future change retargets it). Absent or malformed pointer -> blank cell, no crash. Green: 4 CI suites (2942/890/228/238) + test_reportgen_sut_block.py. --- mlpstorage_py/report_generator.py | 37 +++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/mlpstorage_py/report_generator.py b/mlpstorage_py/report_generator.py index 1280b3a3..70ec4488 100755 --- a/mlpstorage_py/report_generator.py +++ b/mlpstorage_py/report_generator.py @@ -569,6 +569,8 @@ def _workload_result_to_row( 'sut_rus', 'sut_integrated_client_storage', 'sut_usable_capacity_tib', + 'sut_code', + 'sut_logs', ] def _sut_columns( @@ -598,8 +600,43 @@ def _sut_columns( 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. From ef3ccccd22c67f57092a77716d2debd7714f160e Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Mon, 20 Jul 2026 23:04:33 -0700 Subject: [PATCH 05/14] =?UTF-8?q?test(reportgen):=20RED=20=E2=80=94=20VDB?= =?UTF-8?q?=20producer=20summary.json=20+=20reportgen-params=20block=20(Sl?= =?UTF-8?q?ice-VDB-a)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit vdb_summary module (build/write_vdb_summary): locate native query-phase statistics.json (single-node root + MPI vectordb/ subdirs), coerce dict recall to scalar mean_recall, ignore the load phase. Module absent → import error (RED). VectorDBBenchmark.metadata['parameters'] must carry the reportgen VDB identity keys (engine/index_type/num_vectors/dimension); combined_params was never set so the block is {} → engine None (RED). --- tests/unit/test_benchmarks_vectordb.py | 29 +++++ tests/unit/test_vdb_summary.py | 141 +++++++++++++++++++++++++ 2 files changed, 170 insertions(+) create mode 100644 tests/unit/test_vdb_summary.py 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_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() From cf22cf56ac1b2f90a70ee9755b399d06e43ea1c7 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Mon, 20 Jul 2026 23:05:43 -0700 Subject: [PATCH 06/14] feat(reportgen): VDB producer emits summary.json + reportgen params (Slice-VDB-a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add mlpstorage_py/benchmarks/vdb_summary.py with pure build_vdb_summary / write_vdb_summary: locate the native query-phase statistics.json (single-node run root + MPI vectordb/ subdirs), copy it whole, coerce dict recall to scalar mean_recall; the load phase is excluded. Directory-in, dict/path-out so reportgen can call it without constructing a VectorDBBenchmark. VectorDBBenchmark now sets self.combined_params (engine/index_type/ num_vectors/dimension) so base.get_metadata surfaces them in metadata['parameters'] — the VDB final-table identity columns reportgen reads. execute_run best-effort writes summary.json at the run root after a query run (skipped on --what-if; never fatal). GREEN for Slice-VDB-a. --- mlpstorage_py/benchmarks/vdb_summary.py | 110 ++++++++++++++++++++++ mlpstorage_py/benchmarks/vectordbbench.py | 52 +++++++++- 2 files changed, 160 insertions(+), 2 deletions(-) create mode 100644 mlpstorage_py/benchmarks/vdb_summary.py 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.""" From e55216c8046b8293b24a46a8c271133ebb864470 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Mon, 20 Jul 2026 23:08:39 -0700 Subject: [PATCH 07/14] =?UTF-8?q?test(reportgen):=20RED=20=E2=80=94=20VDB?= =?UTF-8?q?=20final-table=20columns=20+=20run=5Fargs=20(Slice-VDB-b)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reportgen _aggregate_vdb must gain: Read B/W (disk_io.total_bytes_read_per_sec ÷1024³ GiB/s), # Client Nodes (disk_io.host_count), Storage IOPs (present-but-blank, unmeasured), recall as percentage (×100, incl. dict→scalar coercion), in-memory legacy backfill from native statistics.json WITHOUT writing into the package, and identity columns (engine/index_type/num_vectors/ dimension) falling back to persisted metadata['args'] when parameters is empty (parameters win when present). BenchmarkRun.run_args must surface metadata['args'] (empty dict when absent). Existing TestVdbPassThrough recall assertions updated to the percentage form. 10 failing. --- tests/unit/test_aggregation.py | 213 ++++++++++++++++++++++++++++++++- 1 file changed, 211 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_aggregation.py b/tests/unit/test_aggregation.py index bb82161c..cc185e10 100644 --- a/tests/unit/test_aggregation.py +++ b/tests/unit/test_aggregation.py @@ -565,7 +565,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 +601,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 == {} # --------------------------------------------------------------------------- # From 1bf87cf680da3c4cdaa45a4a03fd454c2de1b4f3 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Mon, 20 Jul 2026 23:10:28 -0700 Subject: [PATCH 08/14] feat(reportgen): populate VDB final-table columns + args fallback (Slice-VDB-b) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BenchmarkRunData gains run_args (persisted metadata['args'] snapshot), populated at all three extractor sites; BenchmarkRun.run_args exposes it. report_generator._aggregate_vdb now emits the v3.0 VDB final-table columns: - vdb_read_bw_gibps (disk_io.total_bytes_read_per_sec ÷ 1024**3, GiB/s) - vdb_num_client_nodes (disk_io.host_count) - vdb_storage_iops (present-but-blank; not measured) - vdb_recall as a percentage (×100), with dict→scalar mean_recall coercion on both the summary.json and recall_stats.json paths - vdb_num_vectors / vdb_dimension / vdb_engine / vdb_index_type via parameters-first, persisted-args-fallback (blank only when truly absent) New _load_vdb_summary backfills legacy packages IN MEMORY via vdb_summary.build_vdb_summary when summary.json is absent — never writing into the submission package. GREEN for Slice-VDB-b. --- mlpstorage_py/report_generator.py | 93 ++++++++++++++++++++++++++++--- mlpstorage_py/rules/models.py | 17 ++++++ 2 files changed, 102 insertions(+), 8 deletions(-) diff --git a/mlpstorage_py/report_generator.py b/mlpstorage_py/report_generator.py index 70ec4488..836f1a9d 100755 --- a/mlpstorage_py/report_generator.py +++ b/mlpstorage_py/report_generator.py @@ -1236,7 +1236,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", @@ -1250,8 +1250,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): @@ -1259,19 +1263,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 - # Identity columns (D-15). ``BenchmarkRun`` exposes CLI/YAML args - # via ``.parameters``; use ``.get`` so missing keys emit ``None``. + @staticmethod + def _vdb_identity(run: BenchmarkRun, param_key: str, arg_keys: tuple) -> Any: + """Resolve a VDB identity value: parameters first, then persisted args. + + ``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, 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 From 1ecc82edde2fecb9cab07729ecc6a9d48dc948f9 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Tue, 21 Jul 2026 07:31:37 -0700 Subject: [PATCH 09/14] =?UTF-8?q?test(reportgen):=20RED=20=E2=80=94=20kvca?= =?UTF-8?q?che=20#=20Client=20Nodes=20from=20host=5Fcount=20(Slice-KVCache?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/unit/test_aggregation.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/tests/unit/test_aggregation.py b/tests/unit/test_aggregation.py index cc185e10..e4826d5a 100644 --- a/tests/unit/test_aggregation.py +++ b/tests/unit/test_aggregation.py @@ -923,6 +923,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 # From c1ccc8687f167c5bc48d59bb63a97fd571195bf8 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Tue, 21 Jul 2026 07:34:12 -0700 Subject: [PATCH 10/14] feat(reportgen): emit kvcache # Client Nodes from host_count (Slice-KVCache) The v3.0 kvcache final table carries a shared '# Client Nodes' column. Its source is the top-level host_count field kvcache's _write_run_summary already persists; reportgen now surfaces it as kvcache_num_client_nodes. The table's Read/Write B/W (GiB/s) and P95 Read Latency columns already flow verbatim via the existing per-option flattening: the per-option aggregated_read/write_bandwidth_gbps values are already GiB/s binary (kvcache cache.py divides bytes by 1024**3 despite the _gbps name), so no unit conversion is applied. host_count was the only missing field. --- mlpstorage_py/report_generator.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/mlpstorage_py/report_generator.py b/mlpstorage_py/report_generator.py index 836f1a9d..27bd0dd3 100755 --- a/mlpstorage_py/report_generator.py +++ b/mlpstorage_py/report_generator.py @@ -1391,6 +1391,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). From fbdcd6819de1fa46c64002d490fb8bedac862458 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Tue, 21 Jul 2026 07:48:26 -0700 Subject: [PATCH 11/14] =?UTF-8?q?test(reportgen):=20RED=20=E2=80=94=20trai?= =?UTF-8?q?ning=20final-table=20columns=20(Slice-Training)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Read B/W (GiB/s), # Client Nodes, # Simulated Accelerators — sourced from summary.json top-level num_hosts/num_accelerators and DLIO's scalar train_io_mean_MB_per_second (MiB/s, per statscounter.py 'MiB/second' log). --- tests/unit/test_aggregation.py | 110 +++++++++++++++++++++++++++++++++ 1 file changed, 110 insertions(+) diff --git a/tests/unit/test_aggregation.py b/tests/unit/test_aggregation.py index e4826d5a..34b7fcea 100644 --- a/tests/unit/test_aggregation.py +++ b/tests/unit/test_aggregation.py @@ -451,6 +451,116 @@ def test_training_output_keys_use_train_mean_of_prefix(self, tmp_path): ) +# --------------------------------------------------------------------------- # +# 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 + + # --------------------------------------------------------------------------- # # TestCheckpointingAggregation — TEST-15 / D-20 / D-28 / AGG-02 # # --------------------------------------------------------------------------- # From ffb5a515c4781c2d170bbc0c7661f01de1337373 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Tue, 21 Jul 2026 07:52:24 -0700 Subject: [PATCH 12/14] feat(reportgen): populate training final-table columns (Slice-Training) Emits the v3.0 training final-table columns that the list-valued metric aggregation did not carry, read straight from each run's summary.json (the metadata-complete parse path drops them: system_info is never reconstructed and scalar metrics are filtered out): - train_num_client_nodes <- summary top-level num_hosts - train_num_simulated_accelerators <- summary top-level num_accelerators - train_read_bw_gibps <- mean of DLIO's per-run scalar train_io_mean_MB_per_second across non-warmup runs, /1024 Read B/W unit: DLIO's train_io_mean_MB_per_second is MiB/s despite the 'MB' label (statscounter.py computes samples/s * record_size / 1024/1024 and logs it as 'MiB/second'), so GiB/s = value / 1024 (binary), matching the vdb/kvcache slices. Blank-but-present when any run omits the scalar. Updates the D-13 prefix invariant test: metric-mean columns keep train_mean_of_; the new final-table columns keep only the train_ group prefix (preserving column ordering) without mean_of_. --- mlpstorage_py/report_generator.py | 39 +++++++++++++++++++++++++++++++ tests/unit/test_aggregation.py | 27 +++++++++++++++++---- 2 files changed, 62 insertions(+), 4 deletions(-) diff --git a/mlpstorage_py/report_generator.py b/mlpstorage_py/report_generator.py index 27bd0dd3..4b41668f 100755 --- a/mlpstorage_py/report_generator.py +++ b/mlpstorage_py/report_generator.py @@ -1153,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( diff --git a/tests/unit/test_aggregation.py b/tests/unit/test_aggregation.py index 34b7fcea..df755506 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,25 @@ 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)." + ) # --------------------------------------------------------------------------- # From 5ae77a27acf599fd8ca1cc585e0d301334ee6939 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Tue, 21 Jul 2026 08:07:40 -0700 Subject: [PATCH 13/14] =?UTF-8?q?test(reportgen):=20RED=20=E2=80=94=20chec?= =?UTF-8?q?kpointing=20final-table=20columns=20(Slice-Checkpointing)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Write/Read B/W (GiB/s, verbatim — already binary) + Write/Read Duration from DLIO scalars save_/load_checkpoint_io_mean_GB_per_second and _duration_mean_seconds; # Client Nodes from num_hosts; Checkpoint Mode from checkpoint.mode param; DP Instances = int(ClosedGPUs/GPUpDP). --- tests/unit/test_aggregation.py | 132 +++++++++++++++++++++++++++++++++ 1 file changed, 132 insertions(+) diff --git a/tests/unit/test_aggregation.py b/tests/unit/test_aggregation.py index df755506..2d4e4290 100644 --- a/tests/unit/test_aggregation.py +++ b/tests/unit/test_aggregation.py @@ -580,6 +580,138 @@ def test_read_bw_blank_when_scalar_absent(self, tmp_path): 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 + + # --------------------------------------------------------------------------- # # TestCheckpointingAggregation — TEST-15 / D-20 / D-28 / AGG-02 # # --------------------------------------------------------------------------- # From 18431fe78cc269e942d5cfc85ceeb304901ea3a4 Mon Sep 17 00:00:00 2001 From: Curtis Anderson Date: Tue, 21 Jul 2026 08:10:51 -0700 Subject: [PATCH 14/14] feat(reportgen): populate checkpointing final-table columns (Slice-Checkpointing) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Emits the v3.0 checkpointing final-table columns, read from each run's summary.json (the metadata-complete parse path drops the DLIO scalar metric keys via its list-only filter, and they are absent from the fabricated fixtures): - checkpoint_num_client_nodes <- summary top-level num_hosts - checkpoint_mode <- checkpoint.mode param (subset->Subset, default/absent->Full) - checkpoint_dp_instances <- int(ClosedGPUs/GPUpDP) from LLM_ALLOWED_VALUES[model] - checkpoint_write_bw_gibps <- save_checkpoint_io_mean_GB_per_second - checkpoint_write_duration_secs <- save_checkpoint_duration_mean_seconds - checkpoint_read_bw_gibps <- load_checkpoint_io_mean_GB_per_second - checkpoint_read_duration_secs <- load_checkpoint_duration_mean_seconds B/W unit: DLIO's *_io_mean_GB_per_second is already GiB/s binary (checkpoint_size in GiB / seconds, logged 'GiB/second'), so NO conversion — unlike training's MiB-labelled figure. Blank-but-present when any run omits a scalar. Existing checkpoint_mean_of_* columns (built on the fabricated list fixture keys) are left untouched per the agreed scope. --- mlpstorage_py/report_generator.py | 77 ++++++++++++++++++++++++++++++- 1 file changed, 76 insertions(+), 1 deletion(-) diff --git a/mlpstorage_py/report_generator.py b/mlpstorage_py/report_generator.py index 4b41668f..78e5b685 100755 --- a/mlpstorage_py/report_generator.py +++ b/mlpstorage_py/report_generator.py @@ -20,7 +20,7 @@ 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, @@ -1245,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(