refactor: pre-generate data and rewrite benchmarks

.gitignore

@@ -1,3 +1,5 @@
 .idea/
 go.work
 go.work.sum
+bench/run-*/
+/analysis/.ipynb_checkpoints

.python-version

@@ -0,0 +1 @@
+3.13

Makefile

@@ -0,0 +1,15 @@
+install:
+	cd bench && go install ./cmd/gen-changesets
+	cd bench && go install ./cmd/iavl-bench-all
+	cd iavl-v0 && go install .
+	cd iavl-v1 && go install .
+	cd iavl-v2/alpha5 && go install .
+	cd iavl-v2/alpha6 && go install .
+	cd memiavl && go install .
+	cd memiavl/thread-safe && go install .
+	cd memiavl/query-thread-safe && go install .
+	cd store-v1/latest && go install .
+	cd store-v1/iavl-v2 && go install .
+	cd store-v1/memiavl && go install .
+
+PHONY: install

analysis/analysis.py

@@ -0,0 +1,105 @@
+from read_logs import BenchmarkData
+import polars as pl
+import plotly.graph_objects as go
+
+
+def total_ops_per_sec(run: BenchmarkData) -> float:
+    count = run.versions_df['count'].sum()
+    total_duration = run.versions_df['duration'].sum() / 1_000_000_000  # convert from nanoseconds
+    return count / total_duration
+
+
+def max_mem_gb(run: BenchmarkData) -> float:
+    return run.versions_df['mem_gb'].max()
+
+
+def max_disk_gb(run: BenchmarkData) -> float:
+    return run.versions_df['disk_usage_gb'].max()
+
+
+def summary(dataset: dict[str, BenchmarkData], run_names=None) -> pl.DataFrame:
+    if run_names is None:
+        run_names = list(dataset.keys())
+    summary_data = []
+    for name in run_names:
+        run = dataset[name]
+        summary_data.append({
+            'name': name,
+            'ops_per_sec': total_ops_per_sec(run),
+            'max_mem_gb': max_mem_gb(run),
+            'max_disk_gb': max_disk_gb(run),
+        })
+    return pl.DataFrame(summary_data)
+
+
+def calculate_batch_ops_per_sec(versions_df, batch_size=100):
+    """Calculate ops_per_sec for every batch_size versions by summing counts and durations."""
+    return (
+        versions_df
+        .with_columns(
+            ((pl.col("version") / batch_size).ceil() * batch_size).alias("version_batch")
+        )
+        .group_by("version_batch")
+        .agg([
+            pl.col("count").sum().alias("total_count"),
+            pl.col("duration").sum().alias("total_duration")
+        ])
+        .with_columns(
+            (pl.col("total_count") / (pl.col("total_duration") / 1_000_000_000)).alias("ops_per_sec")
+        )
+        .select(["version_batch", "ops_per_sec"])
+        .rename({"version_batch": "version"})
+        .sort("version")
+    )
+
+
+def _create_line_plot(dataset, run_names: list[str], y_axis_title: str, 
+                      data_getter=None, column_name=None):
+    """Generic utility function to create line plots from dataset.
+
+    Either provide data_getter (a function that takes a run and returns a dataframe with 'version' and a y column)
+    or column_name (to directly access run.versions_df[column_name]).
+    """
+    if run_names is None:
+        run_names = list(dataset.keys())
+
+    fig = go.Figure()
+    for name in run_names:
+        run = dataset[name]
+
+        if data_getter:
+            df = data_getter(run)
+            x_data = df['version']
+            y_data = df.select(pl.exclude('version')).to_series()
+        else:
+            x_data = run.versions_df['version']
+            y_data = run.versions_df[column_name]
+
+        fig.add_trace(go.Scatter(
+            x=x_data,
+            y=y_data,
+            mode='lines',
+            name=name
+        ))
+
+    fig.update_layout(
+        xaxis_title="Version",
+        yaxis_title=y_axis_title,
+        hovermode='x unified'
+    )
+    return fig
+
+
+def plot_ops_per_sec(dataset, run_names: list[str] = None, batch_size=100):
+    def get_batched_ops(run):
+        return calculate_batch_ops_per_sec(run.versions_df, batch_size)
+
+    return _create_line_plot(dataset, run_names, 'Ops/Sec', data_getter=get_batched_ops)
+
+
+def plot_mem(dataset, run_names: list[str] = None):
+    return _create_line_plot(dataset, run_names, 'Memory (GB)', column_name='mem_gb')
+
+
+def plot_disk_usage(dataset, run_names: list[str] = None):
+    return _create_line_plot(dataset, run_names, 'Disk Usage (GB)', column_name='disk_usage_gb')

analysis/dashboard.py

