#![allow(dead_code, unused_imports)]

mod common;

use std::hint::black_box;
use std::time::Duration;

use armdb::{
    ConstMap, ConstTree, FixedMap, FixedTree, RapiraCodec, TypedMap, TypedTree, VarMap, VarTree,
};
use criterion::{Criterion, Throughput, criterion_group, criterion_main};
use rand::RngExt;
use tempfile::TempDir;

use common::config::{fixed_config, page_cache_config};
use common::data::{make_const_value, make_key, make_record, make_var_value};
use common::populate::*;
use common::runner::{op_batch, read_seq, write_rand, write_seq};
use common::{HOT_RANGE, QUICK_NUM_ENTRIES, QUICK_WRITE_BATCH, READ_BATCH, Record};

// ---------------------------------------------------------------------------
// seq-write — long-lived DB, monotonic keys
// ---------------------------------------------------------------------------

fn bench_seq_write(c: &mut Criterion) {
    let mut group = c.benchmark_group("seq-write");
    group.throughput(Throughput::Elements(QUICK_WRITE_BATCH));
    group.sample_size(10);

    group.bench_function("const_tree", |b| {
        let dir = TempDir::new().unwrap();
        let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), page_cache_config()).unwrap();
        write_seq(
            b,
            QUICK_WRITE_BATCH,
            |i| {
                const_tree.put(&make_key(i), &make_const_value(i)).unwrap();
            },
            || const_tree.flush_buffers().unwrap(),
        );
    });

    group.bench_function("typed_tree", |b| {
        let dir_t = TempDir::new().unwrap();
        let typed_tree = TypedTree::<[u8; 8], Record, RapiraCodec>::open(
            dir_t.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        write_seq(
            b,
            QUICK_WRITE_BATCH,
            |i| {
                typed_tree.put(&make_key(i), make_record(i)).unwrap();
            },
            || typed_tree.flush_buffers().unwrap(),
        );
    });

    group.bench_function("var_tree", |b| {
        let dir_v = TempDir::new().unwrap();
        let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), page_cache_config()).unwrap();
        write_seq(
            b,
            QUICK_WRITE_BATCH,
            |i| {
                var_tree.put(&make_key(i), &make_var_value(i)).unwrap();
            },
            || var_tree.flush_buffers().unwrap(),
        );
    });

    group.bench_function("fixed_tree", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        write_seq(
            b,
            QUICK_WRITE_BATCH,
            |i| {
                fixed_tree.put(&make_key(i), &make_const_value(i)).unwrap();
            },
            || {},
        ); // Fixed: in-place, no flush
    });

    group.bench_function("const_map", |b| {
        let dir_cm = TempDir::new().unwrap();
        let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), page_cache_config()).unwrap();
        write_seq(
            b,
            QUICK_WRITE_BATCH,
            |i| {
                const_map.put(&make_key(i), &make_const_value(i)).unwrap();
            },
            || const_map.flush_buffers().unwrap(),
        );
    });

    group.bench_function("typed_map", |b| {
        let dir_tm = TempDir::new().unwrap();
        let typed_map = TypedMap::<[u8; 8], Record, RapiraCodec>::open(
            dir_tm.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        write_seq(
            b,
            QUICK_WRITE_BATCH,
            |i| {
                typed_map.put(&make_key(i), make_record(i)).unwrap();
            },
            || typed_map.flush_buffers().unwrap(),
        );
    });

    group.bench_function("var_map", |b| {
        let dir_vm = TempDir::new().unwrap();
        let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), page_cache_config()).unwrap();
        write_seq(
            b,
            QUICK_WRITE_BATCH,
            |i| {
                var_map.put(&make_key(i), &make_var_value(i)).unwrap();
            },
            || var_map.flush_buffers().unwrap(),
        );
    });

    group.bench_function("fixed_map", |b| {
        let dir_fm = TempDir::new().unwrap();
        let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
        write_seq(
            b,
            QUICK_WRITE_BATCH,
            |i| {
                fixed_map.put(&make_key(i), &make_const_value(i)).unwrap();
            },
            || {},
        );
    });

    group.finish();
}

