(
name: str,
rows: int,
bytes_estimate: int,
iterations: int,
fn: Callable[[], Any],
)
| 50 | |
| 51 | |
| 52 | def _measure( |
| 53 | name: str, |
| 54 | rows: int, |
| 55 | bytes_estimate: int, |
| 56 | iterations: int, |
| 57 | fn: Callable[[], Any], |
| 58 | ) -> BenchStats: |
| 59 | timings: list[float] = [] |
| 60 | for _ in range(iterations): |
| 61 | t0 = time.perf_counter() |
| 62 | fn() |
| 63 | timings.append(time.perf_counter() - t0) |
| 64 | timings.sort() |
| 65 | mean_s = statistics.mean(timings) |
| 66 | p50_s = timings[len(timings) // 2] |
| 67 | p95_s = timings[max(int(len(timings) * 0.95) - 1, 0)] |
| 68 | std_s = statistics.pstdev(timings) if len(timings) > 1 else 0.0 |
| 69 | return BenchStats( |
| 70 | name=name, |
| 71 | rows=rows, |
| 72 | bytes_estimate=bytes_estimate, |
| 73 | iterations=iterations, |
| 74 | mean_ms=mean_s * 1000.0, |
| 75 | p50_ms=p50_s * 1000.0, |
| 76 | p95_ms=p95_s * 1000.0, |
| 77 | std_ms=std_s * 1000.0, |
| 78 | rows_per_sec=(rows / mean_s) if mean_s > 0 else 0.0, |
| 79 | mb_per_sec=((bytes_estimate / 1_000_000.0) / mean_s) if mean_s > 0 else 0.0, |
| 80 | ) |
| 81 | |
| 82 | |
| 83 | def _format_rate(v: float) -> str: |
no test coverage detected