// Copyright 2026 PingCAP, Inc.

#![feature(test)]

use std::{
    fs,
    mem::size_of,
    time::{Duration, Instant},
};

use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, black_box};
use kvengine::{
    LOCK_CF, WRITE_CF,
    table::{
        BIT_DELETE, ChecksumType, InnerKey, SnapVersion,
        memtable::{CfTable, WriteBatch},
        sstable::L0Builder,
    },
};

const ENTRY_COUNT: usize = 20_000;
const SCAN_ENTRY_COUNT: usize = 100_000;
const MEMORY_PROFILE_ROUNDS: usize = 10;

#[derive(Clone, Copy)]
enum Backend {
    Arena,
    Crossbeam,
}

#[derive(Clone, Copy)]
enum FlushIteratorKind {
    Regular,
    Flush,
}

impl FlushIteratorKind {
    fn name(self) -> &'static str {
        match self {
            Self::Regular => "regular",
            Self::Flush => "flush",
        }
    }
}

impl Backend {
    fn name(self) -> &'static str {
        match self {
            Self::Arena => "arena",
            Self::Crossbeam => "crossbeam",
        }
    }

    fn new_table(self) -> CfTable {
        CfTable::with_crossbeam_memtable_for_test(matches!(self, Self::Crossbeam))
    }
}

fn keys_from(start: u64, count: usize) -> Vec<[u8; 8]> {
    (start..start + count as u64)
        .map(u64::to_be_bytes)
        .collect()
}

fn keys() -> Vec<[u8; 8]> {
    keys_from(0, ENTRY_COUNT)
}

fn shuffled_keys() -> Vec<[u8; 8]> {
    let mut keys = keys();
    let mut state = 0x9e37_79b9_7f4a_7c15_u64;
    for index in (1..keys.len()).rev() {
        state ^= state << 13;
        state ^= state >> 7;
        state ^= state << 17;
        keys.swap(index, state as usize % (index + 1));
    }
    keys
}

fn write_batch(keys: &[[u8; 8]], value_size: usize) -> WriteBatch {
    let value = vec![b'x'; value_size];
    let mut batch = WriteBatch::new();
    for key in keys {
        batch.put(InnerKey::from_inner_buf(key), 0, &[], 1, &value);
    }
    batch
}

fn versioned_batch(keys: &[[u8; 8]], value_size: usize, versions_per_key: u64) -> WriteBatch {
    let value = vec![b'x'; value_size];
    let mut batch = WriteBatch::new();
    for version in 1..=versions_per_key {
        for key in keys {
            batch.put(InnerKey::from_inner_buf(key), 0, &[], version, &value);
        }
    }
    batch
}

fn delete_batch(keys: &[[u8; 8]], version: u64) -> WriteBatch {
    let mut batch = WriteBatch::new();
    for key in keys {
        batch.put(InnerKey::from_inner_buf(key), BIT_DELETE, &[], version, &[]);
    }
    batch
}

fn churn_batch(
    deleted_keys: &[[u8; 8]],
    inserted_keys: &[[u8; 8]],
    value_size: usize,
) -> WriteBatch {
    let value = vec![b'x'; value_size];
    let mut batch = delete_batch(deleted_keys, 2);
    for key in inserted_keys {
        batch.put(InnerKey::from_inner_buf(key), 0, &[], 1, &value);
    }
    batch
}

fn populated_table(backend: Backend, keys: &[[u8; 8]], value_size: usize) -> CfTable {
    let table = backend.new_table();
    let mut batch = write_batch(keys, value_size);
    table.get_cf(WRITE_CF).put_batch(&mut batch, None, WRITE_CF);
    table
}

fn versioned_table(
    backend: Backend,
    keys: &[[u8; 8]],
    value_size: usize,
    versions_per_key: u64,
) -> CfTable {
    let table = backend.new_table();
    let mut batch = versioned_batch(keys, value_size, versions_per_key);
    table.get_cf(WRITE_CF).put_batch(&mut batch, None, WRITE_CF);
    table
}