// ---------------------------------------------------------------------------
// rand-write — long-lived DB, precomputed random keys
// ---------------------------------------------------------------------------

fn rand_keys(n: u64) -> Vec<u64> {
    let mut rng = rand::rng();
    (0..n).map(|_| rng.random()).collect()
}

fn bench_rand_write(c: &mut Criterion) {
    let mut group = c.benchmark_group("rand-write");
    group.throughput(Throughput::Elements(QUICK_WRITE_BATCH));
    group.sample_size(10);

    group.bench_function("const_tree", |b| {
        // Precomputed key list replayed identically each Criterion iteration
        // (access pattern is fixed, not re-seeded per iteration).
        let keys = rand_keys(QUICK_WRITE_BATCH);
        let dir = TempDir::new().unwrap();
        let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), page_cache_config()).unwrap();
        write_rand(
            b,
            &keys,
            |k| {
                const_tree.put(&make_key(k), &make_const_value(k)).unwrap();
            },
            || const_tree.flush_buffers().unwrap(),
        );
    });

    group.bench_function("typed_tree", |b| {
        let keys = rand_keys(QUICK_WRITE_BATCH);
        let dir_t = TempDir::new().unwrap();
        let typed_tree = TypedTree::<[u8; 8], Record, RapiraCodec>::open(
            dir_t.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        write_rand(
            b,
            &keys,
            |k| {
                typed_tree.put(&make_key(k), make_record(k)).unwrap();
            },
            || typed_tree.flush_buffers().unwrap(),
        );
    });

    group.bench_function("var_tree", |b| {
        let keys = rand_keys(QUICK_WRITE_BATCH);
        let dir_v = TempDir::new().unwrap();
        let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), page_cache_config()).unwrap();
        write_rand(
            b,
            &keys,
            |k| {
                var_tree.put(&make_key(k), &make_var_value(k)).unwrap();
            },
            || var_tree.flush_buffers().unwrap(),
        );
    });

    group.bench_function("fixed_tree", |b| {
        let keys = rand_keys(QUICK_WRITE_BATCH);
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        write_rand(
            b,
            &keys,
            |k| {
                fixed_tree.put(&make_key(k), &make_const_value(k)).unwrap();
            },
            || {},
        );
    });

    group.bench_function("const_map", |b| {
        let keys = rand_keys(QUICK_WRITE_BATCH);
        let dir_cm = TempDir::new().unwrap();
        let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), page_cache_config()).unwrap();
        write_rand(
            b,
            &keys,
            |k| {
                const_map.put(&make_key(k), &make_const_value(k)).unwrap();
            },
            || const_map.flush_buffers().unwrap(),
        );
    });

    group.bench_function("typed_map", |b| {
        let keys = rand_keys(QUICK_WRITE_BATCH);
        let dir_tm = TempDir::new().unwrap();
        let typed_map = TypedMap::<[u8; 8], Record, RapiraCodec>::open(
            dir_tm.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        write_rand(
            b,
            &keys,
            |k| {
                typed_map.put(&make_key(k), make_record(k)).unwrap();
            },
            || typed_map.flush_buffers().unwrap(),
        );
    });

    group.bench_function("var_map", |b| {
        let keys = rand_keys(QUICK_WRITE_BATCH);
        let dir_vm = TempDir::new().unwrap();
        let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), page_cache_config()).unwrap();
        write_rand(
            b,
            &keys,
            |k| {
                var_map.put(&make_key(k), &make_var_value(k)).unwrap();
            },
            || var_map.flush_buffers().unwrap(),
        );
    });

    group.bench_function("fixed_map", |b| {
        let keys = rand_keys(QUICK_WRITE_BATCH);
        let dir_fm = TempDir::new().unwrap();
        let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
        write_rand(
            b,
            &keys,
            |k| {
                fixed_map.put(&make_key(k), &make_const_value(k)).unwrap();
            },
            || {},
        );
    });

    group.finish();
}

