qdrant-edge 0.8.0

use std::collections::{BTreeSet, HashSet};
use std::ops::Range;

use crate::common::bitvec::BitVec;
use crate::common::counter::hardware_accumulator::HwMeasurementAcc;
use crate::common::counter::hardware_counter::HardwareCounterCell;
use crate::common::types::PointOffsetType;
use crate::common::universal_io::MmapFile;
use itertools::Itertools;
use ordered_float::OrderedFloat;
use rand::SeedableRng;
use rand::prelude::StdRng;
use rstest::rstest;
use serde_json::json;
use tempfile::{Builder, TempDir};

use super::builders::{GeoIndexGridstoreBuilder, GeoIndexMmapBuilder};
use super::on_disk_geo_index::OnDiskGeoIndex;
use super::read_ops::GeoIndexRead;
use super::{GEO_QUERY_MAX_REGION, GeoIndex};
use crate::segment::fixtures::payload_fixtures::random_geo_payload;
use crate::segment::index::field_index::geo_hash::{
    GeoHash, circle_hashes, encode_max_precision, polygon_hashes,
};
use crate::segment::index::field_index::{
    CardinalityEstimation, FieldIndexBuilderTrait, PayloadFieldIndex, PayloadFieldIndexRead,
    ValueIndexer,
};
use crate::segment::json_path::JsonPath;
use crate::segment::types::test_utils::build_polygon;
use crate::segment::types::{
    CheckGeoPoint, FieldCondition, GeoBoundingBox, GeoLineString, GeoPoint, GeoPolygon, GeoRadius,
    Memory,
};

/// Generous default size for the deleted-points bitslice used in tests.
///
/// Must be larger than the stored mmap deletion bitslice for any test in
/// this file (which is sized to the highest point id, rounded up to a
/// `usize` boundary). 4096 bits comfortably covers all current tests.
const TEST_DELETED_BITS: usize = 4096;

/// All-zero deletion bitslice for tests that don't care about deletions.
fn empty_deleted() -> BitVec {
    BitVec::repeat(false, TEST_DELETED_BITS)
}

/// Deletion bitslice with specific points marked as deleted.
fn deleted_with(points: &[PointOffsetType]) -> BitVec {
    let mut v = empty_deleted();
    for &p in points {
        v.set(p as usize, true);
    }
    v
}

type Database = ();

#[derive(Clone, Copy, PartialEq, Debug)]
enum IndexType {
    Mutable,
    OnDisk,
    Immutable,
}

#[expect(clippy::large_enum_variant)]
enum IndexBuilder {
    MutableGridstore(GeoIndexGridstoreBuilder),
    Mmap(GeoIndexMmapBuilder),
    Immutable(GeoIndexMmapBuilder),
}

impl IndexBuilder {
    fn add_point(
        &mut self,
        id: PointOffsetType,
        payload: &[&serde_json::Value],
        hw_counter: &HardwareCounterCell,
    ) -> crate::segment::common::operation_error::OperationResult<()> {
        match self {
            IndexBuilder::MutableGridstore(builder) => builder.add_point(id, payload, hw_counter),
            IndexBuilder::Mmap(builder) => builder.add_point(id, payload, hw_counter),
            IndexBuilder::Immutable(builder) => builder.add_point(id, payload, hw_counter),
        }
    }

    fn finalize(self) -> crate::segment::common::operation_error::OperationResult<GeoIndex> {
        match self {
            IndexBuilder::MutableGridstore(builder) => builder.finalize(),
            IndexBuilder::Mmap(builder) => builder.finalize(),
            IndexBuilder::Immutable(builder) => builder.finalize(),
        }
    }
}

const NYC: GeoPoint = GeoPoint::new_unchecked(-73.991516, 40.75798);

const BERLIN: GeoPoint = GeoPoint::new_unchecked(13.41053, 52.52437);

const POTSDAM: GeoPoint = GeoPoint::new_unchecked(13.064473, 52.390569);

const TOKYO: GeoPoint = GeoPoint::new_unchecked(139.691706, 35.689487);

const LOS_ANGELES: GeoPoint = GeoPoint::new_unchecked(-118.243683, 34.052235);

fn condition_for_geo_radius(key: &str, geo_radius: GeoRadius) -> FieldCondition {
    FieldCondition::new_geo_radius(JsonPath::new(key), geo_radius)
}

fn condition_for_geo_polygon(key: &str, geo_polygon: GeoPolygon) -> FieldCondition {
    FieldCondition::new_geo_polygon(JsonPath::new(key), geo_polygon)
}

fn condition_for_geo_box(key: &str, geo_bounding_box: GeoBoundingBox) -> FieldCondition {
    FieldCondition::new_geo_bounding_box(JsonPath::new(key), geo_bounding_box)
}

/// Run a filter query and return the matching point offsets, sorted.
fn filtered_points(index: &GeoIndex, condition: &FieldCondition) -> Vec<PointOffsetType> {
    let hw_acc = HwMeasurementAcc::new();
    let hw_counter = hw_acc.get_counter_cell();
    let mut points: Vec<PointOffsetType> = index
        .filter(condition, &hw_counter)
        .unwrap()
        .unwrap()
        .collect();
    points.sort_unstable();
    points
}

/// Assert that both the geo-radius query and its polygon approximation
/// estimate exactly `expected` matches.
fn assert_radius_cardinality(index: &GeoIndex, center: GeoPoint, meters: f64, expected: usize) {
    let radius = GeoRadius {
        center,
        radius: OrderedFloat(meters),
    };
    let polygon = radius_to_polygon(&radius);
    let hw = HardwareCounterCell::new();
    let check = |cond: FieldCondition| {
        let c = index.estimate_cardinality(&cond, &hw).unwrap().unwrap();
        assert_eq!((c.min, c.exp, c.max), (expected, expected, expected));
    };

    check(condition_for_geo_radius("test", radius));
    check(condition_for_geo_polygon("test", polygon));
}

