dbsp 0.343.0

use std::{
    io::Cursor,
    marker::PhantomData,
    sync::{
        Arc, Mutex,
        atomic::{AtomicUsize, Ordering},
    },
};

use crate::{
    DBWeight, Runtime,
    circuit::{CircuitConfig, CircuitStorageConfig},
    dynamic::{DataTrait, DowncastTrait, DynWeight, Factory, LeanVec, Vector, WithFactory},
    storage::{
        backend::{BlockLocation, StorageBackend},
        buffer_cache::BufferCache,
        file::{
            TouchedWindowCount,
            format::{
                BLOOM_FILTER_BLOCK_MAGIC, BatchMetadata, Compression, FileTrailer,
                INCOMPATIBLE_FEATURE_MODULAR_FILTERS, MODULAR_BLOOM_FILTER_BLOCK_MAGIC,
                ROARING_BITMAP_FILTER_BLOCK_MAGIC,
            },
            reader::{BulkRows, FilteredKeys, Reader},
        },
    },
    trace::{
        BatchReaderFactories, Builder, VecIndexedWSetFactories, VecWSetFactories,
        filter::BatchFilters,
        ord::vec::{indexed_wset_batch::VecIndexedWSetBuilder, wset_batch::VecWSetBuilder},
    },
    utils::{Tup1, test::init_test_logger},
};

use super::{
    Factories, FilterPlan,
    reader::{ColumnSpec, RowGroup},
    writer::{Parameters, Writer1, Writer2},
};

use crate::storage::file::FilterKind;
use crate::storage::{backend::StorageError, buffer_cache::FBuf};
use crate::{
    DBData,
    dynamic::{DynData, Erase},
};
use binrw::BinRead;
use feldera_storage::file::FileId;
use feldera_storage::{FileCommitter, FileReader, FileRw, StoragePath};
use feldera_types::config::{StorageConfig, StorageOptions};
use rand::{Rng, seq::SliceRandom, thread_rng};
use tempfile::tempdir;

feldera_macros::declare_tuple! {
    Tup65<
        T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18,
        T19, T20, T21, T22, T23, T24, T25, T26, T27, T28, T29, T30, T31, T32, T33, T34, T35,
        T36, T37, T38, T39, T40, T41, T42, T43, T44, T45, T46, T47, T48, T49, T50, T51, T52,
        T53, T54, T55, T56, T57, T58, T59, T60, T61, T62, T63, T64
    >
}

type OptString = Option<String>;
type Tup65OptString = Tup65<
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
    OptString,
>;

// Map bits to fields in MSB->LSB order so row indices remain lexicographically sorted.
fn bit_set(bits: u128, idx: usize) -> bool {
    debug_assert!(idx < 65);
    ((bits >> (64 - idx)) & 1) != 0
}

fn opt_str(bit: bool) -> Option<String> {
    if bit { Some("abc".to_string()) } else { None }
}

fn tup65_from_bits(bits: u128) -> Tup65OptString {
    Tup65(
        opt_str(bit_set(bits, 0)),
        opt_str(bit_set(bits, 1)),
        opt_str(bit_set(bits, 2)),
        opt_str(bit_set(bits, 3)),
        opt_str(bit_set(bits, 4)),
        opt_str(bit_set(bits, 5)),
        opt_str(bit_set(bits, 6)),
        opt_str(bit_set(bits, 7)),
        opt_str(bit_set(bits, 8)),
        opt_str(bit_set(bits, 9)),
        opt_str(bit_set(bits, 10)),
        opt_str(bit_set(bits, 11)),
        opt_str(bit_set(bits, 12)),
        opt_str(bit_set(bits, 13)),
        opt_str(bit_set(bits, 14)),
        opt_str(bit_set(bits, 15)),
        opt_str(bit_set(bits, 16)),
        opt_str(bit_set(bits, 17)),
        opt_str(bit_set(bits, 18)),
        opt_str(bit_set(bits, 19)),
        opt_str(bit_set(bits, 20)),
        opt_str(bit_set(bits, 21)),
        opt_str(bit_set(bits, 22)),
        opt_str(bit_set(bits, 23)),
        opt_str(bit_set(bits, 24)),
        opt_str(bit_set(bits, 25)),
        opt_str(bit_set(bits, 26)),
        opt_str(bit_set(bits, 27)),
        opt_str(bit_set(bits, 28)),
        opt_str(bit_set(bits, 29)),
        opt_str(bit_set(bits, 30)),
        opt_str(bit_set(bits, 31)),
        opt_str(bit_set(bits, 32)),
        opt_str(bit_set(bits, 33)),
        opt_str(bit_set(bits, 34)),
        opt_str(bit_set(bits, 35)),
        opt_str(bit_set(bits, 36)),
        opt_str(bit_set(bits, 37)),
        opt_str(bit_set(bits, 38)),
        opt_str(bit_set(bits, 39)),
        opt_str(bit_set(bits, 40)),
        opt_str(bit_set(bits, 41)),
        opt_str(bit_set(bits, 42)),
        opt_str(bit_set(bits, 43)),
        opt_str(bit_set(bits, 44)),
        opt_str(bit_set(bits, 45)),
        opt_str(bit_set(bits, 46)),
        opt_str(bit_set(bits, 47)),
        opt_str(bit_set(bits, 48)),
        opt_str(bit_set(bits, 49)),
        opt_str(bit_set(bits, 50)),
        opt_str(bit_set(bits, 51)),
        opt_str(bit_set(bits, 52)),
        opt_str(bit_set(bits, 53)),
        opt_str(bit_set(bits, 54)),
        opt_str(bit_set(bits, 55)),
        opt_str(bit_set(bits, 56)),
        opt_str(bit_set(bits, 57)),
        opt_str(bit_set(bits, 58)),
        opt_str(bit_set(bits, 59)),
        opt_str(bit_set(bits, 60)),
        opt_str(bit_set(bits, 61)),
        opt_str(bit_set(bits, 62)),
        opt_str(bit_set(bits, 63)),
        opt_str(bit_set(bits, 64)),
    )
}

fn test_buffer_cache() -> Option<Arc<BufferCache>> {
    thread_local! {
        static BUFFER_CACHE: Arc<BufferCache> = Arc::new(BufferCache::new(1024 * 1024));
    }
    Some(BUFFER_CACHE.with(|cache| cache.clone()))
}

fn with_roaring_enabled<F>(f: F)
where
    F: FnOnce() + Clone + Send + 'static,
{
    let mut config = CircuitConfig::default();
    config.dev_tweaks.enable_roaring = Some(true);
    let (handle, ()) = Runtime::init_circuit(config, move |_| {
        f();
        Ok(())
    })
    .unwrap();
    handle.kill().unwrap();
}

/// Runs `f` inside a circuit configured with `rate`, which is what both the
/// writer and the filter loader consult.
///
/// The rate goes through the storage options rather than the deprecated dev
/// tweak, so the tests drive the same path production does. The circuit's own
/// storage is unused: `f` reaches its batch files through its own backend.
fn with_false_positive_rate<F>(rate: f64, f: F)
where
    F: FnOnce() + Clone + Send + 'static,
{
    let tempdir = tempdir().unwrap();
    let config = CircuitConfig::default().with_storage(Some(
        CircuitStorageConfig::for_config(
            StorageConfig {
                path: tempdir.path().to_string_lossy().into_owned(),
                cache: Default::default(),
            },
            StorageOptions {
                bloom_false_positive_rate: Some(rate),
                ..StorageOptions::default()
            },
        )
        .unwrap(),
    ));
    let (handle, ()) = Runtime::init_circuit(config, move |_| {
        f();
        Ok(())
    })
    .unwrap();
    handle.kill().unwrap();
}

fn for_each_compression_type<F>(parameters: Parameters, f: F)
where
    F: Fn(Parameters),
{
    for compression in [None, Some(Compression::Snappy)] {
        print!("\n# testing with compression={compression:?}\n\n");
        f(parameters.clone().with_compression(compression));
    }
}

trait TwoColumns {
    type K0: DBData;
    type A0: DBData;
    type K1: DBData;
    type A1: DBWeight;

    fn n0() -> usize;
    fn key0(row0: usize) -> Self::K0;
    fn near0(row0: usize) -> (Self::K0, Self::K0);
    fn aux0(row0: usize) -> Self::A0;

    fn n1(row0: usize) -> usize;
    fn key1(row0: usize, row1: usize) -> Self::K1;
    fn near1(row0: usize, row1: usize) -> (Self::K1, Self::K1);
    fn aux1(row0: usize, row1: usize) -> Self::A1;
}

struct Column0<T> {
    row: usize,
    _phantom: PhantomData<T>,
}

impl<T> Column0<T> {
    pub fn new() -> Self {
        Self {
            row: 0,
            _phantom: PhantomData,
        }
    }
}