// ---------------------------------------------------------------------------
// seq-read — populate once, flush, read sequentially
// ---------------------------------------------------------------------------

fn bench_seq_read(c: &mut Criterion) {
    let mut group = c.benchmark_group("seq-read");
    group.throughput(Throughput::Elements(READ_BATCH));

    group.bench_function("const_tree", |b| {
        let dir = TempDir::new().unwrap();
        let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), page_cache_config()).unwrap();
        populate_const(&const_tree, QUICK_NUM_ENTRIES);
        read_seq(b, READ_BATCH, |i| {
            black_box(const_tree.get(&make_key(i % QUICK_NUM_ENTRIES)));
        });
        const_tree.close().unwrap();
    });

    group.bench_function("typed_tree", |b| {
        let dir_t = TempDir::new().unwrap();
        let typed_tree = TypedTree::<[u8; 8], Record, RapiraCodec>::open(
            dir_t.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        populate_typed(&typed_tree, QUICK_NUM_ENTRIES);
        read_seq(b, READ_BATCH, |i| {
            black_box(typed_tree.get(&make_key(i % QUICK_NUM_ENTRIES)));
        });
        typed_tree.close().unwrap();
    });

    group.bench_function("var_tree", |b| {
        let dir_v = TempDir::new().unwrap();
        let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), page_cache_config()).unwrap();
        populate_var(&var_tree, QUICK_NUM_ENTRIES);
        read_seq(b, READ_BATCH, |i| {
            black_box(var_tree.get(&make_key(i % QUICK_NUM_ENTRIES)));
        });
        var_tree.close().unwrap();
    });

    group.bench_function("fixed_tree", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        populate_fixed_tree(&fixed_tree, QUICK_NUM_ENTRIES);
        read_seq(b, READ_BATCH, |i| {
            black_box(fixed_tree.get(&make_key(i % QUICK_NUM_ENTRIES)));
        });
        fixed_tree.close().unwrap();
    });

    group.bench_function("const_map", |b| {
        let dir_cm = TempDir::new().unwrap();
        let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), page_cache_config()).unwrap();
        populate_const_map(&const_map, QUICK_NUM_ENTRIES);
        read_seq(b, READ_BATCH, |i| {
            black_box(const_map.get(&make_key(i % QUICK_NUM_ENTRIES)));
        });
        const_map.close().unwrap();
    });

    group.bench_function("typed_map", |b| {
        let dir_tm = TempDir::new().unwrap();
        let typed_map = TypedMap::<[u8; 8], Record, RapiraCodec>::open(
            dir_tm.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        populate_typed_map(&typed_map, QUICK_NUM_ENTRIES);
        read_seq(b, READ_BATCH, |i| {
            black_box(typed_map.get(&make_key(i % QUICK_NUM_ENTRIES)));
        });
        typed_map.close().unwrap();
    });

    group.bench_function("var_map", |b| {
        let dir_vm = TempDir::new().unwrap();
        let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), page_cache_config()).unwrap();
        populate_var_map(&var_map, QUICK_NUM_ENTRIES);
        read_seq(b, READ_BATCH, |i| {
            black_box(var_map.get(&make_key(i % QUICK_NUM_ENTRIES)));
        });
        var_map.close().unwrap();
    });

    group.bench_function("fixed_map", |b| {
        let dir_fm = TempDir::new().unwrap();
        let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
        populate_fixed_map(&fixed_map, QUICK_NUM_ENTRIES);
        read_seq(b, READ_BATCH, |i| {
            black_box(fixed_map.get(&make_key(i % QUICK_NUM_ENTRIES)));
        });
        fixed_map.close().unwrap();
    });

    group.finish();
}

// ---------------------------------------------------------------------------
// rand-read-hot — read from 0..HOT_RANGE (in-cache working set)
// ---------------------------------------------------------------------------

