diff --git a/mlpstorage_py/report_generator.py b/mlpstorage_py/report_generator.py index 11a70a9f..9e707eef 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, MLPSJsonEncoder +from mlpstorage_py.utils import MLPSJsonEncoder from mlpstorage_py.reporting import ( ResultsDirectoryValidator, ValidationMessageFormatter, @@ -82,6 +82,156 @@ def __eq__(self, other: Any) -> bool: and self.text == other.text and self.href == other.href) +# --------------------------------------------------------------------------- # +# Column-parity fixed output schema (Results Table Structure.xlsx, v3.0). # +# # +# ``results.{csv,json}`` is a FIXED contract: exactly the reference webpage # +# columns for all 8 tables (2 training + 4 checkpointing + 1 kvcache + 1 vdb) # +# sharing a System-Under-Test block, plus 3 agreed discriminator columns # +# (Division / Benchmark Type / Model) so a single flat file can carry every # +# table's rows. A MLCommons staff member opens results.csv in Excel and # +# reduces it to any one table by deleting the workload blocks + discriminators # +# that don't apply. # +# # +# The schema is FIXED — never data-driven. reportgen cherry-picks each column # +# from the in-memory row (``_final_row``); a run that lacks a metric leaves # +# that cell blank rather than dropping/adding a column. Metric-block names are # +# unique per column (a row is a dict keyed by column name), so the shared # +# reference names (# Client Nodes, Code, Read B/W, …) are workload-qualified. # +# --------------------------------------------------------------------------- # + +# Model column display labels (reference table titles). Blank for the +# single-table workloads (kvcache, vdb) — the Benchmark Type column +# identifies those. +_MODEL_DISPLAY_LABELS = { + "unet3d": "Unet3D", + "retinanet": "RetinaNet", + "llama3-8b": "8B", + "llama3-70b": "70B", + "llama3-405b": "405B", + "llama3-1t": "1250B", +} + +# Left edge + shared System-Under-Test block. Each entry is +# (final_column_name, internal_row_key_or_None). ``None`` => manual / +# present-but-blank placeholder (Public ID + the 5 submitter-filled cells). +# Division / Benchmark Type / Model are computed in ``_final_row``. +_SUT_FINAL_COLUMNS = [ + ("Public ID", None), + ("Organization", "orgname"), + ("Division", "__division__"), + ("Benchmark Type", "__benchmark_type__"), + ("Model", "__model__"), + ("Name", "sut_name"), + ("Description", "sut_description"), + ("Type", None), + ("Access Protocol", None), + ("Availability", None), + ("RU's", "sut_rus"), + ("Integrated Client Storage", None), + ("Usable Capacity (TiB)", None), +] + +# Per-workload metric/param blocks. ``__code__`` / ``__logs__`` are the +# per-row code-image hyperlinks (routed from the internal ``sut_code`` / +# ``sut_logs`` values into THIS row's own workload block; the reference +# places Code/Logs in each workload's Test-Parameters group because an +# OPEN row may ship its own code). Every other value is an internal +# ``_aggregate_*`` output key. +_TRAINING_BLOCK = [ + ("Accelerator Type", "accelerator"), + ("# Client Nodes", "train_num_client_nodes"), + ("Code", "__code__"), + ("Logs", "__logs__"), + ("# Simulated Accelerators", "train_num_simulated_accelerators"), + ("Read B/W (GiB/s)", "train_read_bw_gibps"), +] +_CHECKPOINTING_BLOCK = [ + ("Checkpoint Mode", "checkpoint_mode"), + ("# Client Nodes", "checkpoint_num_client_nodes"), + ("DP Instances", "checkpoint_dp_instances"), + ("Code", "__code__"), + ("Logs", "__logs__"), + ("Write B/W (GiB/s)", "checkpoint_write_bw_gibps"), + ("Write Duration (secs)", "checkpoint_write_duration_secs"), + ("Read B/W (GiB/s)", "checkpoint_read_bw_gibps"), + ("Read Duration (secs)", "checkpoint_read_duration_secs"), +] +_VDB_BLOCK = [ + ("# Client Nodes", "vdb_num_client_nodes"), + ("Code", "__code__"), + ("Logs", "__logs__"), + ("Vector Count", "vdb_num_vectors"), + ("Vector Dimension", "vdb_dimension"), + ("Index Type", "vdb_index_type"), + ("Queries per Sec", "vdb_throughput_qps"), + ("Query Latency (ms)", "vdb_p99_latency_ms"), + ("Recall Percentage", "vdb_recall"), + ("Storage IOPs", "vdb_storage_iops"), + ("Read B/W (GiB/s)", "vdb_read_bw_gibps"), +] +# KVCache: 3 shared columns + 3 option-groups x 4 metrics. Groups map to +# the fixed kv-cache.py WORKLOAD_PARAMS option numbers (1/2/3). summary.json +# deserializes those keys as strings, so the internal column key uses the +# string option number. +_KVCACHE_SHARED_BLOCK = [ + ("# Client Nodes", "kvcache_num_client_nodes"), + ("Code", "__code__"), + ("Logs", "__logs__"), +] +_KVCACHE_GROUPS = [ + ("1", "llama3.1-8b Storage Only"), + ("2", "llama3.1-8b Storage + Mem"), + ("3", "llama3.1-70b Storage Only"), +] +_KVCACHE_GROUP_METRICS = [ + ("Throughput (tok/s)", "aggregated_avg_throughput_tokens_per_sec"), + ("Read B/W (GiB/s)", "aggregated_read_bandwidth_gbps"), + ("Write B/W (GiB/s)", "aggregated_write_bandwidth_gbps"), + ("P95 Read Latency (ms)", "aggregated_p95_latency_ms"), +] + +# Which workload block a row fills, keyed by the internal benchmark_type +# value. A row fills exactly one block; every other block stays blank. +_WORKLOAD_BLOCK_BY_TYPE = { + "training": ("Training", _TRAINING_BLOCK), + "checkpointing": ("Checkpointing", _CHECKPOINTING_BLOCK), + "vector_database": ("VDB", _VDB_BLOCK), + "kv_cache": ("KVCache", _KVCACHE_SHARED_BLOCK), # + groups, see _final_row +} + + +def _build_final_schema(): + """Assemble the ordered fixed column list from the block definitions. + + Building the header from the SAME structures the projection reads keeps + the two from drifting. + """ + cols = [name for name, _ in _SUT_FINAL_COLUMNS] + cols += [f"Training - {name}" for name, _ in _TRAINING_BLOCK] + cols += [f"Checkpointing - {name}" for name, _ in _CHECKPOINTING_BLOCK] + cols += [f"VDB - {name}" for name, _ in _VDB_BLOCK] + cols += [f"KVCache - {name}" for name, _ in _KVCACHE_SHARED_BLOCK] + for _opt, label in _KVCACHE_GROUPS: + cols += [f"KVCache {label} - {name}" for name, _ in _KVCACHE_GROUP_METRICS] + return cols + + +_FINAL_SCHEMA = _build_final_schema() + + +def _blank_if_nan(value: Any) -> Any: + """Coerce a NaN float to ``""`` (JSON has no NaN literal; CSV wants blank). + + NaN is the only value not equal to itself, so this needs no ``math`` + import. Keeps every fixed column present-but-blank rather than dropping + it (which ``remove_nan_values`` would do, breaking schema completeness). + """ + if isinstance(value, float) and value != value: + return "" + return value + + @dataclass class Result: """Container for a single benchmark run result.""" @@ -593,11 +743,13 @@ def _sut_columns( if category and orgname and systemname: base = f"{category}/{orgname}/systems/{systemname}" - submission_name, rack_units = self._read_system_description( + _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") + # Visible text (user-confirmed 2026-07-21): Name shows the + # system name and links to the system-description.yaml; + # Description shows literal "PDF" and links to the .pdf. + cols['sut_name'] = Hyperlink(systemname, f"{base}.yaml") + cols['sut_description'] = Hyperlink("PDF", f"{base}.pdf") if rack_units is not None: cols['sut_rus'] = rack_units @@ -1295,30 +1447,67 @@ def _first_present(field: str) -> Any: 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]: + # B/W (verbatim GiB/s) + durations — mean of DLIO's per-run scalar, + # taken PER DIRECTION over the invocations that were configured to + # produce it (Rules.md §2.1.23 permits 1-2 timestamp dirs; §4.7.1 + # MANDATES a write→read split when checkpoint-per-node < 3x client + # RAM). In a split, the write dir carries only save_* and the read + # dir only load_*, so a field's None in the OPPOSITE phase is + # expected — not data loss — and must not blank the column. + # + # Each invocation self-declares its phase via + # ``checkpoint.num_checkpoints_{write,read}`` (CLOSED forces each to + # 10 or 0, never both 0 — cli/checkpointing_args.py). We keep the + # loud-failure guard (D-23 / PITFALLS #3: no silent partial-mean) + # but scope it to the producing phase: a save_* missing from an + # invocation that WAS configured to write is real data loss and + # still blanks. When the phase signal is absent (legacy / OPEN + # packages that don't persist the counts) we cannot classify, so we + # fall back to a present-only mean rather than regress those rows. + def _phase_count(run: BenchmarkRun, which: str) -> Any: + params = run.parameters or {} + key = f"num_checkpoints_{which}" + ckpt = params.get("checkpoint") + if isinstance(ckpt, dict) and key in ckpt: + return ckpt.get(key) + return params.get(f"checkpoint.{key}") + + def _is_number(v: Any) -> bool: + return isinstance(v, (int, float)) and not isinstance(v, bool) + + def _directional_mean(field: str, which: str) -> Optional[float]: + # Classify each invocation as a producer of `which` direction + # from its configured count. If ANY invocation lacks the signal + # the set is unclassifiable -> present-only fallback. + counts = [_phase_count(r, which) for r in runs] + if all(_is_number(c) for c in counts): + producers = [ + s for s, c in zip(summaries, counts) if c > 0 + ] + if not producers: + return None + values = [(s.get("metric") or {}).get(field) for s in producers] + # Strict within the producing phase: a missing scalar here + # is data loss -> blank (loud), never a partial mean. + if values and all(_is_number(v) for v in values): + return fmean(values) + return None + # Unclassifiable: mean over whatever's present, blank if none. 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 + present = [v for v in values if _is_number(v)] + return fmean(present) if present else None - out["checkpoint_write_bw_gibps"] = _scalar_mean( - "save_checkpoint_io_mean_GB_per_second" + out["checkpoint_write_bw_gibps"] = _directional_mean( + "save_checkpoint_io_mean_GB_per_second", "write" ) - out["checkpoint_write_duration_secs"] = _scalar_mean( - "save_checkpoint_duration_mean_seconds" + out["checkpoint_write_duration_secs"] = _directional_mean( + "save_checkpoint_duration_mean_seconds", "write" ) - out["checkpoint_read_bw_gibps"] = _scalar_mean( - "load_checkpoint_io_mean_GB_per_second" + out["checkpoint_read_bw_gibps"] = _directional_mean( + "load_checkpoint_io_mean_GB_per_second", "read" ) - out["checkpoint_read_duration_secs"] = _scalar_mean( - "load_checkpoint_duration_mean_seconds" + out["checkpoint_read_duration_secs"] = _directional_mean( + "load_checkpoint_duration_mean_seconds", "read" ) return out @@ -1347,10 +1536,25 @@ def _aggregate_vdb( at ``submission_checker/checks/vdb_checks.py:462-469``. Identity columns (D-15): ``vdb_engine`` and ``vdb_index_type`` - read from ``runs[0].parameters`` (the ``BenchmarkRun`` accessor - for CLI/YAML args on disk). + read from the ``run`` invocation's ``parameters`` (the + ``BenchmarkRun`` accessor for CLI/YAML args on disk), falling + back to the other grouped leaves. + + Run-leaf selection: the VDB workload key (D-06) is + ``(category, orgname, systemname, engine, index_type)`` — it does + NOT include ``command``, so a workload's ``datasize`` / + ``datagen`` / ``run`` leaves all group under one key and arrive + here as a single ``runs`` list in discovery order. Only the + ``run`` leaf carries the native query-phase metrics + (``statistics.json`` / ``summary.json``); the datasize/datagen + leaves have none. Reading ``runs[0]`` blindly left QPS / latency + / recall blank whenever a non-``run`` leaf sorted first, so + select the ``run`` invocation for metrics and recall. Fall back + to ``runs[0]`` only when no ``run`` leaf is present (e.g. a + datasize-only tree) so identity columns still populate. """ - run = runs[0] + # Metrics + recall come from the ``run`` leaf (see docstring). + run = next((r for r in runs if r.command == 'run'), runs[0]) summary = self._load_vdb_summary(run) _VDB_METRIC_FIELDS = ( @@ -1405,13 +1609,24 @@ def _aggregate_vdb( # 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") - ) + # block is empty. Resolve across ALL grouped leaves (the ``run`` leaf + # first, then the datasize/datagen leaves): an identity arg such as + # ``num_vectors`` / ``dimension`` may be recorded only on the + # datasize leaf while the query metrics live on the run leaf. Blank + # only when truly absent from every leaf. + identity_runs = [run] + [r for r in runs if r is not run] + + def _identity(param_key: str, arg_keys: tuple) -> Any: + for candidate_run in identity_runs: + value = self._vdb_identity(candidate_run, param_key, arg_keys) + if value is not None: + return value + return None + + out["vdb_num_vectors"] = _identity("num_vectors", ("num_vectors",)) + out["vdb_dimension"] = _identity("dimension", ("dimension",)) + out["vdb_engine"] = _identity("engine", ("vdb_engine", "engine")) + out["vdb_index_type"] = _identity("index_type", ("index_type", "vdb_index")) return out @staticmethod @@ -2159,77 +2374,89 @@ def _print_workload_details(self, workload_key: tuple, workload_result: Result) print(f"\n {checklist}") + def _final_row(self, row: Dict[str, Any]) -> Dict[str, Any]: + """Project one internal (machine-key) row onto the fixed + ``_FINAL_SCHEMA`` (column-parity contract). + + The emitted column set is decided HERE and nowhere else — it is the + fixed reference schema, never data-driven. Each fixed column is + cherry-picked from the in-memory row; a column that does not apply + to this row's workload (or whose metric is absent) stays blank. + Code/Logs are routed into the row's OWN workload block. The internal + aggregation (``*_mean_of_*``, the D-24 INVALID gate, etc.) is left + untouched upstream — those values are simply not emitted here. + """ + bt = row.get("benchmark_type") or "" + out: Dict[str, Any] = {col: "" for col in _FINAL_SCHEMA} + + # Left edge + shared SUT block. + for final_name, key in _SUT_FINAL_COLUMNS: + if key is None: + continue # manual placeholder — present-but-blank + if key == "__division__": + out[final_name] = (row.get("category") or "").upper() + elif key == "__benchmark_type__": + out[final_name] = bt + elif key == "__model__": + # Reference display label; blank for the single-table + # workloads (kvcache, vdb) — Benchmark Type identifies those. + if bt in ("kv_cache", "vector_database"): + out[final_name] = "" + else: + model = row.get("model") or "" + out[final_name] = _MODEL_DISPLAY_LABELS.get(model, model) + else: + out[final_name] = row.get(key, "") + + # The row's own workload block (exactly one is populated). + block = _WORKLOAD_BLOCK_BY_TYPE.get(bt) + if block is not None: + prefix, columns = block + self._fill_block(out, row, prefix, columns) + if bt == "kv_cache": + # Fixed option-groups: option N -> its reference group. + for opt, label in _KVCACHE_GROUPS: + for metric_name, metric_suffix in _KVCACHE_GROUP_METRICS: + final_name = f"KVCache {label} - {metric_name}" + src = f"kvcache_option_{opt}_{metric_suffix}" + out[final_name] = row.get(src, "") + + return {k: _blank_if_nan(v) for k, v in out.items()} + + @staticmethod + def _fill_block(out: Dict[str, Any], row: Dict[str, Any], + prefix: str, columns) -> None: + """Fill one workload block. ``__code__`` / ``__logs__`` route the + row's per-run code-image hyperlinks (``sut_code`` / ``sut_logs``) + into this block's Code/Logs columns.""" + for name, key in columns: + final_name = f"{prefix} - {name}" + if key == "__code__": + out[final_name] = row.get("sut_code", "") + elif key == "__logs__": + out[final_name] = row.get("sut_logs", "") + else: + out[final_name] = row.get(key, "") + def write_json_file(self, results, target_dir: Optional[str] = None): out_dir = target_dir if target_dir is not None else self.results_dir json_file = os.path.join(out_dir, 'results.json') self.logger.info(f'Writing results to {json_file}') + projected = [self._final_row(r) for r in results] with open(json_file, 'w') as f: - json.dump(results, f, indent=2, cls=MLPSJsonEncoder) + json.dump(projected, 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 csv_file = os.path.join(out_dir, 'results.csv') self.logger.info(f'Writing results to {csv_file}') - flattened_results = [flatten_nested_dict(r) for r in results] - flattened_results = [remove_nan_values(r) for r in flattened_results] - - # D-11 grouped-ordering assembly. Replaces the pure alphabetical - # field-name sort — that would put `checkpoint_*` before - # `train_*` and break the invariant 06-05's TestColumnOrdering - # test-locks. - ordered_fieldnames = self._ordered_fieldnames(flattened_results) - + # Project each internal row onto the fixed column-parity schema. + # ``_final_row`` returns EXACTLY the _FINAL_SCHEMA keys (NaN already + # blanked), so no flatten/field-discovery is needed — the header is + # the fixed schema regardless of input (header-only when empty). + projected = [self._final_row(r) for r in results] with open(csv_file, 'w+', newline='') as file_object: - csv_writer = csv.DictWriter(f=file_object, fieldnames=ordered_fieldnames, lineterminator='\n') + csv_writer = csv.DictWriter( + f=file_object, fieldnames=_FINAL_SCHEMA, lineterminator='\n') csv_writer.writeheader() - csv_writer.writerows(flattened_results) - - def _ordered_fieldnames(self, rows: List[dict]) -> List[str]: - """D-11 grouped-ordering CSV header assembly. - - Layout (exact): - - 1. Fixed 6-column prefix, in this exact order: - ``['category', 'orgname', 'systemname', 'benchmark_type', - 'model', 'accelerator']`` - 2. Sorted ``train_*`` columns. - 3. Sorted ``checkpoint_*`` columns. - 4. Sorted ``vdb_*`` columns. - 5. Sorted ``kvcache_*`` columns. - 6. Any remaining un-prefixed columns, sorted (defensive: catches - new columns future refactors may introduce). - 7. Trailing ``['issues']`` (D-12 last-position invariant). - - For an EMPTY ``rows`` list this returns the prefix + trailing - columns only — the minimal header shape D-03 requires for - empty-model-dir emission (``results.csv`` = header row only). - """ - prefix = ['category', 'orgname', 'systemname', - 'benchmark_type', 'model', 'accelerator'] - trailing = ['issues'] - - all_keys: Set[str] = set() - for r in rows: - all_keys.update(r.keys()) - - prefix_set = set(prefix) - 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 = 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 - # fixed prefix / trailing sets — should be empty in practice. - # Sorted-appending keeps behavior deterministic if a future - # row-shape change introduces such a column. - return prefix + grouped + other + trailing + csv_writer.writerows(projected) diff --git a/tests/integration/test_reportgen_aggregate_end_to_end.py b/tests/integration/test_reportgen_aggregate_end_to_end.py index 937b66e1..c63ef78b 100644 --- a/tests/integration/test_reportgen_aggregate_end_to_end.py +++ b/tests/integration/test_reportgen_aggregate_end_to_end.py @@ -84,15 +84,19 @@ def _read_csv_header(path: pathlib.Path) -> List[str]: return next(reader) -def _row_identity(row: Dict[str, Any]) -> Tuple[str, str, str, str, str, str]: - """Extract the 6-column identity tuple from a row dict.""" +def _row_identity(row: Dict[str, Any]) -> Tuple[str, str, str, str]: + """Extract a stable identity tuple from a fixed-schema row dict. + + Column-parity: the emitted file no longer carries the machine-key + identity prefix. Identity now reads the fixed-schema discriminator + columns (Division / Organization / Benchmark Type / Model). systemname + and accelerator are no longer standalone emitted columns. + """ return ( - str(row.get('category', '') or ''), - str(row.get('orgname', '') or ''), - str(row.get('systemname', '') or ''), - str(row.get('benchmark_type', '') or ''), - str(row.get('model', '') or ''), - str(row.get('accelerator', '') or ''), + str(row.get('Division', '') or ''), + str(row.get('Organization', '') or ''), + str(row.get('Benchmark Type', '') or ''), + str(row.get('Model', '') or ''), ) @@ -241,14 +245,14 @@ def test_deleted_run_subdir_absent_from_next_report(self, tmp_path): f"{len(rows_after)}. Before: {rows_before}. After: {rows_after}" ) - # The deleted workload's identity carried orgname='beta_corp', - # benchmark_type='training', model='unet3d'. Any row bearing - # ('beta_corp', 'training', 'unet3d') must be absent. + # The deleted workload's identity carried Organization='beta_corp', + # Benchmark Type='training', Model='Unet3D' (display label). Any row + # bearing ('beta_corp', 'training', 'Unet3D') must be absent. for row in rows_after: assert not ( - row.get('orgname') == 'beta_corp' - and row.get('benchmark_type') == 'training' - and row.get('model') == 'unet3d' + row.get('Organization') == 'beta_corp' + and row.get('Benchmark Type') == 'training' + and row.get('Model') == 'Unet3D' ), ( f"D-03 violated: deleted workload row still present in " f"top-level after rerun. Row: {row}" @@ -260,9 +264,9 @@ def test_deleted_run_subdir_absent_from_next_report(self, tmp_path): beta_identity = next( ( ident for ident in identities_before - if ident[1] == 'beta_corp' # orgname - and ident[3] == 'training' # benchmark_type - and ident[4] == 'unet3d' # model + if ident[1] == 'beta_corp' # Organization + and ident[2] == 'training' # Benchmark Type + and ident[3] == 'Unet3D' # Model (display label) ), None, ) @@ -327,21 +331,19 @@ def test_empty_model_dir_emits_header_only_csv_and_empty_list_json( f"Directory: {list(per_model_dir.iterdir())}" ) - # CSV: header row only — exactly one line, non-empty, with the - # D-10 6-column prefix present. + # CSV: header row only — exactly one line, non-empty, carrying the + # full fixed webpage-parity schema (column-parity: header is fixed + # even for an empty model dir). + from mlpstorage_py.report_generator import _FINAL_SCHEMA with open(csv_path, "r", newline="") as fh: lines = [line.rstrip("\r\n") for line in fh if line.strip()] assert len(lines) == 1, ( f"D-03 corner violated: expected 1 header line in {csv_path}, " f"got {len(lines)} lines: {lines}" ) - # The header must include the prefix columns. header = _read_csv_header(csv_path) - assert header[:6] == [ - 'category', 'orgname', 'systemname', 'benchmark_type', 'model', 'accelerator' - ], f"empty-model CSV header does not carry D-10 prefix: {header}" - assert header[-1] == 'issues', ( - f"empty-model CSV header does not end with 'issues': {header}" + assert header == _FINAL_SCHEMA, ( + f"empty-model CSV header must be the fixed schema: {header}" ) # JSON: contents == []. diff --git a/tests/unit/test_aggregation.py b/tests/unit/test_aggregation.py index 2d4e4290..9cdda077 100644 --- a/tests/unit/test_aggregation.py +++ b/tests/unit/test_aggregation.py @@ -711,6 +711,119 @@ def test_bw_blank_when_scalar_absent(self, tmp_path): assert result["checkpoint_num_client_nodes"] == 4 assert result["checkpoint_dp_instances"] == 1 # llama3-8b + # ----------------------------------------------------------------- # + # Two-invocation topologies (Rules.md §2.1.23 permits 1-2 timestamp # + # dirs; §4.7.1 MANDATES a write→read split when checkpoint-per-node # + # < 3x client RAM). Each score column must come from the phase that # + # actually produced it — a `None` from the *other* phase is # + # expected, not data loss — while a `None` from a phase that WAS # + # configured to produce the metric stays a loud blank. # + # ----------------------------------------------------------------- # + + def _ckpt_run(self, result_dir, *, model, write, read, ts): + """Build a checkpointing BenchmarkRun whose config self-declares + its phase via ``checkpoint.num_checkpoints_{write,read}`` (CLOSED + forces each to 10 or 0; never both 0 — checkpointing_args.py).""" + return _make_run( + benchmark_type=BENCHMARK_TYPES.checkpointing, + model=model, + result_dir=result_dir, + metrics={}, + parameters={"checkpoint": { + "mode": "default", + "num_checkpoints_write": write, + "num_checkpoints_read": read, + }}, + accelerator=None, + run_datetime=ts, + ) + + def test_split_write_read_invocations_populate_all_four(self, tmp_path): + """Reporter's bug (Alluxio CLOSED v3.0, llama3-8b): a write-phase + dir (write=10, read=0; only ``save_*`` scalars) plus a read-phase + dir (write=0, read=10; only ``load_*`` scalars). All four score + columns must resolve from the phase that produced each — the + current all-summaries-must-be-numeric gate blanks them because the + opposite phase legitimately omits the field.""" + gen = _make_bare_generator(tmp_path) + runs_root = tmp_path / "workload" + write_dir = _write_checkpointing_result_dir( + runs_root, "20260703_100000", + write_bw=4.64, write_dur=22.6, read_bw=None, read_dur=None, + ) + read_dir = _write_checkpointing_result_dir( + runs_root, "20260703_101500", + write_bw=None, write_dur=None, read_bw=5.68, read_dur=18.4, + ) + write_run = self._ckpt_run( + write_dir, model="llama3-8b", write=10, read=0, + ts="20260703_100000") + read_run = self._ckpt_run( + read_dir, model="llama3-8b", write=0, read=10, + ts="20260703_101500") + + result = gen._aggregate_workload_metrics( + [write_run, read_run], warmup_set=set()) + + assert result["checkpoint_write_bw_gibps"] == pytest.approx(4.64) + assert result["checkpoint_read_bw_gibps"] == pytest.approx(5.68) + assert result["checkpoint_write_duration_secs"] == pytest.approx(22.6) + assert result["checkpoint_read_duration_secs"] == pytest.approx(18.4) + assert result["checkpoint_num_client_nodes"] == 4 + assert result["checkpoint_mode"] == "Full" + + def test_two_combined_invocations_average_per_direction(self, tmp_path): + """Two full write+read invocations (both write=read=10). Each + direction is the inter-invocation mean of its producers.""" + gen = _make_bare_generator(tmp_path) + runs_root = tmp_path / "workload" + d1 = _write_checkpointing_result_dir( + runs_root, "20260703_100000", + write_bw=4.0, write_dur=20.0, read_bw=5.0, read_dur=18.0) + d2 = _write_checkpointing_result_dir( + runs_root, "20260703_101500", + write_bw=6.0, write_dur=24.0, read_bw=7.0, read_dur=22.0) + r1 = self._ckpt_run(d1, model="llama3-8b", write=10, read=10, + ts="20260703_100000") + r2 = self._ckpt_run(d2, model="llama3-8b", write=10, read=10, + ts="20260703_101500") + + result = gen._aggregate_workload_metrics([r1, r2], warmup_set=set()) + + assert result["checkpoint_write_bw_gibps"] == pytest.approx(5.0) + assert result["checkpoint_read_bw_gibps"] == pytest.approx(6.0) + assert result["checkpoint_write_duration_secs"] == pytest.approx(22.0) + assert result["checkpoint_read_duration_secs"] == pytest.approx(20.0) + + def test_missing_metric_in_producing_phase_blanks_not_silent(self, tmp_path): + """Loud-failure guard — the behavior the per-direction design + protects. Two invocations BOTH configured to write (write=10): a + ``save_*`` missing from one is real data loss, so the write column + must blank, NEVER silently report the surviving invocation's value + (which a naive 'average over whatever's present' fix would do). The + read direction, present in both, still resolves.""" + gen = _make_bare_generator(tmp_path) + runs_root = tmp_path / "workload" + complete = _write_checkpointing_result_dir( + runs_root, "20260703_100000", + write_bw=4.0, write_dur=20.0, read_bw=5.0, read_dur=18.0) + lost_write = _write_checkpointing_result_dir( + runs_root, "20260703_101500", + write_bw=None, write_dur=None, read_bw=6.0, read_dur=22.0) + r1 = self._ckpt_run(complete, model="llama3-8b", write=10, read=10, + ts="20260703_100000") + r2 = self._ckpt_run(lost_write, model="llama3-8b", write=10, read=10, + ts="20260703_101500") + + result = gen._aggregate_workload_metrics([r1, r2], warmup_set=set()) + + # A producer lost its scalar -> blank, NOT the surviving 4.0. + assert result["checkpoint_write_bw_gibps"] is None + assert result["checkpoint_write_duration_secs"] is None + # Read direction fully present in both -> inter-invocation mean. + assert result["checkpoint_read_bw_gibps"] == pytest.approx(5.5) + assert result["checkpoint_read_duration_secs"] == pytest.approx(20.0) + # --------------------------------------------------------------------------- # # TestCheckpointingAggregation — TEST-15 / D-20 / D-28 / AGG-02 # @@ -1046,6 +1159,71 @@ def test_parameters_win_over_args_for_identity(self, tmp_path): assert result["vdb_num_vectors"] == 10 assert result["vdb_dimension"] == 128 + def test_metrics_read_from_run_leaf_not_first_grouped_leaf(self, tmp_path): + """Regression: metrics must come from the ``run`` leaf of the group. + + The VDB workload key (D-06) is + ``(category, orgname, systemname, engine, index_type)`` — it does + NOT include ``command``, so a workload's ``datasize`` / ``datagen`` + / ``run`` leaves all group under one key and arrive as a single + ``runs`` list (e.g. ``[datasize, datagen, run]``). Only the ``run`` + leaf carries the native ``statistics.json`` query metrics; the + datasize/datagen leaves have none. Reading ``runs[0]`` blindly (the + datasize leaf here) left QPS/latency/recall blank even though the + real values lived in the ``run`` leaf. ``_aggregate_vdb`` must + select the ``run`` invocation. + """ + gen = _make_bare_generator(tmp_path) + root = tmp_path / "vector_database" / "milvus" / "DISKANN" + + # datasize + datagen leaves: metadata-shaped BenchmarkRuns whose + # result dirs carry NO summary.json / statistics.json. + datasize_dir = root / "datasize" / "20260704_090000" + datasize_dir.mkdir(parents=True) + datagen_dir = root / "datagen" / "20260704_093000" + datagen_dir.mkdir(parents=True) + params = {"engine": "milvus", "index_type": "DISKANN", + "num_vectors": 1000000, "dimension": 768} + datasize_run = _make_run( + benchmark_type=BENCHMARK_TYPES.vector_database, model="", + result_dir=str(datasize_dir), parameters=params, + accelerator=None, command="datasize", + ) + datagen_run = _make_run( + benchmark_type=BENCHMARK_TYPES.vector_database, model="", + result_dir=str(datagen_dir), parameters=params, + accelerator=None, command="datagen", + ) + + # run leaf: carries the native statistics.json with the real metrics. + statistics = { + "throughput_qps": 7045.83, + "mean_latency_ms": 7.35, + "p95_latency_ms": 8.07, + "p99_latency_ms": 9.0, + "p999_latency_ms": 11.0, + "recall": {"mean_recall": 0.4009}, + } + run_dir = _write_vdb_result_dir( + root / "run", "20260704_100000", statistics=statistics, + ) + run_run = BenchmarkRun.from_result_dir(str(run_dir)) + + # runs[0] is the datasize leaf — the collision the key cannot prevent. + result = gen._aggregate_workload_metrics( + [datasize_run, datagen_run, run_run], warmup_set=set() + ) + + assert result["vdb_throughput_qps"] == 7045.83 + assert result["vdb_mean_latency_ms"] == 7.35 + assert result["vdb_p95_latency_ms"] == 8.07 + assert result["vdb_p99_latency_ms"] == 9.0 + assert result["vdb_p999_latency_ms"] == 11.0 + assert result["vdb_recall"] == pytest.approx(40.09) + # Identity columns still populate (present on the run leaf). + assert result["vdb_engine"] == "milvus" + assert result["vdb_index_type"] == "DISKANN" + class TestBenchmarkRunArgs: """``BenchmarkRun.run_args`` surfaces the persisted metadata['args'].""" @@ -1989,143 +2167,20 @@ def test_healthy_leaf_datagen_no_warn_no_invalid(self, tmp_path): class TestColumnOrdering: - """CSV column-ordering test-lock (D-14 / D-18). - - Defends against a future PR reverting to pure - ``sorted(fieldnames)`` — which would put ``checkpoint_*`` before - ``train_*`` and break D-11 grouped ordering. Layout invariant: - - 1. Fixed 6-column prefix: ``[category, orgname, systemname, - benchmark_type, model, accelerator]`` (D-10). - 2. Then ``train_*`` cols sorted. - 3. Then ``checkpoint_*`` cols sorted. - 4. Then ``vdb_*`` cols sorted. - 5. Then ``kvcache_*`` cols sorted (including flattened - ``kvcache_option__*`` columns). - 6. Trailing ``issues`` (D-12). + """CSV column-schema test-lock (column-parity contract). + + Supersedes the D-10/D-11 machine-key grouped-ordering lock (and the + removed ``_ordered_fieldnames`` helper). ``write_csv_file`` now emits + the FIXED webpage-parity schema regardless of the input row's keys — + the columns are the reference tables' union + discriminators, never + data-driven. This defends against a future PR reintroducing a + data-driven / machine-key header. """ - def test_ordered_fieldnames_prefix_is_exact(self, tmp_path): - """The 6-column prefix appears at exactly positions [0..5].""" - gen = _make_bare_generator(tmp_path) - rows = [ - { - "category": "closed", - "orgname": "acme", - "systemname": "sys-a", - "benchmark_type": "training", - "model": "unet3d", - "accelerator": "h100", - "train_mean_of_au_percentage": 95.0, - "issues": "", - } - ] - header = gen._ordered_fieldnames(rows) - assert header[:6] == [ - "category", - "orgname", - "systemname", - "benchmark_type", - "model", - "accelerator", - ], f"6-column prefix wrong: {header[:6]!r}" - assert header[-1] == "issues", ( - f"Trailing column must be 'issues'; got {header[-1]!r}" - ) - - def test_ordered_fieldnames_group_order_train_checkpoint_vdb_kvcache(self, tmp_path): - """max(train) < min(checkpoint) < min(vdb) < min(kvcache) (D-11 group order).""" - gen = _make_bare_generator(tmp_path) - rows = [ - { - "category": "closed", - "orgname": "acme", - "systemname": "sys-a", - "benchmark_type": "training", - "model": "unet3d", - "accelerator": "h100", - "train_a": 1.0, - "train_b": 2.0, - "checkpoint_a": 3.0, - "checkpoint_z": 4.0, - "vdb_a": 5.0, - "kvcache_a": 6.0, - "kvcache_option_x_y": 7.0, - "issues": "", - } - ] - header = gen._ordered_fieldnames(rows) - - def indices(prefix: str) -> List[int]: - return [i for i, k in enumerate(header) if k.startswith(prefix)] - - train_idx = indices("train_") - checkpoint_idx = indices("checkpoint_") - vdb_idx = indices("vdb_") - kvcache_idx = indices("kvcache_") - - assert train_idx and checkpoint_idx and vdb_idx and kvcache_idx, ( - f"Every group must contribute at least one column: header={header}" - ) - assert max(train_idx) < min(checkpoint_idx), ( - f"D-11 violated: train indices {train_idx} must precede " - f"checkpoint indices {checkpoint_idx}." - ) - assert max(checkpoint_idx) < min(vdb_idx), ( - f"D-11 violated: checkpoint indices {checkpoint_idx} must precede " - f"vdb indices {vdb_idx}." - ) - assert max(vdb_idx) < min(kvcache_idx), ( - f"D-11 violated: vdb indices {vdb_idx} must precede " - f"kvcache indices {kvcache_idx}." - ) - - def test_within_group_alphabetical(self, tmp_path): - """Within each group, column names are in alphabetical order.""" - gen = _make_bare_generator(tmp_path) - rows = [ - { - "category": "closed", - "orgname": "acme", - "systemname": "sys-a", - "benchmark_type": "training", - "model": "unet3d", - "accelerator": "h100", - "train_c": 1.0, - "train_a": 2.0, - "train_b": 3.0, - "checkpoint_z": 4.0, - "checkpoint_a": 5.0, - "vdb_z": 6.0, - "vdb_a": 7.0, - "kvcache_z": 8.0, - "kvcache_a": 9.0, - "kvcache_option_b_x": 10.0, - "kvcache_option_a_y": 11.0, - "issues": "", - } - ] - header = gen._ordered_fieldnames(rows) - - def group_cols(prefix: str) -> List[str]: - return [k for k in header if k.startswith(prefix)] - - for prefix in ("train_", "checkpoint_", "vdb_", "kvcache_"): - cols = group_cols(prefix) - assert cols == sorted(cols), ( - f"Group {prefix}* not alphabetical: {cols}" - ) - - def test_csv_header_written_by_write_csv_file_uses_ordered_fieldnames(self, tmp_path): - """End-to-end: ``write_csv_file`` writes the D-11 grouped header. - - Reads back the written ``results.csv`` header row and pins - both the 6-column prefix at [0..5] and the trailing ``issues`` - at [-1]. This catches drift in the writer surface (D-10 + - D-12) — the individual ``_ordered_fieldnames`` tests above - pin the helper; this one pins the writer's use of it. - """ + def test_csv_header_written_by_write_csv_file_is_fixed_schema(self, tmp_path): + from mlpstorage_py.report_generator import _FINAL_SCHEMA gen = _make_bare_generator(tmp_path) + # Internal machine-key row (as _workload_result_to_row produces). rows = [ { "category": "closed", @@ -2135,6 +2190,7 @@ def test_csv_header_written_by_write_csv_file_uses_ordered_fieldnames(self, tmp_ "model": "unet3d", "accelerator": "h100", "train_mean_of_au_percentage": 95.0, + "train_read_bw_gibps": 12.5, "issues": "", } ] @@ -2147,15 +2203,14 @@ def test_csv_header_written_by_write_csv_file_uses_ordered_fieldnames(self, tmp_ with open(csv_path, "r") as f: header_line = f.readline().rstrip("\n") columns = header_line.split(",") - assert columns[:6] == [ - "category", - "orgname", - "systemname", - "benchmark_type", - "model", - "accelerator", - ] - assert columns[-1] == "issues" + assert columns == _FINAL_SCHEMA, ( + f"write_csv_file must emit the fixed webpage-parity schema.\n" + f"Expected: {_FINAL_SCHEMA}\nGot: {columns}" + ) + # No machine-key identity/dynamic/issues columns leak into output. + for banned in ("category", "orgname", "systemname", "benchmark_type", + "issues", "train_mean_of_au_percentage"): + assert banned not in columns # --------------------------------------------------------------------------- # diff --git a/tests/unit/test_reportgen_final_schema.py b/tests/unit/test_reportgen_final_schema.py new file mode 100644 index 00000000..631a5d1b --- /dev/null +++ b/tests/unit/test_reportgen_final_schema.py @@ -0,0 +1,325 @@ +"""Column-parity: results.{csv,json} is a FIXED webpage-parity schema. + +The v3.0 user-visible results web page is 8 tables (2 training + 4 +checkpointing + 1 kvcache + 1 vdb) sharing a System-Under-Test block. +``reports reportgen`` emits ONE flat ``results.csv`` / ``results.json`` +that a MLCommons staff member opens in Excel and reduces — by deleting +the workload blocks + discriminator columns that don't apply — to any one +of those 8 tables. + +That requires the emitted file to be a FIXED schema (the exact reference +columns + the 3 agreed discriminators), NOT a data-driven set of columns. +This file pins that contract at the emitted-file boundary: + +- The header is EXACTLY ``FINAL_SCHEMA`` (order + membership), regardless + of which metric keys a given run's summary.json happened to contain. +- Every JSON row dict carries EXACTLY those keys, in order. +- The internal machine columns (``category`` / ``orgname`` / + ``systemname`` / ``benchmark_type`` prefix, ``*_mean_of_*`` dynamic + metric-mean columns, trailing ``issues``) do NOT appear. +- Values are cherry-picked from the in-memory row into the right fixed + column: Division = CLOSED/OPEN, Model = display label, kvcache option + N routed to its fixed group, etc. + +This is the write-layer analog of the internal-structure tests in +tests/unit/test_aggregation.py — those pin the ``_aggregate_*`` machine +keys (UNCHANGED by this task); this pins the projected output file. +""" + +from __future__ import annotations + +import csv +import json +import pathlib +from typing import Any, Dict, List +from unittest.mock import patch + +from mlpstorage_py.report_generator import ReportGenerator, Hyperlink + + +# --------------------------------------------------------------------------- # +# The authoritative fixed schema (Results Table Structure.xlsx, v3.0 + # +# agreed Division/Benchmark Type/Model discriminators). 54 columns. # +# --------------------------------------------------------------------------- # + +FINAL_SCHEMA: List[str] = [ + # Left edge + shared SUT block (13) + "Public ID", + "Organization", + "Division", + "Benchmark Type", + "Model", + "Name", + "Description", + "Type", + "Access Protocol", + "Availability", + "RU's", + "Integrated Client Storage", + "Usable Capacity (TiB)", + # Training block (6) + "Training - Accelerator Type", + "Training - # Client Nodes", + "Training - Code", + "Training - Logs", + "Training - # Simulated Accelerators", + "Training - Read B/W (GiB/s)", + # Checkpointing block (9) + "Checkpointing - Checkpoint Mode", + "Checkpointing - # Client Nodes", + "Checkpointing - DP Instances", + "Checkpointing - Code", + "Checkpointing - Logs", + "Checkpointing - Write B/W (GiB/s)", + "Checkpointing - Write Duration (secs)", + "Checkpointing - Read B/W (GiB/s)", + "Checkpointing - Read Duration (secs)", + # VDB block (11) + "VDB - # Client Nodes", + "VDB - Code", + "VDB - Logs", + "VDB - Vector Count", + "VDB - Vector Dimension", + "VDB - Index Type", + "VDB - Queries per Sec", + "VDB - Query Latency (ms)", + "VDB - Recall Percentage", + "VDB - Storage IOPs", + "VDB - Read B/W (GiB/s)", + # KVCache block (3 shared + 3 groups x 4 = 15) + "KVCache - # Client Nodes", + "KVCache - Code", + "KVCache - Logs", + "KVCache llama3.1-8b Storage Only - Throughput (tok/s)", + "KVCache llama3.1-8b Storage Only - Read B/W (GiB/s)", + "KVCache llama3.1-8b Storage Only - Write B/W (GiB/s)", + "KVCache llama3.1-8b Storage Only - P95 Read Latency (ms)", + "KVCache llama3.1-8b Storage + Mem - Throughput (tok/s)", + "KVCache llama3.1-8b Storage + Mem - Read B/W (GiB/s)", + "KVCache llama3.1-8b Storage + Mem - Write B/W (GiB/s)", + "KVCache llama3.1-8b Storage + Mem - P95 Read Latency (ms)", + "KVCache llama3.1-70b Storage Only - Throughput (tok/s)", + "KVCache llama3.1-70b Storage Only - Read B/W (GiB/s)", + "KVCache llama3.1-70b Storage Only - Write B/W (GiB/s)", + "KVCache llama3.1-70b Storage Only - P95 Read Latency (ms)", +] + + +def _bare_generator(tmp_path: pathlib.Path) -> ReportGenerator: + results_dir = tmp_path / "results" + results_dir.mkdir(exist_ok=True) + with patch.object(ReportGenerator, "accumulate_results"): + with patch.object(ReportGenerator, "print_results"): + return ReportGenerator(str(results_dir), validate_structure=False) + + +def _internal_rows() -> List[Dict[str, Any]]: + """Internal (machine-key) rows exactly as ``_workload_result_to_row`` + produces them — one per workload type. The projection under test must + map these onto ``FINAL_SCHEMA`` and drop everything else. + """ + sut = { + "sut_public_id": "", + "sut_organization": "acme", + "sut_name": Hyperlink("system-a", "closed/acme/systems/system-a.yaml"), + "sut_description": Hyperlink("PDF", "closed/acme/systems/system-a.pdf"), + "sut_type": "", + "sut_access_protocol": "", + "sut_availability": "", + "sut_rus": 14, + "sut_integrated_client_storage": "", + "sut_usable_capacity_tib": "", + "sut_code": Hyperlink("code", "closed/acme/code-abc12345/"), + "sut_logs": Hyperlink("logs", "closed/acme/code-abc12345/"), + } + common = { + "category": "closed", + "orgname": "acme", + "systemname": "system-a", + } + training = { + **common, + "benchmark_type": "training", + "model": "unet3d", + "accelerator": "h100", + **sut, + # dynamic metric-mean columns that MUST be dropped: + "train_mean_of_au_percentage": 95.0, + "train_mean_of_throughput_samples_per_second": 1250.0, + # final-table metric keys: + "train_num_client_nodes": 8, + "train_num_simulated_accelerators": 16, + "train_read_bw_gibps": 12.5, + "issues": "", + } + checkpointing = { + **common, + "benchmark_type": "checkpointing", + "model": "llama3-1t", + "accelerator": "", + **sut, + "checkpoint_mean_of_read_throughput_GB_per_second": 12.4, + "checkpoint_num_client_nodes": 4, + "checkpoint_mode": "Full", + "checkpoint_dp_instances": 2, + "checkpoint_write_bw_gibps": 8.8, + "checkpoint_write_duration_secs": 30.0, + "checkpoint_read_bw_gibps": 9.1, + "checkpoint_read_duration_secs": 27.5, + "issues": "", + } + vdb = { + **common, + "benchmark_type": "vector_database", + "model": "", + "accelerator": "", + **sut, + "vdb_num_client_nodes": 3, + "vdb_num_vectors": 1000000, + "vdb_dimension": 768, + "vdb_index_type": "hnsw", + "vdb_throughput_qps": 4200.0, + "vdb_p99_latency_ms": 1.7, + "vdb_recall": 98.5, + "vdb_storage_iops": None, + "vdb_read_bw_gibps": 5.5, + "issues": "", + } + kvcache = { + **common, + "benchmark_type": "kv_cache", + "model": "llama3.1-8b", + "accelerator": "", + **sut, + "kvcache_num_client_nodes": 2, + "kvcache_option_1_aggregated_avg_throughput_tokens_per_sec": 111.0, + "kvcache_option_1_aggregated_read_bandwidth_gbps": 11.1, + "kvcache_option_1_aggregated_write_bandwidth_gbps": 1.11, + "kvcache_option_1_aggregated_p95_latency_ms": 0.11, + "kvcache_option_2_aggregated_avg_throughput_tokens_per_sec": 222.0, + "kvcache_option_2_aggregated_read_bandwidth_gbps": 22.2, + "kvcache_option_2_aggregated_write_bandwidth_gbps": 2.22, + "kvcache_option_2_aggregated_p95_latency_ms": 0.22, + "kvcache_option_3_aggregated_avg_throughput_tokens_per_sec": 333.0, + "kvcache_option_3_aggregated_read_bandwidth_gbps": 33.3, + "kvcache_option_3_aggregated_write_bandwidth_gbps": 3.33, + "kvcache_option_3_aggregated_p95_latency_ms": 0.33, + "issues": "", + } + return [training, checkpointing, vdb, kvcache] + + +class TestFixedSchemaHeader: + def test_csv_header_is_exactly_final_schema(self, tmp_path): + gen = _bare_generator(tmp_path) + out = tmp_path / "csv_out" + out.mkdir() + gen.write_csv_file(_internal_rows(), target_dir=str(out)) + with open(out / "results.csv", newline="") as fh: + header = next(csv.reader(fh)) + assert header == FINAL_SCHEMA, ( + "results.csv header must be EXACTLY the fixed webpage-parity " + f"schema.\nExpected: {FINAL_SCHEMA}\nGot: {header}" + ) + + def test_empty_input_still_emits_full_fixed_header(self, tmp_path): + gen = _bare_generator(tmp_path) + out = tmp_path / "empty_out" + out.mkdir() + gen.write_csv_file([], target_dir=str(out)) + with open(out / "results.csv", newline="") as fh: + header = next(csv.reader(fh)) + assert header == FINAL_SCHEMA + + def test_json_keys_are_exactly_final_schema_in_order(self, tmp_path): + gen = _bare_generator(tmp_path) + out = tmp_path / "json_out" + out.mkdir() + gen.write_json_file(_internal_rows(), target_dir=str(out)) + rows = json.loads((out / "results.json").read_text()) + assert len(rows) == 4 + for row in rows: + assert list(row.keys()) == FINAL_SCHEMA + + +class TestInternalColumnsRemoved: + def test_machine_and_dynamic_columns_absent(self, tmp_path): + gen = _bare_generator(tmp_path) + out = tmp_path / "absent_out" + out.mkdir() + gen.write_csv_file(_internal_rows(), target_dir=str(out)) + with open(out / "results.csv", newline="") as fh: + header = set(next(csv.reader(fh))) + for banned in ( + "category", + "orgname", + "systemname", + "benchmark_type", + "model", + "accelerator", + "issues", + "sut_organization", + "train_mean_of_au_percentage", + "checkpoint_mean_of_read_throughput_GB_per_second", + ): + assert banned not in header, f"{banned!r} must not appear in results.csv" + + +class TestValueProjection: + def _json_rows(self, tmp_path): + gen = _bare_generator(tmp_path) + out = tmp_path / "vp_out" + out.mkdir() + gen.write_json_file(_internal_rows(), target_dir=str(out)) + return json.loads((out / "results.json").read_text()) + + def test_training_row_values(self, tmp_path): + row = self._json_rows(tmp_path)[0] + assert row["Division"] == "CLOSED" + assert row["Benchmark Type"] == "training" + assert row["Model"] == "Unet3D" + assert row["Organization"] == "acme" + assert row["RU's"] == 14 + assert row["Training - Accelerator Type"] == "h100" + assert row["Training - # Client Nodes"] == 8 + assert row["Training - # Simulated Accelerators"] == 16 + assert row["Training - Read B/W (GiB/s)"] == 12.5 + # Code/Logs live in the workload block, populated for this row. + assert row["Training - Code"] == { + "text": "code", "href": "closed/acme/code-abc12345/"} + # Other workloads' blocks are blank for a training row. + assert row["Checkpointing - Write B/W (GiB/s)"] == "" + assert row["VDB - Vector Count"] == "" + + def test_checkpointing_row_values(self, tmp_path): + row = self._json_rows(tmp_path)[1] + assert row["Benchmark Type"] == "checkpointing" + assert row["Model"] == "1250B" # llama3-1t display label + assert row["Checkpointing - Checkpoint Mode"] == "Full" + assert row["Checkpointing - # Client Nodes"] == 4 + assert row["Checkpointing - DP Instances"] == 2 + assert row["Checkpointing - Write B/W (GiB/s)"] == 8.8 + assert row["Checkpointing - Read Duration (secs)"] == 27.5 + + def test_vdb_row_values(self, tmp_path): + row = self._json_rows(tmp_path)[2] + assert row["Benchmark Type"] == "vector_database" + assert row["Model"] == "" # blank for vdb + assert row["VDB - Vector Count"] == 1000000 + assert row["VDB - Index Type"] == "hnsw" + assert row["VDB - Query Latency (ms)"] == 1.7 + assert row["VDB - Recall Percentage"] == 98.5 + assert row["VDB - Storage IOPs"] in ("", None) + assert row["VDB - Read B/W (GiB/s)"] == 5.5 + + def test_kvcache_group_option_routing(self, tmp_path): + row = self._json_rows(tmp_path)[3] + assert row["Benchmark Type"] == "kv_cache" + assert row["Model"] == "" # blank for kvcache + assert row["KVCache - # Client Nodes"] == 2 + # option 1 -> Storage Only, option 2 -> Storage + Mem, option 3 -> 70b + assert row["KVCache llama3.1-8b Storage Only - Throughput (tok/s)"] == 111.0 + assert row["KVCache llama3.1-8b Storage Only - Read B/W (GiB/s)"] == 11.1 + assert row["KVCache llama3.1-8b Storage + Mem - Throughput (tok/s)"] == 222.0 + assert row["KVCache llama3.1-70b Storage Only - Throughput (tok/s)"] == 333.0 + assert row["KVCache llama3.1-70b Storage Only - P95 Read Latency (ms)"] == 0.33 diff --git a/tests/unit/test_reportgen_output_shape.py b/tests/unit/test_reportgen_output_shape.py index d6b018ee..3ce7c4f2 100644 --- a/tests/unit/test_reportgen_output_shape.py +++ b/tests/unit/test_reportgen_output_shape.py @@ -145,10 +145,21 @@ def _synthetic_rows() -> List[Dict[str, Any]]: # --------------------------------------------------------------------------- # -class TestSixColumnPrefix: - """D-10: fixed 6-column prefix in exact order at CSV header + JSON keys.""" +_LEFT_EDGE = [ + 'Public ID', 'Organization', 'Division', 'Benchmark Type', 'Model', +] - def test_csv_header_starts_with_six_column_prefix(self, tmp_path): + +class TestFixedSchemaLeftEdge: + """Column-parity: fixed left-edge (SUT + discriminators) at CSV + JSON. + + Replaces the SUPERSEDED D-10 machine-key 6-column prefix. The emitted + file is now the fixed webpage-parity schema; the first five columns are + the identity + discriminator block (Public ID, Organization, Division, + Benchmark Type, Model). + """ + + def test_csv_header_starts_with_fixed_left_edge(self, tmp_path): gen = _bare_generator(tmp_path) out_dir = tmp_path / "csv_prefix_out" out_dir.mkdir() @@ -158,15 +169,15 @@ def test_csv_header_starts_with_six_column_prefix(self, tmp_path): assert csv_path.exists(), f"expected {csv_path} to exist" with open(csv_path, "r", newline="") as fh: header = next(csv.reader(fh)) - assert header[:6] == [ - 'category', 'orgname', 'systemname', 'benchmark_type', 'model', 'accelerator' - ], ( - "D-10 6-column prefix violated. First 6 header columns must be exactly " - "['category', 'orgname', 'systemname', 'benchmark_type', 'model', 'accelerator']. " - f"Got: {header[:6]}" + assert header[:5] == _LEFT_EDGE, ( + "Fixed-schema left edge violated. First 5 header columns must be " + f"{_LEFT_EDGE}. Got: {header[:5]}" ) + # The old machine-key identity columns must be gone. + for banned in ('category', 'orgname', 'systemname', 'benchmark_type'): + assert banned not in header - def test_json_row_dict_starts_with_six_column_prefix_keys(self, tmp_path): + def test_json_row_dict_starts_with_fixed_left_edge(self, tmp_path): gen = _bare_generator(tmp_path) out_dir = tmp_path / "json_prefix_out" out_dir.mkdir() @@ -177,21 +188,11 @@ def test_json_row_dict_starts_with_six_column_prefix_keys(self, tmp_path): with open(json_path, "r") as fh: loaded = json.load(fh) - # Every emitted row must at minimum carry the 6 prefix keys. - # Python 3.7+ preserves dict insertion order and json.load - # preserves that order too, so we assert the first 6 KEYS in - # order match the D-10 prefix. - prefix = ['category', 'orgname', 'systemname', - 'benchmark_type', 'model', 'accelerator'] for row in loaded: keys = list(row.keys()) - for col in prefix: - assert col in row, ( - f"D-10 prefix key '{col}' missing from row: {keys}" - ) - assert keys[:6] == prefix, ( - f"D-10 prefix ORDER violated in JSON row. Expected first " - f"6 keys to be {prefix}, got {keys[:6]}." + assert keys[:5] == _LEFT_EDGE, ( + f"Fixed-schema left-edge ORDER violated in JSON row. Expected " + f"first 5 keys to be {_LEFT_EDGE}, got {keys[:5]}." ) @@ -200,10 +201,17 @@ def test_json_row_dict_starts_with_six_column_prefix_keys(self, tmp_path): # --------------------------------------------------------------------------- # -class TestTrailingIssuesColumn: - """D-12: 'issues' is the last column. D-25: '; '-joined at write time.""" +class TestNoIssuesColumn: + """Column-parity: the 'issues' column is NOT emitted. - def test_csv_header_ends_with_issues(self, tmp_path): + Supersedes D-12/D-25 (trailing 'issues' column, '; '-joined). The + fixed webpage-parity schema carries only the reference columns + + discriminators; validation issues are not a reference column. The + internal '; ' join in _workload_result_to_row still runs (it feeds + the on-screen report), it is just no longer emitted to the file. + """ + + def test_csv_header_has_no_issues_column(self, tmp_path): gen = _bare_generator(tmp_path) out_dir = tmp_path / "csv_trailing_out" out_dir.mkdir() @@ -213,41 +221,20 @@ def test_csv_header_ends_with_issues(self, tmp_path): assert csv_path.exists() with open(csv_path, "r", newline="") as fh: header = next(csv.reader(fh)) - assert header[-1] == 'issues', ( - f"D-12 trailing issues column violated. Last header column must be " - f"'issues'; got {header[-1]!r}. Full header: {header}" + assert 'issues' not in header, ( + f"'issues' must not be a results.csv column. Full header: {header}" ) - def test_multi_issue_row_joined_by_semicolon_space(self, tmp_path): - # The '; ' join happens inside _workload_result_to_row (not - # inside write_csv_file). To exercise the join at the writer - # boundary we pre-join in the synthetic row and assert the - # written cell round-trips verbatim. + def test_json_rows_have_no_issues_key(self, tmp_path): gen = _bare_generator(tmp_path) - out_dir = tmp_path / "csv_multi_issue_out" + out_dir = tmp_path / "json_no_issues_out" out_dir.mkdir() + gen.write_json_file(_synthetic_rows(), target_dir=str(out_dir)) - rows = [{ - "category": "INVALID", - "orgname": "acme", - "systemname": "system-a", - "benchmark_type": "training", - "model": "unet3d", - "accelerator": "h100", - "train_mean_of_au_percentage": 95.0, - "issues": "issue1; issue2; issue3", - }] - gen.write_csv_file(rows, target_dir=str(out_dir)) - - csv_path = out_dir / "results.csv" - with open(csv_path, "r", newline="") as fh: - reader = csv.DictReader(fh) - got_rows = list(reader) - assert len(got_rows) == 1 - assert got_rows[0]['issues'] == "issue1; issue2; issue3", ( - f"D-25 '; ' join format violated. Expected 'issue1; issue2; issue3', " - f"got {got_rows[0]['issues']!r}" - ) + with open(out_dir / "results.json", "r") as fh: + loaded = json.load(fh) + for row in loaded: + assert 'issues' not in row, f"'issues' must not be a JSON key: {row}" # --------------------------------------------------------------------------- # @@ -368,30 +355,36 @@ def test_whatif_row_emits_category_whatif(self, tmp_path): with open(top_json, "r") as fh: rows = json.load(fh) - # D-29: at least one row must have category='whatif'. - whatif_rows = [r for r in rows if r.get('category') == 'whatif'] + # D-29: the whatif category surfaces in the fixed-schema Division + # column (category.upper()). Division is the emitted successor to + # the machine-key 'category' prefix column. + whatif_rows = [r for r in rows if r.get('Division') == 'WHATIF'] assert whatif_rows, ( - f"D-29 violated: expected at least one row with category='whatif' " + f"D-29 violated: expected at least one row with Division='WHATIF' " f"in {top_json}. Got rows: {rows}" ) - # D-29 also mandates whatif SKIPS the rules-strict INVALID - # gates. The whatif fixture has 3 runs (not 6) — if the gates - # fired, the D-24 training count-mismatch template would land - # in the issues column. Assert it does NOT. + # D-29 also mandates whatif SKIPS the rules-strict INVALID gates. + # The 'issues' column is no longer emitted (column-parity), so + # assert the skip at the internal Result layer instead: no whatif + # workload Result carries a D-24 INVALID substring. d24_substrings = [ "expected 6 training invocations per Rules.md", "expected exactly 1 warmup invocation to be detected", "expected 10 checkpoint operations per Rules.md", "cannot aggregate", ] - for row in whatif_rows: - issues_val = row.get('issues', '') or '' + for wk, wr in gen.workload_results.items(): + cat = getattr(wr.category, 'value', wr.category) + if str(cat) != 'whatif' and 'whatif' not in str(wk): + continue + issue_text = '; '.join( + str(getattr(i, 'message', i)) for i in (wr.issues or [])) for sub in d24_substrings: - assert sub not in issues_val, ( - f"D-29 violated: whatif row contains D-24 INVALID substring " - f"{sub!r} — whatif MUST skip the rules-strict gates. " - f"Row: {row}" + assert sub not in issue_text, ( + f"D-29 violated: whatif workload {wk} carries D-24 INVALID " + f"substring {sub!r} — whatif MUST skip the rules-strict " + f"gates. Issues: {issue_text}" ) @@ -472,13 +465,13 @@ def test_multi_orgname_produces_single_top_level_file_with_mixed_rows( assert len(beta_rows) == 1, ( f"expected 1 beta_corp workload row in per-model file, got {len(beta_rows)}" ) - assert acme_rows[0].get('orgname') == 'acme', ( - f"expected acme orgname in acme's per-model row, got " - f"{acme_rows[0].get('orgname')!r}" + assert acme_rows[0].get('Organization') == 'acme', ( + f"expected acme in acme's per-model row Organization, got " + f"{acme_rows[0].get('Organization')!r}" ) - assert beta_rows[0].get('orgname') == 'beta_corp', ( - f"expected beta_corp orgname in beta's per-model row, got " - f"{beta_rows[0].get('orgname')!r}" + assert beta_rows[0].get('Organization') == 'beta_corp', ( + f"expected beta_corp in beta's per-model row Organization, got " + f"{beta_rows[0].get('Organization')!r}" ) # D-08 core assertion: the top-level file exists and its rows @@ -507,17 +500,17 @@ def test_multi_orgname_produces_single_top_level_file_with_mixed_rows( # acme OR beta_corp — never blank/other). This still catches # any regression that would COLLAPSE cross-org rows or drop # the orgname column. - top_orgnames = {r.get('orgname') for r in top_rows} + # The fixed-schema 'Organization' column is the emitted successor + # to the machine-key 'orgname' — it distinguishes cross-org rows. + top_orgnames = {r.get('Organization') for r in top_rows} assert top_orgnames, ( f"top-level results.json is empty; expected at least one org's rows" ) - # Every row must carry a real orgname (not empty) — D-08 - # invariant that the orgname column distinguishes rows. assert '' not in top_orgnames, ( - f"D-08 violated: top-level rows include empty orgname. " - f"Orgnames seen: {top_orgnames}" + f"D-08 violated: top-level rows include empty Organization. " + f"Organizations seen: {top_orgnames}" ) assert top_orgnames.issubset({'acme', 'beta_corp'}), ( - f"D-08 violated: top-level rows include unexpected orgnames " + f"D-08 violated: top-level rows include unexpected Organizations " f"{top_orgnames - {'acme', 'beta_corp'}}" ) diff --git a/tests/unit/test_reportgen_sut_block.py b/tests/unit/test_reportgen_sut_block.py index 49092ca2..7998563c 100644 --- a/tests/unit/test_reportgen_sut_block.py +++ b/tests/unit/test_reportgen_sut_block.py @@ -85,21 +85,22 @@ def test_json_sut_columns_populated(self, tmp_path): 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, + assert row["Organization"] == "acme" + assert row["RU's"] == _RACK_UNITS + # Column-parity: Name shows the system name and links to the .yaml; + # Description shows literal "PDF" and links to the .pdf. + assert row["Name"] == { + "text": "system-a", "href": "closed/acme/systems/system-a.yaml", } - assert row["sut_description"] == { - "text": _SUBMISSION_NAME, + assert row["Description"] == { + "text": "PDF", "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"): + for blank in ("Public ID", "Type", "Access Protocol", + "Availability", "Integrated Client Storage", + "Usable Capacity (TiB)"): assert row[blank] == "", f"{blank} should be blank, got {row[blank]!r}" @@ -111,14 +112,14 @@ def test_csv_hyperlinks_render_as_html_anchor(self, tmp_path): rows = list(csv.DictReader(fh)) assert len(rows) == 1 row = rows[0] - assert row["sut_name"] == ( - f'{_SUBMISSION_NAME}' + assert row["Name"] == ( + 'system-a' ) - assert row["sut_description"] == ( - f'{_SUBMISSION_NAME}' + assert row["Description"] == ( + 'PDF' ) - assert row["sut_organization"] == "acme" - assert row["sut_rus"] == str(_RACK_UNITS) + assert row["Organization"] == "acme" + assert row["RU's"] == str(_RACK_UNITS) _CODE_HASH = "0123456789abcdef0123456789abcdef" # md5-tree-v2 32-hex @@ -135,15 +136,17 @@ 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): + # Column-parity: Code/Logs are per-workload. This is a training + # fixture, so they land in the Training block. 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} + assert row["Training - 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} + assert row["Training - Logs"] == {"text": "logs", "href": expected_href} def test_csv_code_logs_anchor(self, tmp_path): root = _prepare_tree(tmp_path) @@ -152,8 +155,8 @@ def test_csv_code_logs_anchor(self, tmp_path): 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' + assert row["Training - Code"] == f'code' + assert row["Training - Logs"] == f'logs' def test_missing_pointer_leaves_code_logs_blank(self, tmp_path): # No pointer injected → graceful blank, no crash. @@ -161,22 +164,22 @@ def test_missing_pointer_leaves_code_logs_blank(self, 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"] == "" + assert row["Training - Code"] == "" + assert row["Training - Logs"] == "" class TestSutBlockDoesNotBreakColumnInvariants: - def test_prefix_first_issues_last_sut_in_between(self, tmp_path): + def test_fixed_schema_left_edge_and_sut_columns(self, tmp_path): + from mlpstorage_py.report_generator import _FINAL_SCHEMA 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 + # Column-parity: the emitted header is EXACTLY the fixed schema. + assert header == _FINAL_SCHEMA, f"header != fixed schema: {header}" + # SUT columns present in the left-edge block; no machine-key + # prefix or trailing issues column. + for col in ("Organization", "Name", "Description", "RU's"): + assert col in header + for banned in ("orgname", "systemname", "benchmark_type", "issues"): + assert banned not in header diff --git a/tests/unit/test_reporting.py b/tests/unit/test_reporting.py index 3ae3d40a..17f1e382 100755 --- a/tests/unit/test_reporting.py +++ b/tests/unit/test_reporting.py @@ -118,10 +118,17 @@ def generator(self, tmp_path): return ReportGenerator(str(results_dir), validate_structure=False) def test_writes_json_file(self, generator): - """Should write results to JSON file.""" + """Should write results to JSON file as fixed-schema rows. + + Column-parity: write_json_file projects each internal row onto the + fixed webpage-parity schema. Every emitted row carries exactly the + _FINAL_SCHEMA keys; internal-only keys (run_id, model) are dropped + unless they map to a fixed column. + """ + from mlpstorage_py.report_generator import _FINAL_SCHEMA results = [ - {'run_id': 'run1', 'model': 'unet3d'}, - {'run_id': 'run2', 'model': 'resnet50'} + {'benchmark_type': 'training', 'orgname': 'acme'}, + {'benchmark_type': 'vector_database', 'orgname': 'beta'}, ] generator.write_json_file(results) @@ -132,7 +139,10 @@ def test_writes_json_file(self, generator): loaded = json.load(f) assert len(loaded) == 2 - assert loaded[0]['run_id'] == 'run1' + assert list(loaded[0].keys()) == _FINAL_SCHEMA + assert loaded[0]['Benchmark Type'] == 'training' + assert loaded[0]['Organization'] == 'acme' + assert 'run_id' not in loaded[0] def test_json_has_proper_formatting(self, generator): """JSON should be properly formatted with indent.""" @@ -177,8 +187,15 @@ def test_writes_csv_file(self, generator): assert len(rows) == 2 - def test_flattens_nested_dicts(self, generator): - """Should flatten nested dictionaries.""" + def test_projects_to_fixed_schema_dropping_unknown_keys(self, generator): + """Column-parity: the writer projects onto the fixed schema. + + Supersedes the old nested-dict flattening behavior. Unknown / + internal-only keys (including nested dicts) are NOT emitted — the + header is exactly the fixed webpage-parity schema regardless of + input row shape. + """ + from mlpstorage_py.report_generator import _FINAL_SCHEMA results = [ {'run_id': 'run1', 'metrics': {'throughput': 100.0, 'au': 95.0}} ] @@ -186,11 +203,11 @@ def test_flattens_nested_dicts(self, generator): csv_file = os.path.join(generator.results_dir, 'results.csv') with open(csv_file, 'r') as f: - reader = csv.DictReader(f) - rows = list(reader) + header = next(csv.reader(f)) - # Should have flattened keys - assert 'metrics.throughput' in rows[0] or 'throughput' in rows[0] + assert header == _FINAL_SCHEMA + for banned in ('metrics', 'metrics.throughput', 'throughput', 'run_id'): + assert banned not in header def test_handles_nan_values(self, generator): """Should remove NaN values."""