impl<T> Iterator for Column0<T>
where
    T: TwoColumns,
{
    type Item = (T::K0, T::A0);

    fn next(&mut self) -> Option<Self::Item> {
        if self.row >= T::n0() {
            None
        } else {
            let retval = Some((T::key0(self.row), T::aux0(self.row)));
            self.row += 1;
            retval
        }
    }
}

struct Column1<T> {
    row0: usize,
    row1: usize,
    _phantom: PhantomData<T>,
}

impl<T> Column1<T>
where
    T: TwoColumns,
{
    fn new() -> Self {
        Self {
            row0: 0,
            row1: 0,
            _phantom: PhantomData,
        }
    }
}

impl<T> Iterator for Column1<T>
where
    T: TwoColumns,
{
    type Item = (T::K1, T::A1);

    fn next(&mut self) -> Option<Self::Item> {
        if self.row0 >= T::n0() {
            None
        } else {
            let retval = Some((T::key1(self.row0, self.row1), T::aux1(self.row0, self.row1)));
            self.row1 += 1;
            if self.row1 >= T::n1(self.row0) {
                self.row0 += 1;
                self.row1 = 0;
            }
            retval
        }
    }
}

fn test_find<K, A, N, T>(
    row_group: &RowGroup<DynData, DynData, N, T>,
    before: &K,
    key: &K,
    after: &K,
    mut aux: A,
) where
    K: DBData,
    A: DBData,
    T: ColumnSpec,
{
    let mut tmp_key = K::default();
    let mut tmp_aux = A::default();
    let (tmp_key, tmp_aux): (&mut DynData, &mut DynData) =
        (tmp_key.erase_mut(), tmp_aux.erase_mut());

    let mut key = key.clone();
    //let (key, aux): (&mut DynData, &mut DynData) = (key.erase_mut(),
    // aux.erase_mut());

    let mut cursor = unsafe { row_group.first().unwrap() };
    unsafe { cursor.advance_to_value_or_larger(key.erase()) }.unwrap();
    assert_eq!(
        unsafe { cursor.item((tmp_key, tmp_aux)) },
        Some((key.erase_mut(), aux.erase_mut()))
    );
    assert_eq!(cursor.key(), Some(key.erase()));

    let mut cursor = unsafe { row_group.first().unwrap() };
    unsafe { cursor.advance_to_value_or_larger(before.erase()) }.unwrap();
    assert_eq!(
        unsafe { cursor.item((tmp_key, tmp_aux)) },
        Some((key.erase_mut(), aux.erase_mut()))
    );
    assert_eq!(cursor.key(), Some(key.erase()));

    let mut cursor = unsafe { row_group.first().unwrap() };
    unsafe { cursor.seek_forward_until(|k| k >= key.erase()) }.unwrap();
    assert_eq!(
        unsafe { cursor.item((tmp_key, tmp_aux)) },
        Some((key.erase_mut(), aux.erase_mut()))
    );
    assert_eq!(cursor.key(), Some(key.erase()));

    let mut cursor = unsafe { row_group.first().unwrap() };
    unsafe { cursor.seek_forward_until(|k| k >= before.erase()) }.unwrap();
    assert_eq!(
        unsafe { cursor.item((tmp_key, tmp_aux)) },
        Some((key.erase_mut(), aux.erase_mut()))
    );
    assert_eq!(cursor.key(), Some(key.erase()));

    let mut cursor = unsafe { row_group.last().unwrap() };
    unsafe { cursor.rewind_to_value_or_smaller(key.erase()) }.unwrap();
    assert_eq!(
        unsafe { cursor.item((tmp_key, tmp_aux)) },
        Some((key.erase_mut(), aux.erase_mut()))
    );
    assert_eq!(cursor.key(), Some(key.erase()));

    let mut cursor = unsafe { row_group.last().unwrap() };
    unsafe { cursor.rewind_to_value_or_smaller(after.erase()) }.unwrap();
    assert_eq!(
        unsafe { cursor.item((tmp_key, tmp_aux)) },
        Some((key.erase_mut(), aux.erase_mut()))
    );
    assert_eq!(cursor.key(), Some(key.erase()));

    let mut cursor = unsafe { row_group.last().unwrap() };
    unsafe { cursor.seek_backward_until(|k| k <= key.erase()) }.unwrap();
    assert_eq!(
        unsafe { cursor.item((tmp_key, tmp_aux)) },
        Some((key.erase_mut(), aux.erase_mut()))
    );
    assert_eq!(cursor.key(), Some(key.erase()));

    let mut cursor = unsafe { row_group.last().unwrap() };
    unsafe { cursor.seek_backward_until(|k| k <= after.erase()) }.unwrap();
    assert_eq!(
        unsafe { cursor.item((tmp_key, tmp_aux)) },
        Some((key.erase_mut(), aux.erase_mut()))
    );
    assert_eq!(cursor.key(), Some(key.erase()));
}