fn bench_rand_read_hot(c: &mut Criterion) {
    let mut group = c.benchmark_group("rand-read-hot");
    group.throughput(Throughput::Elements(READ_BATCH));

    group.bench_function("const_tree", |b| {
        let dir = TempDir::new().unwrap();
        let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), page_cache_config()).unwrap();
        populate_const(&const_tree, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        op_batch(b, READ_BATCH, || {
            black_box(const_tree.get(&make_key(rng.random_range(0..HOT_RANGE))));
        });
        const_tree.close().unwrap();
    });

    group.bench_function("typed_tree", |b| {
        let dir_t = TempDir::new().unwrap();
        let typed_tree = TypedTree::<[u8; 8], Record, RapiraCodec>::open(
            dir_t.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        populate_typed(&typed_tree, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        op_batch(b, READ_BATCH, || {
            black_box(typed_tree.get(&make_key(rng.random_range(0..HOT_RANGE))));
        });
        typed_tree.close().unwrap();
    });

    group.bench_function("var_tree", |b| {
        let dir_v = TempDir::new().unwrap();
        let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), page_cache_config()).unwrap();
        populate_var(&var_tree, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        op_batch(b, READ_BATCH, || {
            black_box(var_tree.get(&make_key(rng.random_range(0..HOT_RANGE))));
        });
        var_tree.close().unwrap();
    });

    group.bench_function("fixed_tree", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        populate_fixed_tree(&fixed_tree, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        op_batch(b, READ_BATCH, || {
            black_box(fixed_tree.get(&make_key(rng.random_range(0..HOT_RANGE))));
        });
        fixed_tree.close().unwrap();
    });

    group.bench_function("const_map", |b| {
        let dir_cm = TempDir::new().unwrap();
        let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), page_cache_config()).unwrap();
        populate_const_map(&const_map, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        op_batch(b, READ_BATCH, || {
            black_box(const_map.get(&make_key(rng.random_range(0..HOT_RANGE))));
        });
        const_map.close().unwrap();
    });

    group.bench_function("typed_map", |b| {
        let dir_tm = TempDir::new().unwrap();
        let typed_map = TypedMap::<[u8; 8], Record, RapiraCodec>::open(
            dir_tm.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        populate_typed_map(&typed_map, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        op_batch(b, READ_BATCH, || {
            black_box(typed_map.get(&make_key(rng.random_range(0..HOT_RANGE))));
        });
        typed_map.close().unwrap();
    });

    group.bench_function("var_map", |b| {
        let dir_vm = TempDir::new().unwrap();
        let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), page_cache_config()).unwrap();
        populate_var_map(&var_map, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        op_batch(b, READ_BATCH, || {
            black_box(var_map.get(&make_key(rng.random_range(0..HOT_RANGE))));
        });
        var_map.close().unwrap();
    });

    group.bench_function("fixed_map", |b| {
        let dir_fm = TempDir::new().unwrap();
        let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
        populate_fixed_map(&fixed_map, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        op_batch(b, READ_BATCH, || {
            black_box(fixed_map.get(&make_key(rng.random_range(0..HOT_RANGE))));
        });
        fixed_map.close().unwrap();
    });

    group.finish();
}

// ---------------------------------------------------------------------------
// read-write-mix — 80% read / 20% write
// ---------------------------------------------------------------------------

fn bench_read_write_mix(c: &mut Criterion) {
    use common::runner::mix_batch;

    let mut group = c.benchmark_group("read-write-mix");
    group.throughput(Throughput::Elements(READ_BATCH));
    group.sample_size(10);

    group.bench_function("const_tree/80r_20w", |b| {
        let dir = TempDir::new().unwrap();
        let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), page_cache_config()).unwrap();
        populate_const(&const_tree, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        mix_batch(
            b,
            800,
            200,
            || {
                black_box(const_tree.get(&make_key(rng.random_range(0..QUICK_NUM_ENTRIES))));
            },
            |s| {
                const_tree
                    .put(
                        &make_key(s % QUICK_NUM_ENTRIES),
                        &make_const_value(QUICK_NUM_ENTRIES + s),
                    )
                    .unwrap();
            },
        );
        const_tree.close().unwrap();
    });

    group.bench_function("typed_tree/80r_20w", |b| {
        let dir_t = TempDir::new().unwrap();
        let typed_tree = TypedTree::<[u8; 8], Record, RapiraCodec>::open(
            dir_t.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        populate_typed(&typed_tree, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        mix_batch(
            b,
            800,
            200,
            || {
                black_box(typed_tree.get(&make_key(rng.random_range(0..QUICK_NUM_ENTRIES))));
            },
            |s| {
                typed_tree
                    .put(
                        &make_key(s % QUICK_NUM_ENTRIES),
                        make_record(QUICK_NUM_ENTRIES + s),
                    )
                    .unwrap();
            },
        );
        typed_tree.close().unwrap();
    });

    group.bench_function("var_tree/80r_20w", |b| {
        let dir_v = TempDir::new().unwrap();
        let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), page_cache_config()).unwrap();
        populate_var(&var_tree, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        mix_batch(
            b,
            800,
            200,
            || {
                black_box(var_tree.get(&make_key(rng.random_range(0..QUICK_NUM_ENTRIES))));
            },
            |s| {
                var_tree
                    .put(
                        &make_key(s % QUICK_NUM_ENTRIES),
                        &make_var_value(QUICK_NUM_ENTRIES + s),
                    )
                    .unwrap();
            },
        );
        var_tree.close().unwrap();
    });

    group.bench_function("fixed_tree/80r_20w", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        populate_fixed_tree(&fixed_tree, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        mix_batch(
            b,
            800,
            200,
            || {
                black_box(fixed_tree.get(&make_key(rng.random_range(0..QUICK_NUM_ENTRIES))));
            },
            |s| {
                fixed_tree
                    .put(
                        &make_key(s % QUICK_NUM_ENTRIES),
                        &make_const_value(QUICK_NUM_ENTRIES + s),
                    )
                    .unwrap();
            },
        );
        fixed_tree.close().unwrap();
    });

    group.bench_function("const_map/80r_20w", |b| {
        let dir_cm = TempDir::new().unwrap();
        let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), page_cache_config()).unwrap();
        populate_const_map(&const_map, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        mix_batch(
            b,
            800,
            200,
            || {
                black_box(const_map.get(&make_key(rng.random_range(0..QUICK_NUM_ENTRIES))));
            },
            |s| {
                const_map
                    .put(
                        &make_key(s % QUICK_NUM_ENTRIES),
                        &make_const_value(QUICK_NUM_ENTRIES + s),
                    )
                    .unwrap();
            },
        );
        const_map.close().unwrap();
    });

    group.bench_function("typed_map/80r_20w", |b| {
        let dir_tm = TempDir::new().unwrap();
        let typed_map = TypedMap::<[u8; 8], Record, RapiraCodec>::open(
            dir_tm.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        populate_typed_map(&typed_map, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        mix_batch(
            b,
            800,
            200,
            || {
                black_box(typed_map.get(&make_key(rng.random_range(0..QUICK_NUM_ENTRIES))));
            },
            |s| {
                typed_map
                    .put(
                        &make_key(s % QUICK_NUM_ENTRIES),
                        make_record(QUICK_NUM_ENTRIES + s),
                    )
                    .unwrap();
            },
        );
        typed_map.close().unwrap();
    });

    group.bench_function("var_map/80r_20w", |b| {
        let dir_vm = TempDir::new().unwrap();
        let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), page_cache_config()).unwrap();
        populate_var_map(&var_map, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        mix_batch(
            b,
            800,
            200,
            || {
                black_box(var_map.get(&make_key(rng.random_range(0..QUICK_NUM_ENTRIES))));
            },
            |s| {
                var_map
                    .put(
                        &make_key(s % QUICK_NUM_ENTRIES),
                        &make_var_value(QUICK_NUM_ENTRIES + s),
                    )
                    .unwrap();
            },
        );
        var_map.close().unwrap();
    });

    group.bench_function("fixed_map/80r_20w", |b| {
        let dir_fm = TempDir::new().unwrap();
        let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
        populate_fixed_map(&fixed_map, QUICK_NUM_ENTRIES);
        let mut rng = rand::rng();
        mix_batch(
            b,
            800,
            200,
            || {
                black_box(fixed_map.get(&make_key(rng.random_range(0..QUICK_NUM_ENTRIES))));
            },
            |s| {
                fixed_map
                    .put(
                        &make_key(s % QUICK_NUM_ENTRIES),
                        &make_const_value(QUICK_NUM_ENTRIES + s),
                    )
                    .unwrap();
            },
        );
        fixed_map.close().unwrap();
    });

    group.finish();
}

// ---------------------------------------------------------------------------
// overwrite — re-put existing keys (in-place put fast path)
// ---------------------------------------------------------------------------

fn bench_overwrite(c: &mut Criterion) {
    let mut group = c.benchmark_group("overwrite");
    group.throughput(Throughput::Elements(READ_BATCH));
    group.sample_size(10);

    group.bench_function("const_tree", |b| {
        let dir = TempDir::new().unwrap();
        let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), page_cache_config()).unwrap();
        populate_const(&const_tree, QUICK_NUM_ENTRIES);
        let mut ctr = 0u64;
        op_batch(b, READ_BATCH, || {
            const_tree
                .put(&make_key(ctr % QUICK_NUM_ENTRIES), &make_const_value(ctr))
                .unwrap();
            ctr += 1;
        });
        const_tree.close().unwrap();
    });

    group.bench_function("typed_tree", |b| {
        let dir_t = TempDir::new().unwrap();
        let typed_tree = TypedTree::<[u8; 8], Record, RapiraCodec>::open(
            dir_t.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        populate_typed(&typed_tree, QUICK_NUM_ENTRIES);
        let mut ctr = 0u64;
        op_batch(b, READ_BATCH, || {
            typed_tree
                .put(&make_key(ctr % QUICK_NUM_ENTRIES), make_record(ctr))
                .unwrap();
            ctr += 1;
        });
        typed_tree.close().unwrap();
    });

    group.bench_function("var_tree", |b| {
        let dir_v = TempDir::new().unwrap();
        let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), page_cache_config()).unwrap();
        populate_var(&var_tree, QUICK_NUM_ENTRIES);
        let mut ctr = 0u64;
        op_batch(b, READ_BATCH, || {
            var_tree
                .put(&make_key(ctr % QUICK_NUM_ENTRIES), &make_var_value(ctr))
                .unwrap();
            ctr += 1;
        });
        var_tree.close().unwrap();
    });

    group.bench_function("fixed_tree", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        populate_fixed_tree(&fixed_tree, QUICK_NUM_ENTRIES);
        let mut ctr = 0u64;
        op_batch(b, READ_BATCH, || {
            fixed_tree
                .put(&make_key(ctr % QUICK_NUM_ENTRIES), &make_const_value(ctr))
                .unwrap();
            ctr += 1;
        });
        fixed_tree.close().unwrap();
    });

    group.bench_function("const_map", |b| {
        let dir_cm = TempDir::new().unwrap();
        let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), page_cache_config()).unwrap();
        populate_const_map(&const_map, QUICK_NUM_ENTRIES);
        let mut ctr = 0u64;
        op_batch(b, READ_BATCH, || {
            const_map
                .put(&make_key(ctr % QUICK_NUM_ENTRIES), &make_const_value(ctr))
                .unwrap();
            ctr += 1;
        });
        const_map.close().unwrap();
    });

    group.bench_function("typed_map", |b| {
        let dir_tm = TempDir::new().unwrap();
        let typed_map = TypedMap::<[u8; 8], Record, RapiraCodec>::open(
            dir_tm.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        populate_typed_map(&typed_map, QUICK_NUM_ENTRIES);
        let mut ctr = 0u64;
        op_batch(b, READ_BATCH, || {
            typed_map
                .put(&make_key(ctr % QUICK_NUM_ENTRIES), make_record(ctr))
                .unwrap();
            ctr += 1;
        });
        typed_map.close().unwrap();
    });

    group.bench_function("var_map", |b| {
        let dir_vm = TempDir::new().unwrap();
        let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), page_cache_config()).unwrap();
        populate_var_map(&var_map, QUICK_NUM_ENTRIES);
        let mut ctr = 0u64;
        op_batch(b, READ_BATCH, || {
            var_map
                .put(&make_key(ctr % QUICK_NUM_ENTRIES), &make_var_value(ctr))
                .unwrap();
            ctr += 1;
        });
        var_map.close().unwrap();
    });

    group.bench_function("fixed_map", |b| {
        let dir_fm = TempDir::new().unwrap();
        let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
        populate_fixed_map(&fixed_map, QUICK_NUM_ENTRIES);
        let mut ctr = 0u64;
        op_batch(b, READ_BATCH, || {
            fixed_map
                .put(&make_key(ctr % QUICK_NUM_ENTRIES), &make_const_value(ctr))
                .unwrap();
            ctr += 1;
        });
        fixed_map.close().unwrap();
    });

    group.finish();
}