fn build_l0(table: &CfTable, iterator_kind: FlushIteratorKind) -> bytes::Bytes {
    let mut builder = L0Builder::new(
        1,
        16 * 1024,
        SnapVersion::new(1, 1),
        ChecksumType::Crc32c,
        None,
    );
    let mut iterator = match iterator_kind {
        FlushIteratorKind::Regular => table.get_cf(WRITE_CF).new_iterator(false),
        FlushIteratorKind::Flush => table.get_cf(WRITE_CF).new_flush_iterator(),
    };
    iterator.rewind();
    while iterator.valid() {
        let key = iterator.key();
        let value = iterator.value();
        builder.add(WRITE_CF, key, &value, None).unwrap();
        iterator.next_all_version();
    }
    builder.finish().unwrap().1
}

#[derive(Clone, Copy)]
struct MemorySnapshot {
    allocated: usize,
    active: usize,
    resident: usize,
    rss: usize,
}

fn allocator_stats() -> MemorySnapshot {
    let stats = tikv_alloc::fetch_stats()
        .expect("failed to fetch allocator stats")
        .expect("jemalloc stats are unavailable");
    let read = |name| {
        stats
            .iter()
            .find_map(|(stat, value)| (*stat == name).then_some(*value))
            .unwrap()
    };
    MemorySnapshot {
        allocated: read("allocated"),
        active: read("active"),
        resident: read("resident"),
        rss: process_rss(),
    }
}

fn process_rss() -> usize {
    let status = fs::read_to_string("/proc/self/status").expect("failed to read process status");
    let rss_kib = status
        .lines()
        .find_map(|line| line.strip_prefix("VmRSS:"))
        .and_then(|value| value.split_whitespace().next())
        .and_then(|value| value.parse::<usize>().ok())
        .expect("VmRSS is missing from process status");
    rss_kib * 1024
}

fn signed_delta(current: usize, baseline: usize) -> i128 {
    current as i128 - baseline as i128
}

fn print_memory_stats(label: &str, baseline: MemorySnapshot, table: Option<&CfTable>) {
    let current = allocator_stats();
    println!(
        "MEMORY,label={label},submitted_bytes={},accounted_bytes={},allocated_delta_bytes={},active_delta_bytes={},resident_delta_bytes={},rss_delta_bytes={}",
        table.map_or(0, |table| table.size()),
        table.map_or(0, |table| table.memory_usage()),
        signed_delta(current.allocated, baseline.allocated),
        signed_delta(current.active, baseline.active),
        signed_delta(current.resident, baseline.resident),
        signed_delta(current.rss, baseline.rss),
    );
}

fn env_usize(name: &str, default: usize) -> usize {
    std::env::var(name)
        .ok()
        .map(|value| {
            value
                .parse::<usize>()
                .unwrap_or_else(|_| panic!("invalid {}: {}", name, value))
        })
        .unwrap_or(default)
}

fn profile_key(id: u64, key_size: usize) -> Vec<u8> {
    assert!(key_size >= size_of::<u64>());
    let mut key = vec![0; key_size];
    key[key_size - size_of::<u64>()..].copy_from_slice(&id.to_be_bytes());
    key
}

fn memory_profile_batch(
    delete_start: Option<u64>,
    insert_start: Option<u64>,
    entry_count: usize,
    key_size: usize,
    value_size: usize,
) -> WriteBatch {
    let value = vec![b'x'; value_size];
    let mut batch = WriteBatch::new();
    if let Some(start) = delete_start {
        for key in start..start + entry_count as u64 {
            batch.put(
                InnerKey::from_inner_buf(&profile_key(key, key_size)),
                BIT_DELETE,
                &[],
                2,
                &[],
            );
        }
    }
    if let Some(start) = insert_start {
        for key in start..start + entry_count as u64 {
            batch.put(
                InnerKey::from_inner_buf(&profile_key(key, key_size)),
                0,
                &[],
                1,
                &value,
            );
        }
    }
    batch
}