#[cfg(feature = "testing")]
fn create_builder(index_type: IndexType) -> (IndexBuilder, TempDir, Database) {
    let temp_dir = Builder::new().prefix("test_dir").tempdir().unwrap();

    let db = ();

    let mut builder = match index_type {
        IndexType::Mutable => IndexBuilder::MutableGridstore(GeoIndex::builder_gridstore(
            temp_dir.path().to_path_buf(),
        )),
        IndexType::OnDisk => IndexBuilder::Mmap(GeoIndex::builder_mmap(
            temp_dir.path(),
            true,
            &empty_deleted(),
        )),
        IndexType::Immutable => IndexBuilder::Immutable(GeoIndex::builder_mmap(
            temp_dir.path(),
            false,
            &empty_deleted(),
        )),
    };
    match &mut builder {
        IndexBuilder::MutableGridstore(builder) => builder.init().unwrap(),
        IndexBuilder::Mmap(builder) => builder.init().unwrap(),
        IndexBuilder::Immutable(builder) => builder.init().unwrap(),
    }
    (builder, temp_dir, db)
}

fn build_random_index(
    num_points: usize,
    num_geo_values: usize,
    index_type: IndexType,
) -> (GeoIndex, TempDir, Database) {
    let mut rnd = StdRng::seed_from_u64(42);
    let (mut builder, temp_dir, db) = create_builder(index_type);

    for idx in 0..num_points {
        let geo_points = random_geo_payload(&mut rnd, num_geo_values..=num_geo_values);
        let array_payload = serde_json::Value::Array(geo_points);
        builder
            .add_point(
                idx as PointOffsetType,
                &[&array_payload],
                &HardwareCounterCell::new(),
            )
            .unwrap();
    }

    let index = builder.finalize().unwrap();
    assert_eq!(index.points_count(), num_points);
    assert_eq!(index.points_values_count(), num_points * num_geo_values);
    (index, temp_dir, db)
}

/// Reopen a persisted index from disk. Mmap indices don't persist runtime
/// deletions, so the deletion bitslice must be re-supplied on reload.
fn reload_index(index_type: IndexType, temp_dir: &TempDir, deleted: &BitVec) -> GeoIndex {
    match index_type {
        IndexType::Mutable => GeoIndex::new_mutable(temp_dir.path().to_path_buf(), true),
        IndexType::OnDisk => GeoIndex::new_immutable(temp_dir.path(), Memory::Cold, deleted),
        IndexType::Immutable => GeoIndex::new_immutable(temp_dir.path(), Memory::Pinned, deleted),
    }
    .unwrap()
    .unwrap()
}

const EARTH_RADIUS_METERS: f64 = 6371.0 * 1000.;
const LON_RANGE: Range<f64> = -180.0..180.0;
const LAT_RANGE: Range<f64> = -90.0..90.0;
const COORD_EPS: f64 = 1e-12;

fn radius_to_polygon(circle: &GeoRadius) -> GeoPolygon {
    let angular_radius: f64 = circle.radius.0 / EARTH_RADIUS_METERS;

    let angular_lat = circle.center.lat.to_radians();
    let mut min_lat = (angular_lat - angular_radius).to_degrees();
    let mut max_lat = (angular_lat + angular_radius).to_degrees();

    let (min_lon, max_lon) = if LAT_RANGE.start < min_lat && max_lat < LAT_RANGE.end {
        let angular_lon = circle.center.lon.to_radians();
        let delta_lon = (angular_radius.sin() / angular_lat.cos()).asin();

        let min_lon = (angular_lon - delta_lon).to_degrees();
        let max_lon = (angular_lon + delta_lon).to_degrees();

        (min_lon, max_lon)
    } else {
        if LAT_RANGE.start > min_lat {
            min_lat = LAT_RANGE.start + COORD_EPS;
        }
        if max_lat > LAT_RANGE.end {
            max_lat = LAT_RANGE.end - COORD_EPS;
        }

        (LON_RANGE.start + COORD_EPS, LON_RANGE.end - COORD_EPS)
    };

    build_polygon(vec![
        (min_lon, min_lat),
        (min_lon, max_lat),
        (max_lon, max_lat),
        (max_lon, min_lat),
        (min_lon, min_lat),
    ])
}