fn test_out_of_range<K, A, N, T>(
    row_group: &RowGroup<DynData, DynData, N, T>,
    before: &K,
    after: &K,
) where
    K: DBData,
    A: DBData,
    T: ColumnSpec,
{
    let mut tmp_key = K::default();
    let mut tmp_aux = A::default();
    let (tmp_key, tmp_aux): (&mut DynData, &mut DynData) =
        (tmp_key.erase_mut(), tmp_aux.erase_mut());

    let mut cursor = unsafe { row_group.first().unwrap() };
    unsafe { cursor.advance_to_value_or_larger(after.erase()) }.unwrap();
    assert_eq!(unsafe { cursor.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(cursor.key(), None);

    unsafe { cursor.move_first() }.unwrap();
    unsafe { cursor.seek_forward_until(|k| k >= after.erase()) }.unwrap();
    assert_eq!(unsafe { cursor.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(cursor.key(), None);

    let mut cursor = unsafe { row_group.last().unwrap() };
    unsafe { cursor.rewind_to_value_or_smaller(before.erase()) }.unwrap();
    assert_eq!(unsafe { cursor.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(cursor.key(), None);

    unsafe { cursor.move_last() }.unwrap();
    unsafe { cursor.seek_backward_until(|k| k <= before.erase()) }.unwrap();
    assert_eq!(unsafe { cursor.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(cursor.key(), None);
}

#[allow(clippy::len_zero)]
fn test_cursor_helper<K, A, N, T>(
    rows: &RowGroup<DynData, DynData, N, T>,
    offset: u64,
    n: usize,
    expected: impl Fn(usize) -> (K, K, K, A),
) where
    K: DBData,
    A: DBData,
    T: ColumnSpec,
{
    let mut tmp_key = K::default();
    let mut tmp_aux = A::default();
    let (tmp_key, tmp_aux): (&mut DynData, &mut DynData) =
        (tmp_key.erase_mut(), tmp_aux.erase_mut());

    assert_eq!(rows.len(), n as u64);

    assert_eq!(rows.len() == 0, rows.is_empty());
    assert_eq!(unsafe { rows.before() }.len(), n as u64);
    assert_eq!(
        unsafe { rows.after() }.len() == 0,
        unsafe { rows.after() }.is_empty()
    );

    if n > 0 {
        let first = unsafe { rows.first().unwrap() };
        let (_before, mut key, _after, mut aux) = expected(0);
        assert_eq!(
            unsafe { first.item((tmp_key, tmp_aux)) },
            Some((key.erase_mut(), aux.erase_mut()))
        );
        assert_eq!(first.key(), Some(key.erase()));

        let last = unsafe { rows.last().unwrap() };
        let (_before, mut key, _after, mut aux) = expected(n - 1);
        assert_eq!(
            unsafe { last.item((tmp_key, tmp_aux)) },
            Some((key.erase_mut(), aux.erase_mut()))
        );
        assert_eq!(last.key(), Some(key.erase()));
    }

    let mut forward = unsafe { rows.before() };
    assert_eq!(unsafe { forward.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(forward.key(), None);
    unsafe { forward.move_prev() }.unwrap();
    assert_eq!(unsafe { forward.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(forward.key(), None);
    unsafe { forward.move_next() }.unwrap();
    for row in 0..n {
        let (_before, mut key, _after, mut aux) = expected(row);
        assert_eq!(
            unsafe { forward.item((tmp_key, tmp_aux)) },
            Some((key.erase_mut(), aux.erase_mut()))
        );
        assert_eq!(forward.key(), Some(key.erase()));
        unsafe { forward.move_next() }.unwrap();
    }
    assert_eq!(unsafe { forward.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(forward.key(), None);
    unsafe { forward.move_next() }.unwrap();
    assert_eq!(unsafe { forward.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(forward.key(), None);

    let mut backward = unsafe { rows.after() };
    assert_eq!(unsafe { backward.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(backward.key(), None);
    unsafe { backward.move_next() }.unwrap();
    assert_eq!(unsafe { backward.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(backward.key(), None);
    unsafe { backward.move_prev() }.unwrap();
    for row in (0..n).rev() {
        let (_before, mut key, _after, mut aux) = expected(row);
        assert_eq!(
            unsafe { backward.item((tmp_key, tmp_aux)) },
            Some((key.erase_mut(), aux.erase_mut()))
        );
        assert_eq!(backward.key(), Some(key.erase()));
        unsafe { backward.move_prev() }.unwrap();
    }
    assert_eq!(unsafe { backward.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(backward.key(), None);
    unsafe { backward.move_prev() }.unwrap();
    assert_eq!(unsafe { backward.item((tmp_key, tmp_aux)) }, None);
    assert_eq!(backward.key(), None);

    for row in 0..n {
        let (before, key, after, aux) = expected(row);
        test_find(rows, &before, &key, &after, aux.clone());
    }

    let mut random = unsafe { rows.before() };
    let mut order: Vec<_> = (0..n + 10).collect();
    order.shuffle(&mut thread_rng());
    for row in order {
        unsafe { random.move_to_row(row as u64) }.unwrap();
        assert_eq!(random.absolute_position(), offset + row.min(n) as u64);
        assert_eq!(random.remaining_rows(), (n - row.min(n)) as u64);
        if row < n {
            let (_before, mut key, _after, mut aux) = expected(row);
            assert_eq!(
                unsafe { random.item((tmp_key, tmp_aux)) },
                Some((key.erase_mut(), aux.erase_mut()))
            );
            assert_eq!(random.key(), Some(key.erase()));
        } else {
            assert_eq!(unsafe { random.item((tmp_key, tmp_aux)) }, None);
            assert_eq!(random.key(), None);
        }
    }

    if n > 0 {
        let (before, _, _, _) = expected(0);
        let (_, _, after, _) = expected(n - 1);
        test_out_of_range::<K, A, N, T>(rows, &before, &after);
    }
}

fn test_cursor<K, A, N, T>(
    rows: &RowGroup<DynData, DynData, N, T>,
    n: usize,
    expected: impl Fn(usize) -> (K, K, K, A),
) where
    K: DBData,
    A: DBData,
    T: ColumnSpec,
{
    let offset = unsafe { rows.before().absolute_position() };
    test_cursor_helper(rows, offset, n, &expected);

    let start = thread_rng().gen_range(0..n);
    let end = thread_rng().gen_range(start..=n);
    let subset = rows.subset(start as u64..end as u64);
    test_cursor_helper(&subset, offset + start as u64, end - start, |index| {
        expected(index + start)
    })
}

fn test_bulk_rows<K, A, N, T>(
    mut bulk: BulkRows<DynData, DynData, N, T>,
    mut expected: impl Iterator<Item = (K, A)>,
) where
    K: DBData,
    A: DBData,
    T: ColumnSpec,
{
    let mut tmp_key = K::default();
    let mut tmp_aux = A::default();
    let (tmp_key, tmp_aux): (&mut DynData, &mut DynData) =
        (tmp_key.erase_mut(), tmp_aux.erase_mut());

    while !bulk.at_eof() {
        bulk.wait().unwrap();
        let (mut key, mut aux) = expected.next().unwrap();
        assert_eq!(
            unsafe { bulk.item((tmp_key, tmp_aux)) },
            Some((key.erase_mut(), aux.erase_mut()))
        );
        bulk.step();
    }
    assert!(expected.next().is_none());
}

fn test_multifetch_zset<K, A>(
    reader: &Reader<(&'static DynData, &'static DynWeight, ())>,
    n: usize,
    expected_fn: impl Fn(usize) -> (K, K, K, A),
) where
    K: DBData,
    A: DBWeight,
{
    let keys_factory: &dyn Factory<dyn Vector<DynData>> = WithFactory::<LeanVec<K>>::FACTORY;

    let vec_wset_factories = VecWSetFactories::new::<K, (), A>();
    let mut expected = VecWSetBuilder::new_builder(&vec_wset_factories);

    let mut keys = keys_factory.default_box();
    for i in 0..n {
        if rand::random() {
            let (_before, key, _after, diff) = (expected_fn)(i);
            keys.push_ref(&key);
            expected.push_val_diff(().erase(), diff.erase());
            expected.push_key(key.erase());
        }
    }
    let expected = expected.done();

    let mut multifetch = reader.fetch_zset(FilteredKeys::all(&*keys)).unwrap();
    while !multifetch.is_done() {
        multifetch.wait().unwrap();
    }
    let output = multifetch.results(vec_wset_factories);
    assert_eq!(&output, &expected);
}

fn test_multifetch_two_columns<T>(
    reader: &Reader<(
        &'static DynData,
        &'static DynData,
        (&'static DynData, &'static DynWeight, ()),
    )>,
) where
    T: TwoColumns,
{
    let keys_factory: &dyn Factory<dyn Vector<DynData>> = WithFactory::<LeanVec<T::K0>>::FACTORY;

    let vec_indexed_wset_factories = VecIndexedWSetFactories::new::<T::K0, T::K1, T::A1>();
    let mut expected = VecIndexedWSetBuilder::new_builder(&vec_indexed_wset_factories);

    let mut keys = keys_factory.default_box();
    for i in 0..T::n0() {
        if rand::random() {
            keys.push_ref(&T::key0(i));

            for j in 0..T::n1(i) {
                let val = T::key1(i, j);
                let weight = T::aux1(i, j);
                expected.push_val_diff(val.erase(), weight.erase());
            }
            let key = T::key0(i);
            expected.push_key(key.erase());
        }
    }
    let expected = expected.done();

    let mut multifetch = reader
        .fetch_indexed_zset(FilteredKeys::all(&*keys))
        .unwrap();
    while !multifetch.is_done() {
        multifetch.wait().unwrap();
    }
    let output = multifetch.results(vec_indexed_wset_factories);
    assert_eq!(&output, &expected);
}

fn test_bloom<K, A>(
    filters: &BatchFilters<DynData>,
    n: usize,
    expected: impl Fn(usize) -> (K, K, K, A),
) where
    K: DBData + Erase<DynData>,
    A: DBData,
{
    let mut false_positives = 0;
    for row in 0..n {
        let (before, key, after, _aux) = expected(row);
        assert!(filters.maybe_contains_key(key.erase(), None));
        if filters.maybe_contains_key(before.erase(), None) {
            false_positives += 1;
        }
        if filters.maybe_contains_key(after.erase(), None) {
            false_positives += 1;
        }
    }
    if n >= 5 {
        // Note that, usually, `after` for row `i` is the same as `before`
        // for row `i + 1`, so the values in the data are not necessarily
        // *unique* values.
        assert!(
            false_positives < n,
            "Out of {} values not in the data, {} appeared in the Bloom filter ({:.1}% false positive rate)",
            2 * n,
            false_positives,
            false_positives as f64 / (2 * n) as f64
        );
    }
}

fn test_key_range<K, A, Aux, N>(
    reader: &Reader<(&'static DynData, &'static Aux, N)>,
    n: usize,
    expected: impl Fn(usize) -> (K, K, K, A),
) where
    K: DBData,
    A: DBData,
    Aux: DataTrait + ?Sized,
    N: ColumnSpec,
{
    let key_range = reader.key_range().unwrap();
    if n == 0 {
        assert!(key_range.is_none());
        return;
    }

    let Some((min, max)) = key_range else {
        panic!("expected non-empty key range");
    };
    let (_, expected_min, _, _) = expected(0);
    let (_, expected_max, _, _) = expected(n - 1);
    assert_eq!(min.downcast_checked::<K>(), &expected_min);
    assert_eq!(max.downcast_checked::<K>(), &expected_max);
}

fn filter_block_magic<K, A, N>(reader: &Reader<(&'static K, &'static A, N)>) -> Option<[u8; 4]>
where
    K: DataTrait + ?Sized,
    A: DataTrait + ?Sized,
    (&'static K, &'static A, N): ColumnSpec,
{
    let file_size = reader.byte_size().unwrap() as usize;
    let trailer_block = reader
        .file_handle()
        .read_block(BlockLocation::new((file_size - 512) as u64, 512).unwrap())
        .unwrap();
    let trailer = FileTrailer::read_le(&mut Cursor::new(trailer_block.as_slice())).unwrap();
    let offset = if trailer.has_filter64() {
        trailer.filter_offset64
    } else {
        trailer.filter_offset
    };
    let size = if trailer.has_filter64() {
        trailer.filter_size64 as usize
    } else {
        trailer.filter_size as usize
    };
    if offset == 0 {
        return None;
    }

    let filter_block = reader
        .file_handle()
        .read_block(BlockLocation::new(offset, size).unwrap())
        .unwrap();
    let mut magic = [0u8; 4];
    magic.copy_from_slice(&filter_block[4..8]);
    Some(magic)
}

fn incompatible_features<K, A, N>(reader: &Reader<(&'static K, &'static A, N)>) -> u64
where
    K: DataTrait + ?Sized,
    A: DataTrait + ?Sized,
    (&'static K, &'static A, N): ColumnSpec,
{
    let file_size = reader.byte_size().unwrap() as usize;
    let trailer_block = reader
        .file_handle()
        .read_block(BlockLocation::new((file_size - 512) as u64, 512).unwrap())
        .unwrap();
    let trailer = FileTrailer::read_le(&mut Cursor::new(trailer_block.as_slice())).unwrap();
    trailer.incompatible_features
}

fn filter_plan_from_keys<K>(keys: &[K]) -> FilterPlan<DynData>
where
    K: DBData + Erase<DynData>,
{
    FilterPlan::from_bounds(keys.first().unwrap().erase(), keys.last().unwrap().erase())
}

fn test_two_columns<T>(parameters: Parameters)
where
    T: TwoColumns,
{
    let factories0 = Factories::<DynData, DynData>::new::<T::K0, T::A0>();
    let factories1 = Factories::<DynData, DynData>::new::<T::K1, T::A1>();

    let tempdir = tempdir().unwrap();
    let storage_backend = <dyn StorageBackend>::new(
        &StorageConfig {
            path: tempdir.path().to_string_lossy().to_string(),
            cache: Default::default(),
        },
        &StorageOptions::default(),
    )
    .unwrap();
    let mut layer_file = Writer2::new(
        &factories0,
        &factories1,
        test_buffer_cache,
        &*storage_backend,
        parameters,
        FilterPlan::<DynData>::decide_filter(None, T::n0()),
    )
    .unwrap();
    let n0 = T::n0();
    for row0 in 0..n0 {
        for row1 in 0..T::n1(row0) {
            layer_file
                .write1((&T::key1(row0, row1), &T::aux1(row0, row1)))
                .unwrap();
        }
        layer_file.write0((&T::key0(row0), &T::aux0(row0))).unwrap();
    }

    let (reader, filters) = layer_file.into_reader(BatchMetadata::default()).unwrap();
    reader.evict();
    let rows0 = reader.rows();
    let expected0 = |row0| {
        let key0 = T::key0(row0);
        let (before0, after0) = T::near0(row0);
        let aux0 = T::aux0(row0);
        (before0, key0, after0, aux0)
    };
    test_cursor(&rows0, n0, expected0);
    test_bloom(&filters, n0, expected0);
    test_key_range(&reader, n0, expected0);

    let expected1 = |row0, row1| {
        let key1 = T::key1(row0, row1);
        let (before1, after1) = T::near1(row0, row1);
        let aux1 = T::aux1(row0, row1);
        (before1, key1, after1, aux1)
    };
    for row0 in 0..n0 {
        let rows1 = rows0.nth(row0 as u64).unwrap().next_column().unwrap();
        let n1 = T::n1(row0);
        test_cursor(&rows1, n1, |row1| expected1(row0, row1));
    }
    test_bulk_rows(reader.bulk_rows().unwrap(), Column0::<T>::new());
    test_bulk_rows(
        reader.bulk_rows().unwrap().next_column().unwrap(),
        Column1::<T>::new(),
    );
}

fn test_two_columns_multifetch<T>(parameters: Parameters)
where
    T: TwoColumns,
{
    let factories0 = Factories::<DynData, DynData>::new::<T::K0, T::A0>();
    let factories1 = Factories::<DynData, DynWeight>::new::<T::K1, T::A1>();

    let tempdir = tempdir().unwrap();
    let storage_backend = <dyn StorageBackend>::new(
        &StorageConfig {
            path: tempdir.path().to_string_lossy().to_string(),
            cache: Default::default(),
        },
        &StorageOptions::default(),
    )
    .unwrap();
    let mut layer_file = Writer2::new(
        &factories0,
        &factories1,
        test_buffer_cache,
        &*storage_backend,
        parameters,
        FilterPlan::<DynData>::decide_filter(None, T::n0()),
    )
    .unwrap();
    let n0 = T::n0();
    for row0 in 0..n0 {
        for row1 in 0..T::n1(row0) {
            layer_file
                .write1((&T::key1(row0, row1), &T::aux1(row0, row1)))
                .unwrap();
        }
        layer_file.write0((&T::key0(row0), &T::aux0(row0))).unwrap();
    }

    let (reader, _filters) = layer_file.into_reader(BatchMetadata::default()).unwrap();
    reader.evict();
    test_multifetch_two_columns::<T>(&reader);
}

fn test_2_columns_helper(parameters: Parameters) {
    struct TwoInts;
    impl TwoColumns for TwoInts {
        type K0 = i32;
        type A0 = u64;
        type K1 = i32;
        type A1 = u64;

        fn n0() -> usize {
            500
        }
        fn key0(row0: usize) -> Self::K0 {
            row0 as i32 * 2
        }
        fn near0(row0: usize) -> (Self::K0, Self::K0) {
            let key0 = Self::key0(row0);
            (key0 - 1, key0 + 1)
        }
        fn aux0(_row0: usize) -> Self::A0 {
            0x1111
        }

        fn n1(_row0: usize) -> usize {
            7
        }
        fn key1(row0: usize, row1: usize) -> Self::K1 {
            (row0 + row1 * 2) as i32
        }
        fn near1(row0: usize, row1: usize) -> (Self::K1, Self::K1) {
            let key1 = Self::key1(row0, row1);
            (key1 - 1, key1 + 1)
        }
        fn aux1(_row0: usize, _row1: usize) -> Self::A1 {
            0x2222
        }
    }
    test_two_columns::<TwoInts>(parameters.clone());
    test_two_columns_multifetch::<TwoInts>(parameters);
}

#[test]
fn two_columns_uncompressed() {
    init_test_logger();
    test_2_columns_helper(Parameters::default().with_compression(None));
}

#[test]
fn two_columns_snappy() {
    init_test_logger();
    test_2_columns_helper(Parameters::default().with_compression(Some(Compression::Snappy)));
}

#[test]
fn two_columns_max_branch_2_uncompressed() {
    init_test_logger();
    test_2_columns_helper(
        Parameters::default()
            .with_max_branch(2)
            .with_compression(None),
    );
}

#[test]
fn two_columns_max_branch_2_snappy() {
    init_test_logger();
    test_2_columns_helper(
        Parameters::default()
            .with_max_branch(2)
            .with_compression(Some(Compression::Snappy)),
    );
}

struct OneColumn<'a, T> {
    expected: &'a T,
    row: usize,
    n: usize,
}

impl<'a, T> OneColumn<'a, T> {
    pub fn new(expected: &'a T, n: usize) -> Self {
        Self {
            expected,
            row: 0,
            n,
        }
    }
}

impl<'a, T, K, A> Iterator for OneColumn<'a, T>
where
    T: Fn(usize) -> (K, K, K, A),
{
    type Item = (K, A);

    fn next(&mut self) -> Option<Self::Item> {
        if self.row >= self.n {
            None
        } else {
            let (_before, key, _after, aux) = (self.expected)(self.row);
            let retval = Some((key, aux));
            self.row += 1;
            retval
        }
    }
}

fn test_one_column<K, A>(n: usize, expected: impl Fn(usize) -> (K, K, K, A), parameters: Parameters)
where
    K: DBData,
    A: DBData,
{
    for reopen in [false, true] {
        let factories = Factories::<DynData, DynData>::new::<K, A>();
        let tempdir = tempdir().unwrap();
        let storage_backend = <dyn StorageBackend>::new(
            &StorageConfig {
                path: tempdir.path().to_string_lossy().to_string(),
                cache: Default::default(),
            },
            &StorageOptions::default(),
        )
        .unwrap();
        let mut writer = Writer1::new(
            &factories,
            test_buffer_cache,
            &*storage_backend,
            parameters.clone(),
            FilterPlan::<DynData>::decide_filter(None, n),
        )
        .unwrap();
        for row in 0..n {
            let (_before, key, _after, aux) = expected(row);
            writer.write0((&key, &aux)).unwrap();
        }

        let (reader, filters) = if reopen {
            println!("closing writer and reopening as reader");
            let path = writer.path().clone();
            let (_file_handle, _key_filter, _key_bounds) =
                writer.close(BatchMetadata::default()).unwrap();
            let (reader, membership_filter) = Reader::open_with_filter(
                &[&factories.any_factories()],
                test_buffer_cache,
                &*storage_backend,
                &path,
            )
            .unwrap();
            let key_range = reader.key_range().unwrap().map(Into::into);
            let filters = BatchFilters::from_file(key_range, membership_filter);
            (reader, filters)
        } else {
            println!("transforming writer into reader");
            let (reader, filters) = writer.into_reader(BatchMetadata::default()).unwrap();
            (reader, filters)
        };
        reader.evict();
        assert_eq!(reader.rows().len(), n as u64);
        test_cursor(&reader.rows(), n, &expected);
        test_bloom(&filters, n, &expected);
        test_key_range(&reader, n, &expected);
        test_bulk_rows(reader.bulk_rows().unwrap(), OneColumn::new(&expected, n));
    }
}

fn test_one_column_zset<K, A>(
    n: usize,
    expected: impl Fn(usize) -> (K, K, K, A),
    parameters: Parameters,
) where
    K: DBData,
    A: DBWeight,
{
    for reopen in [false, true] {
        let factories = Factories::<DynData, DynWeight>::new::<K, A>();
        let tempdir = tempdir().unwrap();
        let storage_backend = <dyn StorageBackend>::new(
            &StorageConfig {
                path: tempdir.path().to_string_lossy().to_string(),
                cache: Default::default(),
            },
            &StorageOptions::default(),
        )
        .unwrap();
        let mut writer = Writer1::new(
            &factories,
            test_buffer_cache,
            &*storage_backend,
            parameters.clone(),
            FilterPlan::<DynData>::decide_filter(None, n),
        )
        .unwrap();
        for row in 0..n {
            let (_before, key, _after, aux) = expected(row);
            writer.write0((&key, &aux)).unwrap();
        }

        let reader = if reopen {
            println!("closing writer and reopening as reader");
            let path = writer.path().clone();
            let (_file_handle, _key_filter, _key_bounds) =
                writer.close(BatchMetadata::default()).unwrap();
            Reader::open(
                &[&factories.any_factories()],
                test_buffer_cache,
                &*storage_backend,
                &path,
            )
            .unwrap()
        } else {
            println!("transforming writer into reader");
            let (reader, _filters) = writer.into_reader(BatchMetadata::default()).unwrap();
            reader
        };
        reader.evict();
        assert_eq!(reader.rows().len(), n as u64);
        test_key_range(&reader, n, &expected);
        test_multifetch_zset(&reader, n, &expected);
    }
}

#[test]
fn one_column_key_range() {
    init_test_logger();

    for reopen in [false, true] {
        for (label, keys) in [
            ("negative", [-30_i64, -20, -10]),
            ("positive", [10_i64, 20, 30]),
            ("mixed", [-30_i64, 0, 20]),
        ] {
            let factories = Factories::<DynData, DynData>::new::<i64, ()>();
            let tempdir = tempdir().unwrap();
            let storage_backend = <dyn StorageBackend>::new(
                &StorageConfig {
                    path: tempdir.path().to_string_lossy().to_string(),
                    cache: Default::default(),
                },
                &StorageOptions::default(),
            )
            .unwrap();
            let mut writer = Writer1::new(
                &factories,
                test_buffer_cache,
                &*storage_backend,
                Parameters::default(),
                FilterPlan::<DynData>::decide_filter(None, keys.len()),
            )
            .unwrap();
            for key in keys {
                writer.write0((&key, &())).unwrap();
            }

            let reader = if reopen {
                let path = writer.path().clone();
                let (_file_handle, _key_filter, _key_bounds) =
                    writer.close(BatchMetadata::default()).unwrap();
                Reader::open(
                    &[&factories.any_factories()],
                    test_buffer_cache,
                    &*storage_backend,
                    &path,
                )
                .unwrap()
            } else {
                let (reader, _filters) = writer.into_reader(BatchMetadata::default()).unwrap();
                reader
            };

            let Some((min, max)) = reader.key_range().unwrap() else {
                panic!("expected non-empty key range for {label}");
            };
            assert_eq!(*min.downcast_checked::<i64>(), keys[0], "{label}");
            assert_eq!(
                *max.downcast_checked::<i64>(),
                keys[keys.len() - 1],
                "{label}"
            );
        }
    }
}

#[test]
fn persists_touched_window_count_in_metadata() {
    let factories = Factories::<DynData, DynData>::new::<u32, ()>();
    let tempdir = tempdir().unwrap();
    let storage_backend = <dyn StorageBackend>::new(
        &StorageConfig {
            path: tempdir.path().to_string_lossy().to_string(),
            cache: Default::default(),
        },
        &StorageOptions::default(),
    )
    .unwrap();
    let mut writer = Writer1::new(
        &factories,
        test_buffer_cache,
        &*storage_backend,
        Parameters::default(),
        FilterPlan::<DynData>::decide_filter(None, 6),
    )
    .unwrap();

    for key in [0u32, 1, 1 << 16, (1 << 16) + 1, 3 << 16, (3 << 16) + 1] {
        writer.write0((&key, &())).unwrap();
    }

    let metadata = BatchMetadata {
        touched_window_count: TouchedWindowCount::new(3),
        ..BatchMetadata::default()
    };
    let (reader, _filters) = writer.into_reader(metadata).unwrap();
    assert_eq!(
        reader.metadata().touched_window_count,
        TouchedWindowCount::new(3)
    );
}

#[test]
fn bloom_filter_roundtrip_and_block_kind() {
    init_test_logger();

    for reopen in [false, true] {
        let factories = Factories::<DynData, DynData>::new::<i64, ()>();
        let tempdir = tempdir().unwrap();
        let storage_backend = <dyn StorageBackend>::new(
            &StorageConfig {
                path: tempdir.path().to_string_lossy().to_string(),
                cache: Default::default(),
            },
            &StorageOptions::default(),
        )
        .unwrap();

        let mut writer = Writer1::new(
            &factories,
            test_buffer_cache,
            &*storage_backend,
            Parameters::default(),
            FilterPlan::<DynData>::decide_filter(None, 3),
        )
        .unwrap();
        for key in [1i64, 3, 7] {
            writer.write0((&key, &())).unwrap();
        }

        let (reader, filters) = if reopen {
            let path = writer.path().clone();
            let (_file_handle, _key_filter, _key_bounds) =
                writer.close(BatchMetadata::default()).unwrap();
            let (reader, membership_filter) = Reader::open_with_filter(
                &[&factories.any_factories()],
                test_buffer_cache,
                &*storage_backend,
                &path,
            )
            .unwrap();
            let key_range = reader.key_range().unwrap().map(Into::into);
            let filters = BatchFilters::from_file(key_range, membership_filter);
            (reader, filters)
        } else {
            writer.into_reader(BatchMetadata::default()).unwrap()
        };

        for key in [1i64, 3, 7] {
            assert!(filters.maybe_contains_key(key.erase(), None));
        }
        assert_eq!(
            filter_block_magic(&reader),
            Some(MODULAR_BLOOM_FILTER_BLOCK_MAGIC)
        );
        // A file carrying more than one module is deliberately unreadable by
        // binaries that predate them, so they fail loudly rather than mis-parse.
        assert_eq!(
            incompatible_features(&reader),
            INCOMPATIBLE_FEATURE_MODULAR_FILTERS
        );
    }
}

#[test]
fn roaring_u32_filter_roundtrip_exact_and_block_kind() {
    init_test_logger();

    with_roaring_enabled(|| {
        for reopen in [false, true] {
            let factories = Factories::<DynData, DynData>::new::<u32, ()>();
            let tempdir = tempdir().unwrap();
            let storage_backend = <dyn StorageBackend>::new(
                &StorageConfig {
                    path: tempdir.path().to_string_lossy().to_string(),
                    cache: Default::default(),
                },
                &StorageOptions::default(),
            )
            .unwrap();

            let filter_plan = filter_plan_from_keys(&[1u32, 3, 7]);
            let mut writer = Writer1::new(
                &factories,
                test_buffer_cache,
                &*storage_backend,
                Parameters::default(),
                FilterPlan::decide_filter(Some(&filter_plan), 3),
            )
            .unwrap();
            for key in [1u32, 3, 7] {
                writer.write0((&key, &())).unwrap();
            }

            let (reader, filters) = if reopen {
                let path = writer.path().clone();
                let (_file_handle, _key_filter, _key_bounds) =
                    writer.close(BatchMetadata::default()).unwrap();
                let (reader, membership_filter) = Reader::open_with_filter(
                    &[&factories.any_factories()],
                    test_buffer_cache,
                    &*storage_backend,
                    &path,
                )
                .unwrap();
                let key_range = reader.key_range().unwrap().map(Into::into);
                let filters = BatchFilters::from_file(key_range, membership_filter);
                (reader, filters)
            } else {
                writer.into_reader(BatchMetadata::default()).unwrap()
            };

            for key in [1u32, 3, 7] {
                assert!(filters.maybe_contains_key(key.erase(), None));
            }
            for key in [0u32, 2, 9] {
                assert!(!filters.maybe_contains_key(key.erase(), None));
            }
            assert_eq!(
                filter_block_magic(&reader),
                Some(ROARING_BITMAP_FILTER_BLOCK_MAGIC)
            );
            assert_ne!(incompatible_features(&reader), 0);
        }
    });
}

#[test]
fn roaring_tup1_i32_filter_roundtrip_exact_and_block_kind() {
    init_test_logger();

    with_roaring_enabled(|| {
        for reopen in [false, true] {
            let factories = Factories::<DynData, DynData>::new::<Tup1<i32>, ()>();
            let tempdir = tempdir().unwrap();
            let storage_backend = <dyn StorageBackend>::new(
                &StorageConfig {
                    path: tempdir.path().to_string_lossy().to_string(),
                    cache: Default::default(),
                },
                &StorageOptions::default(),
            )
            .unwrap();

            let filter_plan = filter_plan_from_keys(&[Tup1(-7i32), Tup1(1), Tup1(3)]);
            let mut writer = Writer1::new(
                &factories,
                test_buffer_cache,
                &*storage_backend,
                Parameters::default(),
                FilterPlan::decide_filter(Some(&filter_plan), 3),
            )
            .unwrap();
            for key in [Tup1(-7i32), Tup1(1), Tup1(3)] {
                writer.write0((&key, &())).unwrap();
            }

            let (reader, filters) = if reopen {
                let path = writer.path().clone();
                let (_file_handle, _key_filter, _key_bounds) =
                    writer.close(BatchMetadata::default()).unwrap();
                let (reader, membership_filter) = Reader::open_with_filter(
                    &[&factories.any_factories()],
                    test_buffer_cache,
                    &*storage_backend,
                    &path,
                )
                .unwrap();
                let key_range = reader.key_range().unwrap().map(Into::into);
                let filters = BatchFilters::from_file(key_range, membership_filter);
                (reader, filters)
            } else {
                writer.into_reader(BatchMetadata::default()).unwrap()
            };

            for key in [Tup1(-7i32), Tup1(1), Tup1(3)] {
                assert!(filters.maybe_contains_key(key.erase(), None));
            }
            for key in [Tup1(-8i32), Tup1(0), Tup1(9)] {
                assert!(!filters.maybe_contains_key(key.erase(), None));
            }
            assert_eq!(
                filter_block_magic(&reader),
                Some(ROARING_BITMAP_FILTER_BLOCK_MAGIC)
            );
            assert_ne!(incompatible_features(&reader), 0);
        }
    });
}

#[test]
fn writer_without_filter_plan_uses_bloom_filter() {
    init_test_logger();

    with_roaring_enabled(|| {
        let factories = Factories::<DynData, DynData>::new::<u32, ()>();
        let tempdir = tempdir().unwrap();
        let storage_backend = <dyn StorageBackend>::new(
            &StorageConfig {
                path: tempdir.path().to_string_lossy().to_string(),
                cache: Default::default(),
            },
            &StorageOptions::default(),
        )
        .unwrap();

        let mut writer = Writer1::new(
            &factories,
            test_buffer_cache,
            &*storage_backend,
            Parameters::default(),
            FilterPlan::<DynData>::decide_filter(None, 2),
        )
        .unwrap();
        for key in [5u32, 8] {
            writer.write0((&key, &())).unwrap();
        }

        let (reader, _filters) = writer.into_reader(BatchMetadata::default()).unwrap();
        assert_eq!(
            filter_block_magic(&reader),
            Some(MODULAR_BLOOM_FILTER_BLOCK_MAGIC)
        );
    });
}

#[test]
fn roaring_i64_filter_roundtrip_uses_batch_min_offset() {
    init_test_logger();

    with_roaring_enabled(|| {
        for reopen in [false, true] {
            let factories = Factories::<DynData, DynData>::new::<i64, ()>();
            let tempdir = tempdir().unwrap();
            let storage_backend = <dyn StorageBackend>::new(
                &StorageConfig {
                    path: tempdir.path().to_string_lossy().to_string(),
                    cache: Default::default(),
                },
                &StorageOptions::default(),
            )
            .unwrap();

            let min = (i64::from(u32::MAX) * 4) + 10;
            let keys = [min, min + 3, min + 7];
            let filter_plan = filter_plan_from_keys(&keys);
            let mut writer = Writer1::new(
                &factories,
                test_buffer_cache,
                &*storage_backend,
                Parameters::default(),
                FilterPlan::decide_filter(Some(&filter_plan), keys.len()),
            )
            .unwrap();
            for key in keys {
                writer.write0((&key, &())).unwrap();
            }

            let (reader, filters) = if reopen {
                let path = writer.path().clone();
                let (_file_handle, _key_filter, _key_bounds) =
                    writer.close(BatchMetadata::default()).unwrap();
                let (reader, membership_filter) = Reader::open_with_filter(
                    &[&factories.any_factories()],
                    test_buffer_cache,
                    &*storage_backend,
                    &path,
                )
                .unwrap();
                let key_range = reader.key_range().unwrap().map(Into::into);
                let filters = BatchFilters::from_file(key_range, membership_filter);
                (reader, filters)
            } else {
                writer.into_reader(BatchMetadata::default()).unwrap()
            };

            for key in keys {
                assert!(filters.maybe_contains_key((&key) as &DynData, None));
            }
            for key in [min - 1, min + 4, min + 9] {
                assert!(!filters.maybe_contains_key((&key) as &DynData, None));
            }
            assert_eq!(
                filter_block_magic(&reader),
                Some(ROARING_BITMAP_FILTER_BLOCK_MAGIC)
            );
        }
    });
}

#[test]
fn roaring_u64_filter_roundtrip_uses_batch_min_offset() {
    init_test_logger();

    with_roaring_enabled(|| {
        let factories = Factories::<DynData, DynData>::new::<u64, ()>();
        let tempdir = tempdir().unwrap();
        let storage_backend = <dyn StorageBackend>::new(
            &StorageConfig {
                path: tempdir.path().to_string_lossy().to_string(),
                cache: Default::default(),
            },
            &StorageOptions::default(),
        )
        .unwrap();

        let base = (u64::from(u32::MAX) << 8) + 11;
        let keys = [base, base + 2, base + 9];
        let filter_plan = filter_plan_from_keys(&keys);
        let mut writer = Writer1::new(
            &factories,
            test_buffer_cache,
            &*storage_backend,
            Parameters::default(),
            FilterPlan::decide_filter(Some(&filter_plan), keys.len()),
        )
        .unwrap();
        for key in keys {
            writer.write0((&key, &())).unwrap();
        }

        let (reader, filters) = writer.into_reader(BatchMetadata::default()).unwrap();
        for key in keys {
            assert!(filters.maybe_contains_key((&key) as &DynData, None));
        }
        for key in [base - 1, base + 3, base + 20] {
            assert!(!filters.maybe_contains_key((&key) as &DynData, None));
        }
        assert_eq!(
            filter_block_magic(&reader),
            Some(ROARING_BITMAP_FILTER_BLOCK_MAGIC)
        );
    });
}

#[test]
fn i64_keys_fallback_to_bloom_when_span_exceeds_u32() {
    init_test_logger();

    with_roaring_enabled(|| {
        let factories = Factories::<DynData, DynData>::new::<i64, ()>();
        let tempdir = tempdir().unwrap();
        let storage_backend = <dyn StorageBackend>::new(
            &StorageConfig {
                path: tempdir.path().to_string_lossy().to_string(),
                cache: Default::default(),
            },
            &StorageOptions::default(),
        )
        .unwrap();

        let max = i64::from(u32::MAX) + 1;
        let filter_plan = FilterPlan::from_bounds((&0i64) as &DynData, (&max) as &DynData);
        let mut writer = Writer1::new(
            &factories,
            test_buffer_cache,
            &*storage_backend,
            Parameters::default(),
            FilterPlan::decide_filter(Some(&filter_plan), 2),
        )
        .unwrap();
        for key in [0i64, max] {
            writer.write0((&key, &())).unwrap();
        }

        let (reader, _filters) = writer.into_reader(BatchMetadata::default()).unwrap();
        assert_eq!(
            filter_block_magic(&reader),
            Some(MODULAR_BLOOM_FILTER_BLOCK_MAGIC)
        );
    });
}

fn test_i64_helper(parameters: Parameters) {
    init_test_logger();
    test_one_column(
        1000,
        |row| (row as u64 * 2, row as u64 * 2 + 1, row as u64 * 2 + 2, ()),
        parameters,
    );
}

#[test]
fn test_i64() {
    for_each_compression_type(Parameters::default(), test_i64_helper);
}

#[test]
fn test_i64_max_branch_32() {
    for_each_compression_type(Parameters::default().with_max_branch(32), test_i64_helper);
}

#[test]
fn test_i64_max_branch_3_uncompressed() {
    test_i64_helper(
        Parameters::default()
            .with_max_branch(3)
            .with_compression(None),
    );
}

#[test]
fn test_i64_max_branch_3_snappy() {
    test_i64_helper(
        Parameters::default()
            .with_max_branch(3)
            .with_compression(Some(Compression::Snappy)),
    );
}

#[test]
fn test_i64_max_branch_2_uncompressed() {
    test_i64_helper(
        Parameters::default()
            .with_max_branch(2)
            .with_compression(None),
    );
}

#[test]
fn test_i64_max_branch_2_snappy() {
    test_i64_helper(
        Parameters::default()
            .with_max_branch(2)
            .with_compression(Some(Compression::Snappy)),
    );
}

#[test]
fn test_string() {
    fn f(x: usize) -> String {
        format!("{x:09}")
    }

    init_test_logger();
    for_each_compression_type(Parameters::default(), |parameters| {
        test_one_column(
            1000,
            |row| (f(row * 2), f(row * 2 + 1), f(row * 2 + 2), ()),
            parameters,
        )
    });
}

#[test]
fn test_tuple() {
    init_test_logger();
    for_each_compression_type(Parameters::default(), |parameters| {
        let expected = |row| {
            (
                (row as u64, 0),
                (row as u64, 1),
                (row as u64, 2),
                row as u64 + 1,
            )
        };
        test_one_column(1000, &expected, parameters.clone());
        test_one_column_zset(1000, &expected, parameters);
    });
}

#[test]
fn test_tup65_option_string() {
    init_test_logger();
    for_each_compression_type(Parameters::default(), |parameters| {
        test_one_column(
            1_000usize,
            |row| {
                let bits = row as u128 * 2 + 1;
                let before = tup65_from_bits(bits - 1);
                let key = tup65_from_bits(bits);
                let after = tup65_from_bits(bits + 1);
                (before, key, after, ())
            },
            parameters,
        );
    });
}

#[test]
fn test_big_values() {
    fn v(row: usize) -> Vec<i64> {
        (0..row as i64).collect()
    }
    init_test_logger();
    for_each_compression_type(Parameters::default(), |parameters| {
        test_one_column(
            500,
            |row| (v(row * 2), v(row * 2 + 1), v(row * 2 + 2), ()),
            parameters,
        )
    });
}

/// Builds a storage backend rooted at `dir`.
fn backend_at(dir: &str) -> Arc<dyn StorageBackend> {
    <dyn StorageBackend>::new(
        &StorageConfig {
            path: dir.to_string(),
            cache: Default::default(),
        },
        &StorageOptions::default(),
    )
    .unwrap()
}

/// Writes `keys` into a batch file under `dir`.
///
/// # Arguments
///
/// - `dir`: directory to write the batch into.
/// - `keys`: keys to write, in order.
///
/// # Returns
///
/// The path of the finished file, which the tests reopen through a fresh
/// backend.
fn write_u32_batch(dir: &str, keys: &[u32]) -> StoragePath {
    let factories = Factories::<DynData, DynData>::new::<u32, ()>();
    let storage_backend = backend_at(dir);
    let mut writer = Writer1::new(
        &factories,
        test_buffer_cache,
        &*storage_backend,
        Parameters::default(),
        FilterPlan::<DynData>::decide_filter(None, keys.len()),
    )
    .unwrap();
    for key in keys {
        writer.write0((key, &())).unwrap();
    }
    let tmp_path = writer.path().clone();
    // The handle must outlive the copy: dropping it removes the file.
    let (_file_handle, _filter, _bounds) = writer.close(BatchMetadata::default()).unwrap();

    // The writer picks its own file name, which a later backend instance has
    // no way to learn, so the batch is renamed to something the tests can ask
    // for by name.
    let stable_path = StoragePath::from("batch.feldera".to_string());
    let content = storage_backend.read(&tmp_path).unwrap();
    storage_backend
        .write(&stable_path, (*content).clone())
        .unwrap();
    storage_backend.delete(&tmp_path).unwrap();
    stable_path
}

/// Opens a batch file written by [`write_u32_batch`].
///
/// # Arguments
///
/// - `dir`: directory holding the batch.
/// - `path`: the batch's path, as returned by [`write_u32_batch`].
///
/// # Returns
///
/// - the batch's filter chain;
/// - the magic of the filter block on disk, or `None` when it has no filter;
/// - the file's incompatible feature bits.
fn open_u32_batch(dir: &str, path: &StoragePath) -> (BatchFilters<DynData>, Option<[u8; 4]>, u64) {
    let factories = Factories::<DynData, DynData>::new::<u32, ()>();
    let storage_backend = backend_at(dir);
    let (reader, membership_filter): (Reader<(&'static DynData, &'static DynData, ())>, _) =
        Reader::open_with_filter(
            &[&factories.any_factories()],
            test_buffer_cache,
            &*storage_backend,
            path,
        )
        .unwrap();
    let magic = filter_block_magic(&reader);
    let features = incompatible_features(&reader);
    let key_range = reader.key_range().unwrap().map(Into::into);
    (
        BatchFilters::from_file(key_range, membership_filter),
        magic,
        features,
    )
}

/// The configured rate selects the module count, and a one-module filter is
/// written in the original encoding so older binaries can still read the file.
#[test]
fn rate_selects_the_module_count_and_the_block_encoding() {
    init_test_logger();

    for (rate, expected_magic) in [
        (1e-4, Some(MODULAR_BLOOM_FILTER_BLOCK_MAGIC)),
        (1e-3, Some(MODULAR_BLOOM_FILTER_BLOCK_MAGIC)),
        (1e-2, Some(MODULAR_BLOOM_FILTER_BLOCK_MAGIC)),
        (1e-1, Some(BLOOM_FILTER_BLOCK_MAGIC)),
        (1.0, None),
    ] {
        let tempdir = tempdir().unwrap();
        let dir = tempdir.path().to_string_lossy().to_string();
        let keys: Vec<u32> = (0..2_000u32).map(|k| k * 2).collect();

        with_false_positive_rate(rate, move || {
            let path = write_u32_batch(&dir, &keys);
            let (filters, magic, features) = open_u32_batch(&dir, &path);
            assert_eq!(magic, expected_magic, "rate {rate:e} wrote the wrong block");

            for key in &keys {
                assert!(
                    filters.maybe_contains_key(key as &DynData, None),
                    "rate {rate:e}: filter rejected stored key {key}"
                );
            }

            // Only a multi-module file is closed to older binaries. A file with
            // one module, or none, stays readable by them.
            let expected_features = if rate < 0.05 {
                INCOMPATIBLE_FEATURE_MODULAR_FILTERS
            } else {
                0
            };
            assert_eq!(
                features, expected_features,
                "rate {rate:e} set the wrong incompatible features"
            );

            let kind = filters.membership_filter_kind();
            if expected_magic.is_none() {
                assert_eq!(
                    kind,
                    FilterKind::None,
                    "rate {rate:e} should write no filter"
                );
            } else {
                assert_eq!(kind, FilterKind::Bloom, "rate {rate:e} should write Bloom");
            }
        });
    }
}

/// Lowering the rate sheds modules when the file is loaded, without rewriting
/// it and without ever losing a key.
#[test]
fn lowering_the_rate_evicts_modules_on_load() {
    init_test_logger();

    let tempdir = tempdir().unwrap();
    let dir = tempdir.path().to_string_lossy().to_string();
    let keys: Vec<u32> = (0..4_000u32).map(|k| k * 2).collect();
    let path = Arc::new(Mutex::new(None));

    // Write at the most accurate rate, so the file holds four modules.
    {
        let (dir, keys, path) = (dir.clone(), keys.clone(), path.clone());
        with_false_positive_rate(1e-4, move || {
            *path.lock().unwrap() = Some(write_u32_batch(&dir, &keys));
        });
    }
    let path = path.lock().unwrap().clone().unwrap();

    // Reload it under progressively coarser rates. Each one keeps fewer bytes
    // resident, and none of them may lose a key.
    let mut previous_bytes = usize::MAX;
    for rate in [1e-4, 1e-3, 1e-2, 1e-1] {
        let (dir, keys, path) = (dir.clone(), keys.clone(), path.clone());
        let bytes = Arc::new(Mutex::new(0));
        {
            let bytes = bytes.clone();
            with_false_positive_rate(rate, move || {
                let (filters, magic, _features) = open_u32_batch(&dir, &path);
                assert_eq!(
                    magic,
                    Some(MODULAR_BLOOM_FILTER_BLOCK_MAGIC),
                    "the file on disk is unchanged by loading it"
                );
                assert_eq!(filters.membership_filter_kind(), FilterKind::Bloom);
                for key in &keys {
                    assert!(
                        filters.maybe_contains_key(key as &DynData, None),
                        "rate {rate:e}: eviction lost key {key}"
                    );
                }
                *bytes.lock().unwrap() = filters.stats().membership_filter.size_byte;
            });
        }
        let bytes = *bytes.lock().unwrap();
        assert!(
            bytes < previous_bytes,
            "rate {rate:e} kept {bytes} bytes resident, no less than the {previous_bytes} before it"
        );
        previous_bytes = bytes;
    }
}

/// A rate of one drops the filter entirely at load time, even for a file that
/// was written with modules.
#[test]
fn a_rate_of_one_drops_an_existing_filter_on_load() {
    init_test_logger();

    let tempdir = tempdir().unwrap();
    let dir = tempdir.path().to_string_lossy().to_string();
    let keys: Vec<u32> = (0..1_000u32).collect();
    let path = Arc::new(Mutex::new(None));

    {
        let (dir, keys, path) = (dir.clone(), keys.clone(), path.clone());
        with_false_positive_rate(1e-4, move || {
            *path.lock().unwrap() = Some(write_u32_batch(&dir, &keys));
        });
    }
    let path = path.lock().unwrap().clone().unwrap();

    with_false_positive_rate(1.0, move || {
        let (filters, _magic, _features) = open_u32_batch(&dir, &path);
        assert_eq!(filters.membership_filter_kind(), FilterKind::None);
        assert_eq!(filters.stats().membership_filter.size_byte, 0);
        // Without a membership filter every in-range key must still be accepted.
        for key in &keys {
            assert!(filters.maybe_contains_key(key as &DynData, None));
        }
    });
}

/// A [`FileReader`] that tallies the bytes the reads it forwards ask for.
#[derive(Debug)]
struct CountingFileReader {
    inner: Arc<dyn FileReader>,
    bytes: Arc<AtomicUsize>,
    calls: Arc<AtomicUsize>,
}

impl FileRw for CountingFileReader {
    fn file_id(&self) -> FileId {
        self.inner.file_id()
    }

    fn path(&self) -> &StoragePath {
        self.inner.path()
    }
}

impl FileCommitter for CountingFileReader {
    fn commit(&self) -> Result<(), StorageError> {
        self.inner.commit()
    }
}

impl FileReader for CountingFileReader {
    fn mark_for_checkpoint(&self) {
        self.inner.mark_for_checkpoint();
    }

    fn read_block(&self, location: BlockLocation) -> Result<Arc<FBuf>, StorageError> {
        self.bytes.fetch_add(location.size, Ordering::Relaxed);
        self.calls.fetch_add(1, Ordering::Relaxed);
        self.inner.read_block(location)
    }

    fn get_size(&self) -> Result<u64, StorageError> {
        self.inner.get_size()
    }
}

/// Opens a batch file, counting what loading its membership filter reads.
///
/// The filter block is read on demand, so counting around that one call leaves
/// out the trailer and index reads and measures the filter alone.
///
/// # Arguments
///
/// - `dir`: directory holding the batch.
/// - `path`: the batch's path, as returned by [`write_u32_batch`].
///
/// # Returns
///
/// - the batch's filter chain;
/// - bytes the reads asked for, filter block only;
/// - number of reads issued.
fn open_u32_batch_counting(dir: &str, path: &StoragePath) -> (BatchFilters<DynData>, usize, usize) {
    let factories = Factories::<DynData, DynData>::new::<u32, ()>();
    let storage_backend = backend_at(dir);
    let bytes = Arc::new(AtomicUsize::new(0));
    let calls = Arc::new(AtomicUsize::new(0));
    let counting = Arc::new(CountingFileReader {
        inner: storage_backend.open(path).unwrap(),
        bytes: bytes.clone(),
        calls: calls.clone(),
    });
    let reader: Reader<(&'static DynData, &'static DynData, ())> =
        Reader::new(&[&factories.any_factories()], test_buffer_cache, counting).unwrap();
    let key_range = reader.key_range().unwrap().map(Into::into);

    let before = (bytes.load(Ordering::Relaxed), calls.load(Ordering::Relaxed));
    let membership_filter = reader.membership_filter().unwrap();
    let filter_bytes = bytes.load(Ordering::Relaxed) - before.0;
    let filter_calls = calls.load(Ordering::Relaxed) - before.1;

    (
        BatchFilters::from_file(key_range, membership_filter),
        filter_bytes,
        filter_calls,
    )
}

/// A coarse rate reads only the modules it needs, not the whole filter block.
///
/// This is what the recorded per-module bit counts buy. Without them a reader
/// would have to read every module to learn what a prefix of them is worth, and
/// dropping modules would save memory but no I/O.
#[test]
fn a_coarse_rate_reads_only_the_modules_it_keeps() {
    init_test_logger();

    let tempdir = tempdir().unwrap();
    let dir = tempdir.path().to_string_lossy().to_string();
    // Enough keys that a module spans many sectors, so the saving cannot hide
    // in the rounding up to the 512-byte read granularity.
    let keys: Vec<u32> = (0..20_000u32).map(|k| k * 3).collect();
    let path = Arc::new(Mutex::new(None));

    {
        let (dir, keys, path) = (dir.clone(), keys.clone(), path.clone());
        with_false_positive_rate(1e-4, move || {
            *path.lock().unwrap() = Some(write_u32_batch(&dir, &keys));
        });
    }
    let path = path.lock().unwrap().clone().unwrap();

    // Coarse first: a shared cache could only make the later, finer open
    // cheaper, so it cannot manufacture the difference asserted below.
    let measure = |rate: f64| {
        let (dir, keys, path) = (dir.clone(), keys.clone(), path.clone());
        let result = Arc::new(Mutex::new((0usize, 0usize, 0usize)));
        {
            let result = result.clone();
            with_false_positive_rate(rate, move || {
                let (filters, bytes, calls) = open_u32_batch_counting(&dir, &path);
                for key in &keys {
                    assert!(
                        filters.maybe_contains_key(key as &DynData, None),
                        "rate {rate:e}: a prefix read lost key {key}"
                    );
                }
                *result.lock().unwrap() =
                    (bytes, filters.stats().membership_filter.size_byte, calls);
            });
        }
        *result.lock().unwrap()
    };

    let (coarse_read, coarse_resident, _coarse_calls) = measure(1e-1);
    let (fine_read, fine_resident, fine_calls) = measure(1e-4);

    // The finest rate cannot shed a module, so it must not pay to find that
    // out: one read of the block, as before any of this existed.
    assert_eq!(
        fine_calls, 1,
        "1e-4 read the filter block in {fine_calls} calls, not the one it needs"
    );

    assert!(
        coarse_resident < fine_resident,
        "1e-1 kept {coarse_resident} bytes resident, 1e-4 kept {fine_resident}"
    );
    assert!(
        coarse_read < fine_read,
        "1e-1 read {coarse_read} bytes of filter, no fewer than the {fine_read} that 1e-4 read"
    );

    // Reads are issued in 512-byte sectors, so a module dropped is a module not
    // read, give or take the sector the header shares with the first of them.
    let unread = fine_read - coarse_read;
    let evicted = fine_resident - coarse_resident;
    assert!(
        unread + 512 >= evicted,
        "evicting {evicted} bytes of modules only avoided reading {unread}"
    );
}

/// A filter written in the original encoding loads whole whatever the rate.
///
/// One module is written as `LFFB`, the encoding that predates modular filters,
/// and such a filter has no ladder to descend: the rate can drop it entirely or
/// leave it alone, and nothing in between. The modular path has its own tests
/// for shedding; this pins that the legacy path does not try to.
#[test]
fn a_legacy_filter_has_no_ladder_to_descend() {
    init_test_logger();

    let tempdir = tempdir().unwrap();
    let dir = tempdir.path().to_string_lossy().to_string();
    let keys: Vec<u32> = (0..4_000u32).map(|k| k * 2).collect();
    let path = Arc::new(Mutex::new(None));

    // 1e-1 is one module, so the writer uses the original encoding.
    {
        let (dir, keys, path) = (dir.clone(), keys.clone(), path.clone());
        with_false_positive_rate(1e-1, move || {
            *path.lock().unwrap() = Some(write_u32_batch(&dir, &keys));
        });
    }
    let path = path.lock().unwrap().clone().unwrap();

    // Loading it at any rate that wants a filter gives back all of it, whether
    // the rate is coarser or finer than the one it was written at.
    let mut sizes = Vec::new();
    for rate in [1e-1, 1e-2, 1e-4] {
        let (dir, keys, path) = (dir.clone(), keys.clone(), path.clone());
        let size = Arc::new(Mutex::new(0));
        {
            let size = size.clone();
            with_false_positive_rate(rate, move || {
                let (filters, magic, _features) = open_u32_batch(&dir, &path);
                assert_eq!(
                    magic,
                    Some(BLOOM_FILTER_BLOCK_MAGIC),
                    "rate {rate:e}: the file on disk is unchanged by loading it"
                );
                assert_eq!(filters.membership_filter_kind(), FilterKind::Bloom);
                for key in &keys {
                    assert!(
                        filters.maybe_contains_key(key as &DynData, None),
                        "rate {rate:e}: a legacy filter lost key {key}"
                    );
                }
                *size.lock().unwrap() = filters.stats().membership_filter.size_byte;
            });
        }
        sizes.push(*size.lock().unwrap());
    }
    assert!(sizes[0] > 0, "the legacy filter measured zero bytes");
    assert!(
        sizes.iter().all(|size| *size == sizes[0]),
        "a legacy filter changed size with the rate: {sizes:?}"
    );

    // A rate of one drops it outright, the one thing the rate can do to it.
    with_false_positive_rate(1.0, move || {
        let (filters, _magic, _features) = open_u32_batch(&dir, &path);
        assert_eq!(filters.membership_filter_kind(), FilterKind::None);
        assert_eq!(filters.stats().membership_filter.size_byte, 0);
        for key in &keys {
            assert!(filters.maybe_contains_key(key as &DynData, None));
        }
    });
}