fn run_memory_profile(backend: Backend) {
    let entry_count = env_usize("MEMTABLE_PROFILE_ENTRIES", ENTRY_COUNT);
    let rounds = env_usize("MEMTABLE_PROFILE_ROUNDS", MEMORY_PROFILE_ROUNDS);
    let key_size = env_usize("MEMTABLE_PROFILE_KEY_SIZE", 16);
    let value_size = env_usize("MEMTABLE_PROFILE_VALUE_SIZE", 256);
    println!(
        "CONFIG,mode=fixed-live-set,backend={},entries={entry_count},rounds={rounds},key_size={key_size},value_size={value_size}",
        backend.name(),
    );

    let table = backend.new_table();
    let baseline = allocator_stats();

    let mut batch = memory_profile_batch(None, Some(0), entry_count, key_size, value_size);
    table.get_cf(LOCK_CF).put_batch(&mut batch, None, LOCK_CF);
    drop(batch);
    print_memory_stats("populated", baseline, Some(&table));

    for round in 0..rounds {
        let old_start = round as u64 * entry_count as u64;
        let new_start = (round + 1) as u64 * entry_count as u64;
        let mut batch = memory_profile_batch(
            Some(old_start),
            Some(new_start),
            entry_count,
            key_size,
            value_size,
        );
        table.get_cf(LOCK_CF).put_batch(&mut batch, None, LOCK_CF);
        drop(batch);
        print_memory_stats(&format!("round-{}", round + 1), baseline, Some(&table));
    }

    let current_start = rounds as u64 * entry_count as u64;
    let mut batch =
        memory_profile_batch(Some(current_start), None, entry_count, key_size, value_size);
    table.get_cf(LOCK_CF).put_batch(&mut batch, None, LOCK_CF);
    drop(batch);
    print_memory_stats("drained", baseline, Some(&table));
    drop(table);
    print_memory_stats("dropped", baseline, None);
}

fn memory_overhead_batch(
    first_id: usize,
    entry_count: usize,
    key_size: usize,
    value_size: usize,
    version: u64,
) -> WriteBatch {
    let value = vec![b'x'; value_size];
    let mut batch = WriteBatch::new();
    for id in first_id..first_id + entry_count {
        batch.put(
            InnerKey::from_inner_buf(&profile_key(id as u64, key_size)),
            0,
            &[],
            version,
            &value,
        );
    }
    batch
}

fn memory_overhead_delete_batch(entry_count: usize, key_size: usize, version: u64) -> WriteBatch {
    let mut batch = WriteBatch::new();
    for id in 0..entry_count as u64 {
        batch.put(
            InnerKey::from_inner_buf(&profile_key(id, key_size)),
            BIT_DELETE,
            &[],
            version,
            &[],
        );
    }
    batch
}

fn run_memory_overhead(backend: Backend) {
    let entry_count = env_usize("MEMTABLE_PROFILE_ENTRIES", 100_000);
    let key_size = env_usize("MEMTABLE_PROFILE_KEY_SIZE", 16);
    let value_size = env_usize("MEMTABLE_PROFILE_VALUE_SIZE", 16);
    let versions_per_key = env_usize("MEMTABLE_PROFILE_VERSIONS", 1);
    let batch_entries = env_usize("MEMTABLE_PROFILE_BATCH_ENTRIES", 10_000);
    assert!(batch_entries > 0);
    println!(
        "CONFIG,mode=memory-overhead,backend={},entries={entry_count},key_size={key_size},value_size={value_size},versions_per_key={versions_per_key},batch_entries={batch_entries}",
        backend.name(),
    );

    let table = backend.new_table();
    let baseline = allocator_stats();
    for version in 1..=versions_per_key as u64 {
        for first_id in (0..entry_count).step_by(batch_entries) {
            let chunk_entries = batch_entries.min(entry_count - first_id);
            let mut batch =
                memory_overhead_batch(first_id, chunk_entries, key_size, value_size, version);
            table.get_cf(WRITE_CF).put_batch(&mut batch, None, WRITE_CF);
        }
    }
    print_memory_stats("populated", baseline, Some(&table));

    let mut batch =
        memory_overhead_delete_batch(entry_count, key_size, versions_per_key as u64 + 1);
    table.get_cf(WRITE_CF).put_batch(&mut batch, None, WRITE_CF);
    drop(batch);
    print_memory_stats("removed", baseline, Some(&table));

    drop(table);
    print_memory_stats("dropped", baseline, None);
}

fn hot_key_batch(version_count: usize, value_size: usize) -> WriteBatch {
    let value = vec![b'x'; value_size];
    let mut batch = WriteBatch::new();
    for version in 1..=version_count as u64 {
        batch.put(
            InnerKey::from_inner_buf(b"hot-key"),
            0,
            &[],
            version,
            &value,
        );
    }
    batch
}