/// Regression: a geo-polygon whose interior ring matches more points than its
/// exterior ring must not underflow the cardinality estimate. `estimate_cardinality`
/// subtracted `usize`s wrapped in a no-op `max(0, ..)`, so when an interior estimate
/// exceeded the exterior estimate the subtraction underflowed (debug panic / release
/// wrap to ~usize::MAX, corrupting query planning).
#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn test_polygon_interior_exceeds_exterior_cardinality(#[case] index_type: IndexType) {
    let (mut builder, _temp_dir, _db) = create_builder(index_type);

    // A single indexed point in New York.
    builder
        .add_point(0, &[&json!([NYC])], &HardwareCounterCell::new())
        .unwrap();
    let index = builder.finalize().unwrap();

    // Exterior ring over Europe (does not contain the point); interior ring around
    // New York (contains the point). The interior cardinality estimate therefore
    // exceeds the exterior one, which previously underflowed `exterior.min - interior.max`.
    let polygon = GeoPolygon {
        exterior: GeoLineString {
            points: vec![
                GeoPoint::new_unchecked(2.0, 51.0),
                GeoPoint::new_unchecked(2.0, 60.0),
                GeoPoint::new_unchecked(20.0, 60.0),
                GeoPoint::new_unchecked(20.0, 51.0),
                GeoPoint::new_unchecked(2.0, 51.0),
            ],
        },
        interiors: Some(vec![GeoLineString {
            points: vec![
                GeoPoint::new_unchecked(-74.5, 40.0),
                GeoPoint::new_unchecked(-74.5, 41.5),
                GeoPoint::new_unchecked(-73.0, 41.5),
                GeoPoint::new_unchecked(-73.0, 40.0),
                GeoPoint::new_unchecked(-74.5, 40.0),
            ],
        }]),
    };

    let hw_counter = HardwareCounterCell::new();
    let card = index
        .estimate_cardinality(&condition_for_geo_polygon("test", polygon), &hw_counter)
        .unwrap()
        .unwrap();

    // No underflow: a self-consistent estimate bounded by the single indexed point.
    assert!(card.min <= card.exp, "min {} > exp {}", card.min, card.exp);
    assert!(card.exp <= card.max, "exp {} > max {}", card.exp, card.max);
    assert!(
        card.max <= index.points_count(),
        "max {} exceeds total points {}",
        card.max,
        index.points_count(),
    );
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn test_polygon_with_exclusion(#[case] index_type: IndexType) {
    let europe = GeoLineString {
        points: vec![
            GeoPoint::new_unchecked(19.415558242000287, 69.18533258102943),
            GeoPoint::new_unchecked(2.4664944437317615, 61.852748225727254),
            GeoPoint::new_unchecked(2.713789718828849, 51.80793869181895),
            GeoPoint::new_unchecked(-8.396395372995187, 46.85848915174239),
            GeoPoint::new_unchecked(-10.508661204875182, 35.64130367692255),
            GeoPoint::new_unchecked(0.9590825812569506, 36.55931431668104),
            GeoPoint::new_unchecked(17.925941188829, 34.89268498908065),
            GeoPoint::new_unchecked(26.378822944221042, 38.87157101630817),
            GeoPoint::new_unchecked(41.568021588510476, 47.7100126473878),
            GeoPoint::new_unchecked(29.149194109528253, 70.96161947624168),
            GeoPoint::new_unchecked(19.415558242000287, 69.18533258102943),
        ],
    };

    let berlin = GeoLineString {
        points: vec![
            GeoPoint::new_unchecked(13.2257943327987, 52.62328249733332),
            GeoPoint::new_unchecked(13.11841750240768, 52.550216162683455),
            GeoPoint::new_unchecked(13.11841750240768, 52.40371784468752),
            GeoPoint::new_unchecked(13.391870497137859, 52.40546474165669),
            GeoPoint::new_unchecked(13.653869963292806, 52.35739986654923),
            GeoPoint::new_unchecked(13.754088338324664, 52.44213360096185),
            GeoPoint::new_unchecked(13.60805584899208, 52.47702797300224),
            GeoPoint::new_unchecked(13.63382628828623, 52.53367235825061),
            GeoPoint::new_unchecked(13.48493041681067, 52.60241883100514),
            GeoPoint::new_unchecked(13.52788114896677, 52.6571647548233),
            GeoPoint::new_unchecked(13.257291536380365, 52.667584785254064),
            GeoPoint::new_unchecked(13.2257943327987, 52.62328249733332),
        ],
    };

    let europe_no_berlin = GeoPolygon {
        exterior: europe,
        interiors: Some(vec![berlin]),
    };
    check_cardinality_match(
        polygon_hashes(&europe_no_berlin, GEO_QUERY_MAX_REGION).unwrap(),
        condition_for_geo_polygon("test", europe_no_berlin.clone()),
        index_type,
    );
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn match_cardinality(#[case] index_type: IndexType) {
    let r_meters = 500_000.0;
    let geo_radius = GeoRadius {
        center: NYC,
        radius: OrderedFloat(r_meters),
    };
    let nyc_hashes = circle_hashes(&geo_radius, GEO_QUERY_MAX_REGION).unwrap();
    check_cardinality_match(
        nyc_hashes,
        condition_for_geo_radius("test", geo_radius),
        index_type,
    );

    let geo_polygon = radius_to_polygon(&geo_radius);
    let polygon_hashes = polygon_hashes(&geo_polygon, GEO_QUERY_MAX_REGION).unwrap();
    check_cardinality_match(
        polygon_hashes,
        condition_for_geo_polygon("test", geo_polygon),
        index_type,
    );
}

/// Assert the estimated cardinality brackets the exact hash-iterator count.
fn check_cardinality_match(
    hashes: Vec<GeoHash>,
    field_condition: FieldCondition,
    index_type: IndexType,
) {
    let (field_index, _, _) = build_random_index(500, 20, index_type);
    let exact_points_for_hashes = field_index.iterator(hashes).unwrap().collect_vec();
    let real_cardinality = exact_points_for_hashes.len();

    let hw_counter = HardwareCounterCell::new();
    let card = field_index
        .estimate_cardinality(&field_condition, &hw_counter)
        .unwrap();
    let card = card.unwrap();

    eprintln!("real_cardinality = {real_cardinality:#?}");
    eprintln!("card = {card:#?}");

    assert!(card.min <= real_cardinality);
    assert!(card.max >= real_cardinality);

    assert!(card.exp >= card.min);
    assert!(card.exp <= card.max);
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn geo_indexed_filtering(#[case] index_type: IndexType) {
    fn check_geo_indexed_filtering<F>(
        field_condition: FieldCondition,
        check_fn: F,
        index_type: IndexType,
    ) where
        F: Fn(&GeoPoint) -> bool + Clone,
    {
        let (field_index, _, _) = build_random_index(1000, 5, index_type);

        let hw_counter = HardwareCounterCell::new();
        let mut matched_points = (0..field_index.count_indexed_points().unwrap()
            as PointOffsetType)
            .filter_map(|idx| {
                if field_index
                    .check_values_any(idx, &hw_counter, &check_fn.clone())
                    .unwrap()
                {
                    Some(idx as PointOffsetType)
                } else {
                    None
                }
            })
            .collect_vec();

        assert!(!matched_points.is_empty());

        let indexed_matched_points = filtered_points(&field_index, &field_condition);

        matched_points.sort_unstable();

        assert_eq!(matched_points, indexed_matched_points);
    }

    let r_meters = 500_000.0;
    let geo_radius = GeoRadius {
        center: NYC,
        radius: OrderedFloat(r_meters),
    };
    check_geo_indexed_filtering(
        condition_for_geo_radius("test", geo_radius),
        |geo_point| geo_radius.check_point(geo_point),
        index_type,
    );

    let geo_polygon: GeoPolygon = build_polygon(vec![
        (-60.0, 37.0),
        (-60.0, 45.0),
        (-50.0, 45.0),
        (-50.0, 37.0),
        (-60.0, 37.0),
    ]);
    check_geo_indexed_filtering(
        condition_for_geo_polygon("test", geo_polygon.clone()),
        |geo_point| geo_polygon.convert().check_point(geo_point),
        index_type,
    );
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn test_payload_blocks(#[case] index_type: IndexType) {
    let (field_index, _, _) = build_random_index(1000, 5, index_type);
    let hw_counter = HardwareCounterCell::new();
    let top_level_points = field_index
        .points_of_hash(Default::default(), &hw_counter)
        .unwrap();
    assert_eq!(top_level_points, 1_000);
    let block_hashes = field_index.large_hashes(100).unwrap().collect_vec();
    assert!(!block_hashes.is_empty());
    for (geohash, size) in block_hashes {
        assert_eq!(geohash.len(), 1);
        assert!(size > 100);
        assert!(size < 1000);
    }

    let mut blocks = Vec::new();
    field_index
        .for_each_payload_block(100, JsonPath::new("test"), &mut |block| {
            blocks.push(block);
            Ok(())
        })
        .unwrap();
    blocks.iter().for_each(|block| {
        let block_points = filtered_points(&field_index, &block.condition);
        assert_eq!(block_points.len(), block.cardinality);
    });
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn match_cardinality_point_with_multi_far_geo_payload(#[case] index_type: IndexType) {
    let (mut builder, _, _) = create_builder(index_type);

    let r_meters = 100.0;
    let geo_values = json!([BERLIN, NYC]);
    let hw_counter = HardwareCounterCell::new();
    builder.add_point(1, &[&geo_values], &hw_counter).unwrap();
    let index = builder.finalize().unwrap();

    assert_radius_cardinality(&index, NYC, r_meters, 1);
    assert_radius_cardinality(&index, BERLIN, r_meters, 1);
    assert_radius_cardinality(&index, TOKYO, r_meters, 0);
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn match_cardinality_point_with_multi_close_geo_payload(#[case] index_type: IndexType) {
    let (mut builder, _, _) = create_builder(index_type);
    let geo_values = json!([BERLIN, POTSDAM]);
    let hw_counter = HardwareCounterCell::new();
    builder.add_point(1, &[&geo_values], &hw_counter).unwrap();
    let index = builder.finalize().unwrap();

    assert_radius_cardinality(&index, BERLIN, 50_000.0, 1);
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn load_from_disk(#[case] index_type: IndexType) {
    let temp_dir = {
        let (mut builder, temp_dir, _) = create_builder(index_type);

        let geo_values = json!([BERLIN, POTSDAM]);
        let hw_counter = HardwareCounterCell::new();
        builder.add_point(1, &[&geo_values], &hw_counter).unwrap();
        builder.finalize().unwrap();
        temp_dir
    };

    let new_index = reload_index(index_type, &temp_dir, &empty_deleted());

    let berlin_geo_radius = GeoRadius {
        center: BERLIN,
        radius: OrderedFloat(50_000.0),
    };

    let field_condition = condition_for_geo_radius("test", berlin_geo_radius);
    let point_offsets = filtered_points(&new_index, &field_condition);
    assert_eq!(point_offsets, vec![1]);

    let field_condition = condition_for_geo_polygon("test", radius_to_polygon(&berlin_geo_radius));
    let point_offsets = filtered_points(&new_index, &field_condition);
    assert_eq!(point_offsets, vec![1]);
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn same_geo_index_between_points_test(#[case] index_type: IndexType) {
    let temp_dir = {
        let (mut builder, temp_dir, _) = create_builder(index_type);

        let geo_values = json!([BERLIN, POTSDAM]);
        let hw_counter = HardwareCounterCell::new();
        let payload = [&geo_values];
        builder.add_point(1, &payload, &hw_counter).unwrap();
        builder.add_point(2, &payload, &hw_counter).unwrap();
        let mut index = builder.finalize().unwrap();

        index.remove_point(1).unwrap();
        index.flusher()().unwrap();

        assert_eq!(index.points_count(), 1);
        if index_type != IndexType::OnDisk {
            assert_eq!(index.points_values_count(), 2);
        }
        drop(index);
        temp_dir
    };

    let new_index = reload_index(index_type, &temp_dir, &deleted_with(&[1]));
    assert_eq!(new_index.points_count(), 1);
    if index_type != IndexType::OnDisk {
        assert_eq!(new_index.points_values_count(), 2);
    }
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn same_geo_index_between_points_with_dups_test(#[case] index_type: IndexType) {
    let temp_dir = {
        let (mut builder, temp_dir, _) = create_builder(index_type);

        let geo_values = json!([BERLIN, BERLIN, POTSDAM]); // Berlin twice
        let hw_counter = HardwareCounterCell::new();
        let payload = [&geo_values];
        builder.add_point(1, &payload, &hw_counter).unwrap();
        builder.add_point(2, &payload, &hw_counter).unwrap();
        let mut index = builder.finalize().unwrap();

        index.remove_point(1).unwrap();
        index.flusher()().unwrap();

        assert_eq!(index.points_count(), 1);
        if index_type != IndexType::OnDisk {
            assert_eq!(index.points_values_count(), 3);
        }
        drop(index);
        temp_dir
    };

    let new_index = reload_index(index_type, &temp_dir, &deleted_with(&[1]));
    assert_eq!(new_index.points_count(), 1);
    if index_type != IndexType::OnDisk {
        assert_eq!(new_index.points_values_count(), 3);

        let hw_counter = HardwareCounterCell::disposable();
        let berlin_hash = encode_max_precision(BERLIN.lon.0, BERLIN.lat.0).unwrap();
        assert_eq!(
            new_index.values_of_hash(berlin_hash, &hw_counter).unwrap(),
            2
        );
        assert_eq!(
            new_index.points_of_hash(berlin_hash, &hw_counter).unwrap(),
            1
        );
    }
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn test_empty_index_cardinality(#[case] index_type: IndexType) {
    let polygon = GeoPolygon {
        exterior: GeoLineString {
            points: vec![
                GeoPoint::new_unchecked(19.415558242000287, 69.18533258102943),
                GeoPoint::new_unchecked(2.4664944437317615, 61.852748225727254),
                GeoPoint::new_unchecked(2.713789718828849, 51.80793869181895),
                GeoPoint::new_unchecked(19.415558242000287, 69.18533258102943),
            ],
        },
        interiors: None,
    };
    let polygon_with_interior = GeoPolygon {
        exterior: polygon.exterior.clone(),
        interiors: Some(vec![GeoLineString {
            points: vec![
                GeoPoint::new_unchecked(13.2257943327987, 52.62328249733332),
                GeoPoint::new_unchecked(13.11841750240768, 52.550216162683455),
                GeoPoint::new_unchecked(13.11841750240768, 52.40371784468752),
                GeoPoint::new_unchecked(13.2257943327987, 52.62328249733332),
            ],
        }]),
    };
    let hashes = polygon_hashes(&polygon, GEO_QUERY_MAX_REGION).unwrap();
    let hashes_with_interior =
        polygon_hashes(&polygon_with_interior, GEO_QUERY_MAX_REGION).unwrap();

    let hw_counter = HardwareCounterCell::new();

    let (field_index, _, _) = build_random_index(0, 0, index_type);
    assert!(
        field_index
            .match_cardinality(&hashes, &hw_counter)
            .unwrap()
            .equals_min_exp_max(&CardinalityEstimation::exact(0)),
    );
    assert!(
        field_index
            .match_cardinality(&hashes_with_interior, &hw_counter)
            .unwrap()
            .equals_min_exp_max(&CardinalityEstimation::exact(0)),
    );

    let (field_index, _, _) = build_random_index(0, 100, index_type);
    assert!(
        field_index
            .match_cardinality(&hashes, &hw_counter)
            .unwrap()
            .equals_min_exp_max(&CardinalityEstimation::exact(0)),
    );
    assert!(
        field_index
            .match_cardinality(&hashes_with_interior, &hw_counter)
            .unwrap()
            .equals_min_exp_max(&CardinalityEstimation::exact(0)),
    );

    let (field_index, _, _) = build_random_index(100, 100, index_type);
    assert!(
        !field_index
            .match_cardinality(&hashes, &hw_counter)
            .unwrap()
            .equals_min_exp_max(&CardinalityEstimation::exact(0)),
    );
    assert!(
        !field_index
            .match_cardinality(&hashes_with_interior, &hw_counter)
            .unwrap()
            .equals_min_exp_max(&CardinalityEstimation::exact(0)),
    );
}

#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn query_across_antimeridian(#[case] index_type: IndexType) {
    let (mut builder, _, _) = create_builder(index_type);
    let geo_values = json!([BERLIN]);
    let hw_counter = HardwareCounterCell::new();

    builder.add_point(1, &[&geo_values], &hw_counter).unwrap();

    let geo_values = json!([LOS_ANGELES]);
    builder.add_point(2, &[&geo_values], &hw_counter).unwrap();

    let geo_values = json!([TOKYO]);
    builder.add_point(3, &[&geo_values], &hw_counter).unwrap();

    let new_index = builder.finalize().unwrap();
    assert_eq!(new_index.points_count(), 3);
    assert_eq!(new_index.points_values_count(), 3);

    let bounding_box = GeoBoundingBox {
        top_left: GeoPoint::new_unchecked(167.0, 74.071028),
        bottom_right: GeoPoint::new_unchecked(-66.885417, 18.7763),
    };

    let field_condition = condition_for_geo_box("test", bounding_box);
    let point_offsets = filtered_points(&new_index, &field_condition);
    assert_eq!(point_offsets, vec![2]);
}

/// Removing a point with duplicate geo values in a multi-value geo field
/// must not produce spurious "no points for hash X was found" warnings.
#[rstest]
#[case(IndexType::Mutable)]
fn test_remove_point_with_duplicate_geo_values(#[case] index_type: IndexType) {
    let (mut builder, _temp_dir, _db) = create_builder(index_type);
    let hw_counter = HardwareCounterCell::new();

    let duplicate_geo = json!([BERLIN, BERLIN]);
    builder
        .add_point(0, &[&duplicate_geo], &hw_counter)
        .unwrap();

    let single_geo = json!(NYC);
    builder.add_point(1, &[&single_geo], &hw_counter).unwrap();

    let mut index = builder.finalize().unwrap();

    assert_eq!(index.points_count(), 2);
    assert_eq!(index.points_values_count(), 3);

    index.remove_point(0).unwrap();

    assert_eq!(index.points_count(), 1);
    assert_eq!(index.points_values_count(), 1);

    let geo_radius = GeoRadius {
        center: NYC,
        radius: OrderedFloat(100.0),
    };
    let field_condition = condition_for_geo_radius("test", geo_radius);
    let results = filtered_points(&index, &field_condition);
    assert_eq!(results, vec![1]);

    index.remove_point(1).unwrap();

    assert_eq!(index.points_count(), 0);
    assert_eq!(index.points_values_count(), 0);
}

/// Regression: `remove_point` must decrement `values_per_hash` once per value,
/// not once per unique geohash. When a point has duplicate geo values (same
/// coordinates producing the same geohash), all increments from `add_many_geo_points`
/// must be reversed on removal. Otherwise `values_per_hash` drifts upward permanently.
#[test]
fn test_values_per_hash_drift_on_duplicate_geo_removal() {
    let (mut builder, _temp_dir, _db) = create_builder(IndexType::Mutable);
    let hw_counter = HardwareCounterCell::new();

    // Point 0 has 3 identical geo values (same geohash produced 3 times).
    let triple_duplicate = json!([BERLIN, BERLIN, BERLIN]);
    builder
        .add_point(0, &[&triple_duplicate], &hw_counter)
        .unwrap();

    let mut index = builder.finalize().unwrap();

    assert_eq!(index.points_count(), 1);
    assert_eq!(index.points_values_count(), 3);

    // After removal, all counters must return to zero.
    index.remove_point(0).unwrap();

    assert_eq!(index.points_count(), 0);
    assert_eq!(index.points_values_count(), 0);

    // The root geohash ("") values_of_hash must be 0 — any leftover means
    // decrement_hash_value_counts was not called the correct number of times.
    let root_hash = GeoHash::default();
    let hw_counter = HardwareCounterCell::new();
    let root_values = index.values_of_hash(root_hash, &hw_counter).unwrap();
    assert_eq!(
        root_values, 0,
        "values_per_hash drifted: expected 0 after removing all points, got {root_values}"
    );

    // Also verify via cardinality estimation: with an empty index, any geo
    // query must estimate zero matches.
    let geo_radius = GeoRadius {
        center: BERLIN,
        radius: OrderedFloat(100.0),
    };
    let field_condition = condition_for_geo_radius("test", geo_radius);
    let hw_counter = HardwareCounterCell::new();
    let card = index
        .estimate_cardinality(&field_condition, &hw_counter)
        .unwrap();
    let card = card.unwrap();
    assert_eq!(
        card.exp, 0,
        "cardinality estimate should be 0 for empty index, got {}",
        card.exp
    );
}

/// Simulate the user's scenario: frequently adding and removing points
/// with geo payloads.
#[rstest]
#[case(IndexType::Mutable)]
fn test_frequent_add_remove_geo_points(#[case] index_type: IndexType) {
    let (mut builder, _temp_dir, _db) = create_builder(index_type);
    let hw_counter = HardwareCounterCell::new();

    let berlin_geo = json!(BERLIN);
    builder.add_point(0, &[&berlin_geo], &hw_counter).unwrap();

    let mut index = builder.finalize().unwrap();
    assert_eq!(index.points_count(), 1);

    for i in 1u32..20 {
        let geo = json!(NYC);
        index.add_point(i, &[&geo], &hw_counter).unwrap();
        assert_eq!(index.points_count(), 2);

        index.remove_point(i).unwrap();
        assert_eq!(index.points_count(), 1);
    }

    let tokyo_geo = json!(TOKYO);
    index.add_point(0, &[&tokyo_geo], &hw_counter).unwrap();
    assert_eq!(index.points_count(), 1);
    assert_eq!(index.points_values_count(), 1);

    let geo_radius = GeoRadius {
        center: TOKYO,
        radius: OrderedFloat(100.0),
    };
    let field_condition = condition_for_geo_radius("test", geo_radius);
    let results = filtered_points(&index, &field_condition);
    assert_eq!(results, vec![0]);

    index.remove_point(0).unwrap();
    assert_eq!(index.points_count(), 0);
    assert_eq!(index.points_values_count(), 0);
}

#[rstest]
#[case(&[IndexType::Mutable, IndexType::OnDisk, IndexType::Immutable], false)]
#[case(&[IndexType::Mutable, IndexType::Immutable], true)]
fn test_congruence(#[case] types: &[IndexType], #[case] deleted: bool) {
    const POINT_COUNT: usize = 500;

    let (mut indices, _data): (Vec<_>, Vec<_>) = types
        .iter()
        .copied()
        .map(|index_type| {
            let (index, temp_dir, db) = build_random_index(POINT_COUNT, 20, index_type);
            (index, (temp_dir, db))
        })
        .unzip();

    let polygon = GeoPolygon {
        exterior: GeoLineString {
            points: vec![
                GeoPoint::new_unchecked(19.415558242000287, 69.18533258102943),
                GeoPoint::new_unchecked(2.4664944437317615, 61.852748225727254),
                GeoPoint::new_unchecked(2.713789718828849, 51.80793869181895),
                GeoPoint::new_unchecked(19.415558242000287, 69.18533258102943),
            ],
        },
        interiors: None,
    };
    let hashes = polygon_hashes(&polygon, GEO_QUERY_MAX_REGION).unwrap();

    if deleted {
        for index in indices.iter_mut() {
            index.remove_point(10).unwrap();
            index.remove_point(11).unwrap();
            index.remove_point(12).unwrap();
            index.remove_point(100).unwrap();
            index.remove_point(150).unwrap();
        }
    }

    for index in &indices[1..] {
        assert_eq!(indices[0].points_count(), index.points_count());
        if !deleted {
            assert_eq!(
                indices[0].points_values_count(),
                index.points_values_count(),
            );
        }
        assert_eq!(
            indices[0].max_values_per_point(),
            index.max_values_per_point(),
        );
        let hw_counter = HardwareCounterCell::disposable();
        for &hash in &hashes {
            assert_eq!(
                indices[0].points_of_hash(hash, &hw_counter).unwrap(),
                index.points_of_hash(hash, &hw_counter).unwrap(),
            );
            assert_eq!(
                indices[0].values_of_hash(hash, &hw_counter),
                index.values_of_hash(hash, &hw_counter),
            );
        }
        assert_eq!(
            indices[0]
                .large_hashes(20)
                .unwrap()
                .map(|(hash, _)| hash)
                .collect::<BTreeSet<_>>(),
            index
                .large_hashes(20)
                .unwrap()
                .map(|(hash, _)| hash)
                .collect::<BTreeSet<_>>(),
        );
        assert_eq!(
            indices[0]
                .iterator(hashes.clone())
                .unwrap()
                .collect::<HashSet<_>>(),
            index
                .iterator(hashes.clone())
                .unwrap()
                .collect::<HashSet<_>>(),
        );
        for point_id in 0..POINT_COUNT {
            assert_eq!(
                indices[0].values_count(point_id as PointOffsetType),
                index.values_count(point_id as PointOffsetType),
            );
            if !deleted {
                assert_eq!(
                    indices[0]
                        .get_values(point_id as PointOffsetType)
                        .unwrap()
                        .collect::<BTreeSet<_>>(),
                    index
                        .get_values(point_id as PointOffsetType)
                        .unwrap()
                        .collect::<BTreeSet<_>>(),
                );
            }
            assert_eq!(
                indices[0].values_is_empty(point_id as PointOffsetType),
                index.values_is_empty(point_id as PointOffsetType),
            );
        }
    }
}

/// Cross-check that [`OnDiskGeoIndex::all_points`] produces the same
/// point set as the mutable and immutable index implementations.
#[test]
fn test_all_points_congruence() {
    const POINT_COUNT: usize = 500;

    let (mut mutable_index, _mutable_tmp, _) =
        build_random_index(POINT_COUNT, 20, IndexType::Mutable);
    let (mut immutable_index, _immutable_tmp, _) =
        build_random_index(POINT_COUNT, 20, IndexType::Immutable);
    let (mut mmap_index, _mmap_tmp, _) = build_random_index(POINT_COUNT, 20, IndexType::OnDisk);

    let GeoIndex::OnDisk(storage_index) = &mmap_index else {
        panic!("expected Mmap variant to build into Storage");
    };

    let europe_polygon = GeoPolygon {
        exterior: GeoLineString {
            points: vec![
                GeoPoint::new_unchecked(19.415558242000287, 69.18533258102943),
                GeoPoint::new_unchecked(2.4664944437317615, 61.852748225727254),
                GeoPoint::new_unchecked(2.713789718828849, 51.80793869181895),
                GeoPoint::new_unchecked(19.415558242000287, 69.18533258102943),
            ],
        },
        interiors: None,
    };
    let europe_hashes = polygon_hashes(&europe_polygon, GEO_QUERY_MAX_REGION).unwrap();

    let nyc_circle = GeoRadius {
        center: NYC,
        radius: OrderedFloat(300_000.0),
    };
    let tokyo_circle = GeoRadius {
        center: TOKYO,
        radius: OrderedFloat(300_000.0),
    };
    let mut disjoint_hashes = circle_hashes(&nyc_circle, GEO_QUERY_MAX_REGION).unwrap();
    disjoint_hashes.extend(circle_hashes(&tokyo_circle, GEO_QUERY_MAX_REGION).unwrap());

    let mut overlapping_hashes = europe_hashes.clone();
    overlapping_hashes.extend(europe_hashes.iter().copied());
    if let Some(first) = europe_hashes.first() {
        let len = first.len();
        if len > 2 {
            overlapping_hashes.push(first.truncate(len - 1));
            overlapping_hashes.push(first.truncate(len - 2));
        }
    }

    let global_polygon = GeoPolygon {
        exterior: GeoLineString {
            points: vec![
                GeoPoint::new_unchecked(-170.0, -80.0),
                GeoPoint::new_unchecked(170.0, -80.0),
                GeoPoint::new_unchecked(170.0, 80.0),
                GeoPoint::new_unchecked(-170.0, 80.0),
                GeoPoint::new_unchecked(-170.0, -80.0),
            ],
        },
        interiors: None,
    };
    let global_hashes = polygon_hashes(&global_polygon, GEO_QUERY_MAX_REGION).unwrap();

    let cases: Vec<(&str, Vec<GeoHash>)> = vec![
        ("europe_single_region", europe_hashes),
        ("nyc_plus_tokyo_disjoint", disjoint_hashes),
        ("overlapping_and_duplicated", overlapping_hashes),
        ("global_large_region", global_hashes),
    ];

    let assert_all_points_match = |mutable_index: &GeoIndex,
                                   immutable_index: &GeoIndex,
                                   storage_index: &OnDiskGeoIndex<MmapFile>,
                                   cases: &[(&str, Vec<GeoHash>)],
                                   phase: &str| {
        for (name, hashes) in cases {
            let hashes = hashes.clone();
            let mutable_result: HashSet<PointOffsetType> =
                mutable_index.iterator(hashes.clone()).unwrap().collect();

            let immutable_result: HashSet<PointOffsetType> =
                immutable_index.iterator(hashes.clone()).unwrap().collect();

            let all_points_result: HashSet<PointOffsetType> = storage_index
                .all_points(hashes)
                .unwrap()
                .into_iter()
                .collect();

            assert_eq!(
                mutable_result, immutable_result,
                "{phase}: case `{name}`: mutable vs immutable differ",
            );
            assert_eq!(
                mutable_result, all_points_result,
                "{phase}: case `{name}`: mutable vs all_points differ",
            );
        }
    };

    assert_all_points_match(
        &mutable_index,
        &immutable_index,
        storage_index,
        &cases,
        "baseline",
    );

    const DELETED_POINT_IDS: &[PointOffsetType] = &[10, 11, 12, 100, 150];
    for &id in DELETED_POINT_IDS {
        mutable_index.remove_point(id).unwrap();
        immutable_index.remove_point(id).unwrap();
        mmap_index.remove_point(id).unwrap();
    }

    let GeoIndex::OnDisk(storage_index) = &mmap_index else {
        panic!("expected Mmap variant to build into Storage");
    };

    assert_all_points_match(
        &mutable_index,
        &immutable_index,
        storage_index,
        &cases,
        "after point deletions",
    );
}

/// Reload contract: runtime deletions are not persisted by the mmap geo
/// index. Callers must re-supply the deletion bitslice on reload.
#[rstest]
#[case(IndexType::Mutable)]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn test_geo_index_reload(#[case] index_type: IndexType) {
    let temp_dir = {
        let (mut builder, temp_dir, _) = create_builder(index_type);

        let hw_counter = HardwareCounterCell::new();

        let berlin = json!(BERLIN);
        let potsdam = json!(POTSDAM);
        let tokyo = json!(TOKYO);
        let nyc = json!(NYC);

        builder.add_point(1, &[&berlin], &hw_counter).unwrap();
        builder.add_point(2, &[&berlin], &hw_counter).unwrap();
        builder.add_point(3, &[&potsdam], &hw_counter).unwrap();
        builder.add_point(4, &[&potsdam], &hw_counter).unwrap();
        builder.add_point(5, &[&tokyo], &hw_counter).unwrap();
        builder.add_point(6, &[&tokyo], &hw_counter).unwrap();
        builder.add_point(7, &[&nyc], &hw_counter).unwrap();

        let mut index = builder.finalize().unwrap();

        index.remove_point(2).unwrap();
        index.remove_point(3).unwrap();
        index.remove_point(6).unwrap();
        index.flusher()().unwrap();
        assert_eq!(index.points_count(), 4);
        drop(index);
        temp_dir
    };

    let new_index = reload_index(index_type, &temp_dir, &deleted_with(&[2, 3, 6]));

    assert_eq!(new_index.points_count(), 4);

    let berlin_radius = GeoRadius {
        center: BERLIN,
        radius: OrderedFloat(50_000.0),
    };
    let field_condition = condition_for_geo_radius("test", berlin_radius);
    let hits = filtered_points(&new_index, &field_condition);
    assert_eq!(hits, vec![1, 4]);

    let tokyo_radius = GeoRadius {
        center: TOKYO,
        radius: OrderedFloat(50_000.0),
    };
    let field_condition = condition_for_geo_radius("test", tokyo_radius);
    let hits = filtered_points(&new_index, &field_condition);
    assert_eq!(hits, vec![5]);

    let nyc_radius = GeoRadius {
        center: NYC,
        radius: OrderedFloat(50_000.0),
    };
    let field_condition = condition_for_geo_radius("test", nyc_radius);
    let hits = filtered_points(&new_index, &field_condition);
    assert_eq!(hits, vec![7]);
}

/// Regression test: when reloading an mmap geo index with a `deleted_points`
/// bitslice shorter than `point_to_values.len()`, missing entries must
/// default to live, not deleted.
#[rstest]
#[case(IndexType::OnDisk)]
#[case(IndexType::Immutable)]
fn test_geo_index_reload_short_deleted_bitslice(#[case] index_type: IndexType) {
    let temp_dir = {
        let (mut builder, temp_dir, _) = create_builder(index_type);

        let hw_counter = HardwareCounterCell::new();

        let berlin = json!(BERLIN);

        builder.add_point(1, &[&berlin], &hw_counter).unwrap();
        builder.add_point(2, &[&berlin], &hw_counter).unwrap();
        builder.add_point(3, &[], &hw_counter).unwrap();
        builder.add_point(4, &[&berlin], &hw_counter).unwrap();

        builder.finalize().unwrap();
        temp_dir
    };

    let mut short_deleted = BitVec::repeat(false, 2);
    short_deleted.set(1, true);
    let new_index = reload_index(index_type, &temp_dir, &short_deleted);

    let berlin_radius = GeoRadius {
        center: BERLIN,
        radius: OrderedFloat(50_000.0),
    };
    let field_condition = condition_for_geo_radius("test", berlin_radius);
    let hits = filtered_points(&new_index, &field_condition);
    assert_eq!(hits, vec![2, 4]);
}

/// The optional block-index sidecars over `counts_per_hash` and `points_map`
/// must be pure accelerators: cardinality estimates and filter results must
/// be identical with the sidecars present and after deleting them (which
/// falls back to plain binary search over the storage).
#[test]
fn test_block_index_fallback_equivalence() {
    fn collect_results(
        index: &GeoIndex,
        conditions: &[FieldCondition],
    ) -> Vec<(usize, usize, usize, Vec<PointOffsetType>)> {
        let hw_counter = HardwareCounterCell::new();
        conditions
            .iter()
            .map(|condition| {
                let estimation = index
                    .estimate_cardinality(condition, &hw_counter)
                    .unwrap()
                    .unwrap();
                let points = filtered_points(index, condition);
                (estimation.min, estimation.exp, estimation.max, points)
            })
            .collect()
    }

    let (index, temp_dir, _db) = build_random_index(2000, 5, IndexType::OnDisk);
    drop(index);

    let counts_sidecar = temp_dir
        .path()
        .join(super::on_disk_geo_index::COUNTS_PER_HASH_BLOCK_INDEX);
    let points_map_sidecar = temp_dir
        .path()
        .join(super::on_disk_geo_index::POINTS_MAP_BLOCK_INDEX);
    assert!(counts_sidecar.exists());
    assert!(points_map_sidecar.exists());

    // Radius queries around random centers (the payload fixture spreads
    // points across the whole globe) plus the named cities, with radii from
    // tight to continent-scale.
    let mut rnd = StdRng::seed_from_u64(7);
    let mut conditions = Vec::new();
    for i in 0..50 {
        let center = GeoPoint::new_unchecked(
            rand::RngExt::random_range(&mut rnd, LON_RANGE),
            rand::RngExt::random_range(&mut rnd, LAT_RANGE),
        );
        let radius_meters = 1_000.0 * 10_f64.powi(i % 4);
        conditions.push(condition_for_geo_radius(
            "test",
            GeoRadius {
                center,
                radius: OrderedFloat(radius_meters),
            },
        ));
    }
    for city in [NYC, BERLIN, POTSDAM, TOKYO, LOS_ANGELES] {
        conditions.push(condition_for_geo_radius(
            "test",
            GeoRadius {
                center: city,
                radius: OrderedFloat(500_000.0),
            },
        ));
    }

    let deleted = empty_deleted();

    let with_block_index = reload_index(IndexType::OnDisk, &temp_dir, &deleted);
    let with_results = collect_results(&with_block_index, &conditions);
    drop(with_block_index);

    fs_err::remove_file(&counts_sidecar).unwrap();
    fs_err::remove_file(&points_map_sidecar).unwrap();
    let without_block_index = reload_index(IndexType::OnDisk, &temp_dir, &deleted);
    let without_results = collect_results(&without_block_index, &conditions);

    assert_eq!(with_results, without_results);
    // Sanity: the query mix actually matches points.
    assert!(
        with_results
            .iter()
            .any(|(_, _, _, points)| !points.is_empty())
    );
}

/// The block-index sidecars must be covered by `preopen`: after
/// `schedule_prefetch`, `open` must be served from the prefetch pool without
/// touching the filesystem again. Conversely, absent sidecars (old segment)
/// must not fail `preopen` and must open in fallback mode.
#[test]
fn test_block_index_preopen() {
    use crate::common::universal_io::{
        CachedFs, CachedReadFs as _, Populate, ReadOnly, UniversalRead, UniversalReadFileOps as _,
    };

    type Storage = ReadOnly<MmapFile>;
    type RoFs = <Storage as UniversalRead>::Fs;

    let (index, temp_dir, _db) = build_random_index(2000, 5, IndexType::OnDisk);
    drop(index);
    let counts_sidecar = temp_dir
        .path()
        .join(super::on_disk_geo_index::COUNTS_PER_HASH_BLOCK_INDEX);
    let points_map_sidecar = temp_dir
        .path()
        .join(super::on_disk_geo_index::POINTS_MAP_BLOCK_INDEX);
    let deleted = empty_deleted();

    // Same order as the segment open path: snapshot, then preopen, then open.
    let fs = RoFs::from_context(Default::default()).unwrap();
    let mut cached_fs = CachedFs::new(fs.clone(), temp_dir.path()).unwrap();
    cached_fs.cache_file_info().unwrap();
    assert!(
        OnDiskGeoIndex::<Storage>::preopen(&cached_fs, temp_dir.path(), Populate::PreferBackground)
            .unwrap()
    );

    // The sidecar reads of `open` must now come from the prefetch pool.
    fs_err::remove_file(&counts_sidecar).unwrap();
    fs_err::remove_file(&points_map_sidecar).unwrap();

    let index = OnDiskGeoIndex::<Storage>::open(
        &cached_fs,
        temp_dir.path(),
        Populate::PreferBackground,
        &deleted,
    )
    .unwrap()
    .unwrap();
    assert!(index.storage.counts_per_hash_block_index.is_some());
    assert!(index.storage.points_map_block_index.is_some());
    drop(index);

    // Absent sidecars (segment built before they were introduced): `preopen`
    // must not fail on the missing files and `open` must fall back.
    let mut cached_fs = CachedFs::new(fs, temp_dir.path()).unwrap();
    cached_fs.cache_file_info().unwrap();
    assert!(
        OnDiskGeoIndex::<Storage>::preopen(&cached_fs, temp_dir.path(), Populate::PreferBackground)
            .unwrap()
    );
    let index = OnDiskGeoIndex::<Storage>::open(
        &cached_fs,
        temp_dir.path(),
        Populate::PreferBackground,
        &deleted,
    )
    .unwrap()
    .unwrap();
    assert!(index.storage.counts_per_hash_block_index.is_none());
    assert!(index.storage.points_map_block_index.is_none());
}