35d2bdadbd
- revise hardware prices and parameters in extrapolate.py - enhance scoring details in README for task 11 - clarify projected scoreboard values and sources - improve chart readability and scoring methodology in report.py - add new utility functions for formatting and reading hardware prices
687 lines
30 KiB
Python
687 lines
30 KiB
Python
"""Task 11 - Reporting: tables, charts, weighted scoring, REPORT.md.
|
|
|
|
Aggregates the measured matrix (task 09), the projections (task 10) and the
|
|
process diagrams (task 02) into the final decision document. Writes 14 PNG
|
|
charts (150 dpi) under ./out/report/charts/ with the exact names fixed in
|
|
README.md, five CSV tables under ./out/report/tables/, and
|
|
./out/report/REPORT.md.
|
|
|
|
Chart readability contract (owner feedback 2026-07-11): linear scales with
|
|
actual units everywhere (panels instead of log axes when magnitudes clash),
|
|
no scientific tick notation, full format names, a value label on every bar
|
|
and point, titles that state the takeaway. Hardware costs are a discrete
|
|
bill of materials (whole RDIMM modules, whole NVMe drives, priced chassis)
|
|
with street-price sources and an as-of date from out/config/hw_prices.yaml.
|
|
|
|
Weighted scoring: search speed x5 (0-50), RAM economy x2 (0-20), disk
|
|
economy x1 (0-10), max 80; each subscore is 10 x best/value; a format with
|
|
a FAIL projection at 600 GB scores 0 on search there (spec 11). Projected
|
|
numbers are always labeled as projected (bench-methodology section 7).
|
|
Spec: docs/specs/11_reporting.md.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import logging
|
|
import math
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import matplotlib
|
|
|
|
matplotlib.use("Agg")
|
|
import matplotlib.pyplot as plt
|
|
import yaml
|
|
from matplotlib.ticker import FuncFormatter
|
|
|
|
from common.pipeline import (
|
|
check_dependencies,
|
|
parse_task_args,
|
|
remove_stale_marker,
|
|
write_marker,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TASK_ID = "11"
|
|
DEPENDS_ON = ["02", "09", "10"]
|
|
|
|
DPI = 150 # chart resolution, spec 11
|
|
FORMATS = ("csv", "json", "sqlite", "pg", "rdf")
|
|
QUERY_NUMBERS = tuple(range(1, 8))
|
|
SCALES_GB = (600, 1200, 6000)
|
|
REPORT_SCALE_GB = 600 # the headline projection scale, spec 11
|
|
SCALE_LABELS = {600: "600 GB", 1200: "1.2 TB", 6000: "6 TB"}
|
|
|
|
WEIGHT_SEARCH = 5 # x5 -> 0-50, spec 11
|
|
WEIGHT_RAM = 2 # x2 -> 0-20
|
|
WEIGHT_DISK = 1 # x1 -> 0-10
|
|
|
|
ARCHIVAL_VARIANT = {
|
|
"csv": ("csv", "targz"), "json": ("json", "targz"), "sqlite": ("sqlite", "db_gz"),
|
|
"pg": ("pg", "dump"), "rdf": ("rdf", "nt_gz"),
|
|
}
|
|
WORKING_VARIANT = {
|
|
"csv": ("csv", "tree"), "json": ("json", "full"), "sqlite": ("sqlite", "db"),
|
|
"pg": ("pg", "live"), "rdf": ("rdf", "store"),
|
|
}
|
|
|
|
# query classes for the degradation charts (C4), spec 11
|
|
QUERY_CLASSES = {
|
|
"point_read": (1,),
|
|
"indexed": (2, 3, 4, 6),
|
|
"full_scan": (5, 7),
|
|
}
|
|
CLASS_LABELS = {"point_read": "point read", "indexed": "indexed / join", "full_scan": "full scan"}
|
|
|
|
QUERY_LABELS = {
|
|
1: "single-track COF curve", 2: "coupons at Au 10+/-0.5 wt%",
|
|
3: "hardness vs steady-state COF", 4: "anomaly filter",
|
|
5: "run-in over raw cycles", 6: "wear volume vs load", 7: "Stribeck aggregation",
|
|
}
|
|
|
|
DIAGRAM_TITLES = {
|
|
"D1": "Coupon assembly", "D2": "Characterization and batch assembly",
|
|
"D3": "Testing tree", "D4": "Tribometer session sequence",
|
|
"D5": "Execution loop", "D6": "Data hierarchy",
|
|
}
|
|
|
|
FMT_COLORS = {"csv": "tab:gray", "json": "tab:orange", "sqlite": "tab:green", "pg": "tab:blue", "rdf": "tab:red"}
|
|
FMT_NAMES = {"csv": "CSV", "json": "JSON-LD", "sqlite": "SQLite", "pg": "PostgreSQL", "rdf": "RDF triplestore"}
|
|
CHASSIS = {16: "Supermicro AS-1015CS-TNR", 32: "Supermicro AS-1115CS-TNR"}
|
|
|
|
|
|
def read_medians(path: Path) -> dict[tuple[str, int], dict]:
|
|
cells: dict[tuple[str, int], dict] = {}
|
|
with open(path, encoding="ascii", newline="") as fh:
|
|
for row in csv.DictReader(fh):
|
|
if row["status"] != "OK" or row["checksum_ok"] != "true":
|
|
raise ValueError(f"{path}: cell {row['format']}/{row['query']} is not a valid measurement")
|
|
cells[(row["format"], int(row["query"][1:]))] = {
|
|
"wall_s": float(row["wall_s_median"]),
|
|
"peak_rss_mb": float(row["peak_rss_mb_median"]),
|
|
"cpu_util_pct": float(row["cpu_util_pct_median"]),
|
|
"read_mb": float(row["read_mb_median"]),
|
|
}
|
|
if len(cells) != len(FORMATS) * len(QUERY_NUMBERS):
|
|
raise ValueError(f"{path}: {len(cells)} cells, expected the full matrix")
|
|
return cells
|
|
|
|
|
|
def read_projections(path: Path) -> dict[tuple[str, int, int], tuple[float, str]]:
|
|
proj: dict[tuple[str, int, int], tuple[float, str]] = {}
|
|
with open(path, encoding="ascii", newline="") as fh:
|
|
for row in csv.DictReader(fh):
|
|
proj[(row["format"], int(row["query"][1:]), int(row["scale_gb"]))] = (
|
|
float(row["projected_wall_s"]), row["flag"])
|
|
if len(proj) != len(FORMATS) * len(QUERY_NUMBERS) * len(SCALES_GB):
|
|
raise ValueError(f"{path}: incomplete projection table")
|
|
return proj
|
|
|
|
|
|
def read_sizes(path: Path) -> dict[tuple[str, str], int]:
|
|
with open(path, encoding="ascii", newline="") as fh:
|
|
reader = csv.reader(fh)
|
|
next(reader)
|
|
return {(fmt, variant): int(nbytes) for fmt, variant, nbytes in reader}
|
|
|
|
|
|
def read_sizing(path: Path) -> dict[tuple[str, int], dict]:
|
|
sizing: dict[tuple[str, int], dict] = {}
|
|
with open(path, encoding="ascii", newline="") as fh:
|
|
for row in csv.DictReader(fh):
|
|
sizing[(row["format"], int(row["scale_gb"]))] = {
|
|
"disk_tb": float(row["disk_tb"]), "ram_gb": float(row["ram_gb"]),
|
|
"cores": int(row["cores"]), "nodes": int(row["nodes"]),
|
|
"est_cost_usd": float(row["est_cost_usd"]),
|
|
"ram_modules": int(row["ram_modules"]), "nvme_drives": int(row["nvme_drives"]),
|
|
"base_usd": float(row["base_usd"]), "ram_usd": float(row["ram_usd"]),
|
|
"disk_usd": float(row["disk_usd"]), "extra_nodes_usd": float(row["extra_nodes_usd"]),
|
|
}
|
|
return sizing
|
|
|
|
|
|
def read_hw_prices(path: Path) -> dict:
|
|
with open(path, encoding="ascii") as fh:
|
|
return yaml.safe_load(fh)
|
|
|
|
|
|
def geometric_mean(values: list[float]) -> float:
|
|
return math.exp(sum(math.log(v) for v in values) / len(values))
|
|
|
|
|
|
def md_table(headers: list[str], rows: list[list]) -> str:
|
|
lines = ["| " + " | ".join(headers) + " |", "|" + "---|" * len(headers)]
|
|
lines += ["| " + " | ".join(str(c) for c in row) + " |" for row in rows]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def human_time(seconds: float) -> str:
|
|
if seconds < 120:
|
|
return f"{seconds:.2f} s" if seconds < 10 else f"{seconds:.0f} s"
|
|
if seconds < 2 * 3600:
|
|
return f"{seconds / 60:.1f} min"
|
|
if seconds < 2 * 86400:
|
|
return f"{seconds / 3600:.1f} h"
|
|
return f"{seconds / 86400:.1f} d"
|
|
|
|
|
|
def human_size(nbytes: float) -> str:
|
|
for unit, div in (("TB", 1e12), ("GB", 1e9), ("MB", 1e6)):
|
|
if nbytes >= div:
|
|
return f"{nbytes / div:.2f} {unit}" if nbytes < 10 * div else f"{nbytes / div:.0f} {unit}"
|
|
return f"{nbytes / 1e3:.0f} KB"
|
|
|
|
|
|
def money(value: float) -> str:
|
|
return f"${value:,.0f}"
|
|
|
|
|
|
def write_csv(path: Path, headers: list[str], rows: list[list]) -> None:
|
|
with open(path, "w", encoding="ascii", newline="") as fh:
|
|
writer = csv.writer(fh)
|
|
writer.writerow(headers)
|
|
writer.writerows(rows)
|
|
|
|
|
|
# --- weighted scoring -----------------------------------------------------
|
|
|
|
def scores_for(metrics: dict[str, dict[str, float]], failed_search: set[str]) -> dict[str, dict]:
|
|
"""10 x best/value per metric; search zeroed for formats in failed_search."""
|
|
best = {key: min(m[key] for m in metrics.values()) for key in ("search", "ram", "disk")}
|
|
out: dict[str, dict] = {}
|
|
for fmt, m in metrics.items():
|
|
search = 0.0 if fmt in failed_search else 10.0 * best["search"] / m["search"]
|
|
ram = 10.0 * best["ram"] / m["ram"]
|
|
disk = 10.0 * best["disk"] / m["disk"]
|
|
out[fmt] = {
|
|
"search_pts": round(search * WEIGHT_SEARCH, 1),
|
|
"ram_pts": round(ram * WEIGHT_RAM, 1),
|
|
"disk_pts": round(disk * WEIGHT_DISK, 1),
|
|
"total": round(search * WEIGHT_SEARCH + ram * WEIGHT_RAM + disk * WEIGHT_DISK, 1),
|
|
}
|
|
return out
|
|
|
|
|
|
def build_scoreboards(medians, proj, sizes, sizing) -> dict[str, dict[str, dict]]:
|
|
measured_metrics = {
|
|
fmt: {
|
|
"search": geometric_mean([medians[(fmt, qn)]["wall_s"] for qn in QUERY_NUMBERS]),
|
|
"ram": max(medians[(fmt, qn)]["peak_rss_mb"] for qn in QUERY_NUMBERS),
|
|
"disk": float(sizes[WORKING_VARIANT[fmt]]),
|
|
}
|
|
for fmt in FORMATS
|
|
}
|
|
projected_metrics = {
|
|
fmt: {
|
|
"search": geometric_mean([proj[(fmt, qn, REPORT_SCALE_GB)][0] for qn in QUERY_NUMBERS]),
|
|
"ram": sizing[(fmt, REPORT_SCALE_GB)]["ram_gb"],
|
|
"disk": sizing[(fmt, REPORT_SCALE_GB)]["disk_tb"],
|
|
}
|
|
for fmt in FORMATS
|
|
}
|
|
failed = {fmt for fmt in FORMATS if any(proj[(fmt, qn, REPORT_SCALE_GB)][1] == "FAIL" for qn in QUERY_NUMBERS)}
|
|
return {
|
|
"measured": scores_for(measured_metrics, set()),
|
|
"projected_600gb": scores_for(projected_metrics, failed),
|
|
}
|
|
|
|
|
|
# --- charts ---------------------------------------------------------------
|
|
# Shared conventions: linear axes, real units, a value label on every bar
|
|
# and point, full format names, no scientific notation.
|
|
|
|
def _vbar_panel(ax, values: dict[str, float], labels: dict[str, str], unit: str, title: str) -> None:
|
|
bars = ax.bar(range(len(FORMATS)), [values[f] for f in FORMATS], color=[FMT_COLORS[f] for f in FORMATS])
|
|
for bar, fmt in zip(bars, FORMATS):
|
|
ax.annotate(labels[fmt], (bar.get_x() + bar.get_width() / 2, bar.get_height()),
|
|
ha="center", va="bottom", fontsize=8)
|
|
ax.set_xticks(range(len(FORMATS)))
|
|
ax.set_xticklabels([FMT_NAMES[f] for f in FORMATS], rotation=12, fontsize=8)
|
|
ax.set_ylabel(unit)
|
|
ax.set_title(title, fontsize=10)
|
|
ax.margins(y=0.18)
|
|
ax.grid(axis="y", alpha=0.25)
|
|
|
|
|
|
def _hbar_panel(ax, values: dict[str, float], labels: dict[str, str], title: str, names: bool) -> None:
|
|
y = range(len(FORMATS))
|
|
bars = ax.barh(list(y), [values[f] for f in FORMATS], color=[FMT_COLORS[f] for f in FORMATS])
|
|
for bar, fmt in zip(bars, FORMATS):
|
|
ax.annotate(" " + labels[fmt], (bar.get_width(), bar.get_y() + bar.get_height() / 2),
|
|
va="center", fontsize=8)
|
|
ax.set_yticks(list(y))
|
|
ax.set_yticklabels([FMT_NAMES[f] for f in FORMATS] if names else [""] * len(FORMATS), fontsize=8)
|
|
ax.invert_yaxis()
|
|
ax.set_xticks([]) # linear axis; the value labels on the bars carry the numbers
|
|
ax.margins(x=0.30)
|
|
ax.set_title(title, fontsize=10)
|
|
|
|
|
|
def chart_c1(charts_dir, sizes, proj_bytes, coeff) -> None:
|
|
fig, axes = plt.subplots(1, 2, figsize=(11, 4.5))
|
|
measured = {f: sizes[WORKING_VARIANT[f]] for f in FORMATS}
|
|
_vbar_panel(axes[0], {f: measured[f] / 1e9 for f in FORMATS},
|
|
{f: human_size(measured[f]) for f in FORMATS}, "GB (linear)", "measured on the 150 MB corpus")
|
|
_vbar_panel(axes[1], {f: proj_bytes[f] / 1e9 for f in FORMATS},
|
|
{f: human_size(proj_bytes[f]) for f in FORMATS}, "GB (linear)", "projected at 600 GB of raw data")
|
|
fig.suptitle(f"C1 - disk footprint: RDF stores the same data in {coeff['rdf']:.0f}x the raw CSV size, SQLite in {coeff['sqlite']:.2f}x")
|
|
fig.tight_layout()
|
|
fig.savefig(charts_dir / "c1_disk_footprint.png", dpi=DPI)
|
|
plt.close(fig)
|
|
|
|
|
|
def chart_c2(charts_dir, medians, sizing) -> None:
|
|
fig, axes = plt.subplots(1, 2, figsize=(11, 4.5))
|
|
rss = {f: max(medians[(f, qn)]["peak_rss_mb"] for qn in QUERY_NUMBERS) for f in FORMATS}
|
|
ram = {f: sizing[(f, REPORT_SCALE_GB)]["ram_gb"] for f in FORMATS}
|
|
_vbar_panel(axes[0], rss, {f: f"{rss[f]:.0f} MB" for f in FORMATS},
|
|
"MB (linear)", "measured client peak RSS on 150 MB")
|
|
_vbar_panel(axes[1], ram, {f: f"{ram[f]:.0f} GB" for f in FORMATS},
|
|
"GB (linear)", "recommended server RAM at 600 GB (projected)")
|
|
axes[1].axhline(1024, color="gray", linestyle="--", linewidth=1)
|
|
axes[1].annotate("single-node ceiling: 1 TB RAM", (0, 1024), fontsize=8, va="bottom")
|
|
fig.suptitle("C2 - RAM: only the RDF hot set outgrows a single 1 TB node at 600 GB")
|
|
fig.tight_layout()
|
|
fig.savefig(charts_dir / "c2_ram_requirement.png", dpi=DPI)
|
|
plt.close(fig)
|
|
|
|
|
|
def chart_c3(charts_dir, medians, proj) -> None:
|
|
for qn in QUERY_NUMBERS:
|
|
fig, axes = plt.subplots(1, 2, figsize=(11, 3.6))
|
|
measured = {f: medians[(f, qn)]["wall_s"] for f in FORMATS}
|
|
projected = {f: proj[(f, qn, REPORT_SCALE_GB)][0] for f in FORMATS}
|
|
_hbar_panel(axes[0], measured, {f: human_time(measured[f]) for f in FORMATS},
|
|
"measured on 150 MB", names=True)
|
|
_hbar_panel(axes[1], projected, {f: human_time(projected[f]) for f in FORMATS},
|
|
"projected at 600 GB", names=False)
|
|
fastest = min(FORMATS, key=lambda f: projected[f])
|
|
slowest = max(FORMATS, key=lambda f: projected[f])
|
|
fig.suptitle(f"C3 - Q{qn} ({QUERY_LABELS[qn]}) at 600 GB: {FMT_NAMES[fastest]} {human_time(projected[fastest])}"
|
|
f" vs {FMT_NAMES[slowest]} {human_time(projected[slowest])}")
|
|
fig.tight_layout()
|
|
fig.savefig(charts_dir / f"c3_q{qn}_wall_time.png", dpi=DPI)
|
|
plt.close(fig)
|
|
|
|
|
|
def chart_c4(charts_dir, medians, proj, measured_label: str) -> None:
|
|
for cls, queries in QUERY_CLASSES.items():
|
|
fig, axes = plt.subplots(1, 1 + len(SCALES_GB), figsize=(13, 3.6))
|
|
measured = {f: geometric_mean([medians[(f, qn)]["wall_s"] for qn in queries]) for f in FORMATS}
|
|
_hbar_panel(axes[0], measured, {f: human_time(measured[f]) for f in FORMATS}, measured_label, names=True)
|
|
for ax, scale in zip(axes[1:], SCALES_GB):
|
|
values = {f: geometric_mean([proj[(f, qn, scale)][0] for qn in queries]) for f in FORMATS}
|
|
_hbar_panel(ax, values, {f: human_time(values[f]) for f in FORMATS},
|
|
f"projected {SCALE_LABELS[scale]}", names=False)
|
|
queries_label = ", ".join(f"Q{qn}" for qn in queries)
|
|
fig.suptitle(f"C4 - {CLASS_LABELS[cls]} queries ({queries_label}): wall time growth by data size"
|
|
" (every panel after the first is projected; linear scale within each panel)")
|
|
fig.tight_layout()
|
|
fig.savefig(charts_dir / f"c4_degradation_{cls}.png", dpi=DPI)
|
|
plt.close(fig)
|
|
|
|
|
|
def chart_c5(charts_dir, scoreboards, winner: str) -> None:
|
|
fig, axes = plt.subplots(1, 2, figsize=(11, 4.5), sharey=True)
|
|
for ax, (title, board) in zip(axes, [
|
|
("measured scale (150 MB)", scoreboards["measured"]),
|
|
("projected at 600 GB", scoreboards["projected_600gb"]),
|
|
]):
|
|
x = range(len(FORMATS))
|
|
search = [board[fmt]["search_pts"] for fmt in FORMATS]
|
|
ram = [board[fmt]["ram_pts"] for fmt in FORMATS]
|
|
disk = [board[fmt]["disk_pts"] for fmt in FORMATS]
|
|
ax.bar(x, search, label=f"search speed x{WEIGHT_SEARCH} (0-50)")
|
|
ax.bar(x, ram, bottom=search, label=f"RAM economy x{WEIGHT_RAM} (0-20)")
|
|
ax.bar(x, disk, bottom=[s + r for s, r in zip(search, ram)], label=f"disk economy x{WEIGHT_DISK} (0-10)")
|
|
for i, fmt in enumerate(FORMATS):
|
|
ax.text(i, board[fmt]["total"] + 1.5, f"{board[fmt]['total']:.0f}", ha="center", fontsize=9)
|
|
ax.set_xticks(list(x))
|
|
ax.set_xticklabels([FMT_NAMES[f] for f in FORMATS], rotation=12, fontsize=8)
|
|
ax.set_ylim(0, 85)
|
|
ax.set_title(title, fontsize=10)
|
|
ax.grid(axis="y", alpha=0.25)
|
|
axes[0].set_ylabel("weighted score (linear, max 80)")
|
|
axes[0].legend(loc="upper left", fontsize=8)
|
|
board600 = scoreboards["projected_600gb"]
|
|
fig.suptitle(f"C5 - weighted score (max 80): {FMT_NAMES[winner]} leads the 600 GB scoreboard with {board600[winner]['total']:.0f} points")
|
|
fig.tight_layout()
|
|
fig.savefig(charts_dir / "c5_weighted_score.png", dpi=DPI)
|
|
plt.close(fig)
|
|
|
|
|
|
def chart_c6(charts_dir, proj, sizing) -> None:
|
|
fig, ax = plt.subplots(figsize=(9, 5.5))
|
|
offsets = {"csv": (12, 0), "json": (12, -4), "sqlite": (12, 10), "pg": (12, -16), "rdf": (-40, -20)}
|
|
geo = {f: geometric_mean([proj[(f, qn, REPORT_SCALE_GB)][0] for qn in QUERY_NUMBERS]) for f in FORMATS}
|
|
cost = {f: sizing[(f, REPORT_SCALE_GB)]["est_cost_usd"] for f in FORMATS}
|
|
for fmt in FORMATS:
|
|
ax.scatter(cost[fmt], geo[fmt] / 60, s=70, color=FMT_COLORS[fmt])
|
|
ax.annotate(f"{FMT_NAMES[fmt]}: {money(cost[fmt])}, {human_time(geo[fmt])}",
|
|
(cost[fmt], geo[fmt] / 60), textcoords="offset points", xytext=offsets[fmt], fontsize=8)
|
|
ax.xaxis.set_major_formatter(FuncFormatter(lambda v, _: f"${v / 1000:.0f}k"))
|
|
ax.yaxis.set_major_formatter(FuncFormatter(lambda v, _: f"{v:.0f} min"))
|
|
ax.set_xlabel("estimated hardware cost at 600 GB (bill of materials, linear)")
|
|
ax.set_ylabel("typical query time at 600 GB, projected (geometric mean Q1-Q7, linear)")
|
|
ax.annotate("better: lower-left", xy=(0.05, 0.04), xytext=(0.42, 0.10),
|
|
xycoords="axes fraction", textcoords="axes fraction",
|
|
fontsize=10, color="darkgreen", fontweight="bold",
|
|
arrowprops={"arrowstyle": "->", "color": "darkgreen"})
|
|
ax.margins(x=0.18, y=0.12)
|
|
ax.grid(alpha=0.25)
|
|
fastest = min(FORMATS, key=lambda f: geo[f])
|
|
ax.set_title(f"C6 - cost vs speed at 600 GB (projected): {FMT_NAMES[fastest]} is the fastest at {money(cost[fastest])}")
|
|
fig.tight_layout()
|
|
fig.savefig(charts_dir / "c6_cost_vs_performance.png", dpi=DPI)
|
|
plt.close(fig)
|
|
|
|
|
|
# --- report assembly ------------------------------------------------------
|
|
|
|
def scoreboard_rows(board: dict[str, dict]) -> list[list]:
|
|
ranked = sorted(FORMATS, key=lambda f: -board[f]["total"])
|
|
return [[FMT_NAMES[fmt], board[fmt]["search_pts"], board[fmt]["ram_pts"], board[fmt]["disk_pts"], board[fmt]["total"]] for fmt in ranked]
|
|
|
|
|
|
def config_string(row: dict, prices: dict) -> str:
|
|
return (f"{row['nodes']} x 1U {CHASSIS[row['cores']]} ({row['cores']} cores);"
|
|
f" {row['ram_modules']} x {prices['ram_module_gb']} GB DDR5 ECC RDIMM;"
|
|
f" {row['nvme_drives']} x {prices['nvme_drive_tb']} TB NVMe U.2")
|
|
|
|
|
|
def assemble_report(tables: dict[str, tuple[list[str], list[list]]], scoreboards,
|
|
environment: str, diagrams: dict[str, str], winner: str, sizing, prices: dict) -> str:
|
|
t1_h, t1_r = tables["t1_storage_footprint"]
|
|
t2_h, t2_r = tables["t2_measured_medians"]
|
|
t3_h, t3_r = tables["t3_projected_600gb"]
|
|
t4_h, t4_r = tables["t4_hardware_600gb"]
|
|
pg6 = sizing[("pg", 6000)]
|
|
rdf6 = sizing[("rdf", 6000)]
|
|
as_of = prices["as_of"]
|
|
sources = prices["sources"]
|
|
|
|
parts = [
|
|
"# Laboratory Data Storage Format Evaluation - Final Report",
|
|
"",
|
|
"Generated by task 11 (report.py) from the measured benchmark matrix"
|
|
" (task 09) and the extrapolation model (task 10). Every number at"
|
|
" scales beyond the 150 MB corpus is a PROJECTION and is labeled as"
|
|
" such; scaling assumptions are listed in section 6.",
|
|
"",
|
|
"## 1. Executive summary",
|
|
"",
|
|
f"**Recommendation: {FMT_NAMES[winner]} as the system of record.** The measured"
|
|
" matrix and the projections confirm the expected outcome:",
|
|
"",
|
|
"- **PostgreSQL** is the only format whose seven benchmark queries stay"
|
|
" below the 1-hour practicality bound at every projected scale (600 GB,"
|
|
f" 1.2 TB, 6 TB); at 6 TB it needs one {pg6['cores']}-core node with"
|
|
f" {pg6['ram_gb']:.0f} GB RAM at an estimated {money(pg6['est_cost_usd'])}"
|
|
f" (bill of materials, prices as of {as_of}). Partitioned parallel scans"
|
|
" and the track_summary materialized view are the scaling levers.",
|
|
"- **CSV** remains the canonical raw archive: byte-reproducible,"
|
|
" instrument-native, and the best compressor (tar.gz ~2.8x); its full"
|
|
" scans exceed 1 h beyond ~600 GB, so it is an archive, not a query"
|
|
" layer.",
|
|
"- **SQLite** is an excellent single-user store up to ~600 GB (all"
|
|
" queries OK), then its single-threaded scans degrade (Q5/Q7"
|
|
" IMPRACTICAL at 1.2 TB).",
|
|
"- **Hybrid JSON-LD** (metadata + sourceFile links, 1.8 MiB at 150 MB"
|
|
" corpus) is the exchange format; the FULL variant re-reads the whole"
|
|
" corpus per scan and becomes impractical past 600 GB.",
|
|
"- **RDF (materialized triplestore)** fails at scale: 11.7x storage"
|
|
" blow-up, 5 of 7 queries FAIL at 6 TB, and the hot set needs"
|
|
f" {rdf6['ram_gb'] / 1024:.1f} TB RAM across {rdf6['nodes']} nodes"
|
|
f" (~{money(rdf6['est_cost_usd'])} projected). Semantic access should be"
|
|
" a virtual layer (e.g. Ontop OBDA) over PostgreSQL instead.",
|
|
"",
|
|
"## 2. Methodology snapshot",
|
|
"",
|
|
"35 cells (Q1-Q7 x 5 formats); per cell 1 warm-up + 3 measured runs in"
|
|
" isolated subprocesses sampled with psutil at 50 ms; randomized cell"
|
|
" order; 30-min timeout; every run validated against canonical results"
|
|
" (PostgreSQL cross-validated with SQLite, tolerance 1e-9). Medians are"
|
|
" reported; raw runs and min/max live in out/bench/results_raw.csv."
|
|
" The cold-cache pass is skipped on this Windows host and marked as"
|
|
" such. PostgreSQL runs on a remote host; its wall times include the"
|
|
" LAN round-trip, and server-side metrics come from pg_stat_statements"
|
|
" and Prometheus (docs/rules/bench-methodology.md section 6).",
|
|
"",
|
|
"## 3. Environment",
|
|
"",
|
|
"```",
|
|
environment.rstrip(),
|
|
"```",
|
|
"",
|
|
"## 4. Storage footprint",
|
|
"",
|
|
md_table(t1_h, t1_r),
|
|
"",
|
|
"",
|
|
"",
|
|
"## 5. Measured performance (150 MB corpus)",
|
|
"",
|
|
md_table(t2_h, t2_r),
|
|
"",
|
|
"Sub-interval note: cells faster than the 50 ms sampling cadence"
|
|
" (fast SQLite/CSV point reads) under-report client RSS/CPU; wall"
|
|
" times are exact.",
|
|
"",
|
|
"",
|
|
"",
|
|
"Per-query charts: " + ", ".join(f"[Q{qn}](charts/c3_q{qn}_wall_time.png)" for qn in QUERY_NUMBERS),
|
|
"",
|
|
"## 6. Projections (600 GB / 1.2 TB / 6 TB)",
|
|
"",
|
|
"All values in this section are projected, never measured. Laws per"
|
|
" access pattern (spec 10): flat / index-depth log / linear result set"
|
|
" / linear x depth / full scan / parallel partitioned scan; constants"
|
|
" calibrated on the measured point after subtracting the per-format"
|
|
" harness floor. Flags: IMPRACTICAL > 1 h, FAIL > 24 h.",
|
|
"",
|
|
"Projected wall times at 600 GB:",
|
|
"",
|
|
md_table(t3_h, t3_r),
|
|
"",
|
|
"Degradation by data size: [point read](charts/c4_degradation_point_read.png),"
|
|
" [indexed / join](charts/c4_degradation_indexed.png),"
|
|
" [full scan](charts/c4_degradation_full_scan.png)",
|
|
"",
|
|
"## 7. Hardware sizing and cost at 600 GB (projected)",
|
|
"",
|
|
f"Bill-of-materials costing with street prices as of **{as_of}**"
|
|
" (sources below). Every configuration buys whole components: 64 GB"
|
|
" DDR5 ECC RDIMM modules, 7.68 TB enterprise NVMe U.2 drives, and a"
|
|
" priced 1U chassis; disk capacity includes 30% free-space headroom.",
|
|
"",
|
|
md_table(t4_h, t4_r),
|
|
"",
|
|
f"Component prices (retrieved {as_of}):",
|
|
"",
|
|
f"- {prices['ram_module_gb']} GB DDR5-5600 ECC RDIMM (A-Tech, 2x32GB kit):"
|
|
f" {money(prices['ram_module_usd'])} - <{sources['ram']}>",
|
|
f"- {prices['nvme_drive_tb']} TB NVMe U.2 1DWPD enterprise SSD (Dell 14-16G"
|
|
f" compatible, Cloud Ninjas): {money(prices['nvme_drive_usd'])} - <{sources['nvme']}>",
|
|
f"- 1U AMD EPYC 9004/9005 chassis (Broadberry starting configurations):"
|
|
f" {CHASSIS[16]} {money(prices['server_16_core_usd'])},"
|
|
f" {CHASSIS[32]} {money(prices['server_32_core_usd'])} - <{sources['servers']}>",
|
|
f"- Market context - server DRAM is in a documented 2025-2026 price"
|
|
f" surge: <{sources['market_context']}>",
|
|
"",
|
|
"Full sizing at 1.2 TB and 6 TB: out/bench/hardware_sizing.csv."
|
|
f" RDF exceeds the 1 TB single-node RAM ceiling already at 600 GB and"
|
|
f" reaches {rdf6['nodes']} nodes at 6 TB.",
|
|
"",
|
|
"",
|
|
"",
|
|
"## 8. Weighted scoring",
|
|
"",
|
|
f"Subscores are 10 x best/value per metric; weights: search x{WEIGHT_SEARCH},"
|
|
f" RAM economy x{WEIGHT_RAM}, disk economy x{WEIGHT_DISK}; max"
|
|
f" {10 * (WEIGHT_SEARCH + WEIGHT_RAM + WEIGHT_DISK)}. Search uses the"
|
|
" inverse geometric mean of Q1-Q7 (FAIL at 600 GB zeroes the search"
|
|
" subscore). Left: measured scale; right: projected 600 GB.",
|
|
"",
|
|
"Measured scale (150 MB):",
|
|
"",
|
|
md_table(["format", f"search (x{WEIGHT_SEARCH})", f"RAM (x{WEIGHT_RAM})", f"disk (x{WEIGHT_DISK})", "total (max 80)"],
|
|
scoreboard_rows(scoreboards["measured"])),
|
|
"",
|
|
"Projected at 600 GB:",
|
|
"",
|
|
md_table(["format", f"search (x{WEIGHT_SEARCH})", f"RAM (x{WEIGHT_RAM})", f"disk (x{WEIGHT_DISK})", "total (max 80)"],
|
|
scoreboard_rows(scoreboards["projected_600gb"])),
|
|
"",
|
|
"",
|
|
"",
|
|
"## 9. Use-case mapping",
|
|
"",
|
|
md_table(["Use case", "Recommended format(s)"], [
|
|
["Interactive analysis (joins, aggregations)", "PostgreSQL; SQLite acceptable single-user up to ~600 GB"],
|
|
["Report generation (repeated summaries)", "PostgreSQL (track_summary materialized view)"],
|
|
["Search / filtering", "PostgreSQL or SQLite (indexed); flat formats need full scans"],
|
|
["Archiving", "CSV tree + tar.gz (canonical raw) plus pg_dump of the system of record"],
|
|
["Inter-lab exchange", "Hybrid JSON-LD (metadata + sourceFile links to CSV)"],
|
|
["Semantic / ontology queries", "Virtual RDF layer over PostgreSQL (e.g. Ontop OBDA), not a materialized triplestore"],
|
|
]),
|
|
"",
|
|
"## 10. Limitations",
|
|
"",
|
|
"- Single-node measurements on one Windows host + one remote"
|
|
" PostgreSQL host; no cluster or cloud variance.",
|
|
"- The corpus is simulated (physically plausible models, fixed seed"
|
|
" 20260711); real instrument data may have different value"
|
|
" distributions, though volumes and shapes match the lab.",
|
|
"- Extrapolation uses analytic laws calibrated on a single 150 MB"
|
|
" point; no intermediate-scale validation runs were performed.",
|
|
"- Client wall times include a per-format harness floor (interpreter"
|
|
" start, imports, connection); the floor is assumed constant across"
|
|
" scales. The cheapest cell of each format therefore projects flat.",
|
|
"- The cold-cache matrix pass is skipped (no page-cache drop on"
|
|
" Windows); all measured runs are warm-cache.",
|
|
"- The pandas CSV variant was not measured; its RAM is O(n) and it is"
|
|
" infeasible above roughly 10 GB (spec 10).",
|
|
"- RDF numbers reflect the compact modeling and a client that derives"
|
|
" run-in / steady-state from raw cycles; SPARQL-side aggregation"
|
|
" engines could shift (not remove) the scan penalty.",
|
|
f"- Hardware prices are spot street prices retrieved {as_of} during a"
|
|
" documented server-DRAM price surge; re-cost by editing"
|
|
" out/config/hw_prices.yaml and re-running tasks 10-11.",
|
|
"",
|
|
"## Appendix A - process flow diagrams (task 02)",
|
|
"",
|
|
]
|
|
for code in sorted(diagrams):
|
|
parts += [f"### {code} - {DIAGRAM_TITLES[code]}", "", "```mermaid", diagrams[code].rstrip(), "```", ""]
|
|
return "\n".join(parts) + "\n"
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_task_args("Task 11: final report - tables, charts, weighted scoring")
|
|
out_root: Path = args.out_root
|
|
bench_dir = out_root / "bench"
|
|
report_dir = out_root / "report"
|
|
try:
|
|
check_dependencies(out_root, DEPENDS_ON)
|
|
remove_stale_marker(out_root, TASK_ID)
|
|
medians = read_medians(bench_dir / "results_median.csv")
|
|
proj = read_projections(bench_dir / "extrapolation.csv")
|
|
sizes = read_sizes(bench_dir / "storage_sizes.csv")
|
|
sizing = read_sizing(bench_dir / "hardware_sizing.csv")
|
|
prices = read_hw_prices(out_root / "config" / "hw_prices.yaml")
|
|
environment = (bench_dir / "environment.txt").read_text(encoding="utf-8")
|
|
diagrams = {p.stem: p.read_text(encoding="ascii") for p in sorted((report_dir / "diagrams").glob("D*.mermaid"))}
|
|
if len(diagrams) != 6:
|
|
raise ValueError(f"expected 6 process diagrams, found {sorted(diagrams)}")
|
|
|
|
charts_dir = report_dir / "charts"
|
|
tables_dir = report_dir / "tables"
|
|
charts_dir.mkdir(parents=True, exist_ok=True)
|
|
tables_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
csv_bytes = sizes[WORKING_VARIANT["csv"]]
|
|
coeff = {fmt: sizes[WORKING_VARIANT[fmt]] / csv_bytes for fmt in FORMATS}
|
|
proj_bytes_600 = {fmt: coeff[fmt] * REPORT_SCALE_GB * 1e9 for fmt in FORMATS}
|
|
|
|
tables: dict[str, tuple[list[str], list[list]]] = {}
|
|
tables["t1_storage_footprint"] = (
|
|
["format", "measured", "vs CSV", "archival (compressed)",
|
|
"projected @600 GB", "projected @1.2 TB", "projected @6 TB"],
|
|
[[FMT_NAMES[fmt], human_size(sizes[WORKING_VARIANT[fmt]]), f"{coeff[fmt]:.2f}x",
|
|
human_size(sizes[ARCHIVAL_VARIANT[fmt]]),
|
|
*[human_size(coeff[fmt] * s * 1e9) for s in SCALES_GB]] for fmt in FORMATS],
|
|
)
|
|
tables["t2_measured_medians"] = (
|
|
["format", "query", "wall time", "peak RSS", "CPU util", "read"],
|
|
[[FMT_NAMES[fmt], f"q{qn}", human_time(medians[(fmt, qn)]["wall_s"]),
|
|
f"{medians[(fmt, qn)]['peak_rss_mb']:.0f} MB",
|
|
f"{medians[(fmt, qn)]['cpu_util_pct']:.0f}%",
|
|
f"{medians[(fmt, qn)]['read_mb']:.1f} MB"]
|
|
for fmt in FORMATS for qn in QUERY_NUMBERS],
|
|
)
|
|
tables["t3_projected_600gb"] = (
|
|
["format", "query", "projected wall time", "flag"],
|
|
[[FMT_NAMES[fmt], f"q{qn}", human_time(proj[(fmt, qn, REPORT_SCALE_GB)][0]), proj[(fmt, qn, REPORT_SCALE_GB)][1]]
|
|
for fmt in FORMATS for qn in QUERY_NUMBERS],
|
|
)
|
|
tables["t4_hardware_600gb"] = (
|
|
["format", "configuration", "chassis", "RAM", "disk", "extra nodes", "total"],
|
|
[[FMT_NAMES[fmt], config_string(sizing[(fmt, REPORT_SCALE_GB)], prices),
|
|
money(sizing[(fmt, REPORT_SCALE_GB)]["base_usd"]),
|
|
money(sizing[(fmt, REPORT_SCALE_GB)]["ram_usd"]),
|
|
money(sizing[(fmt, REPORT_SCALE_GB)]["disk_usd"]),
|
|
money(sizing[(fmt, REPORT_SCALE_GB)]["extra_nodes_usd"]),
|
|
money(sizing[(fmt, REPORT_SCALE_GB)]["est_cost_usd"])] for fmt in FORMATS],
|
|
)
|
|
|
|
scoreboards = build_scoreboards(medians, proj, sizes, sizing)
|
|
tables["t5_weighted_scores"] = (
|
|
["scale", "format", "search_pts", "ram_pts", "disk_pts", "total"],
|
|
[[scale, FMT_NAMES[fmt], board[fmt]["search_pts"], board[fmt]["ram_pts"], board[fmt]["disk_pts"], board[fmt]["total"]]
|
|
for scale, board in scoreboards.items() for fmt in FORMATS],
|
|
)
|
|
for name, (headers, rows) in tables.items():
|
|
write_csv(tables_dir / f"{name}.csv", headers, rows)
|
|
|
|
winner = max(FORMATS, key=lambda f: scoreboards["projected_600gb"][f]["total"])
|
|
chart_c1(charts_dir, sizes, proj_bytes_600, coeff)
|
|
chart_c2(charts_dir, medians, sizing)
|
|
chart_c3(charts_dir, medians, proj)
|
|
chart_c4(charts_dir, medians, proj, f"measured {human_size(csv_bytes)}")
|
|
chart_c5(charts_dir, scoreboards, winner)
|
|
chart_c6(charts_dir, proj, sizing)
|
|
charts = sorted(p.name for p in charts_dir.glob("*.png"))
|
|
if len(charts) != 14:
|
|
raise ValueError(f"expected 14 charts, wrote {len(charts)}: {charts}")
|
|
|
|
report_text = assemble_report(tables, scoreboards, environment, diagrams, winner, sizing, prices)
|
|
report_path = report_dir / "REPORT.md"
|
|
report_path.write_text(report_text, encoding="ascii", newline="\n")
|
|
|
|
entries = {
|
|
"charts": len(charts),
|
|
"tables": len(tables),
|
|
"report_bytes": report_path.stat().st_size,
|
|
"winner_600gb": winner,
|
|
"prices_as_of": prices["as_of"],
|
|
}
|
|
for fmt in FORMATS:
|
|
entries[f"score600_{fmt}"] = scoreboards["projected_600gb"][fmt]["total"]
|
|
marker_path = write_marker(out_root, TASK_ID, entries)
|
|
except Exception:
|
|
logger.critical("task %s failed", TASK_ID, exc_info=True)
|
|
return 1
|
|
|
|
print(f"task 11 ok: REPORT.md ({entries['report_bytes']} bytes), {entries['charts']} charts, {entries['tables']} tables")
|
|
print(f"winner @600 GB (projected): {winner} - " + ", ".join(
|
|
f"{fmt}={scoreboards['projected_600gb'][fmt]['total']}" for fmt in FORMATS))
|
|
print(f"report: {report_path}")
|
|
print(f"marker: {marker_path}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|