fn elapsed_ns(start: Instant) -> u128 {
    start.elapsed().as_nanos()
}

fn run_deep_chain_profile(backend: Backend) {
    let version_count = env_usize("MEMTABLE_PROFILE_VERSIONS", 100_000);
    let value_size = env_usize("MEMTABLE_PROFILE_VALUE_SIZE", 256);
    let repeats = env_usize("MEMTABLE_PROFILE_REPEATS", 5);
    let get_repeats = (10_000_000 / version_count).clamp(20, 20_000);
    println!(
        "CONFIG,mode=deep-chain,backend={},versions={version_count},value_size={value_size},repeats={repeats},get_repeats={get_repeats}",
        backend.name(),
    );

    for repeat in 1..=repeats {
        let mut batch = hot_key_batch(version_count, value_size);
        let table = backend.new_table();

        let start = Instant::now();
        table.get_cf(WRITE_CF).put_batch(&mut batch, None, WRITE_CF);
        let write_ns = elapsed_ns(start);
        drop(batch);

        let mut value_owner = Vec::new();
        let start = Instant::now();
        for _ in 0..get_repeats {
            black_box(table.get_cf(WRITE_CF).get(
                InnerKey::from_inner_buf(b"hot-key"),
                u64::MAX,
                &mut value_owner,
            ));
            value_owner.clear();
        }
        let get_latest_ns = elapsed_ns(start) / get_repeats as u128;

        let start = Instant::now();
        for _ in 0..get_repeats {
            black_box(table.get_cf(WRITE_CF).get(
                InnerKey::from_inner_buf(b"hot-key"),
                1,
                &mut value_owner,
            ));
            value_owner.clear();
        }
        let get_oldest_ns = elapsed_ns(start) / get_repeats as u128;

        let mut iterator = table.get_cf(WRITE_CF).new_iterator(false);
        iterator.rewind();
        let start = Instant::now();
        let mut traversed_versions = 1;
        while iterator.next_version() {
            black_box(iterator.value());
            traversed_versions += 1;
        }
        let traverse_ns = elapsed_ns(start);
        assert_eq!(traversed_versions, version_count);

        let mut delete = WriteBatch::new();
        delete.put(
            InnerKey::from_inner_buf(b"hot-key"),
            BIT_DELETE,
            &[],
            version_count as u64 + 1,
            &[],
        );
        let start = Instant::now();
        table
            .get_cf(WRITE_CF)
            .put_batch(&mut delete, None, WRITE_CF);
        let delete_with_reader_ns = elapsed_ns(start);
        drop(delete);

        let start = Instant::now();
        drop(iterator);
        let last_reader_drop_ns = elapsed_ns(start);

        let start = Instant::now();
        drop(table);
        let table_drop_ns = elapsed_ns(start);

        println!(
            "DEEP_CHAIN,repeat={repeat},write_ns={write_ns},get_latest_ns={get_latest_ns},get_oldest_ns={get_oldest_ns},traverse_ns={traverse_ns},delete_with_reader_ns={delete_with_reader_ns},last_reader_drop_ns={last_reader_drop_ns},table_drop_ns={table_drop_ns}",
        );
    }
}