@@ -0,0 +1,115 @@
+import os
+from pathlib import Path
+
+import humanfriendly
+import polars as pl
+import streamlit as st
+import pandas
+from read_logs import load_benchmark_dir
+
+# get benchmark dir from env var BENCHMARK_RESULTS or panic
+benchmark_dir = os.getenv('BENCHMARK_RESULTS')
+if not benchmark_dir:
+    raise ValueError('BENCHMARK_RESULTS environment variable not set')
+
+all_data = load_benchmark_dir(benchmark_dir)
+all_names = [d.name for d in all_data]
+
+st.title('Benchmark Results Visualization')
+
+summaries = [d.summary for d in all_data if d.summary is not None]
+if len(summaries) != 0:
+    # Show table and bar charts of all summary data
+    st.header('Summary Data')
+
+    summary_df = pandas.DataFrame(summaries)
+    summary_df.index = [d.name for d in all_data if d.summary is not None]
+    tab1, tab2, tab3, tab4 = st.tabs(['Summary', 'Ops/sec', 'Max Mem (GB)', 'Max Disk (GB)'])
+
+    with tab1:
+        st.dataframe(summary_df)
+
+    with tab2:
+        st.bar_chart(summary_df, y='ops_per_sec', stack=False)
+
+    with tab3:
+        st.bar_chart(summary_df, y='max_mem_gb', stack=False)
+
+    with tab4:
+        st.bar_chart(summary_df, y='max_disk_gb', stack=False)
+
+# Show line charts for ops_per_sec, mem_sys, disk_usage over versions for each benchmark
+st.header('Performance Over Time')
+
+names = st.segmented_control("Benchmark Runs", all_names, selection_mode="multi", default=all_names)
+
+if len(names) == 0:
+    st.warning('Please select at least one benchmark run to display')
+    st.stop()
+
+data = [d for d in all_data if d.name in names]
+
+# For now truncate all data to the shortest length
+min_versions = min(len(d.versions) for d in data)
+for d in data:
+    d.versions_df = d.versions_df.head(min_versions)
+
+tab1, tab2, tab3 = st.tabs(['Ops/sec', 'Memory', 'Disk Usage'])
+
+with tab1:
+    ops_per_sec_df = pl.DataFrame({d.name: d.versions_df.select('ops_per_sec').to_series() for d in data})
+    st.line_chart(ops_per_sec_df, x_label='version', y_label='ops/sec')
+
+with tab2:
+    mem_df = pl.DataFrame({d.name: d.versions_df.select('mem_gb').to_series() for d in data})
+    st.line_chart(mem_df, x_label='version', y_label='mem (GB)')
+
+with tab3:
+    disk_df = pl.DataFrame({d.name: d.versions_df.select('disk_usage_gb').to_series() for d in data})
+    st.line_chart(disk_df, x_label='version', y_label='disk (GB)')
+
+# with tab4:
+#     disk_io_df = pl.DataFrame({d.name: d.versions_df.select('disk_io').to_series() for d in data})
+#     st.line_chart(disk_io_df, x_label='version')
+
+st.text(f'Showing data from {len(all_data)} benchmark logs in {Path(benchmark_dir).absolute()}')
+
+init_data0 = all_data[0].init_data
+changeset_dir0 = init_data0.get('changeset_dir') if init_data0 else None
+changeset_info0 = init_data0.get('changeset_info') if init_data0 else None
+st.markdown(f'Changeset Dir: `{changeset_dir0}`')
+st.markdown(f'Changeset Versions: `{changeset_info0.get("versions")}`')
+for store in changeset_info0.get('store_params'):
+    st.markdown(f'Store: `{store["store_key"]}`')
+    st.markdown(
+        f'* Initial Size=`{humanfriendly.format_number(store["initial_size"])}` -> Final Size=`{humanfriendly.format_number(store["final_size"])}` (over `{store["versions"]}` versions)')
+    st.markdown(
+        f'* K mean=`{store["key_mean"]}`, stddev=`{store["key_std_dev"]}`, V mean=`{store["value_mean"]}`, stddev=`{store["value_std_dev"]}`')
+    st.markdown(f'* Change per version=`{store["change_per_version"]}`, delete fraction=`{store["delete_fraction"]}`')
+
+for d in all_data:
+    st.markdown(f'## {d.name}')
+    st.markdown(f'`{len(d.versions)}` Versions Successfully Committed')
+    if d.init_data:
+        if 'changeset_dir' in d.init_data:
+            changeset_dir = d.init_data['changeset_dir']
+            if changeset_dir != changeset_dir0:
+                raise ValueError('Benchmark runs have different changeset dirs')
+        if 'start_version' in d.init_data:
+            start_version = d.init_data['start_version']
+            if start_version != 0:
+                st.markdown(f'Start Version: `{start_version}`')
+        if 'target_version' in d.init_data:
+            target_version = d.init_data['target_version']
+            if target_version != 0:
+                st.markdown(f'Target Version: `{target_version}`')
+        if 'db_options' in d.init_data:
+            db_options = d.init_data['db_options']
+            st.markdown(f'DB Options:')
+            st.json(db_options, expanded=False)
+        if d.memiavl_snapshots is not None:
+            with st.expander('Memiavl Snapshot Details', expanded=False):
+                st.dataframe(d.memiavl_snapshots)
+                st.line_chart(d.memiavl_snapshots.select("version", pl.col("snapshot_duration").dt.total_minutes().alias("snapshot_minutes")),
+                              x="version",
+                              y="snapshot_minutes")

analysis/memiavl_snapshots.py

@@ -0,0 +1,20 @@
+from datetime import datetime
+
+
+def capture_memiavl_snapshot_log(row, snapshots):
+    msg = row.get("msg")
+    time = datetime.fromisoformat(row.get("time"))
+    version = row.get("version")
+    match msg:
+        case "start rewriting snapshot":
+            snapshots += [{"version": version, "start_time": time}]
+        case "finished rewriting snapshot":
+            snapshots[-1]["end_time"] = time
+            snapshots[-1]["snapshot_duration"] = time - snapshots[-1]["start_time"]
+        case "finished best-effort WAL catchup":
+            snapshots[-1]["best_effort_wal_time"] = time
+            snapshots[-1]["best_effort_wal_duration"] = time - snapshots[-1]["end_time"]
+        case "switched to new snapshot":
+            snapshots[-1]["switch_time"] = time
+            snapshots[-1]["wal_sync_duration"] = time - snapshots[-1]["best_effort_wal_time"]
+            snapshots[-1]["switch_version"] = version