cleanup AggregationsSegmentCtx

The metric/cardinality/histogram _mut getters had no callers needing mutation; their two uses already pass the resulting reference as &T. simplify req_data ownership: clone into collectors, Rc only for filter BitSet Replace Vec<Option<Box<T>>> + take/put-back round-trip with Vec<T> + direct clone into collector. Collectors now own their per-segment request data outright, removing the borrow-checker dance that the take/put-back pattern existed to satisfy. The structural clones are cheap (Column<u64> is Arc-internal) except for the filter aggregation, whose DocumentQueryEvaluator carries a precomputed per-segment BitSet sized by max_doc. Wrap that in Rc<DocumentQueryEvaluator> so FilterAggReqData::clone() bumps a refcount instead of duplicating the BitSet. Move SegmentFilterCollector's matching_docs_buffer out of FilterAggReqData so its pre-allocated capacity is preserved per collector instead of being lost on every clone.
2026-06-09 12:00:42 +00:00 · 2026-05-07 11:14:14 +02:00
32 changed files with 551 additions and 1506 deletions
--- a/.github/workflows/coverage.yml
+++ b/.github/workflows/coverage.yml
@@ -20,7 +20,7 @@ jobs:
      contents: read

    steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
      - name: Install Rust
        run: rustup toolchain install nightly-2025-12-01 --profile minimal --component llvm-tools-preview
      - uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1
--- a/.github/workflows/long_running.yml
+++ b/.github/workflows/long_running.yml
@@ -25,7 +25,7 @@ jobs:
      contents: read

    steps:
-    - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+    - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
    - name: Install stable
      uses: actions-rs/toolchain@16499b5e05bf2e26879000db0c1d13f7e13fa3af # v1.0.7
      with:
--- a/.github/workflows/scorecard.yml
+++ b/.github/workflows/scorecard.yml
@@ -22,7 +22,7 @@ jobs:

    steps:
      - name: 'Checkout code'
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
        with:
          persist-credentials: false

@@ -36,7 +36,7 @@ jobs:

      # Upload the results as artifacts.
      - name: 'Upload artifact'
-        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
        with:
          name: SARIF file
          path: results.sarif
@@ -44,6 +44,6 @@ jobs:

      # Upload the results to GitHub's code scanning dashboard.
      - name: 'Upload to code-scanning'
-        uses: github/codeql-action/upload-sarif@8aad20d150bbac5944a9f9d289da16a4b0d87c1e # v4.36.2
+        uses: github/codeql-action/upload-sarif@95e58e9a2cdfd71adc6e0353d5c52f41a045d225 # v4.35.2
        with:
          sarif_file: results.sarif
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -27,7 +27,7 @@ jobs:
      checks: write

    steps:
-    - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+    - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2

    - name: Install nightly
      uses: actions-rs/toolchain@16499b5e05bf2e26879000db0c1d13f7e13fa3af # v1.0.7
@@ -77,7 +77,7 @@ jobs:
    name: test-${{ matrix.features.label}}

    steps:
-    - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+    - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2

    - name: Install stable
      uses: actions-rs/toolchain@16499b5e05bf2e26879000db0c1d13f7e13fa3af # v1.0.7
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -65,7 +65,7 @@ tantivy-bitpacker = { version = "0.10", path = "./bitpacker" }
 common = { version = "0.11", path = "./common/", package = "tantivy-common" }
 tokenizer-api = { version = "0.7", path = "./tokenizer-api", package = "tantivy-tokenizer-api" }
 sketches-ddsketch = { version = "0.4", features = ["use_serde"] }
-datasketches = { version = "0.3.0", features = ["hll"] }
+datasketches = { git = "https://github.com/fulmicoton-dd/datasketches-rust", rev = "7635fb8" }
 futures-util = { version = "0.3.28", optional = true }
 futures-channel = { version = "0.3.28", optional = true }
 fnv = "1.0.7"
@@ -75,7 +75,7 @@ typetag = "0.2.21"
 winapi = "0.3.9"

 [dev-dependencies]
-binggan = "0.17.0"
+binggan = "0.16.1"
 rand = "0.9"
 maplit = "1.0.2"
 matches = "0.1.9"
--- a/bitpacker/Cargo.toml
+++ b/bitpacker/Cargo.toml
@@ -18,10 +18,5 @@ homepage = "https://github.com/quickwit-oss/tantivy"
 bitpacking = { version = "0.9.2", default-features = false, features = ["bitpacker1x"] }

 [dev-dependencies]
-binggan = "0.17.0"
 rand = "0.9"
 proptest = "1"
-
-[[bench]]
-name = "bench"
-harness = false
--- a/bitpacker/benches/bench.rs
+++ b/bitpacker/benches/bench.rs
@@ -1,110 +1,65 @@
-use std::cell::RefCell;
+#![feature(test)]

-use binggan::{BenchRunner, black_box};
-use rand::rng;
-use rand::seq::IteratorRandom;
-use tantivy_bitpacker::{BitPacker, BitUnpacker, BlockedBitpacker};
+extern crate test;