fn run_flush_build_profile(backend: Backend, iterator_kind: FlushIteratorKind) {
    let entry_count = env_usize("MEMTABLE_PROFILE_ENTRIES", 250_000);
    let key_size = env_usize("MEMTABLE_PROFILE_KEY_SIZE", 32);
    let value_size = env_usize("MEMTABLE_PROFILE_VALUE_SIZE", 256);
    let batch_entries = env_usize("MEMTABLE_PROFILE_BATCH_ENTRIES", 1_024);
    let versions_per_key = env_usize("MEMTABLE_PROFILE_VERSIONS_PER_KEY", 1);
    let repeats = env_usize("MEMTABLE_PROFILE_REPEATS", 5);
    assert!(batch_entries > 0);
    assert!(versions_per_key > 0);
    let iterator = iterator_kind.name();
    println!(
        "CONFIG,mode=flush-build,backend={},iterator={iterator},keys={entry_count},key_size={key_size},value_size={value_size},versions_per_key={versions_per_key},batch_entries={batch_entries},repeats={repeats}",
        backend.name(),
    );

    for repeat in 1..=repeats {
        let table = backend.new_table();
        let mut write_ns = 0;
        for version in 1..=versions_per_key as u64 {
            for first_id in (0..entry_count).step_by(batch_entries) {
                let chunk_entries = batch_entries.min(entry_count - first_id);
                let mut batch =
                    memory_overhead_batch(first_id, chunk_entries, key_size, value_size, version);
                let start = Instant::now();
                table.get_cf(WRITE_CF).put_batch(&mut batch, None, WRITE_CF);
                write_ns += elapsed_ns(start);
            }
        }
        table.seal_for_test();

        let start = Instant::now();
        let l0 = build_l0(&table, iterator_kind);
        let build_ns = elapsed_ns(start);
        println!(
            "FLUSH_BUILD,repeat={repeat},write_ns={write_ns},build_ns={build_ns},output_bytes={}",
            l0.len(),
        );
    }
}