// ---------------------------------------------------------------------------
// scan — prefix_iter().take(N) + iter().take(N), tree collections only
// ---------------------------------------------------------------------------

fn bench_scan(c: &mut Criterion) {
    use common::data::make_group_key;
    use common::{NUM_GROUPS, SCAN_TAKE};

    let mut group = c.benchmark_group("scan");
    group.throughput(Throughput::Elements(SCAN_TAKE as u64));

    // Each bench_function is self-contained: populate and close happen inside
    // the closure. On a full run the grouped dataset is populated twice per
    // tree (once for prefix_take_100, once for iter_take_100). That is the
    // accepted cost of filter-correctness — skipped benches do zero setup.

    group.bench_function("const_tree/prefix_take_100", |b| {
        let dir = TempDir::new().unwrap();
        let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), page_cache_config()).unwrap();
        populate_const_grouped(&const_tree, NUM_GROUPS, SCAN_TAKE as u32);
        let mut rng = rand::rng();
        op_batch(b, 1, || {
            let prefix = rng.random_range(0..NUM_GROUPS).to_be_bytes();
            black_box(
                const_tree
                    .prefix_iter(&prefix)
                    .take(SCAN_TAKE)
                    .collect::<Vec<_>>(),
            );
        });
        const_tree.close().unwrap();
    });

    group.bench_function("const_tree/iter_take_100", |b| {
        let dir = TempDir::new().unwrap();
        let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), page_cache_config()).unwrap();
        populate_const_grouped(&const_tree, NUM_GROUPS, SCAN_TAKE as u32);
        op_batch(b, 1, || {
            black_box(const_tree.iter().take(SCAN_TAKE).collect::<Vec<_>>());
        });
        const_tree.close().unwrap();
    });

    group.bench_function("typed_tree/prefix_take_100", |b| {
        let dir_t = TempDir::new().unwrap();
        let typed_tree = TypedTree::<[u8; 8], Record, RapiraCodec>::open(
            dir_t.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        for g in 0..NUM_GROUPS {
            for id in 0..SCAN_TAKE as u32 {
                typed_tree
                    .put(
                        &make_group_key(g, id),
                        make_record(g as u64 * 1000 + id as u64),
                    )
                    .unwrap();
            }
        }
        typed_tree.flush_buffers().unwrap();
        let mut rng = rand::rng();
        op_batch(b, 1, || {
            let prefix = rng.random_range(0..NUM_GROUPS).to_be_bytes();
            black_box(
                typed_tree
                    .prefix_iter(&prefix)
                    .take(SCAN_TAKE)
                    .collect::<Vec<_>>(),
            );
        });
        typed_tree.close().unwrap();
    });

    group.bench_function("typed_tree/iter_take_100", |b| {
        let dir_t = TempDir::new().unwrap();
        let typed_tree = TypedTree::<[u8; 8], Record, RapiraCodec>::open(
            dir_t.path(),
            page_cache_config(),
            RapiraCodec,
        )
        .unwrap();
        for g in 0..NUM_GROUPS {
            for id in 0..SCAN_TAKE as u32 {
                typed_tree
                    .put(
                        &make_group_key(g, id),
                        make_record(g as u64 * 1000 + id as u64),
                    )
                    .unwrap();
            }
        }
        typed_tree.flush_buffers().unwrap();
        op_batch(b, 1, || {
            black_box(typed_tree.iter().take(SCAN_TAKE).collect::<Vec<_>>());
        });
        typed_tree.close().unwrap();
    });

    group.bench_function("var_tree/prefix_take_100", |b| {
        let dir_v = TempDir::new().unwrap();
        let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), page_cache_config()).unwrap();
        for g in 0..NUM_GROUPS {
            for id in 0..SCAN_TAKE as u32 {
                var_tree
                    .put(
                        &make_group_key(g, id),
                        &make_var_value(g as u64 * 1000 + id as u64),
                    )
                    .unwrap();
            }
        }
        var_tree.flush_buffers().unwrap();
        let mut rng = rand::rng();
        op_batch(b, 1, || {
            let prefix = rng.random_range(0..NUM_GROUPS).to_be_bytes();
            black_box(
                var_tree
                    .prefix_iter(&prefix)
                    .take(SCAN_TAKE)
                    .collect::<Vec<_>>(),
            );
        });
        var_tree.close().unwrap();
    });

    group.bench_function("var_tree/iter_take_100", |b| {
        let dir_v = TempDir::new().unwrap();
        let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), page_cache_config()).unwrap();
        for g in 0..NUM_GROUPS {
            for id in 0..SCAN_TAKE as u32 {
                var_tree
                    .put(
                        &make_group_key(g, id),
                        &make_var_value(g as u64 * 1000 + id as u64),
                    )
                    .unwrap();
            }
        }
        var_tree.flush_buffers().unwrap();
        op_batch(b, 1, || {
            black_box(var_tree.iter().take(SCAN_TAKE).collect::<Vec<_>>());
        });
        var_tree.close().unwrap();
    });

    group.bench_function("fixed_tree/prefix_take_100", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        for g in 0..NUM_GROUPS {
            for id in 0..SCAN_TAKE as u32 {
                fixed_tree
                    .put(
                        &make_group_key(g, id),
                        &make_const_value(g as u64 * 1000 + id as u64),
                    )
                    .unwrap();
            }
        }
        let mut rng = rand::rng();
        op_batch(b, 1, || {
            let prefix = rng.random_range(0..NUM_GROUPS).to_be_bytes();
            black_box(
                fixed_tree
                    .prefix_iter(&prefix)
                    .take(SCAN_TAKE)
                    .collect::<Vec<_>>(),
            );
        });
        fixed_tree.close().unwrap();
    });

    group.bench_function("fixed_tree/iter_take_100", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        for g in 0..NUM_GROUPS {
            for id in 0..SCAN_TAKE as u32 {
                fixed_tree
                    .put(
                        &make_group_key(g, id),
                        &make_const_value(g as u64 * 1000 + id as u64),
                    )
                    .unwrap();
            }
        }
        op_batch(b, 1, || {
            black_box(fixed_tree.iter().take(SCAN_TAKE).collect::<Vec<_>>());
        });
        fixed_tree.close().unwrap();
    });

    group.finish();
}

criterion_group! {
    name = benches;
    config = Criterion::default()
        .measurement_time(Duration::from_secs(5))
        .warm_up_time(Duration::from_secs(2));
    targets = bench_seq_write, bench_rand_write, bench_seq_read, bench_rand_read_hot,
        bench_read_write_mix, bench_overwrite, bench_scan,
}

criterion_main!(benches);