-fn create_bitpacked_data(bit_width: u8, num_els: u32) -> Vec<u8> {
-    let mut bitpacker = BitPacker::new();
-    let mut buffer = Vec::new();
-    for _ in 0..num_els {
-        bitpacker.write(0u64, bit_width, &mut buffer).unwrap();
-        bitpacker.flush(&mut buffer).unwrap();
-    }
-    buffer
-}
+#[cfg(test)]
+mod tests {
+    use rand::rng;
+    use rand::seq::IteratorRandom;
+    use tantivy_bitpacker::{BitPacker, BitUnpacker, BlockedBitpacker};
+    use test::Bencher;

-const N: usize = 100_000;
-const MAX_VAL: u64 = 1_000;
-const BIT_WIDTH: u8 = 10; // 2^10 = 1024 > MAX_VAL
-
-fn create_packed_data() -> (BitUnpacker, Vec<u8>) {
-    let mut bitpacker = BitPacker::new();
-    let mut data = Vec::new();
-    for i in 0..N as u64 {
-        let val = i * MAX_VAL / N as u64;
-        bitpacker.write(val, BIT_WIDTH, &mut data).unwrap();
-    }
-    bitpacker.close(&mut data).unwrap();
-    (BitUnpacker::new(BIT_WIDTH), data)
-}
-
-fn bench_bitpacking() {
-    let mut runner = BenchRunner::new();
-    let bit_width = 3;
-    let num_els = 1_000_000u32;
-    let bit_unpacker = BitUnpacker::new(bit_width);
-    let data = create_bitpacked_data(bit_width, num_els);
-    let idxs: Vec<u32> = (0..num_els).choose_multiple(&mut rng(), 100_000);
-    runner.bench_function("bitpacking_read", move |_| {
-        let mut out = 0u64;
-        for &idx in &idxs {
-            out = out.wrapping_add(bit_unpacker.get(idx, &data[..]));
+    #[inline(never)]
+    fn create_bitpacked_data(bit_width: u8, num_els: u32) -> Vec<u8> {
+        let mut bitpacker = BitPacker::new();
+        let mut buffer = Vec::new();
+        for _ in 0..num_els {
+            // the values do not matter.
+            bitpacker.write(0u64, bit_width, &mut buffer).unwrap();
+            bitpacker.flush(&mut buffer).unwrap();
        }
-        black_box(out);
-    });
-}
-
-fn bench_blocked_bitpacker() {
-    let mut runner = BenchRunner::new();
-    let mut blocked_bitpacker = BlockedBitpacker::new();
-    for val in 0..=21500 {
-        blocked_bitpacker.add(val * val);
+        buffer
    }
-    runner.bench_function("blockedbitp_read", move |_| {
-        let mut out = 0u64;
-        for val in 0..=21500 {
-            out = out.wrapping_add(blocked_bitpacker.get(val));
-        }
-        black_box(out);
-    });
-    runner.bench_function("blockedbitp_create", |_| {
+
+    #[bench]
+    fn bench_bitpacking_read(b: &mut Bencher) {
+        let bit_width = 3;
+        let num_els = 1_000_000u32;
+        let bit_unpacker = BitUnpacker::new(bit_width);
+        let data = create_bitpacked_data(bit_width, num_els);
+        let idxs: Vec<u32> = (0..num_els).choose_multiple(&mut rng(), 100_000);
+        b.iter(|| {
+            let mut out = 0u64;
+            for &idx in &idxs {
+                out = out.wrapping_add(bit_unpacker.get(idx, &data[..]));
+            }
+            out
+        });
+    }
+
+    #[bench]
+    fn bench_blockedbitp_read(b: &mut Bencher) {
        let mut blocked_bitpacker = BlockedBitpacker::new();
        for val in 0..=21500 {
            blocked_bitpacker.add(val * val);
        }
-        black_box(blocked_bitpacker);
-    });
-}
-
-fn bench_filter_vec() {
-    let mut runner = BenchRunner::new();
-
-    let (unpacker, data) = create_packed_data();
-    let positions = RefCell::new(Vec::with_capacity(N));
-    runner.bench_function("filter_vec_dense", move |_| {
-        unpacker.get_ids_for_value_range(
-            250..=750,
-            0..N as u32,
-            &data,
-            &mut positions.borrow_mut(),
-        );
-        black_box(positions.borrow().len());
-    });
-
-    let (unpacker, data) = create_packed_data();
-    let positions = RefCell::new(Vec::with_capacity(N));
-    runner.bench_function("filter_vec_sparse", move |_| {
-        unpacker.get_ids_for_value_range(0..=50, 0..N as u32, &data, &mut positions.borrow_mut());
-        black_box(positions.borrow().len());
-    });
-
-    let (unpacker, data) = create_packed_data();
-    let positions = RefCell::new(Vec::with_capacity(N));
-    runner.bench_function("filter_vec_full", move |_| {
-        unpacker.get_ids_for_value_range(
-            0..=MAX_VAL,
-            0..N as u32,
-            &data,
-            &mut positions.borrow_mut(),
-        );
-        black_box(positions.borrow().len());
-    });
-}
-
-fn main() {
-    bench_bitpacking();
-    bench_blocked_bitpacker();
-    bench_filter_vec();
+        b.iter(|| {
+            let mut out = 0u64;
+            for val in 0..=21500 {
+                out = out.wrapping_add(blocked_bitpacker.get(val));
+            }
+            out
+        });
+    }
+
+    #[bench]
+    fn bench_blockedbitp_create(b: &mut Bencher) {
+        b.iter(|| {
+            let mut blocked_bitpacker = BlockedBitpacker::new();
+            for val in 0..=21500 {
+                blocked_bitpacker.add(val * val);
+            }
+            blocked_bitpacker
+        });
+    }
 }
--- a/bitpacker/src/filter_vec/mod.rs
+++ b/bitpacker/src/filter_vec/mod.rs
@@ -1,17 +1,8 @@
-#[cfg(all(target_arch = "aarch64", not(target_vendor = "apple")))]
-use std::arch::is_aarch64_feature_detected;
 use std::ops::RangeInclusive;

 #[cfg(target_arch = "x86_64")]
 mod avx2;

-#[cfg(target_arch = "aarch64")]
-mod neon;
-
-// SVE intrinsics are not exposed on aarch64-apple-darwin.
-#[cfg(all(target_arch = "aarch64", not(target_vendor = "apple")))]
-mod sve;
-
 mod scalar;

 #[derive(Clone, Copy, Eq, PartialEq, Debug)]
@@ -19,10 +10,6 @@ mod scalar;
 enum FilterImplPerInstructionSet {
    #[cfg(target_arch = "x86_64")]
    AVX2 = 0u8,
-    #[cfg(all(target_arch = "aarch64", not(target_vendor = "apple")))]
-    SVE = 3u8,
-    #[cfg(target_arch = "aarch64")]
-    Neon = 2u8,
    Scalar = 1u8,
 }

@@ -32,57 +19,29 @@ impl FilterImplPerInstructionSet {
        match *self {
            #[cfg(target_arch = "x86_64")]
            FilterImplPerInstructionSet::AVX2 => is_x86_feature_detected!("avx2"),
-            #[cfg(all(target_arch = "aarch64", not(target_vendor = "apple")))]
-            FilterImplPerInstructionSet::SVE => is_aarch64_feature_detected!("sve"),
-            // TIL Neon is required on aarch 64.
-            #[cfg(target_arch = "aarch64")]
-            FilterImplPerInstructionSet::Neon => true,
            FilterImplPerInstructionSet::Scalar => true,
        }
    }
 }

-// List of available implementations in preferred order.
+// List of available implementation in preferred order.
 #[cfg(target_arch = "x86_64")]
 const IMPLS: [FilterImplPerInstructionSet; 2] = [
    FilterImplPerInstructionSet::AVX2,
    FilterImplPerInstructionSet::Scalar,
 ];

-// Non-Apple aarch64: try SVE, NEON, Scalar.
-#[cfg(all(target_arch = "aarch64", not(target_vendor = "apple")))]
-const IMPLS: [FilterImplPerInstructionSet; 3] = [
-    FilterImplPerInstructionSet::SVE,
-    FilterImplPerInstructionSet::Neon,
-    FilterImplPerInstructionSet::Scalar,
-];
-
-// Apple aarch64 (M-series): SVE not available; use NEON or Scalar.
-#[cfg(all(target_arch = "aarch64", target_vendor = "apple"))]
-const IMPLS: [FilterImplPerInstructionSet; 2] = [
-    FilterImplPerInstructionSet::Neon,
-    FilterImplPerInstructionSet::Scalar,
-];
-
-#[cfg(not(any(target_arch = "x86_64", target_arch = "aarch64")))]
+#[cfg(not(target_arch = "x86_64"))]
 const IMPLS: [FilterImplPerInstructionSet; 1] = [FilterImplPerInstructionSet::Scalar];

 impl FilterImplPerInstructionSet {
    #[inline]
-    #[allow(unused_variables)]
+    #[allow(unused_variables)] // on non-x86_64, code is unused.
    fn from(code: u8) -> FilterImplPerInstructionSet {
        #[cfg(target_arch = "x86_64")]
        if code == FilterImplPerInstructionSet::AVX2 as u8 {
            return FilterImplPerInstructionSet::AVX2;
        }
-        #[cfg(all(target_arch = "aarch64", not(target_vendor = "apple")))]
-        if code == FilterImplPerInstructionSet::SVE as u8 {
-            return FilterImplPerInstructionSet::SVE;
-        }
-        #[cfg(target_arch = "aarch64")]
-        if code == FilterImplPerInstructionSet::Neon as u8 {
-            return FilterImplPerInstructionSet::Neon;
-        }
        FilterImplPerInstructionSet::Scalar
    }

@@ -91,13 +50,6 @@ impl FilterImplPerInstructionSet {
        match self {
            #[cfg(target_arch = "x86_64")]
            FilterImplPerInstructionSet::AVX2 => avx2::filter_vec_in_place(range, offset, output),
-            #[cfg(all(target_arch = "aarch64", not(target_vendor = "apple")))]
-            // SAFETY: SVE availability was verified by is_available() before selecting this impl.
-            FilterImplPerInstructionSet::SVE => unsafe {
-                sve::filter_vec_in_place(range, offset, output)
-            },
-            #[cfg(target_arch = "aarch64")]
-            FilterImplPerInstructionSet::Neon => neon::filter_vec_in_place(range, offset, output),
            FilterImplPerInstructionSet::Scalar => {
                scalar::filter_vec_in_place(range, offset, output)
            }
@@ -105,12 +57,6 @@ impl FilterImplPerInstructionSet {
    }
 }

-fn available_impls() -> impl Iterator<Item = FilterImplPerInstructionSet> {
-    IMPLS
-        .into_iter()
-        .filter(FilterImplPerInstructionSet::is_available)
-}
-
 #[inline]
 fn get_best_available_instruction_set() -> FilterImplPerInstructionSet {
    use std::sync::atomic::{AtomicU8, Ordering};
@@ -118,7 +64,10 @@ fn get_best_available_instruction_set() -> FilterImplPerInstructionSet {
    let instruction_set_byte: u8 = INSTRUCTION_SET_BYTE.load(Ordering::Relaxed);
    if instruction_set_byte == u8::MAX {
        // Let's initialize the instruction set and cache it.
-        let instruction_set = available_impls().next().unwrap();
+        let instruction_set = IMPLS
+            .into_iter()
+            .find(FilterImplPerInstructionSet::is_available)
+            .unwrap();
        INSTRUCTION_SET_BYTE.store(instruction_set as u8, Ordering::Relaxed);
        return instruction_set;
    }
@@ -131,12 +80,12 @@ pub fn filter_vec_in_place(range: RangeInclusive<u32>, offset: u32, output: &mut

 #[cfg(test)]
 mod tests {
-    use proptest::strategy::Strategy;
-
    use super::*;

    #[test]
    fn test_get_best_available_instruction_set() {
+        // This does not test much unfortunately.
+        // We just make sure the function returns without crashing and returns the same result.
        let instruction_set = get_best_available_instruction_set();
        assert_eq!(get_best_available_instruction_set(), instruction_set);
    }
@@ -153,31 +102,6 @@ mod tests {
        }
    }

-    #[cfg(all(target_arch = "aarch64", not(target_vendor = "apple")))]
-    #[test]
-    fn test_instruction_set_to_code_from_code() {
-        for instruction_set in [
-            FilterImplPerInstructionSet::SVE,
-            FilterImplPerInstructionSet::Neon,
-            FilterImplPerInstructionSet::Scalar,
-        ] {
-            let code = instruction_set as u8;
-            assert_eq!(instruction_set, FilterImplPerInstructionSet::from(code));
-        }
-    }
-
-    #[cfg(all(target_arch = "aarch64", target_vendor = "apple"))]
-    #[test]
-    fn test_instruction_set_to_code_from_code() {
-        for instruction_set in [
-            FilterImplPerInstructionSet::Neon,
-            FilterImplPerInstructionSet::Scalar,
-        ] {
-            let code = instruction_set as u8;
-            assert_eq!(instruction_set, FilterImplPerInstructionSet::from(code));
-        }
-    }
-
    fn test_filter_impl_empty_aux(filter_impl: FilterImplPerInstructionSet) {
        let mut output = vec![];
        filter_impl.filter_vec_in_place(0..=u32::MAX, 0, &mut output);
@@ -202,20 +126,11 @@ mod tests {
        assert_eq!(&output, &[1, 3, 4, 5, 6, 7, 8]);
    }

-    fn test_filter_impl_empty_range_aux(filter_impl: FilterImplPerInstructionSet) {
-        // start > end: RangeInclusive::contains always returns false; output must be empty.
-        // The SVE path's wrapping_sub would otherwise produce a huge range_width.
-        let mut output = vec![3, 2, 1, 5, 11, 2, 5, 10, 2];
-        filter_impl.filter_vec_in_place(10..=5, 0, &mut output);
-        assert_eq!(&output, &[]);
-    }
-
    fn test_filter_impl_test_suite(filter_impl: FilterImplPerInstructionSet) {
        test_filter_impl_empty_aux(filter_impl);
        test_filter_impl_simple_aux(filter_impl);
        test_filter_impl_simple_aux_shifted(filter_impl);
        test_filter_impl_simple_outside_i32_range(filter_impl);
-        test_filter_impl_empty_range_aux(filter_impl);
    }

    #[test]
@@ -226,60 +141,25 @@ mod tests {
        }
    }

-    #[test]
-    #[cfg(all(target_arch = "aarch64", not(target_vendor = "apple")))]
-    fn test_filter_implementation_sve() {
-        if FilterImplPerInstructionSet::SVE.is_available() {
-            test_filter_impl_test_suite(FilterImplPerInstructionSet::SVE);
-        }
-    }
-
-    #[test]
-    #[cfg(target_arch = "aarch64")]
-    fn test_filter_implementation_neon() {
-        test_filter_impl_test_suite(FilterImplPerInstructionSet::Neon);
-    }
-
    #[test]
    fn test_filter_implementation_scalar() {
        test_filter_impl_test_suite(FilterImplPerInstructionSet::Scalar);
    }

-    fn max_val_strategy() -> impl proptest::strategy::Strategy<Value = u32> {
-        proptest::prop_oneof![
-            0u32..10u32,
-            255u32..258u32,
-            proptest::prelude::Just(1u32 << 25),
-            proptest::prelude::Just(u32::MAX - 1),
-            proptest::prelude::Just(u32::MAX),
-        ]
-    }
-
-    fn vals_strategy() -> impl proptest::strategy::Strategy<Value = Vec<u32>> {
-        proptest::prop_oneof![
-            proptest::collection::vec(proptest::prelude::any::<u32>(), 0..300),
-            max_val_strategy()
-                .prop_flat_map(|max_val| { proptest::collection::vec(0..=max_val, 0..300) })
-        ]
-    }
-
+    #[cfg(target_arch = "x86_64")]
    proptest::proptest! {
        #[test]
-        fn test_filter_compare_scalar_and_impls_impl_proptest(
-            start in 0u32..400u32,
-            end in 0u32..400u32,
+        fn test_filter_compare_scalar_and_avx2_impl_proptest(
+            start in proptest::prelude::any::<u32>(),
+            end in proptest::prelude::any::<u32>(),
            offset in 0u32..2u32,
-            vals in vals_strategy()) {
-                for implementation in available_impls() {
-                    if implementation == FilterImplPerInstructionSet::Scalar {
-                        continue;
-                    }
-                    let mut impl_output = vals.clone();
-                    let mut scalar_output = vals.clone();
-                    implementation.filter_vec_in_place(start..=end, offset, &mut impl_output);
-                    FilterImplPerInstructionSet::Scalar.filter_vec_in_place(start..=end, offset, &mut scalar_output);
-                    assert_eq!(&impl_output, &scalar_output);
-                }
+            mut vals in proptest::collection::vec(0..u32::MAX, 0..30)) {
+            if FilterImplPerInstructionSet::AVX2.is_available() {
+                let mut vals_clone = vals.clone();
+                FilterImplPerInstructionSet::AVX2.filter_vec_in_place(start..=end, offset, &mut vals);
+                FilterImplPerInstructionSet::Scalar.filter_vec_in_place(start..=end, offset, &mut vals_clone);
+                assert_eq!(&vals, &vals_clone);
+            }
       }
    }
 }
--- a/bitpacker/src/filter_vec/neon.rs
+++ b/bitpacker/src/filter_vec/neon.rs
@@ -1,118 +0,0 @@
-use std::arch::aarch64::*;
-use std::ops::RangeInclusive;
-
-const NUM_LANES: usize = 4;
-
-// Compacts matching lanes to the front using a byte-level shuffle.
-// `mask` is a 4-bit value: bit k=1 means lane k should appear in the output.
-#[inline]
-#[target_feature(enable = "neon")]
-unsafe fn compact(data: uint32x4_t, mask: u8) -> uint32x4_t {
-    unsafe {
-        // SAFETY: mask is always in [0, 15] by construction (max sum of [1,2,4,8]).
-        // BYTE_SHUFFLE_TABLE has 16 entries, so this is always in bounds.
-        let shuffle = BYTE_SHUFFLE_TABLE.get_unchecked(mask as usize);
-        let shuffle_vec = vld1q_u8(shuffle.as_ptr());
-        vreinterpretq_u32_u8(vqtbl1q_u8(vreinterpretq_u8_u32(data), shuffle_vec))
-    }
-}
-
-// Safe (not unsafe) because NEON is mandatory on aarch64: no runtime feature check needed.
-#[inline(never)]
-pub fn filter_vec_in_place(range: RangeInclusive<u32>, offset: u32, output: &mut Vec<u32>) {
-    let num_words = output.len() / NUM_LANES;
-    let mut output_len = unsafe {
-        filter_vec_neon_aux(
-            output.as_ptr(),
-            range.clone(),
-            output.as_mut_ptr(),
-            offset,
-            num_words,
-        )
-    };
-    let remainder_start = num_words * NUM_LANES;
-    for i in remainder_start..output.len() {
-        let val = output[i];
-        output[output_len] = offset + i as u32;
-        output_len += if range.contains(&val) { 1 } else { 0 };
-    }
-    output.truncate(output_len);
-}
-
-#[target_feature(enable = "neon")]
-unsafe fn filter_vec_neon_aux(
-    input: *const u32,
-    range: RangeInclusive<u32>,
-    output: *mut u32,
-    offset: u32,
-    num_words: usize,
-) -> usize {
-    unsafe {
-        let mut input = input;
-        let mut output_tail = output;
-        let range_start_simd = vdupq_n_u32(*range.start());
-        let range_end_simd = vdupq_n_u32(*range.end());
-        let mut ids = vld1q_u32([offset, offset + 1, offset + 2, offset + 3].as_ptr());
-        let shift = vdupq_n_u32(NUM_LANES as u32);
-        let bit_weights = vld1q_u32([1u32, 2, 4, 8].as_ptr());
-
-        for _ in 0..num_words {
-            let word = vld1q_u32(input);
-
-            // Unsigned compares: CMHS (compare higher or same) tests `word >= start`
-            // and `end >= word`. ANDing both gives the inside-range mask directly,
-            // which is cheaper than computing `outside` and then negating.
-            let ge_start = vcgeq_u32(word, range_start_simd);
-            let le_end = vcleq_u32(word, range_end_simd);
-            // inside[k] = 0xFFFFFFFF if val[k] is in range, 0 otherwise.
-            let inside = vandq_u32(ge_start, le_end);
-
-            // Build the 4-bit mask: AND bit_weights with the inside lane mask, so each
-            // inside lane contributes its bit_weight (1, 2, 4, or 8). Summing yields the
-            // 4-bit mask in one addv.
-            let inside_bits = vandq_u32(bit_weights, inside);
-            let mask = vaddvq_u32(inside_bits) as u8;
-            // mask is mathematically bounded: max value is 1+2+4+8=15 (all lanes match)
-            debug_assert!(mask <= 15, "mask must fit in 4 bits: {}", mask);
-
-            // Count of matching lanes = popcount(mask). Derives the count directly from
-            // the mask instead of running a parallel SIMD reduction over `outside`.
-            let added_len = mask.count_ones() as usize;
-
-            // Safe because mask is guaranteed to be in [0, 15]
-            let filtered_ids = compact(ids, mask);
-            vst1q_u32(output_tail, filtered_ids);
-            output_tail = output_tail.add(added_len);
-            ids = vaddq_u32(ids, shift);
-            input = input.add(NUM_LANES);
-        }
-
-        output_tail.offset_from(output) as usize
-    }
-}
-
-// Byte shuffle patterns to compact matching lanes to the front of the vector.
-// Index is a 4-bit mask: bit k=1 means lane k (bytes 4k..4k+3) is in-range.
-// The j-th set bit determines which input lane goes to output position j.
-const BYTE_SHUFFLE_TABLE: [[u8; 16]; 16] = [
-    [
-        16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
-    ], // 0b0000: none
-    [0, 1, 2, 3, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16], // 0b0001: lane 0
-    [4, 5, 6, 7, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16], // 0b0010: lane 1
-    [0, 1, 2, 3, 4, 5, 6, 7, 16, 16, 16, 16, 16, 16, 16, 16],     // 0b0011: lanes 0,1
-    [8, 9, 10, 11, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16], // 0b0100: lane 2
-    [0, 1, 2, 3, 8, 9, 10, 11, 16, 16, 16, 16, 16, 16, 16, 16],   // 0b0101: lanes 0,2
-    [4, 5, 6, 7, 8, 9, 10, 11, 16, 16, 16, 16, 16, 16, 16, 16],   // 0b0110: lanes 1,2
-    [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 16, 16, 16, 16],       // 0b0111: lanes 0,1,2
-    [
-        12, 13, 14, 15, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
-    ], // 0b1000: lane 3
-    [0, 1, 2, 3, 12, 13, 14, 15, 16, 16, 16, 16, 16, 16, 16, 16], // 0b1001: lanes 0,3
-    [4, 5, 6, 7, 12, 13, 14, 15, 16, 16, 16, 16, 16, 16, 16, 16], // 0b1010: lanes 1,3
-    [0, 1, 2, 3, 4, 5, 6, 7, 12, 13, 14, 15, 16, 16, 16, 16],     // 0b1011: lanes 0,1,3
-    [8, 9, 10, 11, 12, 13, 14, 15, 16, 16, 16, 16, 16, 16, 16, 16], // 0b1100: lanes 2,3
-    [0, 1, 2, 3, 8, 9, 10, 11, 12, 13, 14, 15, 16, 16, 16, 16],   // 0b1101: lanes 0,2,3
-    [4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 16, 16, 16],   // 0b1110: lanes 1,2,3
-    [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15],       // 0b1111: all lanes
-];
--- a/bitpacker/src/filter_vec/sve.rs
+++ b/bitpacker/src/filter_vec/sve.rs
@@ -1,260 +0,0 @@
-use std::ops::RangeInclusive;
-
-// SVE vector length (in u32 lanes) is not a compile-time constant; query at runtime.
-// Safe to call only when SVE is confirmed available via is_aarch64_feature_detected!("sve").
-#[target_feature(enable = "sve")]
-unsafe fn num_lanes() -> usize {
-    let vl: usize;
-    unsafe {
-        core::arch::asm!(
-            "cntw {vl}",
-            vl = out(reg) vl,
-            options(nostack, nomem, preserves_flags),
-        );
-    }
-    vl
-}
-
-// SAFETY: caller must ensure SVE is available (checked via is_aarch64_feature_detected!("sve")).
-// Unlike NEON, SVE is optional on aarch64 and not guaranteed by the target architecture.
-pub unsafe fn filter_vec_in_place(range: RangeInclusive<u32>, offset: u32, output: &mut Vec<u32>) {
-    if range.start() > range.end() {
-        output.clear();
-        return;
-    }
-    let vl = unsafe { num_lanes() };
-    let num_words = output.len() / vl;
-    let range_start = *range.start();
-    // Unsigned subtraction trick: val ∈ [lo, hi] ↔ (val - lo) ≤ᵤ (hi - lo).
-    // Values below lo wrap around to large u32, so the single unsigned ≤ excludes them.
-    let range_width = range.end().wrapping_sub(range_start);
-    let mut output_len = unsafe {
-        filter_vec_sve_aux(
-            output.as_ptr(),
-            range_start,
-            range_width,
-            output.as_mut_ptr(),
-            offset,
-            num_words,
-            vl,
-        )
-    };
-    let remainder_start = num_words * vl;
-    for i in remainder_start..output.len() {
-        let val = output[i];
-        output[output_len] = offset + i as u32;
-        output_len += if range.contains(&val) { 1 } else { 0 };
-    }
-    output.truncate(output_len);
-}
-
-// Register allocation for the asm! blocks:
-//   z0        ids_a (index vector for first half of each pair, advances by step2 each iter)
-//   z1        range_width broadcast
-//   z2        range_start broadcast
-//   z3        step2 broadcast (2 * vl)
-//   z4        ids_b (index vector for second half, = ids_a + step, advances by step2)
-//   z5        scratch: loaded word_a, then compacted_a
-//   z6        scratch: loaded word_b, then compacted_b
-//   p0        all-true predicate (ptrue p0.s)
-//   p1        in-range mask for word_a
-//   p2        in-range mask for word_b
-#[target_feature(enable = "sve")]
-unsafe fn filter_vec_sve_aux(
-    input: *const u32,
-    range_start: u32,
-    range_width: u32,
-    output: *mut u32,
-    offset: u32,
-    num_words: usize,
-    vl: usize,
-) -> usize {
-    let num_pairs = num_words / 2;
-    let mut input_ptr = input;
-    let mut output_tail = output;
-
-    if num_pairs > 0 {
-        unsafe {
-            // We rely on asm! because the SVE intrinsics are not available in stable Rust.
-            // The code that follows was generated by Rustc nightly based on the intrinsics version
-            // at the bottom of this file.
-            core::arch::asm!(
-                // --- Setup ---
-                // All-true predicate for 32-bit lanes.
-                "ptrue p0.s",
-                // ids_a = [offset, offset+1, offset+2, ...]
-                "index z0.s, {offset:w}, #1",
-                // Broadcast scalars into SVE vectors.
-                "mov z1.s, {range_width:w}",
-                "mov z2.s, {range_start:w}",
-                // vl_gpr = number of 32-bit lanes (cntw).
-                "cntw {vl_gpr}",
-                // step2_bytes will first hold 2*vl (for the step2 vector), then 2*VL in bytes.
-                "lsl {step2_bytes}, {vl_gpr}, #1",
-                // z4 = step = [vl, vl, ...]; will become ids_b after the add below.
-                "mov z4.s, {vl_gpr:w}",
-                // z3 = step2 = [2*vl, 2*vl, ...], used to advance both id vectors each iter.
-                "mov z3.s, {step2_bytes:w}",
-                // Repurpose step2_bytes to hold the byte stride for advancing the input pointer
-                // by two full SVE vectors per iteration.
-                "rdvl {step2_bytes}, #2",
-                // ids_b = ids_a + step = [offset+vl, offset+vl+1, ...]
-                "add z4.s, z0.s, z4.s",
-
-                // --- Main loop: process two SVE vectors (ids_a and ids_b) per iteration ---
-                "0:",
-                // Load two consecutive SVE vectors from input.
-                "ld1w {{z5.s}}, p0/z, [{input}]",
-                "ld1w {{z6.s}}, p0/z, [{input}, #1, mul vl]",
-                // Advance input pointer by 2 * VL bytes.
-                "add {input}, {input}, {step2_bytes}",
-                // Unsigned shift: subtract range_start so in-range check becomes a single cmpu ≤.
-                "sub z5.s, z5.s, z2.s",
-                "sub z6.s, z6.s, z2.s",
-                // in_range: shifted value ≤ range_width  (unsigned, so values below lo also fail).
-                "cmphs p1.s, p0/z, z1.s, z5.s",
-                "cmphs p2.s, p0/z, z1.s, z6.s",
-                // Count matching lanes; both cntp calls have independent inputs for OOO parallelism.
-                "cntp {cnt_a}, p0, p1.s",
-                "compact z5.s, p1, z0.s",
-                "compact z6.s, p2, z4.s",
-                "cntp {cnt_b}, p0, p2.s",
-                // Advance id vectors for the next iteration.
-                "add z0.s, z0.s, z3.s",
-                "add z4.s, z4.s, z3.s",
-                // Store compacted ids. Only the first cnt_a / cnt_b slots are valid; the rest
-                // will be overwritten by subsequent iterations before the final truncate.
-                "str z5, [{out}]",
-                "st1w {{z6.s}}, p0, [{out}, {cnt_a}, lsl #2]",
-                "add {out}, {out}, {cnt_a}, lsl #2",
-                "add {out}, {out}, {cnt_b}, lsl #2",
-                "subs {pairs}, {pairs}, #1",
-                "b.ne 0b",
-
-                // --- Operands ---
-                input       = inout(reg) input_ptr,
-                out         = inout(reg) output_tail,
-                pairs       = inout(reg) num_pairs => _,
-                offset      = in(reg) offset,
-                range_start = in(reg) range_start,
-                range_width = in(reg) range_width,
-                vl_gpr      = out(reg) _,
-                step2_bytes = out(reg) _,
-                cnt_a       = out(reg) _,
-                cnt_b       = out(reg) _,
-                out("p0") _, out("p1") _, out("p2") _,
-                out("v0") _, out("v1") _, out("v2") _, out("v3") _,
-                out("v4") _, out("v5") _, out("v6") _,
-                options(nostack),
-            );
-        }
-    }
-
-    // Handle an odd trailing vector.
-    if num_words % 2 == 1 {
-        // ids_a for the odd word starts at offset + num_pairs * 2 * vl.
-        // input_ptr was advanced by the main loop and now points at the odd word.
-        let odd_offset =
-            offset.wrapping_add((num_pairs as u32).wrapping_mul(2).wrapping_mul(vl as u32));
-        unsafe {
-            core::arch::asm!(
-                "ptrue p0.s",
-                "index z0.s, {odd_offset:w}, #1",
-                "mov z1.s, {range_width:w}",
-                "mov z2.s, {range_start:w}",
-                "ld1w {{z3.s}}, p0/z, [{input}]",
-                "sub z3.s, z3.s, z2.s",
-                "cmphs p1.s, p0/z, z1.s, z3.s",
-                "cntp {cnt}, p0, p1.s",
-                "compact z0.s, p1, z0.s",
-                "str z0, [{out}]",
-                "add {out}, {out}, {cnt}, lsl #2",
-                odd_offset  = in(reg) odd_offset,
-                range_width = in(reg) range_width,
-                range_start = in(reg) range_start,
-                input       = in(reg) input_ptr,
-                out         = inout(reg) output_tail,
-                cnt         = out(reg) _,
-                out("p0") _, out("p1") _,
-                out("v0") _, out("v1") _, out("v2") _, out("v3") _,
-                options(nostack),
-            );
-        }
-    }
-
-    unsafe { output_tail.offset_from(output) as usize }
-}
-
-// SVE implements with intrinsics.
-//
-// #[target_feature(enable = "sve")]
-// unsafe fn filter_vec_sve_aux(
-//     input: *const u32,
-//     range_start: u32,
-//     range_width: u32,
-//     output: *mut u32,
-//     offset: u32,
-//     num_words: usize,
-//     vl: usize,
-// ) -> usize {
-//     unsafe {
-//         let all_true = svptrue_b32();
-//         let range_start_simd = svdup_n_u32(range_start);
-//         let range_width_simd = svdup_n_u32(range_width);
-//         // ids_a covers [offset .. offset+vl), ids_b covers the next vl ids.
-//         // Keeping them separate breaks the loop-carried dependency through ids so
-//         // both compact/cntp chains are fully independent within each unrolled body.
-//         let mut ids_a = svindex_u32(offset, 1);
-//         let step = svdup_n_u32(vl as u32);
-//         let step2 = svdup_n_u32(2 * vl as u32);
-//         let mut ids_b = svadd_u32_x(all_true, ids_a, step);
-
-//         let mut input = input;
-//         let mut output_tail = output;
-
-//         // Unrolled ×2: both cntp calls have independent inputs and execute in parallel.
-//         // The two output_tail updates are sequential but together cost 4+1+1=6 cy per
-//         // pair vs 5+5=10 cy for two scalar iterations, breaking the cntp latency chain.
-//         let num_pairs = num_words / 2;
-//         for _ in 0..num_pairs {
-//             let word_a = svld1_u32(all_true, input);
-//             let word_b = svld1_u32(all_true, input.add(vl));
-
-//             let shifted_a = svsub_u32_x(all_true, word_a, range_start_simd);
-//             let shifted_b = svsub_u32_x(all_true, word_b, range_start_simd);
-
-//             let in_range_a = svcmple_u32(all_true, shifted_a, range_width_simd);
-//             let in_range_b = svcmple_u32(all_true, shifted_b, range_width_simd);
-
-//             let compacted_a = svcompact_u32(in_range_a, ids_a);
-//             let compacted_b = svcompact_u32(in_range_b, ids_b);
-//             // cntp_a and cntp_b have independent inputs: OOO engine issues them in parallel.
-//             let added_len_a = svcntp_b32(all_true, in_range_a) as usize;
-//             let added_len_b = svcntp_b32(all_true, in_range_b) as usize;
-
-//             // Write the full vector — only the first added_len slots are valid.
-//             // Subsequent iterations overwrite the trailing zeros before truncate.
-//             svst1_u32(all_true, output_tail, compacted_a);
-//             output_tail = output_tail.add(added_len_a);
-//             svst1_u32(all_true, output_tail, compacted_b);
-//             output_tail = output_tail.add(added_len_b);
-
-//             ids_a = svadd_u32_x(all_true, ids_a, step2);
-//             ids_b = svadd_u32_x(all_true, ids_b, step2);
-//             input = input.add(2 * vl);
-//         }
-
-//         // Handle an odd trailing word.
-//         if num_words % 2 == 1 {
-//             let word = svld1_u32(all_true, input);
-//             let shifted = svsub_u32_x(all_true, word, range_start_simd);
-//             let in_range = svcmple_u32(all_true, shifted, range_width_simd);
-//             let added_len = svcntp_b32(all_true, in_range) as usize;
-//             let compacted_ids = svcompact_u32(in_range, ids_a);
-//             svst1_u32(all_true, output_tail, compacted_ids);
-//             output_tail = output_tail.add(added_len);
-//         }
-
-//         output_tail.offset_from(output) as usize
-//     }
-// }
--- a/columnar/Cargo.toml
+++ b/columnar/Cargo.toml
@@ -23,7 +23,7 @@ downcast-rs = "2.0.1"
 proptest = "1"
 more-asserts = "0.3.1"
 rand = "0.9"
-binggan = "0.17.0"
+binggan = "0.16.1"

 [[bench]]
 name = "bench_merge"
--- a/common/Cargo.toml
+++ b/common/Cargo.toml
@@ -19,6 +19,6 @@ time = { version = "0.3.47", features = ["serde-well-known"] }
 serde = { version = "1.0.136", features = ["derive"] }

 [dev-dependencies]
-binggan = "0.17.0"
+binggan = "0.16.1"
 proptest = "1.0.0"
 rand = "0.9"
--- a/common/src/file_slice.rs
+++ b/common/src/file_slice.rs
@@ -121,7 +121,7 @@ pub struct FileSlice {

 impl fmt::Debug for FileSlice {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "FileSlice({:?}, {:?})", self.data, self.range)
+        write!(f, "FileSlice({:?}, {:?})", &self.data, self.range)
    }
 }

--- a/src/aggregation/agg_data.rs
+++ b/src/aggregation/agg_data.rs
@@ -41,7 +41,7 @@ pub struct AggregationsSegmentCtx {

 impl AggregationsSegmentCtx {
    pub(crate) fn push_term_req_data(&mut self, data: TermsAggReqData) -> usize {
-        self.per_request.term_req_data.push(Some(Box::new(data)));
+        self.per_request.term_req_data.push(data);
        self.per_request.term_req_data.len() - 1
    }
    pub(crate) fn push_cardinality_req_data(&mut self, data: CardinalityAggReqData) -> usize {
@@ -61,31 +61,25 @@ impl AggregationsSegmentCtx {
        self.per_request.missing_term_req_data.len() - 1
    }
    pub(crate) fn push_histogram_req_data(&mut self, data: HistogramAggReqData) -> usize {
-        self.per_request
-            .histogram_req_data
-            .push(Some(Box::new(data)));
+        self.per_request.histogram_req_data.push(data);
        self.per_request.histogram_req_data.len() - 1
    }
    pub(crate) fn push_range_req_data(&mut self, data: RangeAggReqData) -> usize {
-        self.per_request.range_req_data.push(Some(Box::new(data)));
+        self.per_request.range_req_data.push(data);
        self.per_request.range_req_data.len() - 1
    }
    pub(crate) fn push_filter_req_data(&mut self, data: FilterAggReqData) -> usize {
-        self.per_request.filter_req_data.push(Some(Box::new(data)));
+        self.per_request.filter_req_data.push(data);
        self.per_request.filter_req_data.len() - 1
    }
    pub(crate) fn push_composite_req_data(&mut self, data: CompositeAggReqData) -> usize {
-        self.per_request
-            .composite_req_data
-            .push(Some(Box::new(data)));
+        self.per_request.composite_req_data.push(data);
        self.per_request.composite_req_data.len() - 1
    }

    #[inline]
    pub(crate) fn get_term_req_data(&self, idx: usize) -> &TermsAggReqData {
-        self.per_request.term_req_data[idx]
-            .as_deref()
-            .expect("term_req_data slot is empty (taken)")
+        &self.per_request.term_req_data[idx]
    }
    #[inline]
    pub(crate) fn get_cardinality_req_data(&self, idx: usize) -> &CardinalityAggReqData {
@@ -103,116 +97,6 @@ impl AggregationsSegmentCtx {
    pub(crate) fn get_missing_term_req_data(&self, idx: usize) -> &MissingTermAggReqData {
        &self.per_request.missing_term_req_data[idx]
    }
-    #[inline]
-    pub(crate) fn get_histogram_req_data(&self, idx: usize) -> &HistogramAggReqData {
-        self.per_request.histogram_req_data[idx]
-            .as_deref()
-            .expect("histogram_req_data slot is empty (taken)")
-    }
-    #[inline]
-    pub(crate) fn get_range_req_data(&self, idx: usize) -> &RangeAggReqData {
-        self.per_request.range_req_data[idx]
-            .as_deref()
-            .expect("range_req_data slot is empty (taken)")
-    }
-    #[inline]
-    pub(crate) fn get_composite_req_data(&self, idx: usize) -> &CompositeAggReqData {
-        self.per_request.composite_req_data[idx]
-            .as_deref()
-            .expect("composite_req_data slot is empty (taken)")
-    }
-
-    // ---------- mutable getters ----------
-
-    #[inline]
-    pub(crate) fn get_metric_req_data_mut(&mut self, idx: usize) -> &mut MetricAggReqData {
-        &mut self.per_request.stats_metric_req_data[idx]
-    }
-
-    #[inline]
-    pub(crate) fn get_cardinality_req_data_mut(
-        &mut self,
-        idx: usize,
-    ) -> &mut CardinalityAggReqData {
-        &mut self.per_request.cardinality_req_data[idx]
-    }
-
-    #[inline]
-    pub(crate) fn get_histogram_req_data_mut(&mut self, idx: usize) -> &mut HistogramAggReqData {
-        self.per_request.histogram_req_data[idx]
-            .as_deref_mut()
-            .expect("histogram_req_data slot is empty (taken)")
-    }
-
-    // ---------- take / put (terms, histogram, range) ----------
-
-    /// Move out the boxed Histogram request at `idx`, leaving `None`.
-    #[inline]
-    pub(crate) fn take_histogram_req_data(&mut self, idx: usize) -> Box<HistogramAggReqData> {
-        self.per_request.histogram_req_data[idx]
-            .take()
-            .expect("histogram_req_data slot is empty (taken)")
-    }
-
-    /// Put back a Histogram request into an empty slot at `idx`.
-    #[inline]
-    pub(crate) fn put_back_histogram_req_data(
-        &mut self,
-        idx: usize,
-        value: Box<HistogramAggReqData>,
-    ) {
-        debug_assert!(self.per_request.histogram_req_data[idx].is_none());
-        self.per_request.histogram_req_data[idx] = Some(value);
-    }
-
-    /// Move out the boxed Range request at `idx`, leaving `None`.
-    #[inline]
-    pub(crate) fn take_range_req_data(&mut self, idx: usize) -> Box<RangeAggReqData> {
-        self.per_request.range_req_data[idx]
-            .take()
-            .expect("range_req_data slot is empty (taken)")
-    }
-
-    /// Put back a Range request into an empty slot at `idx`.
-    #[inline]
-    pub(crate) fn put_back_range_req_data(&mut self, idx: usize, value: Box<RangeAggReqData>) {
-        debug_assert!(self.per_request.range_req_data[idx].is_none());
-        self.per_request.range_req_data[idx] = Some(value);
-    }
-
-    /// Move out the boxed Filter request at `idx`, leaving `None`.
-    #[inline]
-    pub(crate) fn take_filter_req_data(&mut self, idx: usize) -> Box<FilterAggReqData> {
-        self.per_request.filter_req_data[idx]
-            .take()
-            .expect("filter_req_data slot is empty (taken)")
-    }
-
-    /// Put back a Filter request into an empty slot at `idx`.
-    #[inline]
-    pub(crate) fn put_back_filter_req_data(&mut self, idx: usize, value: Box<FilterAggReqData>) {
-        debug_assert!(self.per_request.filter_req_data[idx].is_none());
-        self.per_request.filter_req_data[idx] = Some(value);
-    }
-
-    /// Move out the Composite request at `idx`.
-    #[inline]
-    pub(crate) fn take_composite_req_data(&mut self, idx: usize) -> Box<CompositeAggReqData> {
-        self.per_request.composite_req_data[idx]
-            .take()
-            .expect("composite_req_data slot is empty (taken)")
-    }
-
-    /// Put back a Composite request into an empty slot at `idx`.
-    #[inline]
-    pub(crate) fn put_back_composite_req_data(
-        &mut self,
-        idx: usize,
-        value: Box<CompositeAggReqData>,
-    ) {
-        debug_assert!(self.per_request.composite_req_data[idx].is_none());
-        self.per_request.composite_req_data[idx] = Some(value);
-    }
 }

 /// Each type of aggregation has its own request data struct. This struct holds
@@ -223,15 +107,14 @@ impl AggregationsSegmentCtx {
 /// for a node with [AggKind::Terms]).
 #[derive(Default)]
 pub struct PerRequestAggSegCtx {
-    // Box for cheap take/put - Only necessary for bucket aggs that have sub-aggregations
    /// TermsAggReqData contains the request data for a terms aggregation.
-    pub term_req_data: Vec<Option<Box<TermsAggReqData>>>,
+    pub term_req_data: Vec<TermsAggReqData>,
    /// HistogramAggReqData contains the request data for a histogram aggregation.
-    pub histogram_req_data: Vec<Option<Box<HistogramAggReqData>>>,
+    pub histogram_req_data: Vec<HistogramAggReqData>,
    /// RangeAggReqData contains the request data for a range aggregation.
-    pub range_req_data: Vec<Option<Box<RangeAggReqData>>>,
+    pub range_req_data: Vec<RangeAggReqData>,
    /// FilterAggReqData contains the request data for a filter aggregation.
-    pub filter_req_data: Vec<Option<Box<FilterAggReqData>>>,
+    pub filter_req_data: Vec<FilterAggReqData>,
    /// Shared by avg, min, max, sum, stats, extended_stats, count
    pub stats_metric_req_data: Vec<MetricAggReqData>,
    /// CardinalityAggReqData contains the request data for a cardinality aggregation.
@@ -241,7 +124,7 @@ pub struct PerRequestAggSegCtx {
    /// MissingTermAggReqData contains the request data for a missing term aggregation.
    pub missing_term_req_data: Vec<MissingTermAggReqData>,
    /// CompositeAggReqData contains the request data for a composite aggregation.
-    pub composite_req_data: Vec<Option<Box<CompositeAggReqData>>>,
+    pub composite_req_data: Vec<CompositeAggReqData>,

    /// Request tree used to build collectors.
    pub agg_tree: Vec<AggRefNode>,
@@ -252,22 +135,22 @@ impl PerRequestAggSegCtx {
    fn get_memory_consumption(&self) -> usize {
        self.term_req_data
            .iter()
-            .map(|b| b.as_ref().unwrap().get_memory_consumption())
+            .map(|t| t.get_memory_consumption())
            .sum::<usize>()
            + self
                .histogram_req_data
                .iter()
-                .map(|b| b.as_ref().unwrap().get_memory_consumption())
+                .map(|t| t.get_memory_consumption())
                .sum::<usize>()
            + self
                .range_req_data
                .iter()
-                .map(|b| b.as_ref().unwrap().get_memory_consumption())
+                .map(|t| t.get_memory_consumption())
                .sum::<usize>()
            + self
                .filter_req_data
                .iter()
-                .map(|b| b.as_ref().unwrap().get_memory_consumption())
+                .map(|t| t.get_memory_consumption())
                .sum::<usize>()
            + self
                .stats_metric_req_data
@@ -292,7 +175,7 @@ impl PerRequestAggSegCtx {
            + self
                .composite_req_data
                .iter()
-                .map(|b| b.as_ref().map(|d| d.get_memory_consumption()).unwrap_or(0))
+                .map(|t| t.get_memory_consumption())
                .sum::<usize>()
            + self.agg_tree.len() * std::mem::size_of::<AggRefNode>()
    }
@@ -301,40 +184,16 @@ impl PerRequestAggSegCtx {
        let idx = node.idx_in_req_data;
        let kind = node.kind;
        match kind {
-            AggKind::Terms => self.term_req_data[idx]
-                .as_deref()
-                .expect("term_req_data slot is empty (taken)")
-                .name
-                .as_str(),
+            AggKind::Terms => self.term_req_data[idx].name.as_str(),
            AggKind::Cardinality => &self.cardinality_req_data[idx].name,
            AggKind::StatsKind(_) => &self.stats_metric_req_data[idx].name,
            AggKind::TopHits => &self.top_hits_req_data[idx].name,
            AggKind::MissingTerm => &self.missing_term_req_data[idx].name,
-            AggKind::Histogram => self.histogram_req_data[idx]
-                .as_deref()
-                .expect("histogram_req_data slot is empty (taken)")
-                .name
-                .as_str(),
-            AggKind::DateHistogram => self.histogram_req_data[idx]
-                .as_deref()
-                .expect("histogram_req_data slot is empty (taken)")
-                .name
-                .as_str(),
-            AggKind::Range => self.range_req_data[idx]
-                .as_deref()
-                .expect("range_req_data slot is empty (taken)")
-                .name
-                .as_str(),
-            AggKind::Filter => self.filter_req_data[idx]
-                .as_deref()
-                .expect("filter_req_data slot is empty (taken)")
-                .name
-                .as_str(),
-            AggKind::Composite => self.composite_req_data[idx]
-                .as_deref()
-                .expect("composite_req_data slot is empty (taken)")
-                .name
-                .as_str(),
+            AggKind::Histogram => self.histogram_req_data[idx].name.as_str(),
+            AggKind::DateHistogram => self.histogram_req_data[idx].name.as_str(),
+            AggKind::Range => self.range_req_data[idx].name.as_str(),
+            AggKind::Filter => self.filter_req_data[idx].name.as_str(),
+            AggKind::Composite => self.composite_req_data[idx].name.as_str(),
        }
    }

@@ -412,7 +271,7 @@ pub(crate) fn build_segment_agg_collector(
            Ok(Box::new(TermMissingAgg::new(req, node)?))
        }
        AggKind::Cardinality => {
-            let req_data = &mut req.get_cardinality_req_data_mut(node.idx_in_req_data);
+            let req_data = req.get_cardinality_req_data(node.idx_in_req_data);
            // For str columns, choose the per-bucket entries representation
            // based on the segment's column.max_value():
            //   * small (< BITSET_MAX_TERM_ORD): `BitSet`, pre-allocated, no promotion machinery.
@@ -459,7 +318,7 @@ pub(crate) fn build_segment_agg_collector(
                    SegmentExtendedStatsCollector::from_req(req_data, sigma),
                )),
                StatsType::Percentiles => {
-                    let req_data = req.get_metric_req_data_mut(node.idx_in_req_data);
+                    let req_data = req.get_metric_req_data(node.idx_in_req_data);
                    Ok(Box::new(
                        SegmentPercentilesCollector::from_req_and_validate(
                            req_data.field_type,
@@ -799,23 +658,18 @@ fn build_nodes(
            let schema = reader.schema();
            let tokenizers = &data.context.tokenizers;
            let query = filter_req.parse_query(schema, tokenizers)?;
-            let evaluator = crate::aggregation::bucket::DocumentQueryEvaluator::new(
-                query,
-                schema.clone(),
-                reader,
-            )?;
-
-            // Pre-allocate buffer for batch filtering
-            let max_doc = reader.max_doc();
-            let buffer_capacity = crate::docset::COLLECT_BLOCK_BUFFER_LEN.min(max_doc as usize);
-            let matching_docs_buffer = Vec::with_capacity(buffer_capacity);
+            let evaluator =
+                std::rc::Rc::new(crate::aggregation::bucket::DocumentQueryEvaluator::new(
+                    query,
+                    schema.clone(),
+                    reader,
+                )?);

            let idx_in_req_data = data.push_filter_req_data(FilterAggReqData {
                name: agg_name.to_string(),
                req: filter_req.clone(),
                segment_reader: reader.clone(),
                evaluator,
-                matching_docs_buffer,
                is_top_level,
            });
            let children = build_children(&req.sub_aggregation, reader, segment_ordinal, data)?;
--- a/src/aggregation/agg_req.rs
+++ b/src/aggregation/agg_req.rs
@@ -115,71 +115,6 @@ pub fn get_fast_field_names(aggs: &Aggregations) -> HashSet<String> {
    fast_field_names
 }

-/// Validates that all fields referenced in the aggregation request exist in the schema
-/// and are configured as fast fields.
-///
-/// This is a convenience function for upfront validation before executing aggregations.
-/// Returns an error if any field doesn't exist or is not a fast field.
-///
-/// Validation is intentionally opt-in rather than baked into aggregation execution: the
-/// default lenient behavior (returning empty results for missing fields) supports
-/// schema evolution and federated queries where the same request runs against segments
-/// or indices with different schemas.
-///
-/// # Example
-/// ```
-/// use tantivy::aggregation::agg_req::{Aggregations, validate_aggregation_fields_exist};
-/// use tantivy::schema::{Schema, FAST};
-/// use tantivy::Index;
-///
-/// # fn main() -> tantivy::Result<()> {
-/// // Create a simple index
-/// let mut schema_builder = Schema::builder();
-/// schema_builder.add_f64_field("price", FAST);
-/// let schema = schema_builder.build();
-/// let index = Index::create_in_ram(schema);
-///
-/// // Parse aggregation request
-/// let agg_req: Aggregations = serde_json::from_str(r#"{
-///     "avg_price": { "avg": { "field": "price" } }
-/// }"#)?;
-///
-/// let reader = index.reader()?;
-/// let searcher = reader.searcher();
-///
-/// // Validate fields before executing
-/// for segment_reader in searcher.segment_readers() {
-///     validate_aggregation_fields_exist(&agg_req, segment_reader)?;
-/// }
-/// # Ok(())
-/// # }
-/// ```
-pub fn validate_aggregation_fields_exist(
-    aggs: &Aggregations,
-    reader: &crate::SegmentReader,
-) -> crate::Result<()> {
-    let field_names = get_fast_field_names(aggs);
-    let schema = reader.schema();
-
-    for field_name in field_names {
-        // Check if the field is either directly in the schema or could be part of a json field
-        // present in the schema, and verify it's a fast field.
-        if let Some((field, _path)) = schema.find_field(&field_name) {
-            let field_type = schema.get_field_entry(field).field_type();
-            if !field_type.is_fast() {
-                return Err(crate::TantivyError::SchemaError(format!(
-                    "Field '{}' is not a fast field. Aggregations require fast fields.",
-                    field_name
-                )));
-            }
-        } else {
-            return Err(crate::TantivyError::FieldNotFound(field_name));
-        }
-    }
-
-    Ok(())
-}
-
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 /// All aggregation types.
 pub enum AggregationVariants {
--- a/src/aggregation/agg_tests.rs
+++ b/src/aggregation/agg_tests.rs
@@ -1436,46 +1436,3 @@ fn test_aggregation_on_json_object_mixed_numerical_segments() {
        )
    );
 }
-
-#[test]
-fn test_aggregation_field_validation_helper() {
-    // Test the standalone validation helper function for field validation
-    let index = get_test_index_2_segments(false).unwrap();
-    let reader = index.reader().unwrap();
-    let searcher = reader.searcher();
-    let segment_reader = searcher.segment_reader(0);
-
-    // Test with invalid field
-    let agg_req: Aggregations = serde_json::from_str(
-        r#"{
-        "avg_test": {
-            "avg": { "field": "nonexistent_field" }
-        }
-    }"#,
-    )
-    .unwrap();
-
-    let result =
-        crate::aggregation::agg_req::validate_aggregation_fields_exist(&agg_req, segment_reader);
-    assert!(result.is_err());
-    match result {
-        Err(crate::TantivyError::FieldNotFound(field_name)) => {
-            assert_eq!(field_name, "nonexistent_field");
-        }
-        _ => panic!("Expected FieldNotFound error, got: {:?}", result),
-    }
-
-    // Test with valid field
-    let agg_req: Aggregations = serde_json::from_str(
-        r#"{
-        "avg_test": {
-            "avg": { "field": "score" }
-        }
-    }"#,
-    )
-    .unwrap();
-
-    let result =
-        crate::aggregation::agg_req::validate_aggregation_fields_exist(&agg_req, segment_reader);
-    assert!(result.is_ok());
-}
--- a/src/aggregation/bucket/composite/accessors.rs
+++ b/src/aggregation/bucket/composite/accessors.rs
@@ -16,6 +16,7 @@ use crate::{SegmentReader, TantivyError};

 /// Contains all information required by the SegmentCompositeCollector to perform the
 /// composite aggregation on a segment.
+#[derive(Debug, Clone)]
 pub struct CompositeAggReqData {
    /// The name of the aggregation.
    pub name: String,
@@ -34,6 +35,7 @@ impl CompositeAggReqData {
 }

 /// Accessors for a single column in a composite source.
+#[derive(Debug, Clone)]
 pub struct CompositeAccessor {
    /// The fast field column
    pub column: Column<u64>,
@@ -48,6 +50,7 @@ pub struct CompositeAccessor {
 }

 /// Accessors to all the columns that belong to the field of a composite source.
+#[derive(Debug, Clone)]
 pub struct CompositeSourceAccessors {
    /// The accessors for this source
    pub accessors: Vec<CompositeAccessor>,
@@ -358,7 +361,7 @@ impl PrecomputedDateInterval {
 ///
 /// Some column types (term, IP) might not have an exact representation of the
 /// specified after key
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum PrecomputedAfterKey {
    /// The after key could be exactly represented in the column space.
    Exact(u64),
--- a/src/aggregation/bucket/composite/collector.rs
+++ b/src/aggregation/bucket/composite/collector.rs
@@ -118,7 +118,7 @@ impl InternalValueRepr {
 pub struct SegmentCompositeCollector {
    /// One DynArrayHeapMap per parent bucket.
    parent_buckets: Vec<DynArrayHeapMap<InternalValueRepr, CompositeBucketCollector>>,
-    accessor_idx: usize,
+    req_data: CompositeAggReqData,
    sub_agg: Option<BufferedSubAggs<HighCardSubAggBuffer>>,
    bucket_id_provider: BucketIdProvider,
    /// Number of sources, needed when creating new DynArrayHeapMaps.
@@ -132,10 +132,7 @@ impl SegmentAggregationCollector for SegmentCompositeCollector {
        results: &mut IntermediateAggregationResults,
        parent_bucket_id: BucketId,
    ) -> crate::Result<()> {
-        let name = agg_data
-            .get_composite_req_data(self.accessor_idx)
-            .name
-            .clone();
+        let name = self.req_data.name.clone();

        let buckets = self.add_intermediate_bucket_result(agg_data, parent_bucket_id)?;
        results.push(
@@ -153,12 +150,11 @@ impl SegmentAggregationCollector for SegmentCompositeCollector {
        agg_data: &mut AggregationsSegmentCtx,
    ) -> crate::Result<()> {
        let mem_pre = self.get_memory_consumption(parent_bucket_id);
-        let composite_agg_data = agg_data.take_composite_req_data(self.accessor_idx);

        for doc in docs {
            let mut visitor = CompositeKeyVisitor {
                doc_id: *doc,
-                composite_agg_data: &composite_agg_data,
+                composite_agg_data: &self.req_data,
                buckets: &mut self.parent_buckets[parent_bucket_id as usize],
                sub_agg: &mut self.sub_agg,
                bucket_id_provider: &mut self.bucket_id_provider,
@@ -166,7 +162,6 @@ impl SegmentAggregationCollector for SegmentCompositeCollector {
            };
            visitor.visit(0, true)?;
        }
-        agg_data.put_back_composite_req_data(self.accessor_idx, composite_agg_data);

        if let Some(sub_agg) = &mut self.sub_agg {
            sub_agg.check_flush_local(agg_data)?;
@@ -221,7 +216,13 @@ impl SegmentCompositeCollector {
        req_data: &mut AggregationsSegmentCtx,
        node: &AggRefNode,
    ) -> crate::Result<Self> {
-        validate_req(req_data, node.idx_in_req_data)?;
+        let composite_req_data =
+            req_data.per_request.composite_req_data[node.idx_in_req_data].clone();
+        validate_req(&composite_req_data)?;
+        req_data
+            .context
+            .limits
+            .add_memory_consumed(composite_req_data.get_memory_consumption() as u64)?;

        let has_sub_aggregations = !node.children.is_empty();
        let sub_agg = if has_sub_aggregations {
@@ -231,12 +232,11 @@ impl SegmentCompositeCollector {
            None
        };

-        let composite_req_data = req_data.get_composite_req_data(node.idx_in_req_data);
        let num_sources = composite_req_data.req.sources.len();

        Ok(SegmentCompositeCollector {
            parent_buckets: vec![DynArrayHeapMap::try_new(num_sources)?],
-            accessor_idx: node.idx_in_req_data,
+            req_data: composite_req_data,
            sub_agg,
            bucket_id_provider: BucketIdProvider::default(),
            num_sources,
@@ -258,7 +258,7 @@ impl SegmentCompositeCollector {
        let mut dict: FxHashMap<Vec<CompositeIntermediateKey>, IntermediateCompositeBucketEntry> =
            Default::default();
        dict.reserve(heap_map.size());
-        let composite_data = agg_data.get_composite_req_data(self.accessor_idx);
+        let composite_data = &self.req_data;
        for (key_internal_repr, agg) in heap_map.into_iter() {
            let key = resolve_key(&key_internal_repr, composite_data)?;
            let mut sub_aggregation_res = IntermediateAggregationResults::default();
@@ -298,8 +298,7 @@ impl SegmentCompositeCollector {
    }
 }

-fn validate_req(req_data: &mut AggregationsSegmentCtx, accessor_idx: usize) -> crate::Result<()> {
-    let composite_data = req_data.get_composite_req_data(accessor_idx);
+fn validate_req(composite_data: &CompositeAggReqData) -> crate::Result<()> {
    let req = &composite_data.req;
    if req.sources.is_empty() {
        return Err(TantivyError::InvalidArgument(
--- a/src/aggregation/bucket/filter.rs
+++ b/src/aggregation/bucket/filter.rs
@@ -1,4 +1,5 @@
 use std::fmt::Debug;
+use std::rc::Rc;

 use common::BitSet;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
@@ -396,6 +397,7 @@ impl PartialEq for FilterAggregation {

 /// Request data for filter aggregation
 /// This struct holds the per-segment data needed to execute a filter aggregation
+#[derive(Clone)]
 pub struct FilterAggReqData {
    /// The name of the filter aggregation
    pub name: String,
@@ -403,22 +405,20 @@ pub struct FilterAggReqData {
    pub req: FilterAggregation,
    /// The segment reader
    pub segment_reader: SegmentReader,
-    /// Document evaluator for the filter query (precomputed BitSet)
-    /// This is built once when the request data is created
-    pub evaluator: DocumentQueryEvaluator,
-    /// Reusable buffer for matching documents to minimize allocations during collection
-    pub matching_docs_buffer: Vec<DocId>,
+    /// Document evaluator for the filter query (precomputed BitSet).
+    /// Wrapped in `Rc` so cloning the request data does not duplicate the (potentially large)
+    /// underlying BitSet.
+    pub evaluator: Rc<DocumentQueryEvaluator>,
    /// True if this filter aggregation is at the top level of the aggregation tree (not nested).
    pub is_top_level: bool,
 }

 impl FilterAggReqData {
    pub(crate) fn get_memory_consumption(&self) -> usize {
-        // Estimate: name + segment reader reference + bitset + buffer capacity
+        // Estimate: name + segment reader reference + bitset
        self.name.len()
        + std::mem::size_of::<SegmentReader>()
        + self.evaluator.bitset.len() / 8 // BitSet memory (bits to bytes)
-        + self.matching_docs_buffer.capacity() * std::mem::size_of::<DocId>()
        + std::mem::size_of::<bool>()
    }
 }
@@ -509,8 +509,10 @@ pub struct SegmentFilterCollector<B: SubAggBuffer> {
    /// Sub-aggregation collectors
    sub_aggregations: Option<BufferedSubAggs<B>>,
    bucket_id_provider: BucketIdProvider,
-    /// Accessor index for this filter aggregation (to access FilterAggReqData)
-    accessor_idx: usize,
+    /// Per-segment filter request data, owned by this collector.
+    req_data: FilterAggReqData,
+    /// Reusable buffer for matching documents to minimize allocations during collection.
+    matching_docs_buffer: Vec<DocId>,
 }

 impl<B: SubAggBuffer> SegmentFilterCollector<B> {
@@ -518,6 +520,7 @@ impl<B: SubAggBuffer> SegmentFilterCollector<B> {
    pub(crate) fn from_req_and_validate(
        req: &mut AggregationsSegmentCtx,
        node: &AggRefNode,
+        req_data: FilterAggReqData,
    ) -> crate::Result<Self> {
        // Build sub-aggregation collectors if any
        let sub_agg_collector = if !node.children.is_empty() {
@@ -527,11 +530,15 @@ impl<B: SubAggBuffer> SegmentFilterCollector<B> {
        };
        let sub_agg_collector = sub_agg_collector.map(BufferedSubAggs::new);

+        let max_doc = req_data.segment_reader.max_doc();
+        let buffer_capacity = crate::docset::COLLECT_BLOCK_BUFFER_LEN.min(max_doc as usize);
+
        Ok(SegmentFilterCollector {
            parent_buckets: Vec::new(),
            sub_aggregations: sub_agg_collector,
-            accessor_idx: node.idx_in_req_data,
+            req_data,
            bucket_id_provider: BucketIdProvider::default(),
+            matching_docs_buffer: Vec::with_capacity(buffer_capacity),
        })
    }
 }
@@ -540,18 +547,23 @@ pub(crate) fn build_segment_filter_collector(
    req: &mut AggregationsSegmentCtx,
    node: &AggRefNode,
 ) -> crate::Result<Box<dyn SegmentAggregationCollector>> {
-    let is_top_level = req.per_request.filter_req_data[node.idx_in_req_data]
-        .as_ref()
-        .expect("filter_req_data slot is empty")
-        .is_top_level;
+    let req_data = req.per_request.filter_req_data[node.idx_in_req_data].clone();
+    req.context
+        .limits
+        .add_memory_consumed(req_data.get_memory_consumption() as u64)?;
+    let is_top_level = req_data.is_top_level;

    if is_top_level {
        Ok(Box::new(
-            SegmentFilterCollector::<LowCardSubAggBuffer>::from_req_and_validate(req, node)?,
+            SegmentFilterCollector::<LowCardSubAggBuffer>::from_req_and_validate(
+                req, node, req_data,
+            )?,
        ))
    } else {
        Ok(Box::new(
-            SegmentFilterCollector::<HighCardSubAggBuffer>::from_req_and_validate(req, node)?,
+            SegmentFilterCollector::<HighCardSubAggBuffer>::from_req_and_validate(
+                req, node, req_data,
+            )?,
        ))
    }
 }
@@ -561,7 +573,7 @@ impl<B: SubAggBuffer> Debug for SegmentFilterCollector<B> {
        f.debug_struct("SegmentFilterCollector")
            .field("buckets", &self.parent_buckets)
            .field("has_sub_aggs", &self.sub_aggregations.is_some())
-            .field("accessor_idx", &self.accessor_idx)
+            .field("name", &self.req_data.name)
            .finish()
    }
 }
@@ -598,11 +610,7 @@ impl<B: SubAggBuffer> SegmentAggregationCollector for SegmentFilterCollector<B>
        };

        // Get the name of this filter aggregation
-        let name = agg_data.per_request.filter_req_data[self.accessor_idx]
-            .as_ref()
-            .expect("filter_req_data slot is empty")
-            .name
-            .clone();
+        let name = self.req_data.name.clone();

        results.push(
            name,
@@ -623,27 +631,24 @@ impl<B: SubAggBuffer> SegmentAggregationCollector for SegmentFilterCollector<B>
        }

        let mut bucket = self.parent_buckets[parent_bucket_id as usize];
-        // Take the request data to avoid borrow checker issues with sub-aggregations
-        let mut req = agg_data.take_filter_req_data(self.accessor_idx);

        // Use batch filtering with O(1) BitSet lookups
-        req.matching_docs_buffer.clear();
-        req.evaluator
-            .filter_batch(docs, &mut req.matching_docs_buffer);
+        self.matching_docs_buffer.clear();
+        self.req_data
+            .evaluator
+            .filter_batch(docs, &mut self.matching_docs_buffer);

-        bucket.doc_count += req.matching_docs_buffer.len() as u64;
+        bucket.doc_count += self.matching_docs_buffer.len() as u64;

        // Batch process sub-aggregations if we have matches
-        if !req.matching_docs_buffer.is_empty() {
+        if !self.matching_docs_buffer.is_empty() {
            if let Some(sub_aggs) = &mut self.sub_aggregations {
-                for &doc_id in &req.matching_docs_buffer {
+                for &doc_id in &self.matching_docs_buffer {
                    sub_aggs.push(bucket.bucket_id, doc_id);
                }
            }
        }

-        // Put the request data back
-        agg_data.put_back_filter_req_data(self.accessor_idx, req);
        if let Some(sub_aggs) = &mut self.sub_aggregations {
            sub_aggs.check_flush_local(agg_data)?;
        }
--- a/src/aggregation/bucket/histogram/histogram.rs
+++ b/src/aggregation/bucket/histogram/histogram.rs
@@ -21,6 +21,7 @@ use crate::TantivyError;

 /// Contains all information required by the SegmentHistogramCollector to perform the
 /// histogram or date_histogram aggregation on a segment.
+#[derive(Debug, Clone)]
 pub struct HistogramAggReqData {
    /// The column accessor to access the fast field values.
    pub accessor: Column<u64>,
@@ -297,7 +298,7 @@ pub struct SegmentHistogramCollector {
    /// One Histogram bucket per parent bucket id.
    parent_buckets: Vec<HistogramBuckets>,
    sub_agg: Option<HighCardBufferedSubAggs>,
-    accessor_idx: usize,
+    req_data: HistogramAggReqData,
    bucket_id_provider: BucketIdProvider,
 }

@@ -308,10 +309,7 @@ impl SegmentAggregationCollector for SegmentHistogramCollector {
        results: &mut IntermediateAggregationResults,
        parent_bucket_id: BucketId,
    ) -> crate::Result<()> {
-        let name = agg_data
-            .get_histogram_req_data(self.accessor_idx)
-            .name
-            .clone();
+        let name = self.req_data.name.clone();
        // TODO: avoid prepare_max_bucket here and handle empty buckets.
        self.prepare_max_bucket(parent_bucket_id, agg_data)?;
        let histogram = std::mem::take(&mut self.parent_buckets[parent_bucket_id as usize]);
@@ -328,10 +326,10 @@ impl SegmentAggregationCollector for SegmentHistogramCollector {
        docs: &[crate::DocId],
        agg_data: &mut AggregationsSegmentCtx,
    ) -> crate::Result<()> {
-        let req = agg_data.take_histogram_req_data(self.accessor_idx);
        let mem_pre = self.get_memory_consumption(parent_bucket_id);
        let buckets = &mut self.parent_buckets[parent_bucket_id as usize].buckets;

+        let req = &self.req_data;
        let bounds = req.bounds;
        let interval = req.req.interval;
        let offset = req.offset;
@@ -361,7 +359,6 @@ impl SegmentAggregationCollector for SegmentHistogramCollector {
                }
            }
        }
-        agg_data.put_back_histogram_req_data(self.accessor_idx, req);

        let mem_delta = self.get_memory_consumption(parent_bucket_id) - mem_pre;
        if mem_delta > 0 {
@@ -427,10 +424,7 @@ impl SegmentHistogramCollector {
        }
        buckets.sort_unstable_by(|b1, b2| b1.key.total_cmp(&b2.key));

-        let is_date_agg = agg_data
-            .get_histogram_req_data(self.accessor_idx)
-            .field_type
-            == ColumnType::DateTime;
+        let is_date_agg = self.req_data.field_type == ColumnType::DateTime;
        Ok(IntermediateBucketResult::Histogram {
            buckets,
            is_date_agg,
@@ -446,7 +440,7 @@ impl SegmentHistogramCollector {
        } else {
            None
        };
-        let req_data = agg_data.get_histogram_req_data_mut(node.idx_in_req_data);
+        let mut req_data = agg_data.per_request.histogram_req_data[node.idx_in_req_data].clone();
        req_data.req.validate()?;
        if req_data.field_type == ColumnType::DateTime && !req_data.is_date_histogram {
            req_data.req.normalize_date_time();
@@ -456,12 +450,16 @@ impl SegmentHistogramCollector {
            max: f64::MAX,
        });
        req_data.offset = req_data.req.offset.unwrap_or(0.0);
+        agg_data
+            .context
+            .limits
+            .add_memory_consumed(req_data.get_memory_consumption() as u64)?;
        let sub_agg = sub_agg.map(BufferedSubAggs::new);

        Ok(Self {
            parent_buckets: Default::default(),
            sub_agg,
-            accessor_idx: node.idx_in_req_data,
+            req_data,
            bucket_id_provider: BucketIdProvider::default(),
        })
    }
--- a/src/aggregation/bucket/range.rs
+++ b/src/aggregation/bucket/range.rs
@@ -23,6 +23,7 @@ use crate::TantivyError;

 /// Contains all information required by the SegmentRangeCollector to perform the
 /// range aggregation on a segment.
+#[derive(Debug, Clone)]
 pub struct RangeAggReqData {
    /// The column accessor to access the fast field values.
    pub accessor: Column<u64>,
@@ -161,7 +162,7 @@ pub struct SegmentRangeCollector<B: SubAggBuffer> {
    /// One for each ParentBucketId
    parent_buckets: Vec<Vec<SegmentRangeAndBucketEntry>>,
    column_type: ColumnType,
-    pub(crate) accessor_idx: usize,
+    pub(crate) req_data: RangeAggReqData,
    sub_agg: Option<BufferedSubAggs<B>>,
    /// Here things get a bit weird. We need to assign unique bucket ids across all
    /// parent buckets. So we keep track of the next available bucket id here.
@@ -184,7 +185,7 @@ impl<B: SubAggBuffer> Debug for SegmentRangeCollector<B> {
        f.debug_struct("SegmentRangeCollector")
            .field("parent_buckets_len", &self.parent_buckets.len())
            .field("column_type", &self.column_type)
-            .field("accessor_idx", &self.accessor_idx)
+            .field("name", &self.req_data.name)
            .field("has_sub_agg", &self.sub_agg.is_some())
            .finish()
    }
@@ -239,10 +240,7 @@ impl<B: SubAggBuffer> SegmentAggregationCollector for SegmentRangeCollector<B> {
    ) -> crate::Result<()> {
        self.prepare_max_bucket(parent_bucket_id, agg_data)?;
        let field_type = self.column_type;
-        let name = agg_data
-            .get_range_req_data(self.accessor_idx)
-            .name
-            .to_string();
+        let name = self.req_data.name.to_string();

        let buckets = std::mem::take(&mut self.parent_buckets[parent_bucket_id as usize]);

@@ -281,17 +279,15 @@ impl<B: SubAggBuffer> SegmentAggregationCollector for SegmentRangeCollector<B> {
        docs: &[crate::DocId],
        agg_data: &mut AggregationsSegmentCtx,
    ) -> crate::Result<()> {
-        let req = agg_data.take_range_req_data(self.accessor_idx);
-
        agg_data
            .column_block_accessor
-            .fetch_block(docs, &req.accessor);
+            .fetch_block(docs, &self.req_data.accessor);

        let buckets = &mut self.parent_buckets[parent_bucket_id as usize];

        for (doc, val) in agg_data
            .column_block_accessor
-            .iter_docid_vals(docs, &req.accessor)
+            .iter_docid_vals(docs, &self.req_data.accessor)
        {
            let bucket_pos = get_bucket_pos(val, buckets);
            let bucket = &mut buckets[bucket_pos];
@@ -301,7 +297,6 @@ impl<B: SubAggBuffer> SegmentAggregationCollector for SegmentRangeCollector<B> {
            }
        }

-        agg_data.put_back_range_req_data(self.accessor_idx, req);
        if let Some(sub_agg) = self.sub_agg.as_mut() {
            sub_agg.check_flush_local(agg_data)?;
        }
@@ -319,10 +314,10 @@ impl<B: SubAggBuffer> SegmentAggregationCollector for SegmentRangeCollector<B> {
    fn prepare_max_bucket(
        &mut self,
        max_bucket: BucketId,
-        agg_data: &AggregationsSegmentCtx,
+        _agg_data: &AggregationsSegmentCtx,
    ) -> crate::Result<()> {
        while self.parent_buckets.len() <= max_bucket as usize {
-            let new_buckets = self.create_new_buckets(agg_data)?;
+            let new_buckets = self.create_new_buckets()?;
            self.parent_buckets.push(new_buckets);
        }

@@ -346,8 +341,11 @@ pub(crate) fn build_segment_range_collector(
    agg_data: &mut AggregationsSegmentCtx,
    node: &AggRefNode,
 ) -> crate::Result<Box<dyn SegmentAggregationCollector>> {
-    let accessor_idx = node.idx_in_req_data;
-    let req_data = agg_data.get_range_req_data(node.idx_in_req_data);
+    let req_data = agg_data.per_request.range_req_data[node.idx_in_req_data].clone();
+    agg_data
+        .context
+        .limits
+        .add_memory_consumed(req_data.get_memory_consumption() as u64)?;
    let field_type = req_data.field_type;

    // TODO: A better metric instead of is_top_level would be the number of buckets expected.
@@ -365,7 +363,7 @@ pub(crate) fn build_segment_range_collector(
        Ok(Box::new(SegmentRangeCollector::<LowCardSubAggBuffer> {
            sub_agg: sub_agg.map(LowCardBufferedSubAggs::new),
            column_type: field_type,
-            accessor_idx,
+            req_data,
            parent_buckets: Vec::new(),
            bucket_id_provider: BucketIdProvider::default(),
            limits: agg_data.context.limits.clone(),
@@ -374,7 +372,7 @@ pub(crate) fn build_segment_range_collector(
        Ok(Box::new(SegmentRangeCollector::<HighCardSubAggBuffer> {
            sub_agg: sub_agg.map(BufferedSubAggs::new),
            column_type: field_type,
-            accessor_idx,
+            req_data,
            parent_buckets: Vec::new(),
            bucket_id_provider: BucketIdProvider::default(),
            limits: agg_data.context.limits.clone(),
@@ -383,12 +381,9 @@ pub(crate) fn build_segment_range_collector(
 }

 impl<B: SubAggBuffer> SegmentRangeCollector<B> {
-    pub(crate) fn create_new_buckets(
-        &mut self,
-        agg_data: &AggregationsSegmentCtx,
-    ) -> crate::Result<Vec<SegmentRangeAndBucketEntry>> {
+    pub(crate) fn create_new_buckets(&mut self) -> crate::Result<Vec<SegmentRangeAndBucketEntry>> {
        let field_type = self.column_type;
-        let req_data = agg_data.get_range_req_data(self.accessor_idx);
+        let req_data = &self.req_data;
        // The range input on the request is f64.
        // We need to convert to u64 ranges, because we read the values as u64.
        // The mapping from the conversion is monotonic so ordering is preserved.
@@ -563,17 +558,16 @@ mod tests {
        get_test_index_with_num_docs,
    };

-    pub fn get_collector_from_ranges(
+    pub fn build_test_buckets(
        ranges: Vec<RangeAggregationRange>,
        field_type: ColumnType,
-    ) -> SegmentRangeCollector<HighCardSubAggBuffer> {
+    ) -> Vec<SegmentRangeAndBucketEntry> {
        let req = RangeAggregation {
            field: "dummy".to_string(),
            ranges,
            ..Default::default()
        };
-        // Build buckets directly as in from_req_and_validate without AggregationsData
-        let buckets: Vec<_> = extend_validate_ranges(&req.ranges, &field_type)
+        extend_validate_ranges(&req.ranges, &field_type)
            .expect("unexpected error in extend_validate_ranges")
            .iter()
            .map(|range| {
@@ -604,16 +598,7 @@ mod tests {
                    },
                }
            })
-            .collect();
-
-        SegmentRangeCollector {
-            parent_buckets: vec![buckets],
-            column_type: field_type,
-            accessor_idx: 0,
-            sub_agg: None,
-            bucket_id_provider: Default::default(),
-            limits: AggregationLimitsGuard::default(),
-        }
+            .collect()
    }

    #[test]
@@ -857,9 +842,9 @@ mod tests {
    #[test]
    fn bucket_test_extend_range_hole() {
        let buckets = vec![(10f64..20f64).into(), (30f64..40f64).into()];
-        let collector = get_collector_from_ranges(buckets, ColumnType::F64);
+        let parent_buckets = vec![build_test_buckets(buckets, ColumnType::F64)];

-        let buckets = collector.parent_buckets[0].clone();
+        let buckets = parent_buckets[0].clone();
        assert_eq!(buckets[0].range.start, u64::MIN);
        assert_eq!(buckets[0].range.end, 10f64.to_u64());
        assert_eq!(buckets[1].range.start, 10f64.to_u64());
@@ -880,9 +865,9 @@ mod tests {
            (10f64..20f64).into(),
            (20f64..f64::MAX).into(),
        ];
-        let collector = get_collector_from_ranges(buckets, ColumnType::F64);
+        let parent_buckets = vec![build_test_buckets(buckets, ColumnType::F64)];

-        let buckets = collector.parent_buckets[0].clone();
+        let buckets = parent_buckets[0].clone();
        assert_eq!(buckets[0].range.start, u64::MIN);
        assert_eq!(buckets[0].range.end, 10f64.to_u64());
        assert_eq!(buckets[1].range.start, 10f64.to_u64());
@@ -895,18 +880,18 @@ mod tests {
    #[test]
    fn bucket_range_test_negative_vals() {
        let buckets = vec![(-10f64..-1f64).into()];
-        let collector = get_collector_from_ranges(buckets, ColumnType::F64);
+        let parent_buckets = vec![build_test_buckets(buckets, ColumnType::F64)];

-        let buckets = collector.parent_buckets[0].clone();
+        let buckets = parent_buckets[0].clone();
        assert_eq!(&buckets[0].bucket.key.to_string(), "*--10");
        assert_eq!(&buckets[buckets.len() - 1].bucket.key.to_string(), "-1-*");
    }
    #[test]
    fn bucket_range_test_positive_vals() {
        let buckets = vec![(0f64..10f64).into()];
-        let collector = get_collector_from_ranges(buckets, ColumnType::F64);
+        let parent_buckets = vec![build_test_buckets(buckets, ColumnType::F64)];

-        let buckets = collector.parent_buckets[0].clone();
+        let buckets = parent_buckets[0].clone();
        assert_eq!(&buckets[0].bucket.key.to_string(), "*-0");
        assert_eq!(&buckets[buckets.len() - 1].bucket.key.to_string(), "10-*");
    }
@@ -914,8 +899,8 @@ mod tests {
    #[test]
    fn range_binary_search_test_u64() {
        let check_ranges = |ranges: Vec<RangeAggregationRange>| {
-            let collector = get_collector_from_ranges(ranges, ColumnType::U64);
-            let search = |val: u64| get_bucket_pos(val, &collector.parent_buckets[0]);
+            let parent_buckets = vec![build_test_buckets(ranges, ColumnType::U64)];
+            let search = |val: u64| get_bucket_pos(val, &parent_buckets[0]);

            assert_eq!(search(u64::MIN), 0);
            assert_eq!(search(9), 0);
@@ -960,8 +945,8 @@ mod tests {
    fn range_binary_search_test_f64() {
        let ranges = vec![(10.0..100.0).into()];

-        let collector = get_collector_from_ranges(ranges, ColumnType::F64);
-        let search = |val: u64| get_bucket_pos(val, &collector.parent_buckets[0]);
+        let parent_buckets = vec![build_test_buckets(ranges, ColumnType::F64)];
+        let search = |val: u64| get_bucket_pos(val, &parent_buckets[0]);

        assert_eq!(search(u64::MIN), 0);
        assert_eq!(search(9f64.to_u64()), 0);
--- a/src/aggregation/metric/cardinality.rs
+++ b/src/aggregation/metric/cardinality.rs
@@ -166,12 +166,8 @@ impl CouponCache {
        let should_use_dense =
            highest_term_ord < 1_000_000u64 || highest_term_ord < num_terms as u64 * 3u64;
        if should_use_dense {
-            // We don't really care about the value here. We will populate all the values we will
-            // read anyway.
-            let uninitialized_coupon = Coupon::from_hash(0);
-            let mut coupon_map: Vec<Coupon> =
-                vec![uninitialized_coupon; highest_term_ord as usize + 1];
-            for (term_ord, coupon) in term_ords.into_iter().zip(coupons) {
+            let mut coupon_map: Vec<Coupon> = vec![Coupon::EMPTY; highest_term_ord as usize + 1];
+            for (term_ord, coupon) in term_ords.into_iter().zip(coupons.into_iter()) {
                coupon_map[term_ord as usize] = coupon;
            }
            CouponCache::Dense {
@@ -825,7 +821,7 @@ impl<'de> Deserialize<'de> for CardinalityCollector {
 impl CardinalityCollector {
    fn new(salt: u8) -> Self {
        Self {
-            sketch: HllSketch::new(LG_K, HllType::Hll8),
+            sketch: HllSketch::new(LG_K, HllType::Hll4),
            salt,
        }
    }
@@ -856,7 +852,7 @@ impl CardinalityCollector {
        let mut union = HllUnion::new(LG_K);
        union.update(&self.sketch);
        union.update(&right.sketch);
-        self.sketch = union.to_sketch(HllType::Hll8);
+        self.sketch = union.to_sketch(HllType::Hll4);
        Ok(())
    }
 }
--- a/src/collector/sort_key/sort_by_static_fast_value.rs
+++ b/src/collector/sort_key/sort_by_static_fast_value.rs
@@ -52,7 +52,7 @@ impl<T: FastValue> SortKeyComputer for SortByStaticFastValue<T> {
        if schema_type != T::to_type() {
            return Err(crate::TantivyError::SchemaError(format!(
                "Field `{}` is of type {schema_type:?}, not of the type {:?}.",
-                self.field,
+                &self.field,
                T::to_type()
            )));
        }
--- a/src/index/segment_reader.rs
+++ b/src/index/segment_reader.rs
@@ -6,7 +6,6 @@ use common::{ByteCount, HasLen};
 use fnv::FnvHashMap;
 use itertools::Itertools;

-use crate::directory::error::OpenReadError;
 use crate::directory::{CompositeFile, FileSlice};
 use crate::error::DataCorruption;
 use crate::fastfield::{intersect_alive_bitsets, AliveBitSet, FacetReader, FastFieldReaders};
@@ -160,10 +159,12 @@ impl SegmentReader {
        let postings_file = segment.open_read(SegmentComponent::Postings)?;
        let postings_composite = CompositeFile::open(&postings_file)?;

-        let positions_composite = match segment.open_read(SegmentComponent::Positions) {
-            Ok(positions_file) => CompositeFile::open(&positions_file)?,
-            Err(OpenReadError::FileDoesNotExist(_)) => CompositeFile::empty(),
-            Err(open_read_error) => return Err(open_read_error.into()),
+        let positions_composite = {
+            if let Ok(positions_file) = segment.open_read(SegmentComponent::Positions) {
+                CompositeFile::open(&positions_file)?
+            } else {
+                CompositeFile::empty()
+            }
        };

        let schema = segment.schema();
@@ -322,7 +323,7 @@ impl SegmentReader {
                            // Without expand dots enabled dots need to be escaped.
                            let escaped_json_path = json_path.replace('.', "\\.");
                            let full_path = format!("{field_name}.{escaped_json_path}");
-                            let full_path_unescaped = format!("{}.{}", field_name, json_path);
+                            let full_path_unescaped = format!("{}.{}", field_name, &json_path);
                            map_to_canonical.insert(full_path_unescaped, full_path.to_string());
                            full_path
                        } else {
--- a/src/postings/recorder.rs
+++ b/src/postings/recorder.rs
@@ -275,9 +275,8 @@ impl Recorder for TfAndPositionRecorder {
 mod tests {

    use common::write_u32_vint;
-    use stacker::MemoryArena;

-    use super::{BufferLender, Recorder, TermFrequencyRecorder, VInt32Reader};
+    use super::{BufferLender, VInt32Reader};

    #[test]
    fn test_buffer_lender() {
@@ -315,98 +314,4 @@ mod tests {
        let res: Vec<u32> = VInt32Reader::new(&buffer[..]).collect();
        assert_eq!(&res[..], &vals[..]);
    }
-
-    // ── TermFrequencyRecorder ─────────────────────────────────────────────────
-
-    #[test]
-    fn term_frequency_recorder_has_term_freq() {
-        let rec = TermFrequencyRecorder::default();
-        assert!(
-            rec.has_term_freq(),
-            "TermFrequencyRecorder must advertise term-frequency support"
-        );
-    }
-
-    #[test]
-    fn term_frequency_recorder_term_doc_freq_single_doc() {
-        let mut arena = MemoryArena::default();
-        let mut rec = TermFrequencyRecorder::default();
-
-        // Record one document with two term occurrences.
-        rec.new_doc(0, &mut arena);
-        rec.record_position(0, &mut arena);
-        rec.record_position(1, &mut arena);
-        rec.close_doc(&mut arena);
-
-        assert_eq!(
-            rec.term_doc_freq(),
-            Some(1),
-            "term_doc_freq should be 1 after recording one document"
-        );
-    }
-
-    #[test]
-    fn term_frequency_recorder_term_doc_freq_multiple_docs() {
-        let mut arena = MemoryArena::default();
-        let mut rec = TermFrequencyRecorder::default();
-
-        // Three documents with 1, 3, and 2 occurrences respectively.
-        for (doc, tf) in [(0u32, 1u32), (5, 3), (10, 2)] {
-            rec.new_doc(doc, &mut arena);
-            for pos in 0..tf {
-                rec.record_position(pos, &mut arena);
-            }
-            rec.close_doc(&mut arena);
-        }
-
-        assert_eq!(
-            rec.term_doc_freq(),
-            Some(3),
-            "term_doc_freq should equal the number of documents recorded"
-        );
-    }
-
-    #[test]
-    fn term_frequency_recorder_zero_docs() {
-        let rec = TermFrequencyRecorder::default();
-        assert_eq!(
-            rec.term_doc_freq(),
-            Some(0),
-            "term_doc_freq should be 0 before any document is recorded"
-        );
-    }
-
-    #[test]
-    fn term_frequency_recorder_single_occurrence_per_doc() {
-        let mut arena = MemoryArena::default();
-        let mut rec = TermFrequencyRecorder::default();
-
-        // Each document has exactly one occurrence — the minimum non-trivial case.
-        for doc in [1u32, 2, 100] {
-            rec.new_doc(doc, &mut arena);
-            rec.record_position(0, &mut arena);
-            rec.close_doc(&mut arena);
-        }
-
-        assert_eq!(rec.term_doc_freq(), Some(3));
-    }
-
-    #[test]
-    fn term_frequency_recorder_high_frequency_doc() {
-        let mut arena = MemoryArena::default();
-        let mut rec = TermFrequencyRecorder::default();
-
-        // A document where the term appears many times.
-        rec.new_doc(42, &mut arena);
-        for pos in 0..1000 {
-            rec.record_position(pos, &mut arena);
-        }
-        rec.close_doc(&mut arena);
-
-        assert_eq!(
-            rec.term_doc_freq(),
-            Some(1),
-            "term_doc_freq counts documents, not occurrences"
-        );
-    }
 }
--- a/src/store/index/skip_index.rs
+++ b/src/store/index/skip_index.rs
@@ -94,7 +94,13 @@ impl SkipIndex {
            byte_range: 0..first_layer_len,
        };
        for layer in &self.layers {
-            cur_checkpoint = layer.seek_start_at_offset(target, cur_checkpoint.byte_range.start)?;
+            if let Some(checkpoint) =
+                layer.seek_start_at_offset(target, cur_checkpoint.byte_range.start)
+            {
+                cur_checkpoint = checkpoint;
+            } else {
+                return None;
+            }
        }
        Some(cur_checkpoint)
    }
--- a/sstable/src/dictionary.rs
+++ b/sstable/src/dictionary.rs
@@ -14,8 +14,11 @@ use itertools::Itertools;
 use tantivy_fst::Automaton;
 use tantivy_fst::automaton::AlwaysMatch;

+use crate::sstable_index_v3::SSTableIndexV3Empty;
 use crate::streamer::{Streamer, StreamerBuilder};
-use crate::{BlockAddr, DeltaReader, Reader, SSTable, SSTableIndex, TermOrdinal, VoidSSTable};
+use crate::{
+    BlockAddr, DeltaReader, Reader, SSTable, SSTableIndex, SSTableIndexV3, TermOrdinal, VoidSSTable,
+};

 /// An SSTable is a sorted map that associates sorted `&[u8]` keys
 /// to any kind of typed values.
@@ -285,7 +288,33 @@ impl<TSSTable: SSTable> Dictionary<TSSTable> {
        let (sstable_slice, index_slice) = main_slice.split(index_offset as usize);
        let sstable_index_bytes = index_slice.read_bytes()?;

-        let sstable_index = SSTableIndex::open(version, index_offset, sstable_index_bytes)?;
+        let sstable_index = match version {
+            2 => SSTableIndex::V2(
+                crate::sstable_index_v2::SSTableIndex::load(sstable_index_bytes).map_err(|_| {
+                    io::Error::new(io::ErrorKind::InvalidData, "SSTable corruption")
+                })?,
+            ),
+            3 => {
+                let (sstable_index_bytes, mut footerv3_len_bytes) = sstable_index_bytes.rsplit(8);
+                let store_offset = u64::deserialize(&mut footerv3_len_bytes)?;
+                if store_offset != 0 {
+                    SSTableIndex::V3(
+                        SSTableIndexV3::load(sstable_index_bytes, store_offset).map_err(|_| {
+                            io::Error::new(io::ErrorKind::InvalidData, "SSTable corruption")
+                        })?,
+                    )
+                } else {
+                    // if store_offset is zero, there is no index, so we build a pseudo-index
+                    // assuming a single block of sstable covering everything.
+                    SSTableIndex::V3Empty(SSTableIndexV3Empty::load(index_offset as usize))
+                }
+            }
+            _ => {
+                return Err(io::Error::other(format!(
+                    "Unsupported sstable version, expected one of [2, 3], found {version}"
+                )));
+            }
+        };

        Ok(Dictionary {
            sstable_slice,
@@ -496,15 +525,10 @@ impl<TSSTable: SSTable> Dictionary<TSSTable> {

        // Open the block for the first ordinal.
        let mut bytes = Vec::new();
-        let (mut current_block_addr, block_id) = self.sstable_index.get_and_locate_with_ord(ord);
+        let mut current_block_addr = self.sstable_index.get_block_with_ord(ord);
        let mut current_sstable_delta_reader =
            self.sstable_delta_reader_block(current_block_addr.clone())?;
        let mut current_block_ordinal = current_block_addr.first_ordinal;
-        let mut current_block_end_bound = self
-            .sstable_index
-            .get_block(block_id + 1)
-            .map(|block_addr| block_addr.first_ordinal)
-            .unwrap_or(u64::MAX);

        loop {
            // move to the ord inside the current block
@@ -533,19 +557,17 @@ impl<TSSTable: SSTable> Dictionary<TSSTable> {
                }
            };

-            if next_ord >= current_block_end_bound {
-                let (new_block_addr, block_id) =
-                    self.sstable_index.get_and_locate_with_ord(next_ord);
+            // TODO optimization: it is silly to do a binary search to get the block every single
+            // time.
+            //
+            // Check if block changed for new term_ord
+            let new_block_addr = self.sstable_index.get_block_with_ord(next_ord);
+            if new_block_addr != current_block_addr {
                current_block_addr = new_block_addr;
                current_block_ordinal = current_block_addr.first_ordinal;
                current_sstable_delta_reader =
                    self.sstable_delta_reader_block(current_block_addr.clone())?;
                bytes.clear();
-                current_block_end_bound = self
-                    .sstable_index
-                    .get_block(block_id + 1)
-                    .map(|block_addr| block_addr.first_ordinal)
-                    .unwrap_or(u64::MAX)
            }
            ord = next_ord;
        }
--- a/sstable/src/index/mod.rs
+++ b/sstable/src/index/mod.rs
@@ -1,319 +0,0 @@
-pub(crate) mod v2;
-pub(crate) mod v3;
-
-use std::io::{self, Read, Write};
-use std::ops::Range;
-
-use common::{BinarySerializable, FixedSize, OwnedBytes};
-use tantivy_fst::{Automaton, MapBuilder};
-
-use crate::{TermOrdinal, common_prefix_len};
-
-#[derive(Debug, Clone)]
-pub enum SSTableIndex {
-    V2(v2::SSTableIndex),
-    V3(v3::SSTableIndexV3),
-    V3Empty(v3::SSTableIndexV3Empty),
-}
-
-impl SSTableIndex {
-    pub(crate) fn open(
-        version: u32,
-        index_offset: u64,
-        index_bytes: OwnedBytes,
-    ) -> io::Result<Self> {
-        let index = match version {
-            2 => {
-                SSTableIndex::V2(v2::SSTableIndex::load(index_bytes).map_err(|_| {
-                    io::Error::new(io::ErrorKind::InvalidData, "SSTable corruption")
-                })?)
-            }
-            3 => {
-                let (index_bytes, mut footerv3_len_bytes) = index_bytes.rsplit(8);
-                let store_offset = u64::deserialize(&mut footerv3_len_bytes)?;
-                if store_offset != 0 {
-                    SSTableIndex::V3(v3::SSTableIndexV3::load(index_bytes, store_offset).map_err(
-                        |_| io::Error::new(io::ErrorKind::InvalidData, "SSTable corruption"),
-                    )?)
-                } else {
-                    // if store_offset is zero, there is no index, so we build a pseudo-index
-                    // assuming a single block of sstable covering everything.
-                    SSTableIndex::V3Empty(v3::SSTableIndexV3Empty::load(index_offset as usize))
-                }
-            }
-            _ => {
-                return Err(io::Error::other(format!(
-                    "Unsupported sstable version, expected one of [2, 3], found {version}"
-                )));
-            }
-        };
-        Ok(index)
-    }
-
-    /// Get the [`BlockAddr`] of the requested block.
-    pub(crate) fn get_block(&self, block_id: u64) -> Option<BlockAddr> {
-        match self {
-            SSTableIndex::V2(v2_index) => v2_index.get_block(block_id as usize),
-            SSTableIndex::V3(v3_index) => v3_index.get_block(block_id),
-            SSTableIndex::V3Empty(v3_empty) => v3_empty.get_block(block_id),
-        }
-    }
-
-    /// Get the block id of the block that would contain `key`.
-    ///
-    /// Returns None if `key` is lexicographically after the last key recorded.
-    pub(crate) fn locate_with_key(&self, key: &[u8]) -> Option<u64> {
-        match self {
-            SSTableIndex::V2(v2_index) => v2_index.locate_with_key(key).map(|i| i as u64),
-            SSTableIndex::V3(v3_index) => v3_index.locate_with_key(key),
-            SSTableIndex::V3Empty(v3_empty) => v3_empty.locate_with_key(key),
-        }
-    }
-
-    /// Get the [`BlockAddr`] of the block that would contain `key`.
-    ///
-    /// Returns None if `key` is lexicographically after the last key recorded.
-    pub fn get_block_with_key(&self, key: &[u8]) -> Option<BlockAddr> {
-        match self {
-            SSTableIndex::V2(v2_index) => v2_index.get_block_with_key(key),
-            SSTableIndex::V3(v3_index) => v3_index.get_block_with_key(key),
-            SSTableIndex::V3Empty(v3_empty) => v3_empty.get_block_with_key(key),
-        }
-    }
-
-    pub(crate) fn locate_with_ord(&self, ord: TermOrdinal) -> u64 {
-        match self {
-            SSTableIndex::V2(v2_index) => v2_index.locate_with_ord(ord) as u64,
-            SSTableIndex::V3(v3_index) => v3_index.locate_with_ord(ord),
-            SSTableIndex::V3Empty(v3_empty) => v3_empty.locate_with_ord(ord),
-        }
-    }
-
-    /// Get the [`BlockAddr`] of the block containing the `ord`-th term.
-    pub(crate) fn get_block_with_ord(&self, ord: TermOrdinal) -> BlockAddr {
-        match self {
-            SSTableIndex::V2(v2_index) => v2_index.get_block_with_ord(ord),
-            SSTableIndex::V3(v3_index) => v3_index.get_block_with_ord(ord),
-            SSTableIndex::V3Empty(v3_empty) => v3_empty.get_block_with_ord(ord),
-        }
-    }
-
-    pub(crate) fn get_and_locate_with_ord(&self, ord: TermOrdinal) -> (BlockAddr, u64) {
-        match self {
-            SSTableIndex::V2(v2_index) => v2_index.get_and_locate_with_ord(ord),
-            SSTableIndex::V3(v3_index) => v3_index.get_and_locate_with_ord(ord),
-            SSTableIndex::V3Empty(v3_empty) => v3_empty.get_and_locate_with_ord(ord),
-        }
-    }
-
-    pub fn get_block_for_automaton<'a>(
-        &'a self,
-        automaton: &'a impl Automaton,
-    ) -> impl Iterator<Item = (u64, BlockAddr)> + 'a {
-        match self {
-            SSTableIndex::V2(v2_index) => {
-                BlockIter::V2(v2_index.get_block_for_automaton(automaton))
-            }
-            SSTableIndex::V3(v3_index) => {
-                BlockIter::V3(v3_index.get_block_for_automaton(automaton))
-            }
-            SSTableIndex::V3Empty(v3_empty) => {
-                BlockIter::V3Empty(std::iter::once((0, v3_empty.block_addr.clone())))
-            }
-        }
-    }
-}
-
-enum BlockIter<V2, V3, T> {
-    V2(V2),
-    V3(V3),
-    V3Empty(std::iter::Once<T>),
-}
-
-impl<V2: Iterator<Item = T>, V3: Iterator<Item = T>, T> Iterator for BlockIter<V2, V3, T> {
-    type Item = T;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        match self {
-            BlockIter::V2(v2) => v2.next(),
-            BlockIter::V3(v3) => v3.next(),
-            BlockIter::V3Empty(once) => once.next(),
-        }
-    }
-}
-
-#[derive(Clone, Eq, PartialEq, Debug)]
-pub struct BlockAddr {
-    pub first_ordinal: u64,
-    pub byte_range: Range<usize>,
-}
-
-impl BlockAddr {
-    fn to_block_start(&self) -> BlockStartAddr {
-        BlockStartAddr {
-            first_ordinal: self.first_ordinal,
-            byte_range_start: self.byte_range.start,
-        }
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-struct BlockStartAddr {
-    first_ordinal: u64,
-    byte_range_start: usize,
-}
-
-impl BlockStartAddr {
-    fn to_block_addr(&self, byte_range_end: usize) -> BlockAddr {
-        BlockAddr {
-            first_ordinal: self.first_ordinal,
-            byte_range: self.byte_range_start..byte_range_end,
-        }
-    }
-}
-
-#[derive(Debug, Clone)]
-pub(crate) struct BlockMeta {
-    /// Any byte string that is lexicographically greater or equal to
-    /// the last key in the block,
-    /// and yet strictly smaller than the first key in the next block.
-    pub last_key_or_greater: Vec<u8>,
-    pub block_addr: BlockAddr,
-}
-
-impl BinarySerializable for BlockStartAddr {
-    fn serialize<W: Write + ?Sized>(&self, writer: &mut W) -> io::Result<()> {
-        let start = self.byte_range_start as u64;
-        start.serialize(writer)?;
-        self.first_ordinal.serialize(writer)
-    }
-
-    fn deserialize<R: Read>(reader: &mut R) -> io::Result<Self> {
-        let byte_range_start = u64::deserialize(reader)? as usize;
-        let first_ordinal = u64::deserialize(reader)?;
-        Ok(BlockStartAddr {
-            first_ordinal,
-            byte_range_start,
-        })
-    }
-
-    // Provided method
-    fn num_bytes(&self) -> u64 {
-        BlockStartAddr::SIZE_IN_BYTES as u64
-    }
-}
-
-impl FixedSize for BlockStartAddr {
-    const SIZE_IN_BYTES: usize = 2 * u64::SIZE_IN_BYTES;
-}
-
-/// Given that left < right,
-/// mutates `left into a shorter byte string left'` that
-/// matches `left <= left' < right`.
-fn find_shorter_str_in_between(left: &mut Vec<u8>, right: &[u8]) {
-    assert!(&left[..] < right);
-    let common_len = common_prefix_len(left, right);
-    if left.len() == common_len {
-        return;
-    }
-    // It is possible to do one character shorter in some case,
-    // but it is not worth the extra complexity
-    for pos in (common_len + 1)..left.len() {
-        if left[pos] != u8::MAX {
-            left[pos] += 1;
-            left.truncate(pos + 1);
-            return;
-        }
-    }
-}
-
-#[derive(Default)]
-pub struct SSTableIndexBuilder {
-    blocks: Vec<BlockMeta>,
-}
-
-impl SSTableIndexBuilder {
-    /// In order to make the index as light as possible, we
-    /// try to find a shorter alternative to the last key of the last block
-    /// that is still smaller than the next key.
-    pub(crate) fn shorten_last_block_key_given_next_key(&mut self, next_key: &[u8]) {
-        if let Some(last_block) = self.blocks.last_mut() {
-            find_shorter_str_in_between(&mut last_block.last_key_or_greater, next_key);
-        }
-    }
-
-    pub fn add_block(&mut self, last_key: &[u8], byte_range: Range<usize>, first_ordinal: u64) {
-        self.blocks.push(BlockMeta {
-            last_key_or_greater: last_key.to_vec(),
-            block_addr: BlockAddr {
-                byte_range,
-                first_ordinal,
-            },
-        })
-    }
-
-    pub fn serialize<W: std::io::Write>(&self, wrt: W) -> io::Result<u64> {
-        if self.blocks.len() <= 1 {
-            return Ok(0);
-        }
-        let counting_writer = common::CountingWriter::wrap(wrt);
-        let mut map_builder = MapBuilder::new(counting_writer).map_err(fst_error_to_io_error)?;
-        for (i, block) in self.blocks.iter().enumerate() {
-            map_builder
-                .insert(&block.last_key_or_greater, i as u64)
-                .map_err(fst_error_to_io_error)?;
-        }
-        let counting_writer = map_builder.into_inner().map_err(fst_error_to_io_error)?;
-        let written_bytes = counting_writer.written_bytes();
-        let mut wrt = counting_writer.finish();
-
-        let mut block_store_writer = v3::BlockAddrStoreWriter::new();
-        for block in &self.blocks {
-            block_store_writer.write_block_meta(block.block_addr.clone())?;
-        }
-        block_store_writer.serialize(&mut wrt)?;
-
-        Ok(written_bytes)
-    }
-}
-
-fn fst_error_to_io_error(error: tantivy_fst::Error) -> io::Error {
-    match error {
-        tantivy_fst::Error::Fst(fst_error) => io::Error::other(fst_error),
-        tantivy_fst::Error::Io(ioerror) => ioerror,
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    #[track_caller]
-    fn test_find_shorter_str_in_between_aux(left: &[u8], right: &[u8]) {
-        let mut left_buf = left.to_vec();
-        super::find_shorter_str_in_between(&mut left_buf, right);
-        assert!(left_buf.len() <= left.len());
-        assert!(left <= &left_buf);
-        assert!(&left_buf[..] < right);
-    }
-
-    #[test]
-    fn test_find_shorter_str_in_between() {
-        test_find_shorter_str_in_between_aux(b"", b"hello");
-        test_find_shorter_str_in_between_aux(b"abc", b"abcd");
-        test_find_shorter_str_in_between_aux(b"abcd", b"abd");
-        test_find_shorter_str_in_between_aux(&[0, 0, 0], &[1]);
-        test_find_shorter_str_in_between_aux(&[0, 0, 0], &[0, 0, 1]);
-        test_find_shorter_str_in_between_aux(&[0, 0, 255, 255, 255, 0u8], &[0, 1]);
-    }
-
-    use proptest::prelude::*;
-
-    proptest! {
-        #![proptest_config(ProptestConfig::with_cases(100))]
-        #[test]
-        fn test_proptest_find_shorter_str(left in any::<Vec<u8>>(), right in any::<Vec<u8>>()) {
-            if left < right {
-                test_find_shorter_str_in_between_aux(&left, &right);
-            }
-        }
-    }
-}
--- a/sstable/src/lib.rs
+++ b/sstable/src/lib.rs
@@ -47,8 +47,9 @@ pub mod merge;
 mod streamer;
 pub mod value;

-mod index;
-pub use index::{BlockAddr, SSTableIndex, SSTableIndexBuilder};
+mod sstable_index_v3;
+pub use sstable_index_v3::{BlockAddr, SSTableIndex, SSTableIndexBuilder, SSTableIndexV3};
+mod sstable_index_v2;
 pub(crate) mod vint;
 pub use dictionary::{Dictionary, TermOrdHit};
 pub use streamer::{Streamer, StreamerBuilder};
--- a/sstable/src/sstable_index_v2.rs
+++ b/sstable/src/sstable_index_v2.rs
@@ -77,13 +77,6 @@ impl SSTableIndex {
        self.get_block(self.locate_with_ord(ord)).unwrap()
    }

-    pub(crate) fn get_and_locate_with_ord(&self, ord: TermOrdinal) -> (BlockAddr, u64) {
-        let location = self.locate_with_ord(ord);
-        // locate_with_ord always returns an index within range
-        let block_addr = self.get_block(location).unwrap();
-        (block_addr, location as u64)
-    }
-
    pub(crate) fn get_block_for_automaton<'a>(
        &'a self,
        automaton: &'a impl Automaton,
--- a/sstable/src/sstable_index_v3.rs
+++ b/sstable/src/sstable_index_v3.rs
@@ -1,14 +1,106 @@
 use std::io::{self, Read, Write};
+use std::ops::Range;
 use std::sync::Arc;

 use common::{BinarySerializable, FixedSize, OwnedBytes};
 use tantivy_bitpacker::{BitPacker, compute_num_bits};
 use tantivy_fst::raw::Fst;
-use tantivy_fst::{Automaton, IntoStreamer, Map, Streamer};
+use tantivy_fst::{Automaton, IntoStreamer, Map, MapBuilder, Streamer};

-use super::{BlockAddr, BlockStartAddr};
 use crate::block_match_automaton::can_block_match_automaton;
-use crate::{SSTableDataCorruption, TermOrdinal};
+use crate::{SSTableDataCorruption, TermOrdinal, common_prefix_len};
+
+#[derive(Debug, Clone)]
+pub enum SSTableIndex {
+    V2(crate::sstable_index_v2::SSTableIndex),
+    V3(SSTableIndexV3),
+    V3Empty(SSTableIndexV3Empty),
+}
+
+impl SSTableIndex {
+    /// Get the [`BlockAddr`] of the requested block.
+    pub(crate) fn get_block(&self, block_id: u64) -> Option<BlockAddr> {
+        match self {
+            SSTableIndex::V2(v2_index) => v2_index.get_block(block_id as usize),
+            SSTableIndex::V3(v3_index) => v3_index.get_block(block_id),
+            SSTableIndex::V3Empty(v3_empty) => v3_empty.get_block(block_id),
+        }
+    }
+
+    /// Get the block id of the block that would contain `key`.
+    ///
+    /// Returns None if `key` is lexicographically after the last key recorded.
+    pub(crate) fn locate_with_key(&self, key: &[u8]) -> Option<u64> {
+        match self {
+            SSTableIndex::V2(v2_index) => v2_index.locate_with_key(key).map(|i| i as u64),
+            SSTableIndex::V3(v3_index) => v3_index.locate_with_key(key),
+            SSTableIndex::V3Empty(v3_empty) => v3_empty.locate_with_key(key),
+        }
+    }
+
+    /// Get the [`BlockAddr`] of the block that would contain `key`.
+    ///
+    /// Returns None if `key` is lexicographically after the last key recorded.
+    pub fn get_block_with_key(&self, key: &[u8]) -> Option<BlockAddr> {
+        match self {
+            SSTableIndex::V2(v2_index) => v2_index.get_block_with_key(key),
+            SSTableIndex::V3(v3_index) => v3_index.get_block_with_key(key),
+            SSTableIndex::V3Empty(v3_empty) => v3_empty.get_block_with_key(key),
+        }
+    }
+
+    pub(crate) fn locate_with_ord(&self, ord: TermOrdinal) -> u64 {
+        match self {
+            SSTableIndex::V2(v2_index) => v2_index.locate_with_ord(ord) as u64,
+            SSTableIndex::V3(v3_index) => v3_index.locate_with_ord(ord),
+            SSTableIndex::V3Empty(v3_empty) => v3_empty.locate_with_ord(ord),
+        }
+    }
+
+    /// Get the [`BlockAddr`] of the block containing the `ord`-th term.
+    pub(crate) fn get_block_with_ord(&self, ord: TermOrdinal) -> BlockAddr {
+        match self {
+            SSTableIndex::V2(v2_index) => v2_index.get_block_with_ord(ord),
+            SSTableIndex::V3(v3_index) => v3_index.get_block_with_ord(ord),
+            SSTableIndex::V3Empty(v3_empty) => v3_empty.get_block_with_ord(ord),
+        }
+    }
+
+    pub fn get_block_for_automaton<'a>(
+        &'a self,
+        automaton: &'a impl Automaton,
+    ) -> impl Iterator<Item = (u64, BlockAddr)> + 'a {
+        match self {
+            SSTableIndex::V2(v2_index) => {
+                BlockIter::V2(v2_index.get_block_for_automaton(automaton))
+            }
+            SSTableIndex::V3(v3_index) => {
+                BlockIter::V3(v3_index.get_block_for_automaton(automaton))
+            }
+            SSTableIndex::V3Empty(v3_empty) => {
+                BlockIter::V3Empty(std::iter::once((0, v3_empty.block_addr.clone())))
+            }
+        }
+    }
+}
+
+enum BlockIter<V2, V3, T> {
+    V2(V2),
+    V3(V3),
+    V3Empty(std::iter::Once<T>),
+}
+
+impl<V2: Iterator<Item = T>, V3: Iterator<Item = T>, T> Iterator for BlockIter<V2, V3, T> {
+    type Item = T;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self {
+            BlockIter::V2(v2) => v2.next(),
+            BlockIter::V3(v3) => v3.next(),
+            BlockIter::V3Empty(once) => once.next(),
+        }
+    }
+}

 #[derive(Debug, Clone)]
 pub struct SSTableIndexV3 {
@@ -68,11 +160,6 @@ impl SSTableIndexV3 {
        self.block_addr_store.binary_search_ord(ord).1
    }

-    pub(crate) fn get_and_locate_with_ord(&self, ord: TermOrdinal) -> (BlockAddr, u64) {
-        let (location, block_addr) = self.block_addr_store.binary_search_ord(ord);
-        (block_addr, location)
-    }
-
    pub(crate) fn get_block_for_automaton<'a>(
        &'a self,
        automaton: &'a impl Automaton,
@@ -129,7 +216,7 @@ impl<A: Automaton> Iterator for GetBlockForAutomaton<'_, A> {

 #[derive(Debug, Clone)]
 pub struct SSTableIndexV3Empty {
-    pub block_addr: BlockAddr,
+    block_addr: BlockAddr,
 }

 impl SSTableIndexV3Empty {
@@ -143,8 +230,8 @@ impl SSTableIndexV3Empty {
    }

    /// Get the [`BlockAddr`] of the requested block.
-    pub(crate) fn get_block(&self, block_id: u64) -> Option<BlockAddr> {
-        (block_id == 0).then(|| self.block_addr.clone())
+    pub(crate) fn get_block(&self, _block_id: u64) -> Option<BlockAddr> {
+        Some(self.block_addr.clone())
    }

    /// Get the block id of the block that would contain `key`.
@@ -169,9 +256,146 @@ impl SSTableIndexV3Empty {
    pub(crate) fn get_block_with_ord(&self, _ord: TermOrdinal) -> BlockAddr {
        self.block_addr.clone()
    }
+}
+#[derive(Clone, Eq, PartialEq, Debug)]
+pub struct BlockAddr {
+    pub first_ordinal: u64,
+    pub byte_range: Range<usize>,
+}

-    pub(crate) fn get_and_locate_with_ord(&self, _ord: TermOrdinal) -> (BlockAddr, u64) {
-        (self.block_addr.clone(), 0)
+impl BlockAddr {
+    fn to_block_start(&self) -> BlockStartAddr {
+        BlockStartAddr {
+            first_ordinal: self.first_ordinal,
+            byte_range_start: self.byte_range.start,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct BlockStartAddr {
+    first_ordinal: u64,
+    byte_range_start: usize,
+}
+
+impl BlockStartAddr {
+    fn to_block_addr(&self, byte_range_end: usize) -> BlockAddr {
+        BlockAddr {
+            first_ordinal: self.first_ordinal,
+            byte_range: self.byte_range_start..byte_range_end,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct BlockMeta {
+    /// Any byte string that is lexicographically greater or equal to
+    /// the last key in the block,
+    /// and yet strictly smaller than the first key in the next block.
+    pub last_key_or_greater: Vec<u8>,
+    pub block_addr: BlockAddr,
+}
+
+impl BinarySerializable for BlockStartAddr {
+    fn serialize<W: Write + ?Sized>(&self, writer: &mut W) -> io::Result<()> {
+        let start = self.byte_range_start as u64;
+        start.serialize(writer)?;
+        self.first_ordinal.serialize(writer)
+    }
+
+    fn deserialize<R: Read>(reader: &mut R) -> io::Result<Self> {
+        let byte_range_start = u64::deserialize(reader)? as usize;
+        let first_ordinal = u64::deserialize(reader)?;
+        Ok(BlockStartAddr {
+            first_ordinal,
+            byte_range_start,
+        })
+    }
+
+    // Provided method
+    fn num_bytes(&self) -> u64 {
+        BlockStartAddr::SIZE_IN_BYTES as u64
+    }
+}
+
+impl FixedSize for BlockStartAddr {
+    const SIZE_IN_BYTES: usize = 2 * u64::SIZE_IN_BYTES;
+}
+
+/// Given that left < right,
+/// mutates `left into a shorter byte string left'` that
+/// matches `left <= left' < right`.
+fn find_shorter_str_in_between(left: &mut Vec<u8>, right: &[u8]) {
+    assert!(&left[..] < right);
+    let common_len = common_prefix_len(left, right);
+    if left.len() == common_len {
+        return;
+    }
+    // It is possible to do one character shorter in some case,
+    // but it is not worth the extra complexity
+    for pos in (common_len + 1)..left.len() {
+        if left[pos] != u8::MAX {
+            left[pos] += 1;
+            left.truncate(pos + 1);
+            return;
+        }
+    }
+}
+
+#[derive(Default)]
+pub struct SSTableIndexBuilder {
+    blocks: Vec<BlockMeta>,
+}
+
+impl SSTableIndexBuilder {
+    /// In order to make the index as light as possible, we
+    /// try to find a shorter alternative to the last key of the last block
+    /// that is still smaller than the next key.
+    pub(crate) fn shorten_last_block_key_given_next_key(&mut self, next_key: &[u8]) {
+        if let Some(last_block) = self.blocks.last_mut() {
+            find_shorter_str_in_between(&mut last_block.last_key_or_greater, next_key);
+        }
+    }
+
+    pub fn add_block(&mut self, last_key: &[u8], byte_range: Range<usize>, first_ordinal: u64) {
+        self.blocks.push(BlockMeta {
+            last_key_or_greater: last_key.to_vec(),
+            block_addr: BlockAddr {
+                byte_range,
+                first_ordinal,
+            },
+        })
+    }
+
+    pub fn serialize<W: std::io::Write>(&self, wrt: W) -> io::Result<u64> {
+        if self.blocks.len() <= 1 {
+            return Ok(0);
+        }
+        let counting_writer = common::CountingWriter::wrap(wrt);
+        let mut map_builder = MapBuilder::new(counting_writer).map_err(fst_error_to_io_error)?;
+        for (i, block) in self.blocks.iter().enumerate() {
+            map_builder
+                .insert(&block.last_key_or_greater, i as u64)
+                .map_err(fst_error_to_io_error)?;
+        }
+        let counting_writer = map_builder.into_inner().map_err(fst_error_to_io_error)?;
+        let written_bytes = counting_writer.written_bytes();
+        let mut wrt = counting_writer.finish();
+
+        let mut block_store_writer = BlockAddrStoreWriter::new();
+        for block in &self.blocks {
+            block_store_writer.write_block_meta(block.block_addr.clone())?;
+        }
+        block_store_writer.serialize(&mut wrt)?;
+
+        Ok(written_bytes)
+    }
+}
+
+fn fst_error_to_io_error(error: tantivy_fst::Error) -> io::Error {
+    match error {
+        tantivy_fst::Error::Fst(fst_error) => io::Error::other(fst_error),
+        tantivy_fst::Error::Io(ioerror) => ioerror,
    }
 }

@@ -423,14 +647,14 @@ fn binary_search(max: u64, cmp_fn: impl Fn(u64) -> std::cmp::Ordering) -> Result
    Err(left)
 }

-pub(crate) struct BlockAddrStoreWriter {
+struct BlockAddrStoreWriter {
    buffer_block_metas: Vec<u8>,
    buffer_addrs: Vec<u8>,
    block_addrs: Vec<BlockAddr>,
 }

 impl BlockAddrStoreWriter {
-    pub(crate) fn new() -> Self {
+    fn new() -> Self {
        BlockAddrStoreWriter {
            buffer_block_metas: Vec::new(),
            buffer_addrs: Vec::new(),
@@ -438,7 +662,7 @@ impl BlockAddrStoreWriter {
        }
    }

-    pub(crate) fn flush_block(&mut self) -> io::Result<()> {
+    fn flush_block(&mut self) -> io::Result<()> {
        if self.block_addrs.is_empty() {
            return Ok(());
        }
@@ -517,7 +741,7 @@ impl BlockAddrStoreWriter {
        Ok(())
    }

-    pub(crate) fn write_block_meta(&mut self, block_addr: BlockAddr) -> io::Result<()> {
+    fn write_block_meta(&mut self, block_addr: BlockAddr) -> io::Result<()> {
        self.block_addrs.push(block_addr);
        if self.block_addrs.len() >= STORE_BLOCK_LEN {
            self.flush_block()?;
@@ -525,7 +749,7 @@ impl BlockAddrStoreWriter {
        Ok(())
    }

-    pub(crate) fn serialize<W: std::io::Write>(&mut self, wrt: &mut W) -> io::Result<()> {
+    fn serialize<W: std::io::Write>(&mut self, wrt: &mut W) -> io::Result<()> {
        self.flush_block()?;
        let len = self.buffer_block_metas.len() as u64;
        len.serialize(wrt)?;
@@ -600,9 +824,8 @@ mod tests {
    use common::OwnedBytes;

    use super::*;
+    use crate::SSTableDataCorruption;
    use crate::block_match_automaton::tests::EqBuffer;
-    use crate::index::BlockMeta;
-    use crate::{SSTableDataCorruption, SSTableIndexBuilder};

    #[test]
    fn test_sstable_index() {
@@ -651,7 +874,36 @@ mod tests {
        assert!(matches!(data_corruption_err, SSTableDataCorruption));
    }

-    //    use proptest::prelude::*;
+    #[track_caller]
+    fn test_find_shorter_str_in_between_aux(left: &[u8], right: &[u8]) {
+        let mut left_buf = left.to_vec();
+        super::find_shorter_str_in_between(&mut left_buf, right);
+        assert!(left_buf.len() <= left.len());
+        assert!(left <= &left_buf);
+        assert!(&left_buf[..] < right);
+    }
+
+    #[test]
+    fn test_find_shorter_str_in_between() {
+        test_find_shorter_str_in_between_aux(b"", b"hello");
+        test_find_shorter_str_in_between_aux(b"abc", b"abcd");
+        test_find_shorter_str_in_between_aux(b"abcd", b"abd");
+        test_find_shorter_str_in_between_aux(&[0, 0, 0], &[1]);
+        test_find_shorter_str_in_between_aux(&[0, 0, 0], &[0, 0, 1]);
+        test_find_shorter_str_in_between_aux(&[0, 0, 255, 255, 255, 0u8], &[0, 1]);
+    }
+
+    use proptest::prelude::*;
+
+    proptest! {
+        #![proptest_config(ProptestConfig::with_cases(100))]
+        #[test]
+        fn test_proptest_find_shorter_str(left in any::<Vec<u8>>(), right in any::<Vec<u8>>()) {
+            if left < right {
+                test_find_shorter_str_in_between_aux(&left, &right);
+            }
+        }
+    }

    #[test]
    fn test_find_best_slop() {
--- a/stacker/Cargo.toml
+++ b/stacker/Cargo.toml
@@ -8,7 +8,7 @@ repository = "https://github.com/quickwit-oss/tantivy"
 description = "term hashmap used for indexing"

 [dependencies]
-murmurhash32 = "0.4"
+murmurhash32 = "0.3"
 common = { version = "0.11", path = "../common/", package = "tantivy-common" }
 ahash = { version = "0.8.11", default-features = false, optional = true }

@@ -27,7 +27,7 @@ rand = "0.9"
 zipf = "7.0.0"
 rustc-hash = "2.1.0"
 proptest = "1.2.0"
-binggan = { version = "0.17.0" }
+binggan = { version = "0.16.1" }
 rand_distr = "0.5"

 [features]