fn main() {
    let flush_mode = std::env::args().find_map(|argument| match argument.as_str() {
        "flush-build-arena" => Some((Backend::Arena, FlushIteratorKind::Flush)),
        "flush-build-crossbeam-regular" => Some((Backend::Crossbeam, FlushIteratorKind::Regular)),
        "flush-build-crossbeam" => Some((Backend::Crossbeam, FlushIteratorKind::Flush)),
        _ => None,
    });
    if let Some((backend, iterator_kind)) = flush_mode {
        run_flush_build_profile(backend, iterator_kind);
        return;
    }

    let mode = std::env::args().find_map(|argument| match argument.as_str() {
        "memory-profile-arena" => Some(("memory-profile", Backend::Arena)),
        "memory-profile-crossbeam" => Some(("memory-profile", Backend::Crossbeam)),
        "memory-overhead-arena" => Some(("memory-overhead", Backend::Arena)),
        "memory-overhead-crossbeam" => Some(("memory-overhead", Backend::Crossbeam)),
        "deep-chain-arena" => Some(("deep-chain", Backend::Arena)),
        "deep-chain-crossbeam" => Some(("deep-chain", Backend::Crossbeam)),
        _ => None,
    });
    if let Some((mode, backend)) = mode {
        match mode {
            "memory-profile" => run_memory_profile(backend),
            "memory-overhead" => run_memory_overhead(backend),
            "deep-chain" => run_deep_chain_profile(backend),
            _ => unreachable!(),
        }
        return;
    }

    let mut criterion = Criterion::default()
        .warm_up_time(Duration::from_secs(1))
        .measurement_time(Duration::from_secs(3))
        .sample_size(20)
        .configure_from_args();
    let ordered_keys = keys();
    let shuffled_keys = shuffled_keys();

    for value_size in [16, 256, 4096] {
        let mut group = criterion.benchmark_group(format!("memtable/write_batch/{value_size}"));
        group.throughput(Throughput::Elements(ENTRY_COUNT as u64));
        group.bench_function("build", |bencher| {
            bencher.iter_with_large_drop(|| write_batch(&ordered_keys, value_size));
        });
        group.finish();
    }

    for (order, keys) in [
        ("ordered", ordered_keys.as_slice()),
        ("shuffled", shuffled_keys.as_slice()),
    ] {
        for value_size in [16, 256, 4096] {
            let mut group =
                criterion.benchmark_group(format!("memtable/write/{order}/{value_size}"));
            group.throughput(Throughput::Elements(ENTRY_COUNT as u64));
            for backend in [Backend::Arena, Backend::Crossbeam] {
                group.bench_with_input(
                    BenchmarkId::from_parameter(backend.name()),
                    &backend,
                    |bencher, backend| {
                        bencher.iter_batched(
                            || (backend.new_table(), write_batch(keys, value_size)),
                            |(table, mut batch)| {
                                table.get_cf(WRITE_CF).put_batch(&mut batch, None, WRITE_CF);
                                (table, batch)
                            },
                            BatchSize::LargeInput,
                        );
                    },
                );
            }
            group.finish();
        }
    }

    let value_size = 256;
    let mut group = criterion.benchmark_group("memtable/write/repeated_versions/256");
    group.throughput(Throughput::Elements(ENTRY_COUNT as u64));
    let repeated_version_keys = &ordered_keys[..ENTRY_COUNT / 10];
    for backend in [Backend::Arena, Backend::Crossbeam] {
        group.bench_with_input(
            BenchmarkId::from_parameter(backend.name()),
            &backend,
            |bencher, backend| {
                bencher.iter_batched(
                    || {
                        (
                            backend.new_table(),
                            versioned_batch(repeated_version_keys, value_size, 10),
                        )
                    },
                    |(table, mut batch)| {
                        table.get_cf(WRITE_CF).put_batch(&mut batch, None, WRITE_CF);
                        (table, batch)
                    },
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();

    let mut group = criterion.benchmark_group("memtable/delete/256");
    group.throughput(Throughput::Elements(ENTRY_COUNT as u64));
    for backend in [Backend::Arena, Backend::Crossbeam] {
        group.bench_with_input(
            BenchmarkId::from_parameter(backend.name()),
            &backend,
            |bencher, backend| {
                bencher.iter_batched(
                    || {
                        (
                            populated_table(*backend, &ordered_keys, value_size),
                            delete_batch(&ordered_keys, 2),
                        )
                    },
                    |(table, mut batch)| {
                        table.get_cf(WRITE_CF).put_batch(&mut batch, None, WRITE_CF);
                        (table, batch)
                    },
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();

    let churn_count = ENTRY_COUNT / 2;
    let inserted_keys = keys_from(ENTRY_COUNT as u64, churn_count);
    let mut group = criterion.benchmark_group("memtable/churn/256");
    group.throughput(Throughput::Elements((churn_count * 2) as u64));
    for backend in [Backend::Arena, Backend::Crossbeam] {
        group.bench_with_input(
            BenchmarkId::from_parameter(backend.name()),
            &backend,
            |bencher, backend| {
                bencher.iter_batched(
                    || {
                        (
                            populated_table(*backend, &ordered_keys, value_size),
                            churn_batch(&ordered_keys[..churn_count], &inserted_keys, value_size),
                        )
                    },
                    |(table, mut batch)| {
                        table.get_cf(WRITE_CF).put_batch(&mut batch, None, WRITE_CF);
                        (table, batch)
                    },
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();

    let mut group = criterion.benchmark_group("memtable/drop/256");
    group.throughput(Throughput::Elements(ENTRY_COUNT as u64));
    for backend in [Backend::Arena, Backend::Crossbeam] {
        group.bench_with_input(
            BenchmarkId::from_parameter(backend.name()),
            &backend,
            |bencher, backend| {
                bencher.iter_batched(
                    || populated_table(*backend, &ordered_keys, value_size),
                    drop,
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();

    let scan_keys = keys_from(0, SCAN_ENTRY_COUNT);
    let missing_keys = keys_from(SCAN_ENTRY_COUNT as u64, SCAN_ENTRY_COUNT);
    let mut group = criterion.benchmark_group("memtable/read");
    group.throughput(Throughput::Elements(SCAN_ENTRY_COUNT as u64));
    for backend in [Backend::Arena, Backend::Crossbeam] {
        let table = populated_table(backend, &scan_keys, value_size);

        group.bench_function(BenchmarkId::new("point_get", backend.name()), |bencher| {
            bencher.iter(|| {
                let mut value_owner = Vec::new();
                for key in &scan_keys {
                    let value = table.get_cf(WRITE_CF).get(
                        InnerKey::from_inner_buf(key),
                        u64::MAX,
                        &mut value_owner,
                    );
                    black_box(value.get_value());
                    value_owner.clear();
                }
            });
        });

        group.bench_function(
            BenchmarkId::new("point_get_miss", backend.name()),
            |bencher| {
                bencher.iter(|| {
                    let mut value_owner = Vec::new();
                    for key in &missing_keys {
                        let value = table.get_cf(WRITE_CF).get(
                            InnerKey::from_inner_buf(key),
                            u64::MAX,
                            &mut value_owner,
                        );
                        black_box(value.get_value());
                        value_owner.clear();
                    }
                });
            },
        );

        group.bench_function(BenchmarkId::new("scan", backend.name()), |bencher| {
            bencher.iter(|| {
                let mut iterator = table.get_cf(WRITE_CF).new_iterator(false);
                iterator.rewind();
                while iterator.valid() {
                    black_box(iterator.key());
                    black_box(iterator.value());
                    iterator.next();
                }
            });
        });

        group.bench_function(
            BenchmarkId::new("scan_reverse", backend.name()),
            |bencher| {
                bencher.iter(|| {
                    let mut iterator = table.get_cf(WRITE_CF).new_iterator(true);
                    iterator.rewind();
                    while iterator.valid() {
                        black_box(iterator.key());
                        black_box(iterator.value());
                        iterator.next();
                    }
                });
            },
        );
    }
    group.finish();

    let hot_key = [[0_u8; 8]];
    for versions_per_key in [1, 10, 100, 1_000] {
        let mut group =
            criterion.benchmark_group(format!("memtable/read_versions/{versions_per_key}"));
        group.throughput(Throughput::Elements(ENTRY_COUNT as u64));
        for backend in [Backend::Arena, Backend::Crossbeam] {
            let table = versioned_table(backend, &hot_key, value_size, versions_per_key);
            group.bench_function(BenchmarkId::new("latest", backend.name()), |bencher| {
                bencher.iter(|| {
                    let mut value_owner = Vec::new();
                    for _ in 0..ENTRY_COUNT {
                        let value = table.get_cf(WRITE_CF).get(
                            InnerKey::from_inner_buf(&hot_key[0]),
                            u64::MAX,
                            &mut value_owner,
                        );
                        black_box(value.get_value());
                        value_owner.clear();
                    }
                });
            });
            group.bench_function(BenchmarkId::new("oldest", backend.name()), |bencher| {
                bencher.iter(|| {
                    let mut value_owner = Vec::new();
                    for _ in 0..ENTRY_COUNT {
                        let value = table.get_cf(WRITE_CF).get(
                            InnerKey::from_inner_buf(&hot_key[0]),
                            1,
                            &mut value_owner,
                        );
                        black_box(value.get_value());
                        value_owner.clear();
                    }
                });
            });
        }
        group.finish();
    }

    let mut group = criterion.benchmark_group("memtable/scan_versions/10");
    group.throughput(Throughput::Elements(SCAN_ENTRY_COUNT as u64));
    for backend in [Backend::Arena, Backend::Crossbeam] {
        let table = versioned_table(backend, &scan_keys, 16, 10);
        for (direction, reversed) in [("forward", false), ("reverse", true)] {
            group.bench_function(BenchmarkId::new(direction, backend.name()), |bencher| {
                bencher.iter(|| {
                    let mut iterator = table.get_cf(WRITE_CF).new_iterator(reversed);
                    iterator.rewind();
                    while iterator.valid() {
                        black_box(iterator.key());
                        black_box(iterator.value());
                        iterator.next();
                    }
                });
            });
        }
    }
    group.finish();

    const SEEK_COUNT: usize = 100;
    for scan_length in [100, 1_000] {
        let seek_span = SCAN_ENTRY_COUNT - 2 * scan_length;
        let seek_keys: Vec<_> = (0..SEEK_COUNT)
            .map(|index| (scan_length + index * 7_919 % seek_span) as u64)
            .map(u64::to_be_bytes)
            .collect();
        let mut group = criterion.benchmark_group(format!("memtable/seek_scan/{scan_length}"));
        group.throughput(Throughput::Elements((SEEK_COUNT * scan_length) as u64));
        for backend in [Backend::Arena, Backend::Crossbeam] {
            let table = populated_table(backend, &scan_keys, value_size);
            for (direction, reversed) in [("forward", false), ("reverse", true)] {
                group.bench_function(BenchmarkId::new(direction, backend.name()), |bencher| {
                    bencher.iter(|| {
                        let mut iterator = table.get_cf(WRITE_CF).new_iterator(reversed);
                        for key in &seek_keys {
                            iterator.seek(InnerKey::from_inner_buf(key));
                            for _ in 0..scan_length {
                                assert!(iterator.valid());
                                black_box(iterator.key());
                                black_box(iterator.value());
                                iterator.next();
                            }
                        }
                    });
                });
            }
        }
        group.finish();
    }

    criterion.final_summary();
}