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
296 lines
13 KiB
Python
296 lines
13 KiB
Python
"""Task 10 - Extrapolation to 600 GB / 1.2 TB / 6 TB.
|
|
|
|
Calibrates per-cell scaling laws on the measured 150 MB point (task 09
|
|
medians) and projects wall times and hardware needs per format x scale.
|
|
Model (spec 10, docs/rules/bench-methodology.md section 7):
|
|
- per-format fixed overhead = the format's cheapest measured median
|
|
(interpreter start, imports, connection); the remainder is the
|
|
data-dependent calibration constant;
|
|
- growth laws per access pattern: o1 (flat), log (index depth,
|
|
1 + ln r / ln n0 with n0 = measured friction-cycle rows), logk (result
|
|
set grows linearly with data), k_log_n (linear x index depth), scan
|
|
(linear), scan_parallel (linear x base_cores / cores_at_scale; range
|
|
partitions outnumber cores at every scale);
|
|
- RDF join penalties are already embedded in the measured constants;
|
|
- RDF q3/q5/q7 are scans because the implementation derives run-in /
|
|
steady-state from raw cycles (no summary triples exist by design).
|
|
Every number written here is a PROJECTION, never a measurement; rows are
|
|
flagged IMPRACTICAL (> 1 h) and FAIL (> 24 h).
|
|
Spec: docs/specs/10_extrapolation_model.md.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import logging
|
|
import math
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import psycopg
|
|
import yaml
|
|
|
|
from common.pg import get_dsn
|
|
from common.pipeline import (
|
|
check_dependencies,
|
|
load_lab_config,
|
|
parse_task_args,
|
|
remove_stale_marker,
|
|
write_marker,
|
|
)
|
|
from common.relational import expected_counts
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TASK_ID = "10"
|
|
DEPENDS_ON = ["09"]
|
|
|
|
SCALES_GB = (600, 1200, 6000) # raw-CSV target scales (decimal GB), spec 10
|
|
IMPRACTICAL_S = 3600.0 # projection > 1 h, spec 10
|
|
FAIL_S = 24 * 3600.0 # projection > 24 h, spec 10
|
|
DISK_HEADROOM = 1.3 # 30% free-space headroom, spec 10
|
|
CACHE_INDEX_FRACTION = 0.10 # relational hot-set: 10% of index size, spec 10
|
|
WORKING_SET_GB = 4.0 # relational session working set, spec 10 default
|
|
RDF_HOT_FRACTION = 0.20 # dictionary+index working set, 15-25% midpoint, spec 10
|
|
STREAM_RAM_GB = 16.0 # flat streaming baseline (csv/json servers)
|
|
MIN_RAM_GB = 16.0 # smallest sensible server RAM
|
|
NODE_RAM_CEILING_GB = 1024.0 # single-node RAM threshold, spec 10 RDF note
|
|
BASE_CORES = 4 # cores behind the measured PG parallel scans (task 09 host)
|
|
CORE_OPTIONS = (16, 32) # priced server tiers, hw_prices.yaml
|
|
|
|
FORMATS = ("csv", "json", "sqlite", "pg", "rdf")
|
|
QUERY_NUMBERS = tuple(range(1, 8))
|
|
|
|
# working (queryable) footprint variant per format for storage scaling
|
|
SIZE_VARIANT = {
|
|
"csv": ("csv", "tree"), "json": ("json", "full"), "sqlite": ("sqlite", "db"),
|
|
"pg": ("pg", "live"), "rdf": ("rdf", "store"),
|
|
}
|
|
|
|
# access-pattern law per (format, query) - spec 10 table, adjusted to the
|
|
# ACTUAL implementations (rdf q3/q5/q7 read raw cycles; see module docstring)
|
|
LAWS = {
|
|
("csv", 1): "o1", ("json", 1): "o1", ("sqlite", 1): "log", ("pg", 1): "log", ("rdf", 1): "log",
|
|
("csv", 2): "scan", ("json", 2): "scan", ("sqlite", 2): "logk", ("pg", 2): "logk", ("rdf", 2): "logk",
|
|
("csv", 3): "scan", ("json", 3): "scan", ("sqlite", 3): "k_log_n", ("pg", 3): "k_log_n", ("rdf", 3): "scan",
|
|
("csv", 4): "scan", ("json", 4): "scan", ("sqlite", 4): "logk", ("pg", 4): "logk", ("rdf", 4): "logk",
|
|
("csv", 5): "scan", ("json", 5): "scan", ("sqlite", 5): "scan", ("pg", 5): "scan_parallel", ("rdf", 5): "scan",
|
|
("csv", 6): "scan", ("json", 6): "scan", ("sqlite", 6): "k_log_n", ("pg", 6): "k_log_n", ("rdf", 6): "k_log_n",
|
|
("csv", 7): "scan", ("json", 7): "scan", ("sqlite", 7): "scan", ("pg", 7): "scan_parallel", ("rdf", 7): "scan",
|
|
}
|
|
|
|
HW_PRICES_DEFAULT = """# Hardware cost parameters for extrapolation (task 10).
|
|
# Street prices retrieved 2026-07-11 from the sources below (server DRAM is
|
|
# in a documented 2025-2026 price surge - see market_context). The cost
|
|
# model buys whole components: 64 GB RDIMM modules and 7.68 TB NVMe drives.
|
|
# Operator-editable; re-run extrapolate.py after changes (code-config-yaml).
|
|
as_of: "2026-07-11"
|
|
ram_module_gb: 64
|
|
ram_module_usd: 1887.0 # A-Tech 64GB (2x32GB) DDR5-5600 ECC RDIMM
|
|
nvme_drive_tb: 7.68
|
|
nvme_drive_usd: 3995.0 # Cloud Ninjas NEW 7.68TB NVMe U.2 1DWPD (Dell 14-16G)
|
|
server_16_core_usd: 4618.0 # Supermicro AS-1015CS-TNR 1U (EPYC 9004/9005), Broadberry starting config
|
|
server_32_core_usd: 6272.0 # Supermicro AS-1115CS-TNR 1U, Broadberry starting config
|
|
extra_node_usd: 4618.0 # one additional entry 1U node (AS-1015CS-TNR chassis)
|
|
sources:
|
|
ram: https://atechmemory.com/collections/ddr5-memory-ram
|
|
nvme: https://cloudninjas.com/products/new-7-68tb-nvme-u-2-1dwpd-sie-2-5-enterprise-solid-state-drive-for-14th-15th-16th-gen-dell
|
|
servers: https://www.broadberry.com/amd-epyc-9004-supermicro-servers
|
|
market_context: https://www.techpowerup.com/342331/server-dram-pricing-jumps-50-only-70-of-orders-getting-filled
|
|
"""
|
|
HW_PRICE_KEYS = (
|
|
"as_of", "ram_module_gb", "ram_module_usd", "nvme_drive_tb", "nvme_drive_usd",
|
|
"server_16_core_usd", "server_32_core_usd", "extra_node_usd", "sources",
|
|
)
|
|
|
|
|
|
def read_medians(path: Path) -> dict[tuple[str, int], float]:
|
|
"""Median walls of the 35 cells; abort unless every cell is OK + checksummed
|
|
(projections built on invalid measurements are forbidden, bench-methodology).
|
|
"""
|
|
medians: dict[tuple[str, int], float] = {}
|
|
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 {row['status']}/checksum={row['checksum_ok']} - cannot calibrate on it")
|
|
medians[(row["format"], int(row["query"][1:]))] = float(row["wall_s_median"])
|
|
expected_cells = {(fmt, qn) for fmt in FORMATS for qn in QUERY_NUMBERS}
|
|
if set(medians) != expected_cells:
|
|
raise ValueError(f"{path}: {len(medians)} cells, expected the full 35-cell matrix")
|
|
return medians
|
|
|
|
|
|
def read_sizes(path: Path) -> dict[tuple[str, str], int]:
|
|
sizes: dict[tuple[str, str], int] = {}
|
|
with open(path, encoding="ascii", newline="") as fh:
|
|
reader = csv.reader(fh)
|
|
next(reader)
|
|
for fmt, variant, nbytes in reader:
|
|
sizes[(fmt, variant)] = int(nbytes)
|
|
missing = set(SIZE_VARIANT.values()) - set(sizes)
|
|
if missing:
|
|
raise ValueError(f"{path}: missing rows {sorted(missing)}")
|
|
return sizes
|
|
|
|
|
|
def pg_index_share(dsn: str) -> tuple[float, int]:
|
|
"""Index share of the live PG footprint (read-only catalog query)."""
|
|
with psycopg.connect(dsn) as conn:
|
|
index_bytes, total_bytes = conn.execute(
|
|
"SELECT COALESCE(sum(pg_indexes_size(c.oid)), 0)::bigint,"
|
|
" COALESCE(sum(pg_total_relation_size(c.oid)), 0)::bigint"
|
|
" FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace"
|
|
" WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p', 'm')"
|
|
).fetchone()
|
|
if total_bytes == 0:
|
|
raise ValueError("pg catalog reports zero relation bytes - is the database loaded?")
|
|
return index_bytes / total_bytes, index_bytes
|
|
|
|
|
|
def load_hw_prices(config_dir: Path) -> dict:
|
|
path = config_dir / "hw_prices.yaml"
|
|
if not path.exists():
|
|
path.write_text(HW_PRICES_DEFAULT, encoding="ascii", newline="\n")
|
|
logger.info("seeded editable defaults: %s", path)
|
|
with open(path, encoding="ascii") as fh:
|
|
prices = yaml.safe_load(fh)
|
|
missing = [k for k in HW_PRICE_KEYS if k not in prices]
|
|
if missing:
|
|
raise ValueError(f"{path}: missing keys {missing}")
|
|
return prices
|
|
|
|
|
|
def growth(law: str, r: float, n0: int, cores: int) -> float:
|
|
depth = 1.0 + math.log(r) / math.log(n0)
|
|
if law == "o1":
|
|
return 1.0
|
|
if law == "log":
|
|
return depth
|
|
if law == "logk":
|
|
return r
|
|
if law == "k_log_n":
|
|
return r * depth
|
|
if law == "scan":
|
|
return r
|
|
if law == "scan_parallel":
|
|
return r * BASE_CORES / cores
|
|
logger.error("growth: unknown law=%r", law)
|
|
raise ValueError(f"unknown scaling law: {law!r}")
|
|
|
|
|
|
def flag_of(seconds: float) -> str:
|
|
if seconds > FAIL_S:
|
|
return "FAIL"
|
|
if seconds > IMPRACTICAL_S:
|
|
return "IMPRACTICAL"
|
|
return "OK"
|
|
|
|
|
|
def project(fmt: str, qn: int, medians: dict, overhead: dict, r: float, n0: int, cores: int) -> float:
|
|
variable = max(0.0, medians[(fmt, qn)] - overhead[fmt])
|
|
return overhead[fmt] + variable * growth(LAWS[(fmt, qn)], r, n0, cores)
|
|
|
|
|
|
def pick_pg_cores(medians: dict, overhead: dict, r: float, n0: int) -> int:
|
|
"""Smallest priced tier; escalate only when it fixes an IMPRACTICAL flag
|
|
on a parallel-scan cell (the only law that benefits from cores)."""
|
|
for cores in CORE_OPTIONS:
|
|
worst = max(project("pg", qn, medians, overhead, r, n0, cores) for qn in QUERY_NUMBERS if LAWS[("pg", qn)] == "scan_parallel")
|
|
if worst <= IMPRACTICAL_S:
|
|
return cores
|
|
return CORE_OPTIONS[-1]
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_task_args("Task 10: extrapolate measurements to 600 GB / 1.2 TB / 6 TB")
|
|
out_root: Path = args.out_root
|
|
bench_dir = out_root / "bench"
|
|
try:
|
|
check_dependencies(out_root, DEPENDS_ON)
|
|
remove_stale_marker(out_root, TASK_ID)
|
|
medians = read_medians(bench_dir / "results_median.csv")
|
|
sizes = read_sizes(bench_dir / "storage_sizes.csv")
|
|
prices = load_hw_prices(out_root / "config")
|
|
n0 = expected_counts(load_lab_config(out_root))["friction_cycles"]
|
|
csv_bytes = sizes[SIZE_VARIANT["csv"]]
|
|
coeff = {fmt: sizes[SIZE_VARIANT[fmt]] / csv_bytes for fmt in FORMATS}
|
|
index_share, pg_index_bytes = pg_index_share(get_dsn())
|
|
logger.info("storage coefficients vs csv: %s; pg index share %.3f",
|
|
", ".join(f"{fmt}={coeff[fmt]:.2f}" for fmt in FORMATS), index_share)
|
|
overhead = {fmt: min(medians[(fmt, qn)] for qn in QUERY_NUMBERS) for fmt in FORMATS}
|
|
|
|
flags = {"OK": 0, "IMPRACTICAL": 0, "FAIL": 0}
|
|
with open(bench_dir / "extrapolation.csv", "w", encoding="ascii", newline="") as fh:
|
|
writer = csv.writer(fh)
|
|
writer.writerow(["format", "query", "scale_gb", "projected_wall_s", "flag"])
|
|
for scale_gb in SCALES_GB:
|
|
r = scale_gb * 1e9 / csv_bytes
|
|
cores = {fmt: CORE_OPTIONS[0] for fmt in FORMATS}
|
|
cores["pg"] = pick_pg_cores(medians, overhead, r, n0)
|
|
for fmt in FORMATS:
|
|
for qn in QUERY_NUMBERS:
|
|
seconds = project(fmt, qn, medians, overhead, r, n0, cores[fmt])
|
|
flag = flag_of(seconds)
|
|
flags[flag] += 1
|
|
writer.writerow([fmt, f"q{qn}", scale_gb, round(seconds, 3), flag])
|
|
|
|
with open(bench_dir / "hardware_sizing.csv", "w", encoding="ascii", newline="") as fh:
|
|
writer = csv.writer(fh)
|
|
writer.writerow(["format", "scale_gb", "disk_tb", "ram_gb", "cores", "nodes",
|
|
"est_cost_usd", "ram_modules", "nvme_drives", "base_usd", "ram_usd", "disk_usd", "extra_nodes_usd"])
|
|
for scale_gb in SCALES_GB:
|
|
r = scale_gb * 1e9 / csv_bytes
|
|
for fmt in FORMATS:
|
|
fmt_bytes = coeff[fmt] * scale_gb * 1e9
|
|
disk_tb = fmt_bytes * DISK_HEADROOM / 1e12
|
|
if fmt in ("csv", "json"):
|
|
ram_gb = STREAM_RAM_GB
|
|
elif fmt in ("sqlite", "pg"):
|
|
# same index share assumed for sqlite: identical rows,
|
|
# keys, and indexes as the PG schema (variant B)
|
|
ram_gb = max(MIN_RAM_GB, CACHE_INDEX_FRACTION * index_share * fmt_bytes / 1e9 + WORKING_SET_GB)
|
|
else:
|
|
ram_gb = max(MIN_RAM_GB, RDF_HOT_FRACTION * fmt_bytes / 1e9)
|
|
cores = pick_pg_cores(medians, overhead, r, n0) if fmt == "pg" else CORE_OPTIONS[0]
|
|
nodes = max(1, math.ceil(ram_gb / NODE_RAM_CEILING_GB))
|
|
# bill of materials: whole RDIMM modules and whole NVMe drives
|
|
ram_modules = math.ceil(ram_gb / prices["ram_module_gb"])
|
|
nvme_drives = math.ceil(disk_tb / prices["nvme_drive_tb"])
|
|
base_usd = prices["server_16_core_usd"] if cores == 16 else prices["server_32_core_usd"]
|
|
ram_usd = ram_modules * prices["ram_module_usd"]
|
|
disk_usd = nvme_drives * prices["nvme_drive_usd"]
|
|
extra_nodes_usd = (nodes - 1) * prices["extra_node_usd"]
|
|
cost = base_usd + ram_usd + disk_usd + extra_nodes_usd
|
|
writer.writerow([fmt, scale_gb, round(disk_tb, 3), round(ram_gb, 1), cores, nodes,
|
|
round(cost), ram_modules, nvme_drives,
|
|
round(base_usd), round(ram_usd), round(disk_usd), round(extra_nodes_usd)])
|
|
|
|
entries = {
|
|
"extrapolation_rows": len(SCALES_GB) * len(FORMATS) * len(QUERY_NUMBERS),
|
|
"sizing_rows": len(SCALES_GB) * len(FORMATS),
|
|
"flags_ok": flags["OK"],
|
|
"flags_impractical": flags["IMPRACTICAL"],
|
|
"flags_fail": flags["FAIL"],
|
|
"pg_index_share": round(index_share, 3),
|
|
"pg_index_mb": round(pg_index_bytes / 1048576, 1),
|
|
"prices_as_of": prices["as_of"],
|
|
"note_pandas": "pandas CSV variant not measured; its RAM is O(n) - infeasible above ~10 GB (spec 10)",
|
|
}
|
|
for fmt in FORMATS:
|
|
entries[f"coeff_{fmt}"] = round(coeff[fmt], 3)
|
|
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 10 ok: {entries['extrapolation_rows']} projections "
|
|
f"(ok={flags['OK']}, impractical={flags['IMPRACTICAL']}, fail={flags['FAIL']}), "
|
|
f"{entries['sizing_rows']} sizing rows")
|
|
print(f"outputs: {bench_dir / 'extrapolation.csv'}, {bench_dir / 'hardware_sizing.csv'}")
|
|
print(f"marker: {marker_path}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|