Making stemming optional. (#2791)

Fixed code and CI to run on no default features.

Co-authored-by: Paul Masurel <paul.masurel@datadoghq.com>

.github/workflows/coverage.yml

@@ -15,11 +15,11 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - name: Install Rust
-        run: rustup toolchain install nightly-2024-07-01 --profile minimal --component llvm-tools-preview
+        run: rustup toolchain install nightly-2025-12-01 --profile minimal --component llvm-tools-preview
       - uses: Swatinem/rust-cache@v2
       - uses: taiki-e/install-action@cargo-llvm-cov
       - name: Generate code coverage
-        run: cargo +nightly-2024-07-01 llvm-cov --all-features --workspace --doctests --lcov --output-path lcov.info
+        run: cargo +nightly-2025-12-01 llvm-cov --all-features --workspace --doctests --lcov --output-path lcov.info
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v3
         continue-on-error: true

.github/workflows/test.yml

@@ -39,11 +39,11 @@ jobs:
 
     - name: Check Formatting
       run: cargo +nightly fmt --all -- --check
-    
+
     - name: Check Stable Compilation
       run: cargo build --all-features
 
-    
+
     - name: Check Bench Compilation
       run: cargo +nightly bench --no-run --profile=dev --all-features
 
@@ -59,10 +59,10 @@ jobs:
 
     strategy:
       matrix:
-        features: [
-            { label: "all", flags: "mmap,stopwords,lz4-compression,zstd-compression,failpoints" },
-            { label: "quickwit", flags: "mmap,quickwit,failpoints" }
-        ]
+        features:
+          - { label: "all", flags: "mmap,stopwords,lz4-compression,zstd-compression,failpoints,stemmer" }
+          - { label: "quickwit", flags: "mmap,quickwit,failpoints" }
+          - { label: "none", flags: "" }
 
     name: test-${{ matrix.features.label}}
 
@@ -80,7 +80,21 @@ jobs:
     - uses: Swatinem/rust-cache@v2
 
     - name: Run tests
-      run: cargo +stable nextest run --features ${{ matrix.features.flags }} --verbose --workspace
+      run: |
+        # if matrix.feature.flags is empty then run on --lib to avoid compiling examples
+        # (as most of them rely on mmap) otherwise run all
+        if [ -z "${{ matrix.features.flags }}" ]; then
+          cargo +stable nextest run --lib --no-default-features --verbose --workspace
+        else
+          cargo +stable nextest run --features ${{ matrix.features.flags }} --no-default-features --verbose --workspace
+        fi
 
     - name: Run doctests
-      run: cargo +stable test --doc --features ${{ matrix.features.flags }} --verbose --workspace
+      run: |
+        # if matrix.feature.flags is empty then run on --lib to avoid compiling examples
+        # (as most of them rely on mmap) otherwise run all
+        if [ -z "${{ matrix.features.flags }}" ]; then
+          echo "no doctest for no feature flag"
+        else
+          cargo +stable test --doc --features ${{ matrix.features.flags }} --verbose --workspace
+        fi

Cargo.toml

@@ -37,7 +37,7 @@ fs4 = { version = "0.13.1", optional = true }
 levenshtein_automata = "0.2.1"
 uuid = { version = "1.0.0", features = ["v4", "serde"] }
 crossbeam-channel = "0.5.4"
-rust-stemmers = "1.2.0"
+rust-stemmers = { version = "1.2.0", optional = true }
 downcast-rs = "2.0.1"
 bitpacking = { version = "0.9.2", default-features = false, features = [
     "bitpacker4x",
@@ -75,12 +75,12 @@ typetag = "0.2.21"
 winapi = "0.3.9"
 
 [dev-dependencies]
-binggan = "0.14.0"
+binggan = "0.14.2"
 rand = "0.8.5"
 maplit = "1.0.2"
 matches = "0.1.9"
 pretty_assertions = "1.2.1"
-proptest = "1.0.0"
+proptest = "1.7.0"
 test-log = "0.2.10"
 futures = "0.3.21"
 paste = "1.0.11"
@@ -113,7 +113,8 @@ debug-assertions = true
 overflow-checks = true
 
 [features]
-default = ["mmap", "stopwords", "lz4-compression", "columnar-zstd-compression"]
+default = ["mmap", "stopwords", "lz4-compression", "columnar-zstd-compression", "stemmer"]
+stemmer = ["rust-stemmers"]
 mmap = ["fs4", "tempfile", "memmap2"]
 stopwords = []
 
@@ -173,6 +174,18 @@ harness = false
 name = "exists_json"
 harness = false
 
+[[bench]]
+name = "range_query"
+harness = false
+
 [[bench]]
 name = "and_or_queries"
 harness = false
+
+[[bench]]
+name = "range_queries"
+harness = false
+
+[[bench]]
+name = "bool_queries_with_range"
+harness = false

benches/bool_queries_with_range.rs

@@ -0,0 +1,288 @@
+use binggan::{black_box, BenchGroup, BenchRunner};
+use rand::prelude::*;
+use rand::rngs::StdRng;
+use rand::SeedableRng;
+use tantivy::collector::{Collector, Count, DocSetCollector, TopDocs};
+use tantivy::query::{Query, QueryParser};
+use tantivy::schema::{Schema, FAST, INDEXED, TEXT};
+use tantivy::{doc, Index, Order, ReloadPolicy, Searcher};
+
+#[derive(Clone)]
+struct BenchIndex {
+    #[allow(dead_code)]
+    index: Index,
+    searcher: Searcher,
+    query_parser: QueryParser,
+}
+
+fn build_shared_indices(num_docs: usize, p_title_a: f32, distribution: &str) -> BenchIndex {
+    // Unified schema
+    let mut schema_builder = Schema::builder();
+    let f_title = schema_builder.add_text_field("title", TEXT);
+    let f_num_rand = schema_builder.add_u64_field("num_rand", INDEXED);
+    let f_num_asc = schema_builder.add_u64_field("num_asc", INDEXED);
+    let f_num_rand_fast = schema_builder.add_u64_field("num_rand_fast", INDEXED | FAST);
+    let f_num_asc_fast = schema_builder.add_u64_field("num_asc_fast", INDEXED | FAST);
+    let schema = schema_builder.build();
+    let index = Index::create_in_ram(schema.clone());
+
+    // Populate index with stable RNG for reproducibility.
+    let mut rng = StdRng::from_seed([7u8; 32]);
+
+    {
+        let mut writer = index.writer_with_num_threads(1, 4_000_000_000).unwrap();
+
+        match distribution {
+            "dense" => {
+                for doc_id in 0..num_docs {
+                    // Always add title to avoid empty documents
+                    let title_token = if rng.gen_bool(p_title_a as f64) {
+                        "a"
+                    } else {
+                        "b"
+                    };
+
+                    let num_rand = rng.gen_range(0u64..1000u64);
+
+                    let num_asc = (doc_id / 10000) as u64;
+
+                    writer
+                        .add_document(doc!(
+                            f_title=>title_token,
+                            f_num_rand=>num_rand,
+                            f_num_asc=>num_asc,
+                            f_num_rand_fast=>num_rand,
+                            f_num_asc_fast=>num_asc,
+                        ))
+                        .unwrap();
+                }
+            }
+            "sparse" => {
+                for doc_id in 0..num_docs {
+                    // Always add title to avoid empty documents
+                    let title_token = if rng.gen_bool(p_title_a as f64) {
+                        "a"
+                    } else {
+                        "b"
+                    };
+
+                    let num_rand = rng.gen_range(0u64..10000000u64);
+
+                    let num_asc = doc_id as u64;
+
+                    writer
+                        .add_document(doc!(
+                            f_title=>title_token,
+                            f_num_rand=>num_rand,
+                            f_num_asc=>num_asc,
+                            f_num_rand_fast=>num_rand,
+                            f_num_asc_fast=>num_asc,
+                        ))
+                        .unwrap();
+                }
+            }
+            _ => {
+                panic!("Unsupported distribution type");
+            }
+        }
+        writer.commit().unwrap();
+    }
+
+    // Prepare reader/searcher once.
+    let reader = index
+        .reader_builder()
+        .reload_policy(ReloadPolicy::Manual)
+        .try_into()
+        .unwrap();
+    let searcher = reader.searcher();
+
+    // Build query parser for title field
+    let qp_title = QueryParser::for_index(&index, vec![f_title]);
+
+    BenchIndex {
+        index,
+        searcher,
+        query_parser: qp_title,
+    }
+}
+
+fn main() {
+    // Prepare corpora with varying scenarios
+    let scenarios = vec![
+        (
+            "dense and 99% a".to_string(),
+            10_000_000,
+            0.99,
+            "dense",
+            0,
+            9,
+        ),
+        (
+            "dense and 99% a".to_string(),
+            10_000_000,
+            0.99,
+            "dense",
+            990,
+            999,
+        ),
+        (
+            "sparse and 99% a".to_string(),
+            10_000_000,
+            0.99,
+            "sparse",
+            0,
+            9,
+        ),
+        (
+            "sparse and 99% a".to_string(),
+            10_000_000,
+            0.99,
+            "sparse",
+            9_999_990,
+            9_999_999,
+        ),
+    ];
+
+    let mut runner = BenchRunner::new();
+    for (scenario_id, n, p_title_a, num_rand_distribution, range_low, range_high) in scenarios {
+        // Build index for this scenario
+        let bench_index = build_shared_indices(n, p_title_a, num_rand_distribution);
+
+        // Create benchmark group
+        let mut group = runner.new_group();
+
+        // Now set the name (this moves scenario_id)
+        group.set_name(scenario_id);
+
+        // Define all four field types
+        let field_names = ["num_rand", "num_asc", "num_rand_fast", "num_asc_fast"];
+
+        // Define the three terms we want to test with
+        let terms = ["a", "b", "z"];
+
+        // Generate all combinations of terms and field names
+        let mut queries = Vec::new();
+        for &term in &terms {
+            for &field_name in &field_names {
+                let query_str = format!(
+                    "{} AND {}:[{} TO {}]",
+                    term, field_name, range_low, range_high
+                );
+                queries.push((query_str, field_name.to_string()));
+            }
+        }
+
+        let query_str = format!(
+            "{}:[{} TO {}] AND {}:[{} TO {}]",
+            "num_rand_fast", range_low, range_high, "num_asc_fast", range_low, range_high
+        );
+        queries.push((query_str, "num_asc_fast".to_string()));
+
+        // Run all benchmark tasks for each query and its corresponding field name
+        for (query_str, field_name) in queries {
+            run_benchmark_tasks(&mut group, &bench_index, &query_str, &field_name);
+        }
+
+        group.run();
+    }
+}
+
+/// Run all benchmark tasks for a given query string and field name
+fn run_benchmark_tasks(
+    bench_group: &mut BenchGroup,
+    bench_index: &BenchIndex,
+    query_str: &str,
+    field_name: &str,
+) {
+    // Test count
+    add_bench_task(bench_group, bench_index, query_str, Count, "count");
+
+    // Test all results
+    add_bench_task(
+        bench_group,
+        bench_index,
+        query_str,
+        DocSetCollector,
+        "all results",
+    );
+
+    // Test top 100 by the field (if it's a FAST field)
+    if field_name.ends_with("_fast") {
+        // Ascending order
+        {
+            let collector_name = format!("top100_by_{}_asc", field_name);
+            let field_name_owned = field_name.to_string();
+            add_bench_task(
+                bench_group,
+                bench_index,
+                query_str,
+                TopDocs::with_limit(100).order_by_fast_field::<u64>(field_name_owned, Order::Asc),
+                &collector_name,
+            );
+        }
+
+        // Descending order
+        {
+            let collector_name = format!("top100_by_{}_desc", field_name);
+            let field_name_owned = field_name.to_string();
+            add_bench_task(
+                bench_group,
+                bench_index,
+                query_str,
+                TopDocs::with_limit(100).order_by_fast_field::<u64>(field_name_owned, Order::Desc),
+                &collector_name,
+            );
+        }
+    }
+}
+
+fn add_bench_task<C: Collector + 'static>(
+    bench_group: &mut BenchGroup,
+    bench_index: &BenchIndex,
+    query_str: &str,
+    collector: C,
+    collector_name: &str,
+) {
+    let task_name = format!("{}_{}", query_str.replace(" ", "_"), collector_name);
+    let query = bench_index.query_parser.parse_query(query_str).unwrap();
+    let search_task = SearchTask {
+        searcher: bench_index.searcher.clone(),
+        collector,
+        query,
+    };
+    bench_group.register(task_name, move |_| black_box(search_task.run()));
+}
+
+struct SearchTask<C: Collector> {
+    searcher: Searcher,
+    collector: C,
+    query: Box<dyn Query>,
+}
+
+impl<C: Collector> SearchTask<C> {
+    #[inline(never)]
+    pub fn run(&self) -> usize {
+        let result = self.searcher.search(&self.query, &self.collector).unwrap();
+        if let Some(count) = (&result as &dyn std::any::Any).downcast_ref::<usize>() {
+            *count
+        } else if let Some(top_docs) = (&result as &dyn std::any::Any)
+            .downcast_ref::<Vec<(Option<u64>, tantivy::DocAddress)>>()
+        {
+            top_docs.len()
+        } else if let Some(top_docs) =
+            (&result as &dyn std::any::Any).downcast_ref::<Vec<(u64, tantivy::DocAddress)>>()
+        {
+            top_docs.len()
+        } else if let Some(doc_set) = (&result as &dyn std::any::Any)
+            .downcast_ref::<std::collections::HashSet<tantivy::DocAddress>>()
+        {
+            doc_set.len()
+        } else {
+            eprintln!(
+                "Unknown collector result type: {:?}",
+                std::any::type_name::<C::Fruit>()
+            );
+            0
+        }
+    }
+}

benches/range_queries.rs

@@ -0,0 +1,365 @@
+use std::ops::Bound;
+
+use binggan::{black_box, BenchGroup, BenchRunner};
+use rand::prelude::*;
+use rand::rngs::StdRng;
+use rand::SeedableRng;
+use tantivy::collector::{Count, DocSetCollector, TopDocs};
+use tantivy::query::RangeQuery;
+use tantivy::schema::{Schema, FAST, INDEXED};
+use tantivy::{doc, Index, Order, ReloadPolicy, Searcher, Term};
+
+#[derive(Clone)]
+struct BenchIndex {
+    #[allow(dead_code)]
+    index: Index,
+    searcher: Searcher,
+}
+
+fn build_shared_indices(num_docs: usize, distribution: &str) -> BenchIndex {
+    // Schema with fast fields only
+    let mut schema_builder = Schema::builder();
+    let f_num_rand_fast = schema_builder.add_u64_field("num_rand_fast", INDEXED | FAST);
+    let f_num_asc_fast = schema_builder.add_u64_field("num_asc_fast", INDEXED | FAST);
+    let schema = schema_builder.build();
+    let index = Index::create_in_ram(schema.clone());
+
+    // Populate index with stable RNG for reproducibility.
+    let mut rng = StdRng::from_seed([7u8; 32]);
+
+    {
+        let mut writer = index.writer_with_num_threads(1, 4_000_000_000).unwrap();
+
+        match distribution {
+            "dense" => {
+                for doc_id in 0..num_docs {
+                    let num_rand = rng.gen_range(0u64..1000u64);
+                    let num_asc = (doc_id / 10000) as u64;
+
+                    writer
+                        .add_document(doc!(
+                            f_num_rand_fast=>num_rand,
+                            f_num_asc_fast=>num_asc,
+                        ))
+                        .unwrap();
+                }
+            }
+            "sparse" => {
+                for doc_id in 0..num_docs {
+                    let num_rand = rng.gen_range(0u64..10000000u64);
+                    let num_asc = doc_id as u64;
+
+                    writer
+                        .add_document(doc!(
+                            f_num_rand_fast=>num_rand,
+                            f_num_asc_fast=>num_asc,
+                        ))
+                        .unwrap();
+                }
+            }
+            _ => {
+                panic!("Unsupported distribution type");
+            }
+        }
+        writer.commit().unwrap();
+    }
+
+    // Prepare reader/searcher once.
+    let reader = index
+        .reader_builder()
+        .reload_policy(ReloadPolicy::Manual)
+        .try_into()
+        .unwrap();
+    let searcher = reader.searcher();
+
+    BenchIndex { index, searcher }
+}
+
+fn main() {
+    // Prepare corpora with varying scenarios
+    let scenarios = vec![
+        // Dense distribution - random values in small range (0-999)
+        (
+            "dense_values_search_low_value_range".to_string(),
+            10_000_000,
+            "dense",
+            0,
+            9,
+        ),
+        (
+            "dense_values_search_high_value_range".to_string(),
+            10_000_000,
+            "dense",
+            990,
+            999,
+        ),
+        (
+            "dense_values_search_out_of_range".to_string(),
+            10_000_000,
+            "dense",
+            1000,
+            1002,
+        ),
+        (
+            "sparse_values_search_low_value_range".to_string(),
+            10_000_000,
+            "sparse",
+            0,
+            9,
+        ),
+        (
+            "sparse_values_search_high_value_range".to_string(),
+            10_000_000,
+            "sparse",
+            9_999_990,
+            9_999_999,
+        ),
+        (
+            "sparse_values_search_out_of_range".to_string(),
+            10_000_000,
+            "sparse",
+            10_000_000,
+            10_000_002,
+        ),
+    ];
+
+    let mut runner = BenchRunner::new();
+    for (scenario_id, n, num_rand_distribution, range_low, range_high) in scenarios {
+        // Build index for this scenario
+        let bench_index = build_shared_indices(n, num_rand_distribution);
+
+        // Create benchmark group
+        let mut group = runner.new_group();
+
+        // Now set the name (this moves scenario_id)
+        group.set_name(scenario_id);
+
+        // Define fast field types
+        let field_names = ["num_rand_fast", "num_asc_fast"];
+
+        // Generate range queries for fast fields
+        for &field_name in &field_names {
+            // Create the range query
+            let field = bench_index.searcher.schema().get_field(field_name).unwrap();
+            let lower_term = Term::from_field_u64(field, range_low);
+            let upper_term = Term::from_field_u64(field, range_high);
+
+            let query = RangeQuery::new(Bound::Included(lower_term), Bound::Included(upper_term));
+
+            run_benchmark_tasks(
+                &mut group,
+                &bench_index,
+                query,
+                field_name,
+                range_low,
+                range_high,
+            );
+        }
+
+        group.run();
+    }
+}
+
+/// Run all benchmark tasks for a given range query and field name
+fn run_benchmark_tasks(
+    bench_group: &mut BenchGroup,
+    bench_index: &BenchIndex,
+    query: RangeQuery,
+    field_name: &str,
+    range_low: u64,
+    range_high: u64,
+) {
+    // Test count
+    add_bench_task_count(
+        bench_group,
+        bench_index,
+        query.clone(),
+        "count",
+        field_name,
+        range_low,
+        range_high,
+    );
+
+    // Test top 100 by the field (ascending order)
+    {
+        let collector_name = format!("top100_by_{}_asc", field_name);
+        let field_name_owned = field_name.to_string();
+        add_bench_task_top100_asc(
+            bench_group,
+            bench_index,
+            query.clone(),
+            &collector_name,
+            field_name,
+            range_low,
+            range_high,
+            field_name_owned,
+        );
+    }
+
+    // Test top 100 by the field (descending order)
+    {
+        let collector_name = format!("top100_by_{}_desc", field_name);
+        let field_name_owned = field_name.to_string();
+        add_bench_task_top100_desc(
+            bench_group,
+            bench_index,
+            query,
+            &collector_name,
+            field_name,
+            range_low,
+            range_high,
+            field_name_owned,
+        );
+    }
+}
+
+fn add_bench_task_count(
+    bench_group: &mut BenchGroup,
+    bench_index: &BenchIndex,
+    query: RangeQuery,
+    collector_name: &str,
+    field_name: &str,
+    range_low: u64,
+    range_high: u64,
+) {
+    let task_name = format!(
+        "range_{}_[{} TO {}]_{}",
+        field_name, range_low, range_high, collector_name
+    );
+
+    let search_task = CountSearchTask {
+        searcher: bench_index.searcher.clone(),
+        query,
+    };
+    bench_group.register(task_name, move |_| black_box(search_task.run()));
+}
+
+fn add_bench_task_docset(
+    bench_group: &mut BenchGroup,
+    bench_index: &BenchIndex,
+    query: RangeQuery,
+    collector_name: &str,
+    field_name: &str,
+    range_low: u64,
+    range_high: u64,
+) {
+    let task_name = format!(
+        "range_{}_[{} TO {}]_{}",
+        field_name, range_low, range_high, collector_name
+    );
+
+    let search_task = DocSetSearchTask {
+        searcher: bench_index.searcher.clone(),
+        query,
+    };
+    bench_group.register(task_name, move |_| black_box(search_task.run()));
+}
+
+fn add_bench_task_top100_asc(
+    bench_group: &mut BenchGroup,
+    bench_index: &BenchIndex,
+    query: RangeQuery,
+    collector_name: &str,
+    field_name: &str,
+    range_low: u64,
+    range_high: u64,
+    field_name_owned: String,
+) {
+    let task_name = format!(
+        "range_{}_[{} TO {}]_{}",
+        field_name, range_low, range_high, collector_name
+    );
+
+    let search_task = Top100AscSearchTask {
+        searcher: bench_index.searcher.clone(),
+        query,
+        field_name: field_name_owned,
+    };
+    bench_group.register(task_name, move |_| black_box(search_task.run()));
+}
+
+fn add_bench_task_top100_desc(
+    bench_group: &mut BenchGroup,
+    bench_index: &BenchIndex,
+    query: RangeQuery,
+    collector_name: &str,
+    field_name: &str,
+    range_low: u64,
+    range_high: u64,
+    field_name_owned: String,
+) {
+    let task_name = format!(
+        "range_{}_[{} TO {}]_{}",
+        field_name, range_low, range_high, collector_name
+    );
+
+    let search_task = Top100DescSearchTask {
+        searcher: bench_index.searcher.clone(),
+        query,
+        field_name: field_name_owned,
+    };
+    bench_group.register(task_name, move |_| black_box(search_task.run()));
+}
+
+struct CountSearchTask {
+    searcher: Searcher,
+    query: RangeQuery,
+}
+
+impl CountSearchTask {
+    #[inline(never)]
+    pub fn run(&self) -> usize {
+        self.searcher.search(&self.query, &Count).unwrap()
+    }
+}
+
+struct DocSetSearchTask {
+    searcher: Searcher,
+    query: RangeQuery,
+}
+
+impl DocSetSearchTask {
+    #[inline(never)]
+    pub fn run(&self) -> usize {
+        let result = self.searcher.search(&self.query, &DocSetCollector).unwrap();
+        result.len()
+    }
+}
+
+struct Top100AscSearchTask {
+    searcher: Searcher,
+    query: RangeQuery,
+    field_name: String,
+}
+
+impl Top100AscSearchTask {
+    #[inline(never)]
+    pub fn run(&self) -> usize {
+        let collector =
+            TopDocs::with_limit(100).order_by_fast_field::<u64>(&self.field_name, Order::Asc);
+        let result = self.searcher.search(&self.query, &collector).unwrap();
+        for (_score, doc_address) in &result {
+            let _doc: tantivy::TantivyDocument = self.searcher.doc(*doc_address).unwrap();
+        }
+        result.len()
+    }
+}
+
+struct Top100DescSearchTask {
+    searcher: Searcher,
+    query: RangeQuery,
+    field_name: String,
+}
+
+impl Top100DescSearchTask {
+    #[inline(never)]
+    pub fn run(&self) -> usize {
+        let collector =
+            TopDocs::with_limit(100).order_by_fast_field::<u64>(&self.field_name, Order::Desc);
+        let result = self.searcher.search(&self.query, &collector).unwrap();
+        for (_score, doc_address) in &result {
+            let _doc: tantivy::TantivyDocument = self.searcher.doc(*doc_address).unwrap();
+        }
+        result.len()
+    }
+}

benches/range_query.rs

@@ -0,0 +1,260 @@
+use std::fmt::Display;
+use std::net::Ipv6Addr;
+use std::ops::RangeInclusive;
+
+use binggan::plugins::PeakMemAllocPlugin;
+use binggan::{black_box, BenchRunner, OutputValue, PeakMemAlloc, INSTRUMENTED_SYSTEM};
+use columnar::MonotonicallyMappableToU128;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use tantivy::collector::{Count, TopDocs};
+use tantivy::query::QueryParser;
+use tantivy::schema::*;
+use tantivy::{doc, Index};
+
+#[global_allocator]
+pub static GLOBAL: &PeakMemAlloc<std::alloc::System> = &INSTRUMENTED_SYSTEM;
+
+fn main() {
+    bench_range_query();
+}
+
+fn bench_range_query() {
+    let index = get_index_0_to_100();
+    let mut runner = BenchRunner::new();
+    runner.add_plugin(PeakMemAllocPlugin::new(GLOBAL));
+
+    runner.set_name("range_query on u64");
+    let field_name_and_descr: Vec<_> = vec![
+        ("id", "Single Valued Range Field"),
+        ("ids", "Multi Valued Range Field"),
+    ];
+    let range_num_hits = vec![
+        ("90_percent", get_90_percent()),
+        ("10_percent", get_10_percent()),
+        ("1_percent", get_1_percent()),
+    ];
+
+    test_range(&mut runner, &index, &field_name_and_descr, range_num_hits);
+
+    runner.set_name("range_query on ip");
+    let field_name_and_descr: Vec<_> = vec![
+        ("ip", "Single Valued Range Field"),
+        ("ips", "Multi Valued Range Field"),
+    ];
+    let range_num_hits = vec![
+        ("90_percent", get_90_percent_ip()),
+        ("10_percent", get_10_percent_ip()),
+        ("1_percent", get_1_percent_ip()),
+    ];
+
+    test_range(&mut runner, &index, &field_name_and_descr, range_num_hits);
+}
+
+fn test_range<T: Display>(
+    runner: &mut BenchRunner,
+    index: &Index,
+    field_name_and_descr: &[(&str, &str)],
+    range_num_hits: Vec<(&str, RangeInclusive<T>)>,
+) {
+    for (field, suffix) in field_name_and_descr {
+        let term_num_hits = vec![
+            ("", ""),
+            ("1_percent", "veryfew"),
+            ("10_percent", "few"),
+            ("90_percent", "most"),
+        ];
+        let mut group = runner.new_group();
+        group.set_name(suffix);
+        // all intersect combinations
+        for (range_name, range) in &range_num_hits {
+            for (term_name, term) in &term_num_hits {
+                let index = &index;
+                let test_name = if term_name.is_empty() {
+                    format!("id_range_hit_{}", range_name)
+                } else {
+                    format!(
+                        "id_range_hit_{}_intersect_with_term_{}",
+                        range_name, term_name
+                    )
+                };
+                group.register(test_name, move |_| {
+                    let query = if term_name.is_empty() {
+                        "".to_string()
+                    } else {
+                        format!("AND id_name:{}", term)
+                    };
+                    black_box(execute_query(field, range, &query, index));
+                });
+            }
+        }
+        group.run();
+    }
+}
+
+fn get_index_0_to_100() -> Index {
+    let mut rng = StdRng::from_seed([1u8; 32]);
+    let num_vals = 100_000;
+    let docs: Vec<_> = (0..num_vals)
+        .map(|_i| {
+            let id_name = if rng.gen_bool(0.01) {
+                "veryfew".to_string() // 1%
+            } else if rng.gen_bool(0.1) {
+                "few".to_string() // 9%
+            } else {
+                "most".to_string() // 90%
+            };
+            Doc {
+                id_name,
+                id: rng.gen_range(0..100),
+                // Multiply by 1000, so that we create most buckets in the compact space
+                // The benches depend on this range to select n-percent of elements with the
+                // methods below.
+                ip: Ipv6Addr::from_u128(rng.gen_range(0..100) * 1000),
+            }
+        })
+        .collect();
+
+    create_index_from_docs(&docs)
+}
+
+#[derive(Clone, Debug)]
+pub struct Doc {
+    pub id_name: String,
+    pub id: u64,
+    pub ip: Ipv6Addr,
+}
+
+pub fn create_index_from_docs(docs: &[Doc]) -> Index {
+    let mut schema_builder = Schema::builder();
+    let id_u64_field = schema_builder.add_u64_field("id", INDEXED | STORED | FAST);
+    let ids_u64_field =
+        schema_builder.add_u64_field("ids", NumericOptions::default().set_fast().set_indexed());
+
+    let id_f64_field = schema_builder.add_f64_field("id_f64", INDEXED | STORED | FAST);
+    let ids_f64_field = schema_builder.add_f64_field(
+        "ids_f64",
+        NumericOptions::default().set_fast().set_indexed(),
+    );
+
+    let id_i64_field = schema_builder.add_i64_field("id_i64", INDEXED | STORED | FAST);
+    let ids_i64_field = schema_builder.add_i64_field(
+        "ids_i64",
+        NumericOptions::default().set_fast().set_indexed(),
+    );
+
+    let text_field = schema_builder.add_text_field("id_name", STRING | STORED);
+    let text_field2 = schema_builder.add_text_field("id_name_fast", STRING | STORED | FAST);
+
+    let ip_field = schema_builder.add_ip_addr_field("ip", FAST);
+    let ips_field = schema_builder.add_ip_addr_field("ips", FAST);
+
+    let schema = schema_builder.build();
+
+    let index = Index::create_in_ram(schema);
+
+    {
+        let mut index_writer = index.writer_with_num_threads(1, 50_000_000).unwrap();
+        for doc in docs.iter() {
+            index_writer
+                .add_document(doc!(
+                    ids_i64_field => doc.id as i64,
+                    ids_i64_field => doc.id as i64,
+                    ids_f64_field => doc.id as f64,
+                    ids_f64_field => doc.id as f64,
+                    ids_u64_field => doc.id,
+                    ids_u64_field => doc.id,
+                    id_u64_field => doc.id,
+                    id_f64_field => doc.id as f64,
+                    id_i64_field => doc.id as i64,
+                    text_field => doc.id_name.to_string(),
+                    text_field2 => doc.id_name.to_string(),
+                    ips_field => doc.ip,
+                    ips_field => doc.ip,
+                    ip_field => doc.ip,
+                ))
+                .unwrap();
+        }
+
+        index_writer.commit().unwrap();
+    }
+    index
+}
+
+fn get_90_percent() -> RangeInclusive<u64> {
+    0..=90
+}
+
+fn get_10_percent() -> RangeInclusive<u64> {
+    0..=10
+}
+
+fn get_1_percent() -> RangeInclusive<u64> {
+    10..=10
+}
+
+fn get_90_percent_ip() -> RangeInclusive<Ipv6Addr> {
+    let start = Ipv6Addr::from_u128(0);
+    let end = Ipv6Addr::from_u128(90 * 1000);
+    start..=end
+}
+
+fn get_10_percent_ip() -> RangeInclusive<Ipv6Addr> {
+    let start = Ipv6Addr::from_u128(0);
+    let end = Ipv6Addr::from_u128(10 * 1000);
+    start..=end
+}
+
+fn get_1_percent_ip() -> RangeInclusive<Ipv6Addr> {
+    let start = Ipv6Addr::from_u128(10 * 1000);
+    let end = Ipv6Addr::from_u128(10 * 1000);
+    start..=end
+}
+
+struct NumHits {
+    count: usize,
+}
+impl OutputValue for NumHits {
+    fn column_title() -> &'static str {
+        "NumHits"
+    }
+    fn format(&self) -> Option<String> {
+        Some(self.count.to_string())
+    }
+}
+
+fn execute_query<T: Display>(
+    field: &str,
+    id_range: &RangeInclusive<T>,
+    suffix: &str,
+    index: &Index,
+) -> NumHits {
+    let gen_query_inclusive = |from: &T, to: &T| {
+        format!(
+            "{}:[{} TO {}] {}",
+            field,
+            &from.to_string(),
+            &to.to_string(),
+            suffix
+        )
+    };
+
+    let query = gen_query_inclusive(id_range.start(), id_range.end());
+    execute_query_(&query, index)
+}
+
+fn execute_query_(query: &str, index: &Index) -> NumHits {
+    let query_from_text = |text: &str| {
+        QueryParser::for_index(index, vec![])
+            .parse_query(text)
+            .unwrap()
+    };
+    let query = query_from_text(query);
+    let reader = index.reader().unwrap();
+    let searcher = reader.searcher();
+    let num_hits = searcher
+        .search(&query, &(TopDocs::with_limit(10).order_by_score(), Count))
+        .unwrap()
+        .1;
+    NumHits { count: num_hits }
+}

columnar/src/column_values/u64_based/bitpacked.rs

@@ -41,12 +41,6 @@ fn transform_range_before_linear_transformation(
     if range.is_empty() {
         return None;
     }
-    if stats.min_value > *range.end() {
-        return None;
-    }
-    if stats.max_value < *range.start() {
-        return None;
-    }
     let shifted_range =
         range.start().saturating_sub(stats.min_value)..=range.end().saturating_sub(stats.min_value);
     let start_before_gcd_multiplication: u64 = div_ceil(*shifted_range.start(), stats.gcd);

src/aggregation/accessor_helpers.rs

@@ -55,44 +55,22 @@ pub(crate) fn get_numeric_or_date_column_types() -> &'static [ColumnType] {
     ]
 }
 
-/// Get fast field reader or return an error if the field doesn't exist.
+/// Get fast field reader or empty as default.
 pub(crate) fn get_ff_reader(
     reader: &SegmentReader,
     field_name: &str,
     allowed_column_types: Option<&[ColumnType]>,
 ) -> crate::Result<(columnar::Column<u64>, ColumnType)> {
     let ff_fields = reader.fast_fields();
-    let ff_field_with_type = ff_fields.u64_lenient_for_type(allowed_column_types, field_name)?;
-
-    match ff_field_with_type {
-        Some(field) => Ok(field),
-        None => {
-            // Check if the field exists in the schema but is not a fast field
-            let schema = reader.schema();
-            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
-                    )));
-                }
-            }
-
-            // Field doesn't exist at all or has no values in this segment
-            // Check if it exists in schema to provide a better error message
-            if schema.find_field(field_name).is_none() {
-                return Err(crate::TantivyError::FieldNotFound(field_name.to_string()));
-            }
-
-            // Field exists in schema and is a fast field, but has no values in this segment
-            // This is acceptable - return an empty column
-            Ok((
+    let ff_field_with_type = ff_fields
+        .u64_lenient_for_type(allowed_column_types, field_name)?
+        .unwrap_or_else(|| {
+            (
                 Column::build_empty_column(reader.num_docs()),
                 ColumnType::U64,
-            ))
-        }
-    }
+            )
+        });
+    Ok(ff_field_with_type)
 }
 
 pub(crate) fn get_dynamic_columns(
@@ -111,7 +89,6 @@ pub(crate) fn get_dynamic_columns(
 /// Get all fast field reader or empty as default.
 ///
 /// Is guaranteed to return at least one column.
-/// Returns an error if the field doesn't exist in the schema or is not a fast field.
 pub(crate) fn get_all_ff_reader_or_empty(
     reader: &SegmentReader,
     field_name: &str,
@@ -121,25 +98,7 @@ pub(crate) fn get_all_ff_reader_or_empty(
     let ff_fields = reader.fast_fields();
     let mut ff_field_with_type =
         ff_fields.u64_lenient_for_type_all(allowed_column_types, field_name)?;
-
     if ff_field_with_type.is_empty() {
-        // Check if the field exists in the schema but is not a fast field
-        let schema = reader.schema();
-        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 {
-            // Field doesn't exist in the schema at all
-            return Err(crate::TantivyError::FieldNotFound(field_name.to_string()));
-        }
-
-        // Field exists in schema and is a fast field, but has no values in this segment
-        // This is acceptable - return an empty column
         ff_field_with_type.push((Column::build_empty_column(reader.num_docs()), fallback_type));
     }
     Ok(ff_field_with_type)

src/aggregation/agg_data.rs

@@ -1057,7 +1057,7 @@ mod tests {
             "avg": {"field": "score"}
         }));
         let terms_string_with_child = agg_from_json(json!({
-            "terms": {"field": "text"},
+            "terms": {"field": "string_id"},
             "aggs": {
                 "histo": {"histogram": {"field": "score", "interval": 10.0}}
             }

src/aggregation/agg_tests.rs

@@ -1005,123 +1005,3 @@ fn test_aggregation_on_json_object_mixed_numerical_segments() {
         )
     );
 }
-
-#[test]
-fn test_aggregation_invalid_field_returns_error() {
-    // Test that aggregations return an error when given an invalid field name
-    let index = get_test_index_2_segments(false).unwrap();
-    let reader = index.reader().unwrap();
-    let searcher = reader.searcher();
-
-    // Test with a field that doesn't exist at all
-    let agg_req_str = r#"
-    {
-        "date_histogram_test": {
-            "date_histogram": {
-                "field": "not_valid_field",
-                "fixed_interval": "30d"
-            }
-        }
-    }"#;
-    let agg: Aggregations = serde_json::from_str(agg_req_str).unwrap();
-    let collector = get_collector(agg);
-    let result = searcher.search(&AllQuery, &collector);
-
-    assert!(result.is_err());
-    match result {
-        Err(crate::TantivyError::FieldNotFound(field_name)) => {
-            assert_eq!(field_name, "not_valid_field");
-        }
-        _ => panic!("Expected FieldNotFound error, got: {:?}", result),
-    }
-
-    // Test with histogram aggregation on invalid field
-    let agg_req_str = r#"
-    {
-        "histogram_test": {
-            "histogram": {
-                "field": "invalid_histogram_field",
-                "interval": 10.0
-            }
-        }
-    }"#;
-    let agg: Aggregations = serde_json::from_str(agg_req_str).unwrap();
-    let collector = get_collector(agg);
-    let result = searcher.search(&AllQuery, &collector);
-
-    assert!(result.is_err());
-    match result {
-        Err(crate::TantivyError::FieldNotFound(field_name)) => {
-            assert_eq!(field_name, "invalid_histogram_field");
-        }
-        _ => panic!("Expected FieldNotFound error, got: {:?}", result),
-    }
-
-    // Test with terms aggregation on invalid field
-    let agg_req_str = r#"
-    {
-        "terms_test": {
-            "terms": {
-                "field": "invalid_terms_field"
-            }
-        }
-    }"#;
-    let agg: Aggregations = serde_json::from_str(agg_req_str).unwrap();
-    let collector = get_collector(agg);
-    let result = searcher.search(&AllQuery, &collector);
-
-    assert!(result.is_err());
-    match result {
-        Err(crate::TantivyError::FieldNotFound(field_name)) => {
-            assert_eq!(field_name, "invalid_terms_field");
-        }
-        _ => panic!("Expected FieldNotFound error, got: {:?}", result),
-    }
-
-    // Test with avg metric aggregation on invalid field
-    let agg_req_str = r#"
-    {
-        "avg_test": {
-            "avg": {
-                "field": "invalid_avg_field"
-            }
-        }
-    }"#;
-    let agg: Aggregations = serde_json::from_str(agg_req_str).unwrap();
-    let collector = get_collector(agg);
-    let result = searcher.search(&AllQuery, &collector);
-
-    assert!(result.is_err());
-    match result {
-        Err(crate::TantivyError::FieldNotFound(field_name)) => {
-            assert_eq!(field_name, "invalid_avg_field");
-        }
-        _ => panic!("Expected FieldNotFound error, got: {:?}", result),
-    }
-
-    // Test with range aggregation on invalid field
-    let agg_req_str = r#"
-    {
-        "range_test": {
-            "range": {
-                "field": "invalid_range_field",
-                "ranges": [
-                    { "to": 10.0 },
-                    { "from": 10.0, "to": 20.0 },
-                    { "from": 20.0 }
-                ]
-            }
-        }
-    }"#;
-    let agg: Aggregations = serde_json::from_str(agg_req_str).unwrap();
-    let collector = get_collector(agg);
-    let result = searcher.search(&AllQuery, &collector);
-
-    assert!(result.is_err());
-    match result {
-        Err(crate::TantivyError::FieldNotFound(field_name)) => {
-            assert_eq!(field_name, "invalid_range_field");
-        }
-        _ => panic!("Expected FieldNotFound error, got: {:?}", result),
-    }
-}

src/aggregation/bucket/term_missing_agg.rs

@@ -255,7 +255,6 @@ mod tests {
     fn terms_aggregation_missing_mult_seg_empty() -> crate::Result<()> {
         let mut schema_builder = Schema::builder();
         let score = schema_builder.add_f64_field("score", FAST);
-        schema_builder.add_json_field("json", FAST);
         let schema = schema_builder.build();
         let index = Index::create_in_ram(schema);
         let mut index_writer: IndexWriter = index.writer_for_tests().unwrap();
@@ -303,7 +302,6 @@ mod tests {
     fn terms_aggregation_missing_single_seg_empty() -> crate::Result<()> {
         let mut schema_builder = Schema::builder();
         let score = schema_builder.add_f64_field("score", FAST);
-        schema_builder.add_json_field("json", FAST);
         let schema = schema_builder.build();
         let index = Index::create_in_ram(schema);
         let mut index_writer: IndexWriter = index.writer_for_tests().unwrap();

src/collector/sort_key/mod.rs

@@ -1,25 +1,48 @@
 mod order;
+mod sort_by_erased_type;
 mod sort_by_score;
 mod sort_by_static_fast_value;
 mod sort_by_string;
 mod sort_key_computer;
 
 pub use order::*;
+pub use sort_by_erased_type::SortByErasedType;
 pub use sort_by_score::SortBySimilarityScore;
 pub use sort_by_static_fast_value::SortByStaticFastValue;
 pub use sort_by_string::SortByString;
 pub use sort_key_computer::{SegmentSortKeyComputer, SortKeyComputer};
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
+
+    // By spec, regardless of whether ascending or descending order was requested, in presence of a
+    // tie, we sort by ascending doc id/doc address.
+    pub(crate) fn sort_hits<TSortKey: Ord, D: Ord>(
+        hits: &mut [ComparableDoc<TSortKey, D>],
+        order: Order,
+    ) {
+        if order.is_asc() {
+            hits.sort_by(|l, r| l.sort_key.cmp(&r.sort_key).then(l.doc.cmp(&r.doc)));
+        } else {
+            hits.sort_by(|l, r| {
+                l.sort_key
+                    .cmp(&r.sort_key)
+                    .reverse() // This is descending
+                    .then(l.doc.cmp(&r.doc))
+            });
+        }
+    }
+
     use std::collections::HashMap;
     use std::ops::Range;
 
-    use crate::collector::sort_key::{SortBySimilarityScore, SortByStaticFastValue, SortByString};
+    use crate::collector::sort_key::{
+        SortByErasedType, SortBySimilarityScore, SortByStaticFastValue, SortByString,
+    };
     use crate::collector::{ComparableDoc, DocSetCollector, TopDocs};
     use crate::indexer::NoMergePolicy;
     use crate::query::{AllQuery, QueryParser};
-    use crate::schema::{Schema, FAST, TEXT};
+    use crate::schema::{OwnedValue, Schema, FAST, TEXT};
     use crate::{DocAddress, Document, Index, Order, Score, Searcher};
 
     fn make_index() -> crate::Result<Index> {
@@ -294,11 +317,9 @@ mod tests {
                 (SortBySimilarityScore, score_order),
                 (SortByString::for_field("city"), city_order),
             ));
-            Ok(searcher
-                .search(&AllQuery, &top_collector)?
-                .into_iter()
-                .map(|(f, doc)| (f, ids[&doc]))
-                .collect())
+            let results: Vec<((Score, Option<String>), DocAddress)> =
+                searcher.search(&AllQuery, &top_collector)?;
+            Ok(results.into_iter().map(|(f, doc)| (f, ids[&doc])).collect())
         }
 
         assert_eq!(
@@ -323,6 +344,51 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_order_by_score_then_owned_value() -> crate::Result<()> {
+        let index = make_index()?;
+
+        type SortKey = (Score, OwnedValue);
+
+        fn query(
+            index: &Index,
+            score_order: Order,
+            city_order: Order,
+        ) -> crate::Result<Vec<(SortKey, u64)>> {
+            let searcher = index.reader()?.searcher();
+            let ids = id_mapping(&searcher);
+
+            let top_collector = TopDocs::with_limit(4).order_by::<(Score, OwnedValue)>((
+                (SortBySimilarityScore, score_order),
+                (SortByErasedType::for_field("city"), city_order),
+            ));
+            let results: Vec<((Score, OwnedValue), DocAddress)> =
+                searcher.search(&AllQuery, &top_collector)?;
+            Ok(results.into_iter().map(|(f, doc)| (f, ids[&doc])).collect())
+        }
+
+        assert_eq!(
+            &query(&index, Order::Asc, Order::Asc)?,
+            &[
+                ((1.0, OwnedValue::Str("austin".to_owned())), 0),
+                ((1.0, OwnedValue::Str("greenville".to_owned())), 1),
+                ((1.0, OwnedValue::Str("tokyo".to_owned())), 2),
+                ((1.0, OwnedValue::Null), 3),
+            ]
+        );
+
+        assert_eq!(
+            &query(&index, Order::Asc, Order::Desc)?,
+            &[
+                ((1.0, OwnedValue::Str("tokyo".to_owned())), 2),
+                ((1.0, OwnedValue::Str("greenville".to_owned())), 1),
+                ((1.0, OwnedValue::Str("austin".to_owned())), 0),
+                ((1.0, OwnedValue::Null), 3),
+            ]
+        );
+        Ok(())
+    }
+
     use proptest::prelude::*;
 
     proptest! {
@@ -372,15 +438,10 @@ mod tests {
 
             // Using the TopDocs collector should always be equivalent to sorting, skipping the
             // offset, and then taking the limit.
-            let sorted_docs: Vec<_> = if order.is_desc() {
-                let mut comparable_docs: Vec<ComparableDoc<_, _, true>> =
+            let sorted_docs: Vec<_> = {
+                let mut comparable_docs: Vec<ComparableDoc<_, _>> =
                     all_results.into_iter().map(|(sort_key, doc)| ComparableDoc { sort_key, doc}).collect();
-                comparable_docs.sort();
-                comparable_docs.into_iter().map(|cd| (cd.sort_key, cd.doc)).collect()
-            } else {
-                let mut comparable_docs: Vec<ComparableDoc<_, _, false>> =
-                    all_results.into_iter().map(|(sort_key, doc)| ComparableDoc { sort_key, doc}).collect();
-                comparable_docs.sort();
+                sort_hits(&mut comparable_docs, order);
                 comparable_docs.into_iter().map(|cd| (cd.sort_key, cd.doc)).collect()
             };
             let expected_docs = sorted_docs.into_iter().skip(offset).take(limit).collect::<Vec<_>>();

src/collector/sort_key/order.rs

@@ -1,36 +1,116 @@
 use std::cmp::Ordering;
 
+use columnar::MonotonicallyMappableToU64;
 use serde::{Deserialize, Serialize};
 
 use crate::collector::{SegmentSortKeyComputer, SortKeyComputer};
-use crate::schema::Schema;
+use crate::schema::{OwnedValue, Schema};
 use crate::{DocId, Order, Score};
 
+fn compare_owned_value<const NULLS_FIRST: bool>(lhs: &OwnedValue, rhs: &OwnedValue) -> Ordering {
+    match (lhs, rhs) {
+        (OwnedValue::Null, OwnedValue::Null) => Ordering::Equal,
+        (OwnedValue::Null, _) => {
+            if NULLS_FIRST {
+                Ordering::Less
+            } else {
+                Ordering::Greater
+            }
+        }
+        (_, OwnedValue::Null) => {
+            if NULLS_FIRST {
+                Ordering::Greater
+            } else {
+                Ordering::Less
+            }
+        }
+        (OwnedValue::Str(a), OwnedValue::Str(b)) => a.cmp(b),
+        (OwnedValue::PreTokStr(a), OwnedValue::PreTokStr(b)) => a.cmp(b),
+        (OwnedValue::U64(a), OwnedValue::U64(b)) => a.cmp(b),
+        (OwnedValue::I64(a), OwnedValue::I64(b)) => a.cmp(b),
+        (OwnedValue::F64(a), OwnedValue::F64(b)) => a.to_u64().cmp(&b.to_u64()),
+        (OwnedValue::Bool(a), OwnedValue::Bool(b)) => a.cmp(b),
+        (OwnedValue::Date(a), OwnedValue::Date(b)) => a.cmp(b),
+        (OwnedValue::Facet(a), OwnedValue::Facet(b)) => a.cmp(b),
+        (OwnedValue::Bytes(a), OwnedValue::Bytes(b)) => a.cmp(b),
+        (OwnedValue::IpAddr(a), OwnedValue::IpAddr(b)) => a.cmp(b),
+        (OwnedValue::U64(a), OwnedValue::I64(b)) => {
+            if *b < 0 {
+                Ordering::Greater
+            } else {
+                a.cmp(&(*b as u64))
+            }
+        }
+        (OwnedValue::I64(a), OwnedValue::U64(b)) => {
+            if *a < 0 {
+                Ordering::Less
+            } else {
+                (*a as u64).cmp(b)
+            }
+        }
+        (OwnedValue::U64(a), OwnedValue::F64(b)) => (*a as f64).to_u64().cmp(&b.to_u64()),
+        (OwnedValue::F64(a), OwnedValue::U64(b)) => a.to_u64().cmp(&(*b as f64).to_u64()),
+        (OwnedValue::I64(a), OwnedValue::F64(b)) => (*a as f64).to_u64().cmp(&b.to_u64()),
+        (OwnedValue::F64(a), OwnedValue::I64(b)) => a.to_u64().cmp(&(*b as f64).to_u64()),
+        (a, b) => {
+            let ord = a.discriminant_value().cmp(&b.discriminant_value());
+            // If the discriminant is equal, it's because a new type was added, but hasn't been
+            // included in this `match` statement.
+            assert!(
+                ord != Ordering::Equal,
+                "Unimplemented comparison for type of {a:?}, {b:?}"
+            );
+            ord
+        }
+    }
+}
+
 /// Comparator trait defining the order in which documents should be ordered.
 pub trait Comparator<T>: Send + Sync + std::fmt::Debug + Default {
     /// Return the order between two values.
     fn compare(&self, lhs: &T, rhs: &T) -> Ordering;
 }
 
-/// With the natural comparator, the top k collector will return
-/// the top documents in decreasing order.
+/// Compare values naturally (e.g. 1 < 2).
+///
+/// When used with `TopDocs`, which reverses the order, this results in a
+/// "Descending" sort (Greatest values first).
+///
+/// `None` (or Null for `OwnedValue`) values are considered to be smaller than any other value,
+/// and will therefore appear last in a descending sort (e.g. `[Some(20), Some(10), None]`).
 #[derive(Debug, Copy, Clone, Default, Serialize, Deserialize)]
 pub struct NaturalComparator;
 
 impl<T: PartialOrd> Comparator<T> for NaturalComparator {
     #[inline(always)]
     fn compare(&self, lhs: &T, rhs: &T) -> Ordering {
-        lhs.partial_cmp(rhs).unwrap()
+        lhs.partial_cmp(rhs).unwrap_or(Ordering::Equal)
     }
 }
 
-/// Sorts document in reverse order.
+/// A (partial) implementation of comparison for OwnedValue.
 ///
-/// If the sort key is None, it will considered as the lowest value, and will therefore appear
-/// first.
+/// Intended for use within columns of homogenous types, and so will panic for OwnedValues with
+/// mismatched types. The one exception is Null, for which we do define all comparisons.
+impl Comparator<OwnedValue> for NaturalComparator {
+    #[inline(always)]
+    fn compare(&self, lhs: &OwnedValue, rhs: &OwnedValue) -> Ordering {
+        compare_owned_value::</* NULLS_FIRST= */ true>(lhs, rhs)
+    }
+}
+
+/// Compare values in reverse (e.g. 2 < 1).
+///
+/// When used with `TopDocs`, which reverses the order, this results in an
+/// "Ascending" sort (Smallest values first).
+///
+/// `None` is considered smaller than `Some` in the underlying comparator, but because the
+/// comparison is reversed, `None` is effectively treated as the lowest value in the resulting
+/// Ascending sort (e.g. `[None, Some(10), Some(20)]`).
 ///
 /// The ReverseComparator does not necessarily imply that the sort order is reversed compared
-/// to the NaturalComparator. In presence of a tie, both version will retain the higher doc ids.
+/// to the NaturalComparator. In presence of a tie on the sort key, documents will always be
+/// sorted by ascending `DocId`/`DocAddress` in TopN results, regardless of the sort key's order.
 #[derive(Debug, Copy, Clone, Default, Serialize, Deserialize)]
 pub struct ReverseComparator;
 
@@ -43,11 +123,15 @@ where NaturalComparator: Comparator<T>
     }
 }
 
-/// Sorts document in reverse order, but considers None as having the lowest value.
+/// Compare values in reverse, but treating `None` as lower than `Some`.
+///
+/// When used with `TopDocs`, which reverses the order, this results in an
+/// "Ascending" sort (Smallest values first), but with `None` values appearing last
+/// (e.g. `[Some(10), Some(20), None]`).
 ///
 /// This is usually what is wanted when sorting by a field in an ascending order.
-/// For instance, in a e-commerce website, if I sort by price ascending, I most likely want the
-/// cheapest items first, and the items without a price at last.
+/// For instance, in an e-commerce website, if sorting by price ascending,
+/// the cheapest items would appear first, and items without a price would appear last.
 #[derive(Debug, Copy, Clone, Default)]
 pub struct ReverseNoneIsLowerComparator;
 
@@ -107,6 +191,84 @@ impl Comparator<String> for ReverseNoneIsLowerComparator {
     }
 }
 
+impl Comparator<OwnedValue> for ReverseNoneIsLowerComparator {
+    #[inline(always)]
+    fn compare(&self, lhs: &OwnedValue, rhs: &OwnedValue) -> Ordering {
+        compare_owned_value::</* NULLS_FIRST= */ false>(rhs, lhs)
+    }
+}
+
+/// Compare values naturally, but treating `None` as higher than `Some`.
+///
+/// When used with `TopDocs`, which reverses the order, this results in a
+/// "Descending" sort (Greatest values first), but with `None` values appearing first
+/// (e.g. `[None, Some(20), Some(10)]`).
+#[derive(Debug, Copy, Clone, Default, Serialize, Deserialize)]
+pub struct NaturalNoneIsHigherComparator;
+
+impl<T> Comparator<Option<T>> for NaturalNoneIsHigherComparator
+where NaturalComparator: Comparator<T>
+{
+    #[inline(always)]
+    fn compare(&self, lhs_opt: &Option<T>, rhs_opt: &Option<T>) -> Ordering {
+        match (lhs_opt, rhs_opt) {
+            (None, None) => Ordering::Equal,
+            (None, Some(_)) => Ordering::Greater,
+            (Some(_), None) => Ordering::Less,
+            (Some(lhs), Some(rhs)) => NaturalComparator.compare(lhs, rhs),
+        }
+    }
+}
+
+impl Comparator<u32> for NaturalNoneIsHigherComparator {
+    #[inline(always)]
+    fn compare(&self, lhs: &u32, rhs: &u32) -> Ordering {
+        NaturalComparator.compare(lhs, rhs)
+    }
+}
+
+impl Comparator<u64> for NaturalNoneIsHigherComparator {
+    #[inline(always)]
+    fn compare(&self, lhs: &u64, rhs: &u64) -> Ordering {
+        NaturalComparator.compare(lhs, rhs)
+    }
+}
+
+impl Comparator<f64> for NaturalNoneIsHigherComparator {
+    #[inline(always)]
+    fn compare(&self, lhs: &f64, rhs: &f64) -> Ordering {
+        NaturalComparator.compare(lhs, rhs)
+    }
+}
+
+impl Comparator<f32> for NaturalNoneIsHigherComparator {
+    #[inline(always)]
+    fn compare(&self, lhs: &f32, rhs: &f32) -> Ordering {
+        NaturalComparator.compare(lhs, rhs)
+    }
+}
+
+impl Comparator<i64> for NaturalNoneIsHigherComparator {
+    #[inline(always)]
+    fn compare(&self, lhs: &i64, rhs: &i64) -> Ordering {
+        NaturalComparator.compare(lhs, rhs)
+    }
+}
+
+impl Comparator<String> for NaturalNoneIsHigherComparator {
+    #[inline(always)]
+    fn compare(&self, lhs: &String, rhs: &String) -> Ordering {
+        NaturalComparator.compare(lhs, rhs)
+    }
+}
+
+impl Comparator<OwnedValue> for NaturalNoneIsHigherComparator {
+    #[inline(always)]
+    fn compare(&self, lhs: &OwnedValue, rhs: &OwnedValue) -> Ordering {
+        compare_owned_value::</* NULLS_FIRST= */ false>(lhs, rhs)
+    }
+}
+
 /// An enum representing the different sort orders.
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Default)]
 pub enum ComparatorEnum {
@@ -115,8 +277,10 @@ pub enum ComparatorEnum {
     Natural,
     /// Reverse order (See [ReverseComparator])
     Reverse,
-    /// Reverse order by treating None as the lowest value.(See [ReverseNoneLowerComparator])
+    /// Reverse order by treating None as the lowest value. (See [ReverseNoneLowerComparator])
     ReverseNoneLower,
+    /// Natural order but treating None as the highest value. (See [NaturalNoneIsHigherComparator])
+    NaturalNoneHigher,
 }
 
 impl From<Order> for ComparatorEnum {
@@ -133,6 +297,7 @@ where
     ReverseNoneIsLowerComparator: Comparator<T>,
     NaturalComparator: Comparator<T>,
     ReverseComparator: Comparator<T>,
+    NaturalNoneIsHigherComparator: Comparator<T>,
 {
     #[inline(always)]
     fn compare(&self, lhs: &T, rhs: &T) -> Ordering {
@@ -140,6 +305,7 @@ where
             ComparatorEnum::Natural => NaturalComparator.compare(lhs, rhs),
             ComparatorEnum::Reverse => ReverseComparator.compare(lhs, rhs),
             ComparatorEnum::ReverseNoneLower => ReverseNoneIsLowerComparator.compare(lhs, rhs),
+            ComparatorEnum::NaturalNoneHigher => NaturalNoneIsHigherComparator.compare(lhs, rhs),
         }
     }
 }
@@ -322,11 +488,12 @@ impl<TSegmentSortKeyComputer, TSegmentSortKey, TComparator> SegmentSortKeyComput
     for SegmentSortKeyComputerWithComparator<TSegmentSortKeyComputer, TComparator>
 where
     TSegmentSortKeyComputer: SegmentSortKeyComputer<SegmentSortKey = TSegmentSortKey>,
-    TSegmentSortKey: PartialOrd + Clone + 'static + Sync + Send,
+    TSegmentSortKey: Clone + 'static + Sync + Send,
     TComparator: Comparator<TSegmentSortKey> + 'static + Sync + Send,
 {
     type SortKey = TSegmentSortKeyComputer::SortKey;
     type SegmentSortKey = TSegmentSortKey;
+    type SegmentComparator = TComparator;
 
     fn segment_sort_key(&mut self, doc: DocId, score: Score) -> Self::SegmentSortKey {
         self.segment_sort_key_computer.segment_sort_key(doc, score)
@@ -346,3 +513,55 @@ where
             .convert_segment_sort_key(sort_key)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::schema::OwnedValue;
+
+    #[test]
+    fn test_natural_none_is_higher() {
+        let comp = NaturalNoneIsHigherComparator;
+        let null = None;
+        let v1 = Some(1_u64);
+        let v2 = Some(2_u64);
+
+        // NaturalNoneIsGreaterComparator logic:
+        // 1. Delegates to NaturalComparator for non-nulls.
+        // NaturalComparator compare(2, 1) -> 2.cmp(1) -> Greater.
+        assert_eq!(comp.compare(&v2, &v1), Ordering::Greater);
+
+        // 2. Treats None (Null) as Greater than any value.
+        // compare(None, Some(2)) should be Greater.
+        assert_eq!(comp.compare(&null, &v2), Ordering::Greater);
+
+        // compare(Some(1), None) should be Less.
+        assert_eq!(comp.compare(&v1, &null), Ordering::Less);
+
+        // compare(None, None) should be Equal.
+        assert_eq!(comp.compare(&null, &null), Ordering::Equal);
+    }
+
+    #[test]
+    fn test_mixed_ownedvalue_compare() {
+        let u = OwnedValue::U64(10);
+        let i = OwnedValue::I64(10);
+        let f = OwnedValue::F64(10.0);
+
+        let nc = NaturalComparator;
+        assert_eq!(nc.compare(&u, &i), Ordering::Equal);
+        assert_eq!(nc.compare(&u, &f), Ordering::Equal);
+        assert_eq!(nc.compare(&i, &f), Ordering::Equal);
+
+        let u2 = OwnedValue::U64(11);
+        assert_eq!(nc.compare(&u2, &f), Ordering::Greater);
+
+        let s = OwnedValue::Str("a".to_string());
+        // Str < U64
+        assert_eq!(nc.compare(&s, &u), Ordering::Less);
+        // Str < I64
+        assert_eq!(nc.compare(&s, &i), Ordering::Less);
+        // Str < F64
+        assert_eq!(nc.compare(&s, &f), Ordering::Less);
+    }
+}

src/collector/sort_key/sort_by_erased_type.rs

@@ -0,0 +1,361 @@
+use columnar::{ColumnType, MonotonicallyMappableToU64};
+
+use crate::collector::sort_key::{
+    NaturalComparator, SortBySimilarityScore, SortByStaticFastValue, SortByString,
+};
+use crate::collector::{SegmentSortKeyComputer, SortKeyComputer};
+use crate::fastfield::FastFieldNotAvailableError;
+use crate::schema::OwnedValue;
+use crate::{DateTime, DocId, Score};
+
+/// Sort by the boxed / OwnedValue representation of either a fast field, or of the score.
+///
+/// Using the OwnedValue representation allows for type erasure, and can be useful when sort orders
+/// are not known until runtime. But it comes with a performance cost: wherever possible, prefer to
+/// use a SortKeyComputer implementation with a known-type at compile time.
+#[derive(Debug, Clone)]
+pub enum SortByErasedType {
+    /// Sort by a fast field
+    Field(String),
+    /// Sort by score
+    Score,
+}
+
+impl SortByErasedType {
+    /// Creates a new sort key computer which will sort by the given fast field column, with type
+    /// erasure.
+    pub fn for_field(column_name: impl ToString) -> Self {
+        Self::Field(column_name.to_string())
+    }
+
+    /// Creates a new sort key computer which will sort by score, with type erasure.
+    pub fn for_score() -> Self {
+        Self::Score
+    }
+}
+
+trait ErasedSegmentSortKeyComputer: Send + Sync {
+    fn segment_sort_key(&mut self, doc: DocId, score: Score) -> Option<u64>;
+    fn convert_segment_sort_key(&self, sort_key: Option<u64>) -> OwnedValue;
+}
+
+struct ErasedSegmentSortKeyComputerWrapper<C, F> {
+    inner: C,
+    converter: F,
+}
+
+impl<C, F> ErasedSegmentSortKeyComputer for ErasedSegmentSortKeyComputerWrapper<C, F>
+where
+    C: SegmentSortKeyComputer<SegmentSortKey = Option<u64>> + Send + Sync,
+    F: Fn(C::SortKey) -> OwnedValue + Send + Sync + 'static,
+{
+    fn segment_sort_key(&mut self, doc: DocId, score: Score) -> Option<u64> {
+        self.inner.segment_sort_key(doc, score)
+    }
+
+    fn convert_segment_sort_key(&self, sort_key: Option<u64>) -> OwnedValue {
+        let val = self.inner.convert_segment_sort_key(sort_key);
+        (self.converter)(val)
+    }
+}
+
+struct ScoreSegmentSortKeyComputer {
+    segment_computer: SortBySimilarityScore,
+}
+
+impl ErasedSegmentSortKeyComputer for ScoreSegmentSortKeyComputer {
+    fn segment_sort_key(&mut self, doc: DocId, score: Score) -> Option<u64> {
+        let score_value: f64 = self.segment_computer.segment_sort_key(doc, score).into();
+        Some(score_value.to_u64())
+    }
+
+    fn convert_segment_sort_key(&self, sort_key: Option<u64>) -> OwnedValue {
+        let score_value: u64 = sort_key.expect("This implementation always produces a score.");
+        OwnedValue::F64(f64::from_u64(score_value))
+    }
+}
+
+impl SortKeyComputer for SortByErasedType {
+    type SortKey = OwnedValue;
+    type Child = ErasedColumnSegmentSortKeyComputer;
+    type Comparator = NaturalComparator;
+
+    fn requires_scoring(&self) -> bool {
+        matches!(self, Self::Score)
+    }
+
+    fn segment_sort_key_computer(
+        &self,
+        segment_reader: &crate::SegmentReader,
+    ) -> crate::Result<Self::Child> {
+        let inner: Box<dyn ErasedSegmentSortKeyComputer> = match self {
+            Self::Field(column_name) => {
+                let fast_fields = segment_reader.fast_fields();
+                // TODO: We currently double-open the column to avoid relying on the implementation
+                // details of `SortByString` or `SortByStaticFastValue`. Once
+                // https://github.com/quickwit-oss/tantivy/issues/2776 is resolved, we should
+                // consider directly constructing the appropriate `SegmentSortKeyComputer` type for
+                // the column that we open here.
+                let (_column, column_type) =
+                    fast_fields.u64_lenient(column_name)?.ok_or_else(|| {
+                        FastFieldNotAvailableError {
+                            field_name: column_name.to_owned(),
+                        }
+                    })?;
+
+                match column_type {
+                    ColumnType::Str => {
+                        let computer = SortByString::for_field(column_name);
+                        let inner = computer.segment_sort_key_computer(segment_reader)?;
+                        Box::new(ErasedSegmentSortKeyComputerWrapper {
+                            inner,
+                            converter: |val: Option<String>| {
+                                val.map(OwnedValue::Str).unwrap_or(OwnedValue::Null)
+                            },
+                        })
+                    }
+                    ColumnType::U64 => {
+                        let computer = SortByStaticFastValue::<u64>::for_field(column_name);
+                        let inner = computer.segment_sort_key_computer(segment_reader)?;
+                        Box::new(ErasedSegmentSortKeyComputerWrapper {
+                            inner,
+                            converter: |val: Option<u64>| {
+                                val.map(OwnedValue::U64).unwrap_or(OwnedValue::Null)
+                            },
+                        })
+                    }
+                    ColumnType::I64 => {
+                        let computer = SortByStaticFastValue::<i64>::for_field(column_name);
+                        let inner = computer.segment_sort_key_computer(segment_reader)?;
+                        Box::new(ErasedSegmentSortKeyComputerWrapper {
+                            inner,
+                            converter: |val: Option<i64>| {
+                                val.map(OwnedValue::I64).unwrap_or(OwnedValue::Null)
+                            },
+                        })
+                    }
+                    ColumnType::F64 => {
+                        let computer = SortByStaticFastValue::<f64>::for_field(column_name);
+                        let inner = computer.segment_sort_key_computer(segment_reader)?;
+                        Box::new(ErasedSegmentSortKeyComputerWrapper {
+                            inner,
+                            converter: |val: Option<f64>| {
+                                val.map(OwnedValue::F64).unwrap_or(OwnedValue::Null)
+                            },
+                        })
+                    }
+                    ColumnType::Bool => {
+                        let computer = SortByStaticFastValue::<bool>::for_field(column_name);
+                        let inner = computer.segment_sort_key_computer(segment_reader)?;
+                        Box::new(ErasedSegmentSortKeyComputerWrapper {
+                            inner,
+                            converter: |val: Option<bool>| {
+                                val.map(OwnedValue::Bool).unwrap_or(OwnedValue::Null)
+                            },
+                        })
+                    }
+                    ColumnType::DateTime => {
+                        let computer = SortByStaticFastValue::<DateTime>::for_field(column_name);
+                        let inner = computer.segment_sort_key_computer(segment_reader)?;
+                        Box::new(ErasedSegmentSortKeyComputerWrapper {
+                            inner,
+                            converter: |val: Option<DateTime>| {
+                                val.map(OwnedValue::Date).unwrap_or(OwnedValue::Null)
+                            },
+                        })
+                    }
+                    column_type => {
+                        return Err(crate::TantivyError::SchemaError(format!(
+                            "Field `{}` is of type {column_type:?}, which is not supported for \
+                             sorting by owned value yet.",
+                            column_name
+                        )))
+                    }
+                }
+            }
+            Self::Score => Box::new(ScoreSegmentSortKeyComputer {
+                segment_computer: SortBySimilarityScore,
+            }),
+        };
+        Ok(ErasedColumnSegmentSortKeyComputer { inner })
+    }
+}
+
+pub struct ErasedColumnSegmentSortKeyComputer {
+    inner: Box<dyn ErasedSegmentSortKeyComputer>,
+}
+
+impl SegmentSortKeyComputer for ErasedColumnSegmentSortKeyComputer {
+    type SortKey = OwnedValue;
+    type SegmentSortKey = Option<u64>;
+    type SegmentComparator = NaturalComparator;
+
+    #[inline(always)]
+    fn segment_sort_key(&mut self, doc: DocId, score: Score) -> Option<u64> {
+        self.inner.segment_sort_key(doc, score)
+    }
+
+    fn convert_segment_sort_key(&self, segment_sort_key: Self::SegmentSortKey) -> OwnedValue {
+        self.inner.convert_segment_sort_key(segment_sort_key)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::collector::sort_key::{ComparatorEnum, SortByErasedType};
+    use crate::collector::TopDocs;
+    use crate::query::AllQuery;
+    use crate::schema::{OwnedValue, Schema, FAST, TEXT};
+    use crate::Index;
+
+    #[test]
+    fn test_sort_by_owned_u64() {
+        let mut schema_builder = Schema::builder();
+        let id_field = schema_builder.add_u64_field("id", FAST);
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        let mut writer = index.writer_for_tests().unwrap();
+        writer.add_document(doc!(id_field => 10u64)).unwrap();
+        writer.add_document(doc!(id_field => 2u64)).unwrap();
+        writer.add_document(doc!()).unwrap();
+        writer.commit().unwrap();
+
+        let reader = index.reader().unwrap();
+        let searcher = reader.searcher();
+
+        let collector = TopDocs::with_limit(10)
+            .order_by((SortByErasedType::for_field("id"), ComparatorEnum::Natural));
+        let top_docs = searcher.search(&AllQuery, &collector).unwrap();
+
+        let values: Vec<OwnedValue> = top_docs.into_iter().map(|(key, _)| key).collect();
+
+        assert_eq!(
+            values,
+            vec![OwnedValue::U64(10), OwnedValue::U64(2), OwnedValue::Null]
+        );
+
+        let collector = TopDocs::with_limit(10).order_by((
+            SortByErasedType::for_field("id"),
+            ComparatorEnum::ReverseNoneLower,
+        ));
+        let top_docs = searcher.search(&AllQuery, &collector).unwrap();
+
+        let values: Vec<OwnedValue> = top_docs.into_iter().map(|(key, _)| key).collect();
+
+        assert_eq!(
+            values,
+            vec![OwnedValue::U64(2), OwnedValue::U64(10), OwnedValue::Null]
+        );
+    }
+
+    #[test]
+    fn test_sort_by_owned_string() {
+        let mut schema_builder = Schema::builder();
+        let city_field = schema_builder.add_text_field("city", FAST | TEXT);
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        let mut writer = index.writer_for_tests().unwrap();
+        writer.add_document(doc!(city_field => "tokyo")).unwrap();
+        writer.add_document(doc!(city_field => "austin")).unwrap();
+        writer.add_document(doc!()).unwrap();
+        writer.commit().unwrap();
+
+        let reader = index.reader().unwrap();
+        let searcher = reader.searcher();
+
+        let collector = TopDocs::with_limit(10).order_by((
+            SortByErasedType::for_field("city"),
+            ComparatorEnum::ReverseNoneLower,
+        ));
+        let top_docs = searcher.search(&AllQuery, &collector).unwrap();
+
+        let values: Vec<OwnedValue> = top_docs.into_iter().map(|(key, _)| key).collect();
+
+        assert_eq!(
+            values,
+            vec![
+                OwnedValue::Str("austin".to_string()),
+                OwnedValue::Str("tokyo".to_string()),
+                OwnedValue::Null
+            ]
+        );
+    }
+
+    #[test]
+    fn test_sort_by_owned_reverse() {
+        let mut schema_builder = Schema::builder();
+        let id_field = schema_builder.add_u64_field("id", FAST);
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        let mut writer = index.writer_for_tests().unwrap();
+        writer.add_document(doc!(id_field => 10u64)).unwrap();
+        writer.add_document(doc!(id_field => 2u64)).unwrap();
+        writer.add_document(doc!()).unwrap();
+        writer.commit().unwrap();
+
+        let reader = index.reader().unwrap();
+        let searcher = reader.searcher();
+
+        let collector = TopDocs::with_limit(10)
+            .order_by((SortByErasedType::for_field("id"), ComparatorEnum::Reverse));
+        let top_docs = searcher.search(&AllQuery, &collector).unwrap();
+
+        let values: Vec<OwnedValue> = top_docs.into_iter().map(|(key, _)| key).collect();
+
+        assert_eq!(
+            values,
+            vec![OwnedValue::Null, OwnedValue::U64(2), OwnedValue::U64(10)]
+        );
+    }
+
+    #[test]
+    fn test_sort_by_owned_score() {
+        let mut schema_builder = Schema::builder();
+        let body_field = schema_builder.add_text_field("body", TEXT);
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        let mut writer = index.writer_for_tests().unwrap();
+        writer.add_document(doc!(body_field => "a a")).unwrap();
+        writer.add_document(doc!(body_field => "a")).unwrap();
+        writer.commit().unwrap();
+
+        let reader = index.reader().unwrap();
+        let searcher = reader.searcher();
+        let query_parser = crate::query::QueryParser::for_index(&index, vec![body_field]);
+        let query = query_parser.parse_query("a").unwrap();
+
+        // Sort by score descending (Natural)
+        let collector = TopDocs::with_limit(10)
+            .order_by((SortByErasedType::for_score(), ComparatorEnum::Natural));
+        let top_docs = searcher.search(&query, &collector).unwrap();
+
+        let values: Vec<f64> = top_docs
+            .into_iter()
+            .map(|(key, _)| match key {
+                OwnedValue::F64(val) => val,
+                _ => panic!("Wrong type {key:?}"),
+            })
+            .collect();
+
+        assert_eq!(values.len(), 2);
+        assert!(values[0] > values[1]);
+
+        // Sort by score ascending (ReverseNoneLower)
+        let collector = TopDocs::with_limit(10).order_by((
+            SortByErasedType::for_score(),
+            ComparatorEnum::ReverseNoneLower,
+        ));
+        let top_docs = searcher.search(&query, &collector).unwrap();
+
+        let values: Vec<f64> = top_docs
+            .into_iter()
+            .map(|(key, _)| match key {
+                OwnedValue::F64(val) => val,
+                _ => panic!("Wrong type {key:?}"),
+            })
+            .collect();
+
+        assert_eq!(values.len(), 2);
+        assert!(values[0] < values[1]);
+    }
+}

src/collector/sort_key/sort_by_score.rs

@@ -63,8 +63,8 @@ impl SortKeyComputer for SortBySimilarityScore {
 
 impl SegmentSortKeyComputer for SortBySimilarityScore {
     type SortKey = Score;
-
     type SegmentSortKey = Score;
+    type SegmentComparator = NaturalComparator;
 
     #[inline(always)]
     fn segment_sort_key(&mut self, _doc: DocId, score: Score) -> Score {

src/collector/sort_key/sort_by_static_fast_value.rs

@@ -34,9 +34,7 @@ impl<T: FastValue> SortByStaticFastValue<T> {
 
 impl<T: FastValue> SortKeyComputer for SortByStaticFastValue<T> {
     type Child = SortByFastValueSegmentSortKeyComputer<T>;
-
     type SortKey = Option<T>;
-
     type Comparator = NaturalComparator;
 
     fn check_schema(&self, schema: &crate::schema::Schema) -> crate::Result<()> {
@@ -84,8 +82,8 @@ pub struct SortByFastValueSegmentSortKeyComputer<T> {
 
 impl<T: FastValue> SegmentSortKeyComputer for SortByFastValueSegmentSortKeyComputer<T> {
     type SortKey = Option<T>;
-
     type SegmentSortKey = Option<u64>;
+    type SegmentComparator = NaturalComparator;
 
     #[inline(always)]
     fn segment_sort_key(&mut self, doc: DocId, _score: Score) -> Self::SegmentSortKey {

src/collector/sort_key/sort_by_string.rs

@@ -30,9 +30,7 @@ impl SortByString {
 
 impl SortKeyComputer for SortByString {
     type SortKey = Option<String>;
-
     type Child = ByStringColumnSegmentSortKeyComputer;
-
     type Comparator = NaturalComparator;
 
     fn segment_sort_key_computer(
@@ -50,8 +48,8 @@ pub struct ByStringColumnSegmentSortKeyComputer {
 
 impl SegmentSortKeyComputer for ByStringColumnSegmentSortKeyComputer {
     type SortKey = Option<String>;
-
     type SegmentSortKey = Option<TermOrdinal>;
+    type SegmentComparator = NaturalComparator;
 
     #[inline(always)]
     fn segment_sort_key(&mut self, doc: DocId, _score: Score) -> Option<TermOrdinal> {
@@ -60,6 +58,8 @@ impl SegmentSortKeyComputer for ByStringColumnSegmentSortKeyComputer {
     }
 
     fn convert_segment_sort_key(&self, term_ord_opt: Option<TermOrdinal>) -> Option<String> {
+        // TODO: Individual lookups to the dictionary like this are very likely to repeatedly
+        // decompress the same blocks. See https://github.com/quickwit-oss/tantivy/issues/2776
         let term_ord = term_ord_opt?;
         let str_column = self.str_column_opt.as_ref()?;
         let mut bytes = Vec::new();

src/collector/sort_key/sort_key_computer.rs

@@ -12,13 +12,21 @@ use crate::{DocAddress, DocId, Result, Score, SegmentReader};
 /// It is the segment local version of the [`SortKeyComputer`].
 pub trait SegmentSortKeyComputer: 'static {
     /// The final score being emitted.
-    type SortKey: 'static + PartialOrd + Send + Sync + Clone;
+    type SortKey: 'static + Send + Sync + Clone;
 
     /// Sort key used by at the segment level by the `SegmentSortKeyComputer`.
     ///
     /// It is typically small like a `u64`, and is meant to be converted
     /// to the final score at the end of the collection of the segment.
-    type SegmentSortKey: 'static + PartialOrd + Clone + Send + Sync + Clone;
+    type SegmentSortKey: 'static + Clone + Send + Sync + Clone;
+
+    /// Comparator type.
+    type SegmentComparator: Comparator<Self::SegmentSortKey> + 'static;
+
+    /// Returns the segment sort key comparator.
+    fn segment_comparator(&self) -> Self::SegmentComparator {
+        Self::SegmentComparator::default()
+    }
 
     /// Computes the sort key for the given document and score.
     fn segment_sort_key(&mut self, doc: DocId, score: Score) -> Self::SegmentSortKey;
@@ -47,7 +55,7 @@ pub trait SegmentSortKeyComputer: 'static {
         left: &Self::SegmentSortKey,
         right: &Self::SegmentSortKey,
     ) -> Ordering {
-        NaturalComparator.compare(left, right)
+        self.segment_comparator().compare(left, right)
     }
 
     /// Implementing this method makes it possible to avoid computing
@@ -81,7 +89,7 @@ pub trait SegmentSortKeyComputer: 'static {
 /// the sort key at a segment scale.
 pub trait SortKeyComputer: Sync {
     /// The sort key type.
-    type SortKey: 'static + Send + Sync + PartialOrd + Clone + std::fmt::Debug;
+    type SortKey: 'static + Send + Sync + Clone + std::fmt::Debug;
     /// Type of the associated [`SegmentSortKeyComputer`].
     type Child: SegmentSortKeyComputer<SortKey = Self::SortKey>;
     /// Comparator type.
@@ -136,10 +144,7 @@ where
     HeadSortKeyComputer: SortKeyComputer,
     TailSortKeyComputer: SortKeyComputer,
 {
-    type SortKey = (
-        <HeadSortKeyComputer::Child as SegmentSortKeyComputer>::SortKey,
-        <TailSortKeyComputer::Child as SegmentSortKeyComputer>::SortKey,
-    );
+    type SortKey = (HeadSortKeyComputer::SortKey, TailSortKeyComputer::SortKey);
     type Child = (HeadSortKeyComputer::Child, TailSortKeyComputer::Child);
 
     type Comparator = (
@@ -188,6 +193,11 @@ where
         TailSegmentSortKeyComputer::SegmentSortKey,
     );
 
+    type SegmentComparator = (
+        HeadSegmentSortKeyComputer::SegmentComparator,
+        TailSegmentSortKeyComputer::SegmentComparator,
+    );
+
     /// A SegmentSortKeyComputer maps to a SegmentSortKey, but it can also decide on
     /// its ordering.
     ///
@@ -269,11 +279,12 @@ impl<T, PreviousScore, NewScore> SegmentSortKeyComputer
     for MappedSegmentSortKeyComputer<T, PreviousScore, NewScore>
 where
     T: SegmentSortKeyComputer<SortKey = PreviousScore>,
-    PreviousScore: 'static + Clone + Send + Sync + PartialOrd,
-    NewScore: 'static + Clone + Send + Sync + PartialOrd,
+    PreviousScore: 'static + Clone + Send + Sync,
+    NewScore: 'static + Clone + Send + Sync,
 {
     type SortKey = NewScore;
     type SegmentSortKey = T::SegmentSortKey;
+    type SegmentComparator = T::SegmentComparator;
 
     fn segment_sort_key(&mut self, doc: DocId, score: Score) -> Self::SegmentSortKey {
         self.sort_key_computer.segment_sort_key(doc, score)
@@ -463,6 +474,7 @@ where
 {
     type SortKey = TSortKey;
     type SegmentSortKey = TSortKey;
+    type SegmentComparator = NaturalComparator;
 
     fn segment_sort_key(&mut self, doc: DocId, _score: Score) -> TSortKey {
         (self)(doc)

src/collector/top_collector.rs

@@ -1,64 +1,22 @@
-use std::cmp::Ordering;
-
 use serde::{Deserialize, Serialize};
 
 /// Contains a feature (field, score, etc.) of a document along with the document address.
 ///
-/// It guarantees stable sorting: in case of a tie on the feature, the document
-/// address is used.
-///
-/// The REVERSE_ORDER generic parameter controls whether the by-feature order
-/// should be reversed, which is useful for achieving for example largest-first
-/// semantics without having to wrap the feature in a `Reverse`.
-#[derive(Clone, Default, Serialize, Deserialize)]
-pub struct ComparableDoc<T, D, const REVERSE_ORDER: bool = false> {
+/// Used only by TopNComputer, which implements the actual comparison via a `Comparator`.
+#[derive(Clone, Default, Eq, PartialEq, Serialize, Deserialize)]
+pub struct ComparableDoc<T, D> {
     /// The feature of the document. In practice, this is
-    /// is any type that implements `PartialOrd`.
+    /// is a type which can be compared with a `Comparator<T>`.
     pub sort_key: T,
-    /// The document address. In practice, this is any
-    /// type that implements `PartialOrd`, and is guaranteed
-    /// to be unique for each document.
+    /// The document address. In practice, this is either a `DocId` or `DocAddress`.
     pub doc: D,
 }
-impl<T: std::fmt::Debug, D: std::fmt::Debug, const R: bool> std::fmt::Debug
-    for ComparableDoc<T, D, R>
-{
+
+impl<T: std::fmt::Debug, D: std::fmt::Debug> std::fmt::Debug for ComparableDoc<T, D> {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        f.debug_struct(format!("ComparableDoc<_, _ {R}").as_str())
+        f.debug_struct("ComparableDoc")
             .field("feature", &self.sort_key)
             .field("doc", &self.doc)
             .finish()
     }
 }
-
-impl<T: PartialOrd, D: PartialOrd, const R: bool> PartialOrd for ComparableDoc<T, D, R> {
-    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
-        Some(self.cmp(other))
-    }
-}
-
-impl<T: PartialOrd, D: PartialOrd, const R: bool> Ord for ComparableDoc<T, D, R> {
-    #[inline]
-    fn cmp(&self, other: &Self) -> Ordering {
-        let by_feature = self
-            .sort_key
-            .partial_cmp(&other.sort_key)
-            .map(|ord| if R { ord.reverse() } else { ord })
-            .unwrap_or(Ordering::Equal);
-
-        let lazy_by_doc_address = || self.doc.partial_cmp(&other.doc).unwrap_or(Ordering::Equal);
-
-        // In case of a tie on the feature, we sort by ascending
-        // `DocAddress` in order to ensure a stable sorting of the
-        // documents.
-        by_feature.then_with(lazy_by_doc_address)
-    }
-}
-
-impl<T: PartialOrd, D: PartialOrd, const R: bool> PartialEq for ComparableDoc<T, D, R> {
-    fn eq(&self, other: &Self) -> bool {
-        self.cmp(other) == Ordering::Equal
-    }
-}
-
-impl<T: PartialOrd, D: PartialOrd, const R: bool> Eq for ComparableDoc<T, D, R> {}

src/collector/top_score_collector.rs

@@ -23,10 +23,9 @@ use crate::{DocAddress, DocId, Order, Score, SegmentReader};
 /// The theoretical complexity for collecting the top `K` out of `N` documents
 /// is `O(N + K)`.
 ///
-/// This collector does not guarantee a stable sorting in case of a tie on the
-/// document score, for stable sorting `PartialOrd` needs to resolve on other fields
-/// like docid in case of score equality.
-/// Only then, it is suitable for pagination.
+/// This collector guarantees a stable sorting in case of a tie on the
+/// document score/sort key: The document address (`DocAddress`) is used as a tie breaker.
+/// In case of a tie on the sort key, documents are always sorted by ascending `DocAddress`.
 ///
 /// ```rust
 /// use tantivy::collector::TopDocs;
@@ -325,7 +324,7 @@ impl TopDocs {
         sort_key_computer: impl SortKeyComputer<SortKey = TSortKey> + Send + 'static,
     ) -> impl Collector<Fruit = Vec<(TSortKey, DocAddress)>>
     where
-        TSortKey: 'static + Clone + Send + Sync + PartialOrd + std::fmt::Debug,
+        TSortKey: 'static + Clone + Send + Sync + std::fmt::Debug,
     {
         TopBySortKeyCollector::new(sort_key_computer, self.doc_range())
     }
@@ -446,7 +445,7 @@ where
     F: 'static + Send + Sync + Fn(&SegmentReader) -> TTweakScoreSortKeyFn,
     TTweakScoreSortKeyFn: 'static + Fn(DocId, Score) -> TSortKey,
     TweakScoreSegmentSortKeyComputer<TTweakScoreSortKeyFn>:
-        SegmentSortKeyComputer<SortKey = TSortKey>,
+        SegmentSortKeyComputer<SortKey = TSortKey, SegmentSortKey = TSortKey>,
     TSortKey: 'static + PartialOrd + Clone + Send + Sync + std::fmt::Debug,
 {
     type SortKey = TSortKey;
@@ -481,6 +480,7 @@ where
 {
     type SortKey = TSortKey;
     type SegmentSortKey = TSortKey;
+    type SegmentComparator = NaturalComparator;
 
     fn segment_sort_key(&mut self, doc: DocId, score: Score) -> TSortKey {
         (self.sort_key_fn)(doc, score)
@@ -500,8 +500,13 @@ where
 ///
 /// For TopN == 0, it will be relative expensive.
 ///
-/// When using the natural comparator, the top N computer returns the top N elements in
-/// descending order, as expected for a top N.
+/// The TopNComputer will tiebreak by using ascending `D` (DocId or DocAddress):
+/// i.e., in case of a tie on the sort key, the `DocId|DocAddress` are always sorted in
+/// ascending order, regardless of the `Comparator` used for the `Score` type.
+///
+/// NOTE: Items must be `push`ed to the TopNComputer in ascending `DocId|DocAddress` order, as the
+/// threshold used to eliminate docs does not include the `DocId` or `DocAddress`: this provides
+/// the ascending `DocId|DocAddress` tie-breaking behavior without additional comparisons.
 #[derive(Serialize, Deserialize)]
 #[serde(from = "TopNComputerDeser<Score, D, C>")]
 pub struct TopNComputer<Score, D, C> {
@@ -580,6 +585,18 @@ where
     }
 }
 
+#[inline(always)]
+fn compare_for_top_k<TSortKey, D: Ord, C: Comparator<TSortKey>>(
+    c: &C,
+    lhs: &ComparableDoc<TSortKey, D>,
+    rhs: &ComparableDoc<TSortKey, D>,
+) -> std::cmp::Ordering {
+    c.compare(&lhs.sort_key, &rhs.sort_key)
+        .reverse() // Reverse here because we want top K.
+        .then_with(|| lhs.doc.cmp(&rhs.doc)) // Regardless of asc/desc, in presence of a tie, we
+                                             // sort by doc id
+}
+
 impl<TSortKey, D, C> TopNComputer<TSortKey, D, C>
 where
     D: Ord,
@@ -600,10 +617,13 @@ where
 
     /// Push a new document to the top n.
     /// If the document is below the current threshold, it will be ignored.
+    ///
+    /// NOTE: `push` must be called in ascending `DocId`/`DocAddress` order.
     #[inline]
     pub fn push(&mut self, sort_key: TSortKey, doc: D) {
         if let Some(last_median) = &self.threshold {
-            if self.comparator.compare(&sort_key, last_median) == Ordering::Less {
+            // See the struct docs for an explanation of why this comparison is strict.
+            if self.comparator.compare(&sort_key, last_median) != Ordering::Greater {
                 return;
             }
         }
@@ -629,9 +649,7 @@ where
     fn truncate_top_n(&mut self) -> TSortKey {
         // Use select_nth_unstable to find the top nth score
         let (_, median_el, _) = self.buffer.select_nth_unstable_by(self.top_n, |lhs, rhs| {
-            self.comparator
-                .compare(&rhs.sort_key, &lhs.sort_key)
-                .then_with(|| lhs.doc.cmp(&rhs.doc))
+            compare_for_top_k(&self.comparator, lhs, rhs)
         });
 
         let median_score = median_el.sort_key.clone();
@@ -646,11 +664,8 @@ where
         if self.buffer.len() > self.top_n {
             self.truncate_top_n();
         }
-        self.buffer.sort_unstable_by(|left, right| {
-            self.comparator
-                .compare(&right.sort_key, &left.sort_key)
-                .then_with(|| left.doc.cmp(&right.doc))
-        });
+        self.buffer
+            .sort_unstable_by(|lhs, rhs| compare_for_top_k(&self.comparator, lhs, rhs));
         self.buffer
     }
 
@@ -755,6 +770,33 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_topn_computer_duplicates() {
+        let mut computer: TopNComputer<u32, u32, NaturalComparator> =
+            TopNComputer::new_with_comparator(2, NaturalComparator);
+
+        computer.push(1u32, 1u32);
+        computer.push(1u32, 2u32);
+        computer.push(1u32, 3u32);
+        computer.push(1u32, 4u32);
+        computer.push(1u32, 5u32);
+
+        // In the presence of duplicates, DocIds are always ascending order.
+        assert_eq!(
+            computer.into_sorted_vec(),
+            &[
+                ComparableDoc {
+                    sort_key: 1u32,
+                    doc: 1u32,
+                },
+                ComparableDoc {
+                    sort_key: 1u32,
+                    doc: 2u32,
+                }
+            ]
+        );
+    }
+
     #[test]
     fn test_topn_computer_no_panic() {
         for top_n in 0..10 {
@@ -772,14 +814,17 @@ mod tests {
         #[test]
         fn test_topn_computer_asc_prop(
           limit in 0..10_usize,
-          docs in proptest::collection::vec((0..100_u64, 0..100_u64), 0..100_usize),
+          mut docs in proptest::collection::vec((0..100_u64, 0..100_u64), 0..100_usize),
         ) {
+            // NB: TopNComputer must receive inputs in ascending DocId order.
+            docs.sort_by_key(|(_, doc_id)| *doc_id);
             let mut computer: TopNComputer<_, _, ReverseComparator> = TopNComputer::new_with_comparator(limit, ReverseComparator);
             for (feature, doc) in &docs {
                 computer.push(*feature, *doc);
             }
-            let mut comparable_docs: Vec<ComparableDoc<u64, u64>> = docs.into_iter().map(|(sort_key, doc)| ComparableDoc { sort_key, doc }).collect::<Vec<_>>();
-            comparable_docs.sort();
+            let mut comparable_docs: Vec<ComparableDoc<u64, u64>> =
+                docs.into_iter().map(|(sort_key, doc)| ComparableDoc { sort_key, doc }).collect();
+            crate::collector::sort_key::tests::sort_hits(&mut comparable_docs, Order::Asc);
             comparable_docs.truncate(limit);
             prop_assert_eq!(
                 computer.into_sorted_vec(),
@@ -1406,15 +1451,10 @@ mod tests {
 
             // Using the TopDocs collector should always be equivalent to sorting, skipping the
             // offset, and then taking the limit.
-            let sorted_docs: Vec<_> = if order.is_desc() {
-                let mut comparable_docs: Vec<ComparableDoc<_, _, true>> =
+            let sorted_docs: Vec<_> = {
+                let mut comparable_docs: Vec<ComparableDoc<_, _>> =
                     all_results.into_iter().map(|(sort_key, doc)| ComparableDoc { sort_key, doc}).collect();
-                comparable_docs.sort();
-                comparable_docs.into_iter().map(|cd| (cd.sort_key, cd.doc)).collect()
-            } else {
-                let mut comparable_docs: Vec<ComparableDoc<_, _, false>> =
-                    all_results.into_iter().map(|(sort_key, doc)| ComparableDoc { sort_key, doc}).collect();
-                comparable_docs.sort();
+                crate::collector::sort_key::tests::sort_hits(&mut comparable_docs, order);
                 comparable_docs.into_iter().map(|cd| (cd.sort_key, cd.doc)).collect()
             };
             let expected_docs = sorted_docs.into_iter().skip(offset).take(limit).collect::<Vec<_>>();

src/directory/mmap_directory/mod.rs

@@ -1,3 +1,5 @@
+mod file_watcher;
+
 use std::collections::HashMap;
 use std::fmt;
 use std::fs::{self, File, OpenOptions};
@@ -7,6 +9,7 @@ use std::path::{Path, PathBuf};
 use std::sync::{Arc, RwLock, Weak};
 
 use common::StableDeref;
+use file_watcher::FileWatcher;
 use fs4::fs_std::FileExt;
 #[cfg(all(feature = "mmap", unix))]
 pub use memmap2::Advice;
@@ -18,7 +21,6 @@ use crate::core::META_FILEPATH;
 use crate::directory::error::{
     DeleteError, LockError, OpenDirectoryError, OpenReadError, OpenWriteError,
 };
-use crate::directory::file_watcher::FileWatcher;
 use crate::directory::{
     AntiCallToken, Directory, DirectoryLock, FileHandle, Lock, OwnedBytes, TerminatingWrite,
     WatchCallback, WatchHandle, WritePtr,

src/directory/mod.rs

@@ -5,7 +5,6 @@ mod mmap_directory;
 
 mod directory;
 mod directory_lock;
-mod file_watcher;
 pub mod footer;
 mod managed_directory;
 mod ram_directory;

src/docset.rs

@@ -40,6 +40,8 @@ pub trait DocSet: Send {
     /// of `DocSet` should support it.
     ///
     /// Calling `seek(TERMINATED)` is also legal and is the normal way to consume a `DocSet`.
+    ///
+    /// `target` has to be larger or equal to `.doc()` when calling `seek`.
     fn seek(&mut self, target: DocId) -> DocId {
         let mut doc = self.doc();
         debug_assert!(doc <= target);
@@ -49,6 +51,33 @@ pub trait DocSet: Send {
         doc
     }
 
+    /// Seeks to the target if possible and returns true if the target is in the DocSet.
+    ///
+    /// DocSets that already have an efficient `seek` method don't need to implement
+    /// `seek_into_the_danger_zone`. All wrapper DocSets should forward
+    /// `seek_into_the_danger_zone` to the underlying DocSet.
+    ///
+    /// ## API Behaviour
+    /// If `seek_into_the_danger_zone` is returning true, a call to `doc()` has to return target.
+    /// If `seek_into_the_danger_zone` is returning false, a call to `doc()` may return any doc
+    /// between the last doc that matched and target or a doc that is a valid next hit after
+    /// target. The DocSet is considered to be in an invalid state until
+    /// `seek_into_the_danger_zone` returns true again.
+    ///
+    /// `target` needs to be equal or larger than `doc` when in a valid state.
+    ///
+    /// Consecutive calls are not allowed to have decreasing `target` values.
+    ///
+    /// # Warning
+    /// This is an advanced API used by intersection. The API contract is tricky, avoid using it.
+    fn seek_into_the_danger_zone(&mut self, target: DocId) -> bool {
+        let current_doc = self.doc();
+        if current_doc < target {
+            self.seek(target);
+        }
+        self.doc() == target
+    }
+
     /// Fills a given mutable buffer with the next doc ids from the
     /// `DocSet`
     ///
@@ -94,6 +123,15 @@ pub trait DocSet: Send {
     /// which would be the number of documents in the DocSet.
     ///
     /// By default this returns `size_hint()`.
+    ///
+    /// DocSets may have vastly different cost depending on their type,
+    /// e.g. an intersection with 10 hits is much cheaper than
+    /// a phrase search with 10 hits, since it needs to load positions.
+    ///
+    /// ### Future Work
+    /// We may want to differentiate `DocSet` costs more more granular, e.g.
+    /// creation_cost, advance_cost, seek_cost on to get a good estimation
+    /// what query types to choose.
     fn cost(&self) -> u64 {
         self.size_hint() as u64
     }
@@ -137,6 +175,10 @@ impl DocSet for &mut dyn DocSet {
         (**self).seek(target)
     }
 
+    fn seek_into_the_danger_zone(&mut self, target: DocId) -> bool {
+        (**self).seek_into_the_danger_zone(target)
+    }
+
     fn doc(&self) -> u32 {
         (**self).doc()
     }
@@ -169,6 +211,11 @@ impl<TDocSet: DocSet + ?Sized> DocSet for Box<TDocSet> {
         unboxed.seek(target)
     }
 
+    fn seek_into_the_danger_zone(&mut self, target: DocId) -> bool {
+        let unboxed: &mut TDocSet = self.borrow_mut();
+        unboxed.seek_into_the_danger_zone(target)
+    }
+
     fn fill_buffer(&mut self, buffer: &mut [DocId; COLLECT_BLOCK_BUFFER_LEN]) -> usize {
         let unboxed: &mut TDocSet = self.borrow_mut();
         unboxed.fill_buffer(buffer)

src/index/index_meta.rs

@@ -13,9 +13,9 @@ use crate::store::Compressor;
 use crate::{Inventory, Opstamp, TrackedObject};
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
-struct DeleteMeta {
+pub struct DeleteMeta {
     num_deleted_docs: u32,
-    opstamp: Opstamp,
+    pub opstamp: Opstamp,
 }
 
 #[derive(Clone, Default)]
@@ -213,7 +213,7 @@ impl SegmentMeta {
 struct InnerSegmentMeta {
     segment_id: SegmentId,
     max_doc: u32,
-    deletes: Option<DeleteMeta>,
+    pub deletes: Option<DeleteMeta>,
     /// If you want to avoid the SegmentComponent::TempStore file to be covered by
     /// garbage collection and deleted, set this to true. This is used during merge.
     #[serde(skip)]
@@ -404,7 +404,10 @@ mod tests {
             schema_builder.build()
         };
         let index_metas = IndexMeta {
-            index_settings: IndexSettings::default(),
+            index_settings: IndexSettings {
+                docstore_compression: Compressor::None,
+                ..Default::default()
+            },
             segments: Vec::new(),
             schema,
             opstamp: 0u64,
@@ -413,7 +416,7 @@ mod tests {
         let json = serde_json::ser::to_string(&index_metas).expect("serialization failed");
         assert_eq!(
             json,
-            r#"{"index_settings":{"docstore_compression":"lz4","docstore_blocksize":16384},"segments":[],"schema":[{"name":"text","type":"text","options":{"indexing":{"record":"position","fieldnorms":true,"tokenizer":"default"},"stored":false,"fast":false}}],"opstamp":0}"#
+            r#"{"index_settings":{"docstore_compression":"none","docstore_blocksize":16384},"segments":[],"schema":[{"name":"text","type":"text","options":{"indexing":{"record":"position","fieldnorms":true,"tokenizer":"default"},"stored":false,"fast":false}}],"opstamp":0}"#
         );
 
         let deser_meta: UntrackedIndexMeta = serde_json::from_str(&json).unwrap();
@@ -494,6 +497,8 @@ mod tests {
     #[test]
     #[cfg(feature = "lz4-compression")]
     fn test_index_settings_default() {
+        use crate::store::Compressor;
+
         let mut index_settings = IndexSettings::default();
         assert_eq!(
             index_settings,

src/index/segment.rs

@@ -46,7 +46,7 @@ impl Segment {
     ///
     /// This method is only used when updating `max_doc` from 0
     /// as we finalize a fresh new segment.
-    pub(crate) fn with_max_doc(self, max_doc: u32) -> Segment {
+    pub fn with_max_doc(self, max_doc: u32) -> Segment {
         Segment {
             index: self.index,
             meta: self.meta.with_max_doc(max_doc),

src/indexer/delete_queue.rs

@@ -4,38 +4,37 @@ use std::sync::{Arc, RwLock, Weak};
 use super::operation::DeleteOperation;
 use crate::Opstamp;
 
-// The DeleteQueue is similar in conceptually to a multiple
-// consumer single producer broadcast channel.
-//
-// All consumer will receive all messages.
-//
-// Consumer of the delete queue are holding a `DeleteCursor`,
-// which points to a specific place of the `DeleteQueue`.
-//
-// New consumer can be created in two ways
-// - calling `delete_queue.cursor()` returns a cursor, that will include all future delete operation
-//   (and some or none of the past operations... The client is in charge of checking the opstamps.).
-// - cloning an existing cursor returns a new cursor, that is at the exact same position, and can
-//   now advance independently from the original cursor.
+/// The DeleteQueue is similar in conceptually to a multiple
+/// consumer single producer broadcast channel.
+///
+/// All consumer will receive all messages.
+///
+/// Consumer of the delete queue are holding a `DeleteCursor`,
+/// which points to a specific place of the `DeleteQueue`.
+///
+/// New consumer can be created in two ways
+/// - calling `delete_queue.cursor()` returns a cursor, that will include all future delete
+///   operation (and some or none of the past operations... The client is in charge of checking the
+///   opstamps.).
+/// - cloning an existing cursor returns a new cursor, that is at the exact same position, and can
+///   now advance independently from the original cursor.
 #[derive(Default)]
 struct InnerDeleteQueue {
     writer: Vec<DeleteOperation>,
     last_block: Weak<Block>,
 }
 
-#[derive(Clone)]
+/// The delete queue is a linked list storing delete operations.
+///
+/// Several consumers can hold a reference to it. Delete operations
+/// get dropped/gc'ed when no more consumers are holding a reference
+/// to them.
+#[derive(Clone, Default)]
 pub struct DeleteQueue {
     inner: Arc<RwLock<InnerDeleteQueue>>,
 }
 
 impl DeleteQueue {
-    // Creates a new delete queue.
-    pub fn new() -> DeleteQueue {
-        DeleteQueue {
-            inner: Arc::default(),
-        }
-    }
-
     fn get_last_block(&self) -> Arc<Block> {
         {
             // try get the last block with simply acquiring the read lock.
@@ -58,10 +57,10 @@ impl DeleteQueue {
         block
     }
 
-    // Creates a new cursor that makes it possible to
-    // consume future delete operations.
-    //
-    // Past delete operations are not accessible.
+    /// Creates a new cursor that makes it possible to
+    /// consume future delete operations.
+    ///
+    /// Past delete operations are not accessible.
     pub fn cursor(&self) -> DeleteCursor {
         let last_block = self.get_last_block();
         let operations_len = last_block.operations.len();
@@ -71,7 +70,7 @@ impl DeleteQueue {
         }
     }
 
-    // Appends a new delete operations.
+    /// Appends a new delete operations.
     pub fn push(&self, delete_operation: DeleteOperation) {
         self.inner
             .write()
@@ -169,6 +168,7 @@ struct Block {
     next: NextBlock,
 }
 
+/// As we process delete operations, keeps track of our position.
 #[derive(Clone)]
 pub struct DeleteCursor {
     block: Arc<Block>,
@@ -261,7 +261,7 @@ mod tests {
 
     #[test]
     fn test_deletequeue() {
-        let delete_queue = DeleteQueue::new();
+        let delete_queue = DeleteQueue::default();
 
         let make_op = |i: usize| DeleteOperation {
             opstamp: i as u64,

src/indexer/index_writer.rs

@@ -128,7 +128,7 @@ fn compute_deleted_bitset(
 /// is `==` target_opstamp.
 /// For instance, there was no delete operation between the state of the `segment_entry` and
 /// the `target_opstamp`, `segment_entry` is not updated.
-pub(crate) fn advance_deletes(
+pub fn advance_deletes(
     mut segment: Segment,
     segment_entry: &mut SegmentEntry,
     target_opstamp: Opstamp,
@@ -303,7 +303,7 @@ impl<D: Document> IndexWriter<D> {
         let (document_sender, document_receiver) =
             crossbeam_channel::bounded(PIPELINE_MAX_SIZE_IN_DOCS);
 
-        let delete_queue = DeleteQueue::new();
+        let delete_queue = DeleteQueue::default();
 
         let current_opstamp = index.load_metas()?.opstamp;
 

src/indexer/mod.rs

@@ -4,6 +4,7 @@
 //! `IndexWriter` is the main entry point for that, which created from
 //! [`Index::writer`](crate::Index::writer).
 
+/// Delete queue implementation for broadcasting delete operations to consumers.
 pub(crate) mod delete_queue;
 pub(crate) mod path_to_unordered_id;
 
@@ -32,12 +33,11 @@ mod stamper;
 use crossbeam_channel as channel;
 use smallvec::SmallVec;
 
-pub use self::index_writer::{IndexWriter, IndexWriterOptions};
+pub use self::index_writer::{advance_deletes, IndexWriter, IndexWriterOptions};
 pub use self::log_merge_policy::LogMergePolicy;
 pub use self::merge_operation::MergeOperation;
 pub use self::merge_policy::{MergeCandidate, MergePolicy, NoMergePolicy};
-use self::operation::AddOperation;
-pub use self::operation::UserOperation;
+pub use self::operation::{AddOperation, DeleteOperation, UserOperation};
 pub use self::prepared_commit::PreparedCommit;
 pub use self::segment_entry::SegmentEntry;
 pub(crate) use self::segment_serializer::SegmentSerializer;

src/indexer/operation.rs

@@ -5,14 +5,20 @@ use crate::Opstamp;
 
 /// Timestamped Delete operation.
 pub struct DeleteOperation {
+    /// Operation stamp.
+    /// It is used to check whether the delete operation
+    /// applies to an added document operation.
     pub opstamp: Opstamp,
+    /// Weight is used to define the set of documents to be deleted.
     pub target: Box<dyn Weight>,
 }
 
 /// Timestamped Add operation.
 #[derive(Eq, PartialEq, Debug)]
 pub struct AddOperation<D: Document = TantivyDocument> {
+    /// Operation stamp.
     pub opstamp: Opstamp,
+    /// Document to be added.
     pub document: D,
 }
 

src/indexer/segment_register.rs

@@ -117,7 +117,7 @@ mod tests {
     #[test]
     fn test_segment_register() {
         let inventory = SegmentMetaInventory::default();
-        let delete_queue = DeleteQueue::new();
+        let delete_queue = DeleteQueue::default();
 
         let mut segment_register = SegmentRegister::default();
         let segment_id_a = SegmentId::generate_random();

src/indexer/segment_writer.rs

@@ -421,10 +421,9 @@ fn remap_and_write(
 #[cfg(test)]
 mod tests {
     use std::collections::BTreeMap;
-    use std::path::{Path, PathBuf};
+    use std::path::Path;
 
     use columnar::ColumnType;
-    use tempfile::TempDir;
 
     use crate::collector::{Count, TopDocs};
     use crate::directory::RamDirectory;
@@ -1067,10 +1066,7 @@ mod tests {
         let mut schema_builder = Schema::builder();
         schema_builder.add_text_field("title", text_options);
         let schema = schema_builder.build();
-        let tempdir = TempDir::new().unwrap();
-        let tempdir_path = PathBuf::from(tempdir.path());
-        Index::create_in_dir(&tempdir_path, schema).unwrap();
-        let index = Index::open_in_dir(tempdir_path).unwrap();
+        let index = Index::create_in_ram(schema);
         let schema = index.schema();
         let mut index_writer = index.writer(50_000_000).unwrap();
         let title = schema.get_field("title").unwrap();

src/lib.rs

@@ -17,6 +17,7 @@
 //!
 //! ```rust
 //! # use std::path::Path;
+//! # use std::fs;
 //! # use tempfile::TempDir;
 //! # use tantivy::collector::TopDocs;
 //! # use tantivy::query::QueryParser;
@@ -27,8 +28,11 @@
 //! #     // Let's create a temporary directory for the
 //! #     // sake of this example
 //! #     if let Ok(dir) = TempDir::new() {
-//! #         run_example(dir.path()).unwrap();
-//! #         dir.close().unwrap();
+//! #         let index_path = dir.path().join("index");
+//! #         // In case the directory already exists, we remove it
+//! #         let _ = fs::remove_dir_all(&index_path);
+//! #         fs::create_dir_all(&index_path).unwrap();
+//! #         run_example(&index_path).unwrap();
 //! #     }
 //! # }
 //! #
@@ -203,6 +207,7 @@ mod docset;
 mod reader;
 
 #[cfg(test)]
+#[cfg(feature = "mmap")]
 mod compat_tests;
 
 pub use self::reader::{IndexReader, IndexReaderBuilder, ReloadPolicy, Warmer};
@@ -1170,12 +1175,11 @@ pub mod tests {
 
     #[test]
     fn test_validate_checksum() -> crate::Result<()> {
-        let index_path = tempfile::tempdir().expect("dir");
         let mut builder = Schema::builder();
         let body = builder.add_text_field("body", TEXT | STORED);
         let schema = builder.build();
-        let index = Index::create_in_dir(&index_path, schema)?;
-        let mut writer: IndexWriter = index.writer(50_000_000)?;
+        let index = Index::create_in_ram(schema);
+        let mut writer: IndexWriter = index.writer_for_tests()?;
         writer.set_merge_policy(Box::new(NoMergePolicy));
         for _ in 0..5000 {
             writer.add_document(doc!(body => "foo"))?;

src/postings/compression/mod.rs

@@ -1,12 +1,15 @@
 use bitpacking::{BitPacker, BitPacker4x};
-use common::FixedSize;
 
 pub const COMPRESSION_BLOCK_SIZE: usize = BitPacker4x::BLOCK_LEN;
-const COMPRESSED_BLOCK_MAX_SIZE: usize = COMPRESSION_BLOCK_SIZE * u32::SIZE_IN_BYTES;
+// in vint encoding, each byte stores 7 bits of data, so we need at most 32 / 7 = 4.57 bytes to
+// store a u32 in the worst case, rounding up to 5 bytes total
+const MAX_VINT_SIZE: usize = 5;
+const COMPRESSED_BLOCK_MAX_SIZE: usize = COMPRESSION_BLOCK_SIZE * MAX_VINT_SIZE;
 
 mod vint;
 
 /// Returns the size in bytes of a compressed block, given `num_bits`.
+#[inline]
 pub fn compressed_block_size(num_bits: u8) -> usize {
     (num_bits as usize) * COMPRESSION_BLOCK_SIZE / 8
 }
@@ -267,7 +270,6 @@ impl VIntDecoder for BlockDecoder {
 
 #[cfg(test)]
 pub(crate) mod tests {
-
     use super::*;
     use crate::TERMINATED;
 
@@ -372,6 +374,13 @@ pub(crate) mod tests {
             }
         }
     }
+
+    #[test]
+    fn test_compress_vint_unsorted_does_not_overflow() {
+        let mut encoder = BlockEncoder::new();
+        let input: Vec<u32> = vec![u32::MAX; COMPRESSION_BLOCK_SIZE];
+        encoder.compress_vint_unsorted(&input);
+    }
 }
 
 #[cfg(all(test, feature = "unstable"))]

src/postings/mod.rs

@@ -527,6 +527,7 @@ pub(crate) mod tests {
     }
 
     impl<TScorer: Scorer> Scorer for UnoptimizedDocSet<TScorer> {
+        #[inline]
         fn score(&mut self) -> Score {
             self.0.score()
         }

src/postings/skip.rs

@@ -6,17 +6,21 @@ use crate::{DocId, Score, TERMINATED};
 
 // doc num bits uses the following encoding:
 // given 0b a b cdefgh
-//         |1|2|   3  |
+//         |1|2|3|  4  |
 // - 1: unused
 // - 2: is delta-1 encoded. 0 if not, 1, if yes
-// - 3: a 6 bit number in 0..=32, the actual bitwidth
+// - 3: unused
+// - 4: a 5 bit number in 0..32, the actual bitwidth. Bitpacking could in theory say this is 32
+//   (requiring a 6th bit), but the biggest doc_id we can want to encode is TERMINATED-1, which can
+//   be represented on 31b without delta encoding.
 fn encode_bitwidth(bitwidth: u8, delta_1: bool) -> u8 {
+    assert!(bitwidth < 32);
     bitwidth | ((delta_1 as u8) << 6)
 }
 
 fn decode_bitwidth(raw_bitwidth: u8) -> (u8, bool) {
     let delta_1 = ((raw_bitwidth >> 6) & 1) != 0;
-    let bitwidth = raw_bitwidth & 0x3f;
+    let bitwidth = raw_bitwidth & 0x1f;
     (bitwidth, delta_1)
 }
 
@@ -430,7 +434,7 @@ mod tests {
 
     #[test]
     fn test_encode_decode_bitwidth() {
-        for bitwidth in 0..=32 {
+        for bitwidth in 0..32 {
             for delta_1 in [false, true] {
                 assert_eq!(
                     (bitwidth, delta_1),

src/query/all_query.rs

@@ -23,7 +23,11 @@ pub struct AllWeight;
 impl Weight for AllWeight {
     fn scorer(&self, reader: &SegmentReader, boost: Score) -> crate::Result<Box<dyn Scorer>> {
         let all_scorer = AllScorer::new(reader.max_doc());
-        Ok(Box::new(BoostScorer::new(all_scorer, boost)))
+        if boost != 1.0 {
+            Ok(Box::new(BoostScorer::new(all_scorer, boost)))
+        } else {
+            Ok(Box::new(all_scorer))
+        }
     }
 
     fn explain(&self, reader: &SegmentReader, doc: DocId) -> crate::Result<Explanation> {
@@ -58,6 +62,15 @@ impl DocSet for AllScorer {
         self.doc
     }
 
+    fn seek(&mut self, target: DocId) -> DocId {
+        debug_assert!(target >= self.doc);
+        self.doc = target;
+        if self.doc >= self.max_doc {
+            self.doc = TERMINATED;
+        }
+        self.doc
+    }
+
     fn fill_buffer(&mut self, buffer: &mut [DocId; COLLECT_BLOCK_BUFFER_LEN]) -> usize {
         if self.doc() == TERMINATED {
             return 0;
@@ -92,6 +105,7 @@ impl DocSet for AllScorer {
 }
 
 impl Scorer for AllScorer {
+    #[inline]
     fn score(&mut self) -> Score {
         1.0
     }

src/query/boolean_query/block_wand.rs

@@ -483,7 +483,7 @@ mod tests {
             let checkpoints_for_each_pruning =
                 compute_checkpoints_for_each_pruning(term_scorers.clone(), top_k);
             let checkpoints_manual =
-                compute_checkpoints_manual(term_scorers.clone(), top_k, 100_000);
+                compute_checkpoints_manual(term_scorers.clone(), top_k, max_doc as u32);
             assert_eq!(checkpoints_for_each_pruning.len(), checkpoints_manual.len());
             for (&(left_doc, left_score), &(right_doc, right_score)) in checkpoints_for_each_pruning
                 .iter()

src/query/boolean_query/boolean_weight.rs

@@ -97,6 +97,65 @@ fn into_box_scorer<TScoreCombiner: ScoreCombiner>(
     }
 }
 
+/// Returns the effective MUST scorer, accounting for removed AllScorers.
+///
+/// When AllScorer instances are removed from must_scorers as an optimization,
+/// we must restore the "match all" semantics if the list becomes empty.
+fn effective_must_scorer(
+    must_scorers: Vec<Box<dyn Scorer>>,
+    removed_all_scorer_count: usize,
+    max_doc: DocId,
+    num_docs: u32,
+) -> Option<Box<dyn Scorer>> {
+    if must_scorers.is_empty() {
+        if removed_all_scorer_count > 0 {
+            // Had AllScorer(s) only - all docs match
+            Some(Box::new(AllScorer::new(max_doc)))
+        } else {
+            // No MUST constraint at all
+            None
+        }
+    } else {
+        Some(intersect_scorers(must_scorers, num_docs))
+    }
+}
+
+/// Returns a SHOULD scorer with AllScorer union if any were removed.
+///
+/// For union semantics (OR): if any SHOULD clause was an AllScorer, the result
+/// should include all documents. We restore this by unioning with AllScorer.
+///
+/// When `scoring_enabled` is false, we can just return AllScorer alone since
+/// we don't need score contributions from the should_scorer.
+fn effective_should_scorer_for_union<TScoreCombiner: ScoreCombiner>(
+    should_scorer: SpecializedScorer,
+    removed_all_scorer_count: usize,
+    max_doc: DocId,
+    num_docs: u32,
+    score_combiner_fn: impl Fn() -> TScoreCombiner,
+    scoring_enabled: bool,
+) -> SpecializedScorer {
+    if removed_all_scorer_count > 0 {
+        if scoring_enabled {
+            // Need to union to get score contributions from both
+            let all_scorers: Vec<Box<dyn Scorer>> = vec![
+                into_box_scorer(should_scorer, &score_combiner_fn, num_docs),
+                Box::new(AllScorer::new(max_doc)),
+            ];
+            SpecializedScorer::Other(Box::new(BufferedUnionScorer::build(
+                all_scorers,
+                score_combiner_fn,
+                num_docs,
+            )))
+        } else {
+            // Scoring disabled - AllScorer alone is sufficient
+            SpecializedScorer::Other(Box::new(AllScorer::new(max_doc)))
+        }
+    } else {
+        should_scorer
+    }
+}
+
 enum ShouldScorersCombinationMethod {
     // Should scorers are irrelevant.
     Ignored,
@@ -193,18 +252,18 @@ impl<TScoreCombiner: ScoreCombiner> BooleanWeight<TScoreCombiner> {
             return Ok(SpecializedScorer::Other(Box::new(EmptyScorer)));
         }
 
-        let minimum_number_should_match = self
+        let effective_minimum_number_should_match = self
             .minimum_number_should_match
             .saturating_sub(should_special_scorer_counts.num_all_scorers);
 
         let should_scorers: ShouldScorersCombinationMethod = {
             let num_of_should_scorers = should_scorers.len();
-            if minimum_number_should_match > num_of_should_scorers {
+            if effective_minimum_number_should_match > num_of_should_scorers {
                 // We don't have enough scorers to satisfy the minimum number of should matches.
                 // The request will match no documents.
                 return Ok(SpecializedScorer::Other(Box::new(EmptyScorer)));
             }
-            match minimum_number_should_match {
+            match effective_minimum_number_should_match {
                 0 if num_of_should_scorers == 0 => ShouldScorersCombinationMethod::Ignored,
                 0 => ShouldScorersCombinationMethod::Optional(scorer_union(
                     should_scorers,
@@ -226,7 +285,7 @@ impl<TScoreCombiner: ScoreCombiner> BooleanWeight<TScoreCombiner> {
                     scorer_disjunction(
                         should_scorers,
                         score_combiner_fn(),
-                        self.minimum_number_should_match,
+                        effective_minimum_number_should_match,
                     ),
                 )),
             }
@@ -246,53 +305,78 @@ impl<TScoreCombiner: ScoreCombiner> BooleanWeight<TScoreCombiner> {
 
         let include_scorer = match (should_scorers, must_scorers) {
             (ShouldScorersCombinationMethod::Ignored, must_scorers) => {
-                let boxed_scorer: Box<dyn Scorer> = if must_scorers.is_empty() {
-                    // We do not have any should scorers, nor all scorers.
-                    // There are still two cases here.
-                    //
-                    // If this follows the removal of some AllScorers in the should/must clauses,
-                    // then we match all documents.
-                    //
-                    // Otherwise, it is really just an EmptyScorer.
-                    if must_special_scorer_counts.num_all_scorers
-                        + should_special_scorer_counts.num_all_scorers
-                        > 0
-                    {
-                        Box::new(AllScorer::new(reader.max_doc()))
-                    } else {
-                        Box::new(EmptyScorer)
-                    }
-                } else {
-                    intersect_scorers(must_scorers, num_docs)
-                };
+                // No SHOULD clauses (or they were absorbed into MUST).
+                // Result depends entirely on MUST + any removed AllScorers.
+                let combined_all_scorer_count = must_special_scorer_counts.num_all_scorers
+                    + should_special_scorer_counts.num_all_scorers;
+                let boxed_scorer: Box<dyn Scorer> = effective_must_scorer(
+                    must_scorers,
+                    combined_all_scorer_count,
+                    reader.max_doc(),
+                    num_docs,
+                )
+                .unwrap_or_else(|| Box::new(EmptyScorer));
                 SpecializedScorer::Other(boxed_scorer)
             }
             (ShouldScorersCombinationMethod::Optional(should_scorer), must_scorers) => {
-                if must_scorers.is_empty() && must_special_scorer_counts.num_all_scorers == 0 {
-                    // Optional options are promoted to required if no must scorers exists.
-                    should_scorer
-                } else {
-                    let must_scorer = intersect_scorers(must_scorers, num_docs);
-                    if self.scoring_enabled {
-                        SpecializedScorer::Other(Box::new(RequiredOptionalScorer::<
-                            _,
-                            _,
-                            TScoreCombiner,
-                        >::new(
-                            must_scorer,
-                            into_box_scorer(should_scorer, &score_combiner_fn, num_docs),
-                        )))
-                    } else {
-                        SpecializedScorer::Other(must_scorer)
+                // Optional SHOULD: contributes to scoring but not required for matching.
+                match effective_must_scorer(
+                    must_scorers,
+                    must_special_scorer_counts.num_all_scorers,
+                    reader.max_doc(),
+                    num_docs,
+                ) {
+                    None => {
+                        // No MUST constraint: promote SHOULD to required.
+                        // Must preserve any removed AllScorers from SHOULD via union.
+                        effective_should_scorer_for_union(
+                            should_scorer,
+                            should_special_scorer_counts.num_all_scorers,
+                            reader.max_doc(),
+                            num_docs,
+                            &score_combiner_fn,
+                            self.scoring_enabled,
+                        )
+                    }
+                    Some(must_scorer) => {
+                        // Has MUST constraint: SHOULD only affects scoring.
+                        if self.scoring_enabled {
+                            SpecializedScorer::Other(Box::new(RequiredOptionalScorer::<
+                                _,
+                                _,
+                                TScoreCombiner,
+                            >::new(
+                                must_scorer,
+                                into_box_scorer(should_scorer, &score_combiner_fn, num_docs),
+                            )))
+                        } else {
+                            SpecializedScorer::Other(must_scorer)
+                        }
                     }
                 }
             }
-            (ShouldScorersCombinationMethod::Required(should_scorer), mut must_scorers) => {
-                if must_scorers.is_empty() {
-                    should_scorer
-                } else {
-                    must_scorers.push(into_box_scorer(should_scorer, &score_combiner_fn, num_docs));
-                    SpecializedScorer::Other(intersect_scorers(must_scorers, num_docs))
+            (ShouldScorersCombinationMethod::Required(should_scorer), must_scorers) => {
+                // Required SHOULD: at least `minimum_number_should_match` must match.
+                // Semantics: (MUST constraint) AND (SHOULD constraint)
+                match effective_must_scorer(
+                    must_scorers,
+                    must_special_scorer_counts.num_all_scorers,
+                    reader.max_doc(),
+                    num_docs,
+                ) {
+                    None => {
+                        // No MUST constraint: SHOULD alone determines matching.
+                        should_scorer
+                    }
+                    Some(must_scorer) => {
+                        // Has MUST constraint: intersect MUST with SHOULD.
+                        let should_boxed =
+                            into_box_scorer(should_scorer, &score_combiner_fn, num_docs);
+                        SpecializedScorer::Other(intersect_scorers(
+                            vec![must_scorer, should_boxed],
+                            num_docs,
+                        ))
+                    }
                 }
             }
         };

src/query/boolean_query/mod.rs

@@ -9,12 +9,14 @@ pub use self::boolean_weight::BooleanWeight;
 #[cfg(test)]
 mod tests {
 
+    use std::ops::Bound;
+
     use super::*;
     use crate::collector::tests::TEST_COLLECTOR_WITH_SCORE;
-    use crate::collector::TopDocs;
+    use crate::collector::{Count, TopDocs};
     use crate::query::term_query::TermScorer;
     use crate::query::{
-        AllScorer, EmptyScorer, EnableScoring, Intersection, Occur, Query, QueryParser,
+        AllScorer, EmptyScorer, EnableScoring, Intersection, Occur, Query, QueryParser, RangeQuery,
         RequiredOptionalScorer, Scorer, SumCombiner, TermQuery,
     };
     use crate::schema::*;
@@ -374,4 +376,466 @@ mod tests {
         }
         Ok(())
     }
+
+    #[test]
+    pub fn test_min_should_match_with_all_query() -> crate::Result<()> {
+        let mut schema_builder = Schema::builder();
+        let text_field = schema_builder.add_text_field("text", TEXT);
+        let num_field =
+            schema_builder.add_i64_field("num", NumericOptions::default().set_fast().set_indexed());
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        let mut index_writer: IndexWriter = index.writer_for_tests()?;
+
+        index_writer.add_document(doc!(text_field => "apple", num_field => 10i64))?;
+        index_writer.add_document(doc!(text_field => "banana", num_field => 20i64))?;
+        index_writer.commit()?;
+
+        let searcher = index.reader()?.searcher();
+
+        let effective_all_match_query: Box<dyn Query> = Box::new(RangeQuery::new(
+            Bound::Excluded(Term::from_field_i64(num_field, 0)),
+            Bound::Unbounded,
+        ));
+        let term_query: Box<dyn Query> = Box::new(TermQuery::new(
+            Term::from_field_text(text_field, "apple"),
+            IndexRecordOption::Basic,
+        ));
+
+        // in some previous version, we would remove the 2 all_match, but then say we need *4*
+        // matches out of the 3 term queries, which matches nothing.
+        let mut bool_query = BooleanQuery::new(vec![
+            (Occur::Should, effective_all_match_query.box_clone()),
+            (Occur::Should, effective_all_match_query.box_clone()),
+            (Occur::Should, term_query.box_clone()),
+            (Occur::Should, term_query.box_clone()),
+            (Occur::Should, term_query.box_clone()),
+        ]);
+        bool_query.set_minimum_number_should_match(4);
+        let count = searcher.search(&bool_query, &Count)?;
+        assert_eq!(count, 1);
+
+        Ok(())
+    }
+
+    // =========================================================================
+    // AllScorer Preservation Regression Tests
+    // =========================================================================
+    //
+    // These tests verify the fix for a bug where AllScorer instances (produced by
+    // queries matching all documents, such as range queries covering all values)
+    // were incorrectly removed from Boolean query processing, causing documents
+    // to be unexpectedly excluded from results.
+    //
+    // The bug manifested in several scenarios:
+    // 1. SHOULD + SHOULD where one clause is AllScorer
+    // 2. MUST (AllScorer) + SHOULD
+    // 3. Range queries in Boolean clauses when all documents match the range
+
+    /// Regression test: SHOULD clause with AllScorer combined with other SHOULD clauses.
+    ///
+    /// When a SHOULD clause produces an AllScorer (e.g., from a range query matching
+    /// all documents), the Boolean query should still match all documents.
+    ///
+    /// Bug before fix: AllScorer was removed during optimization, leaving only the
+    /// other SHOULD clauses, which incorrectly excluded documents.
+    #[test]
+    pub fn test_should_with_all_scorer_regression() -> crate::Result<()> {
+        let mut schema_builder = Schema::builder();
+        let text_field = schema_builder.add_text_field("text", TEXT);
+        let num_field =
+            schema_builder.add_i64_field("num", NumericOptions::default().set_fast().set_indexed());
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        let mut index_writer: IndexWriter = index.writer_for_tests()?;
+
+        // All docs have num > 0, so range query will return AllScorer
+        index_writer.add_document(doc!(text_field => "hello", num_field => 10i64))?;
+        index_writer.add_document(doc!(text_field => "world", num_field => 20i64))?;
+        index_writer.add_document(doc!(text_field => "hello world", num_field => 30i64))?;
+        index_writer.add_document(doc!(text_field => "foo", num_field => 40i64))?;
+        index_writer.add_document(doc!(text_field => "bar", num_field => 50i64))?;
+        index_writer.add_document(doc!(text_field => "baz", num_field => 60i64))?;
+        index_writer.commit()?;
+
+        let searcher = index.reader()?.searcher();
+
+        // Range query matching all docs (returns AllScorer)
+        let all_match_query: Box<dyn Query> = Box::new(RangeQuery::new(
+            Bound::Excluded(Term::from_field_i64(num_field, 0)),
+            Bound::Unbounded,
+        ));
+        let term_query: Box<dyn Query> = Box::new(TermQuery::new(
+            Term::from_field_text(text_field, "hello"),
+            IndexRecordOption::Basic,
+        ));
+
+        // Verify range matches all 6 docs
+        assert_eq!(searcher.search(all_match_query.as_ref(), &Count)?, 6);
+
+        // RangeQuery(all) OR TermQuery should match all 6 docs
+        let bool_query = BooleanQuery::new(vec![
+            (Occur::Should, all_match_query.box_clone()),
+            (Occur::Should, term_query.box_clone()),
+        ]);
+        let count = searcher.search(&bool_query, &Count)?;
+        assert_eq!(count, 6, "SHOULD with AllScorer should match all docs");
+
+        // Order should not matter
+        let bool_query_reversed = BooleanQuery::new(vec![
+            (Occur::Should, term_query.box_clone()),
+            (Occur::Should, all_match_query.box_clone()),
+        ]);
+        let count_reversed = searcher.search(&bool_query_reversed, &Count)?;
+        assert_eq!(
+            count_reversed, 6,
+            "Order of SHOULD clauses should not matter"
+        );
+
+        Ok(())
+    }
+
+    /// Regression test: MUST clause with AllScorer combined with SHOULD clause.
+    ///
+    /// When MUST contains an AllScorer, all documents satisfy the MUST constraint.
+    /// The SHOULD clause should only affect scoring, not filtering.
+    ///
+    /// Bug before fix: AllScorer was removed, leaving an empty must_scorers vector.
+    /// intersect_scorers([]) incorrectly returned EmptyScorer, matching 0 documents.
+    #[test]
+    pub fn test_must_all_with_should_regression() -> crate::Result<()> {
+        let mut schema_builder = Schema::builder();
+        let text_field = schema_builder.add_text_field("text", TEXT);
+        let num_field =
+            schema_builder.add_i64_field("num", NumericOptions::default().set_fast().set_indexed());
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        let mut index_writer: IndexWriter = index.writer_for_tests()?;
+
+        // All docs have num > 0, so range query will return AllScorer
+        index_writer.add_document(doc!(text_field => "apple", num_field => 10i64))?;
+        index_writer.add_document(doc!(text_field => "banana", num_field => 20i64))?;
+        index_writer.add_document(doc!(text_field => "cherry", num_field => 30i64))?;
+        index_writer.add_document(doc!(text_field => "date", num_field => 40i64))?;
+        index_writer.commit()?;
+
+        let searcher = index.reader()?.searcher();
+
+        // Range query matching all docs (returns AllScorer)
+        let all_match_query: Box<dyn Query> = Box::new(RangeQuery::new(
+            Bound::Excluded(Term::from_field_i64(num_field, 0)),
+            Bound::Unbounded,
+        ));
+        let term_query: Box<dyn Query> = Box::new(TermQuery::new(
+            Term::from_field_text(text_field, "apple"),
+            IndexRecordOption::Basic,
+        ));
+
+        // Verify range matches all 4 docs
+        assert_eq!(searcher.search(all_match_query.as_ref(), &Count)?, 4);
+
+        // MUST(range matching all) AND SHOULD(term) should match all 4 docs
+        let bool_query = BooleanQuery::new(vec![
+            (Occur::Must, all_match_query.box_clone()),
+            (Occur::Should, term_query.box_clone()),
+        ]);
+        let count = searcher.search(&bool_query, &Count)?;
+        assert_eq!(count, 4, "MUST AllScorer + SHOULD should match all docs");
+
+        Ok(())
+    }
+
+    /// Regression test: Range queries in Boolean clauses when all documents match.
+    ///
+    /// Range queries can return AllScorer as an optimization when all indexed values
+    /// fall within the range. This test ensures such queries work correctly in
+    /// Boolean combinations.
+    ///
+    /// This is the most common real-world manifestation of the bug, occurring in
+    /// queries like: (age > 50 OR name = 'Alice') AND status = 'active'
+    /// when all documents have age > 50.
+    #[test]
+    pub fn test_range_query_all_match_in_boolean() -> crate::Result<()> {
+        let mut schema_builder = Schema::builder();
+        let name_field = schema_builder.add_text_field("name", TEXT);
+        let age_field =
+            schema_builder.add_i64_field("age", NumericOptions::default().set_fast().set_indexed());
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        let mut index_writer: IndexWriter = index.writer_for_tests()?;
+
+        // All documents have age > 50, so range query will return AllScorer
+        index_writer.add_document(doc!(name_field => "alice", age_field => 55_i64))?;
+        index_writer.add_document(doc!(name_field => "bob", age_field => 60_i64))?;
+        index_writer.add_document(doc!(name_field => "charlie", age_field => 70_i64))?;
+        index_writer.add_document(doc!(name_field => "diana", age_field => 80_i64))?;
+        index_writer.commit()?;
+
+        let searcher = index.reader()?.searcher();
+
+        let range_query: Box<dyn Query> = Box::new(RangeQuery::new(
+            Bound::Excluded(Term::from_field_i64(age_field, 50)),
+            Bound::Unbounded,
+        ));
+        let term_query: Box<dyn Query> = Box::new(TermQuery::new(
+            Term::from_field_text(name_field, "alice"),
+            IndexRecordOption::Basic,
+        ));
+
+        // Verify preconditions
+        assert_eq!(searcher.search(range_query.as_ref(), &Count)?, 4);
+        assert_eq!(searcher.search(term_query.as_ref(), &Count)?, 1);
+
+        // SHOULD(range) OR SHOULD(term): range matches all, so result is 4
+        let should_query = BooleanQuery::new(vec![
+            (Occur::Should, range_query.box_clone()),
+            (Occur::Should, term_query.box_clone()),
+        ]);
+        assert_eq!(
+            searcher.search(&should_query, &Count)?,
+            4,
+            "SHOULD range OR term should match all"
+        );
+
+        // MUST(range) AND SHOULD(term): range matches all, term is optional
+        let must_should_query = BooleanQuery::new(vec![
+            (Occur::Must, range_query.box_clone()),
+            (Occur::Should, term_query.box_clone()),
+        ]);
+        assert_eq!(
+            searcher.search(&must_should_query, &Count)?,
+            4,
+            "MUST range + SHOULD term should match all"
+        );
+
+        Ok(())
+    }
+
+    /// Test multiple AllScorer instances in different clause types.
+    ///
+    /// Verifies correct behavior when AllScorers appear in multiple positions.
+    #[test]
+    pub fn test_multiple_all_scorers() -> crate::Result<()> {
+        let mut schema_builder = Schema::builder();
+        let text_field = schema_builder.add_text_field("text", TEXT);
+        let num_field =
+            schema_builder.add_i64_field("num", NumericOptions::default().set_fast().set_indexed());
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        let mut index_writer: IndexWriter = index.writer_for_tests()?;
+
+        // All docs have num > 0, so range queries will return AllScorer
+        index_writer.add_document(doc!(text_field => "doc1", num_field => 10i64))?;
+        index_writer.add_document(doc!(text_field => "doc2", num_field => 20i64))?;
+        index_writer.add_document(doc!(text_field => "doc3", num_field => 30i64))?;
+        index_writer.commit()?;
+
+        let searcher = index.reader()?.searcher();
+
+        // Two different range queries that both match all docs (return AllScorer)
+        let all_query1: Box<dyn Query> = Box::new(RangeQuery::new(
+            Bound::Excluded(Term::from_field_i64(num_field, 0)),
+            Bound::Unbounded,
+        ));
+        let all_query2: Box<dyn Query> = Box::new(RangeQuery::new(
+            Bound::Excluded(Term::from_field_i64(num_field, 5)),
+            Bound::Unbounded,
+        ));
+        let term_query: Box<dyn Query> = Box::new(TermQuery::new(
+            Term::from_field_text(text_field, "doc1"),
+            IndexRecordOption::Basic,
+        ));
+
+        // Multiple AllScorers in SHOULD
+        let multi_all_should = BooleanQuery::new(vec![
+            (Occur::Should, all_query1.box_clone()),
+            (Occur::Should, all_query2.box_clone()),
+            (Occur::Should, term_query.box_clone()),
+        ]);
+        assert_eq!(
+            searcher.search(&multi_all_should, &Count)?,
+            3,
+            "Multiple AllScorers in SHOULD"
+        );
+
+        // AllScorer in both MUST and SHOULD
+        let all_must_and_should = BooleanQuery::new(vec![
+            (Occur::Must, all_query1.box_clone()),
+            (Occur::Should, all_query2.box_clone()),
+        ]);
+        assert_eq!(
+            searcher.search(&all_must_and_should, &Count)?,
+            3,
+            "AllScorer in both MUST and SHOULD"
+        );
+
+        Ok(())
+    }
+}
+
+/// A proptest which generates arbitrary permutations of a simple boolean AST, and then matches
+/// the result against an index which contains all permutations of documents with N fields.
+#[cfg(test)]
+mod proptest_boolean_query {
+    use std::collections::{BTreeMap, HashSet};
+    use std::ops::{Bound, Range};
+
+    use proptest::collection::vec;
+    use proptest::prelude::*;
+
+    use crate::collector::DocSetCollector;
+    use crate::query::{AllQuery, BooleanQuery, Occur, Query, RangeQuery, TermQuery};
+    use crate::schema::{Field, NumericOptions, OwnedValue, Schema, TEXT};
+    use crate::{DocId, Index, Term};
+
+    #[derive(Debug, Clone)]
+    enum BooleanQueryAST {
+        /// Matches all documents via AllQuery (wraps AllScorer in BoostScorer)
+        All,
+        /// Matches all documents via RangeQuery (returns bare AllScorer)
+        /// This is the actual trigger for the AllScorer preservation bug
+        RangeAll,
+        /// Matches documents where the field has value "true"
+        Leaf {
+            field_idx: usize,
+        },
+        Union(Vec<BooleanQueryAST>),
+        Intersection(Vec<BooleanQueryAST>),
+    }
+
+    impl BooleanQueryAST {
+        fn matches(&self, doc_id: DocId) -> bool {
+            match self {
+                BooleanQueryAST::All => true,
+                BooleanQueryAST::RangeAll => true,
+                BooleanQueryAST::Leaf { field_idx } => Self::matches_field(doc_id, *field_idx),
+                BooleanQueryAST::Union(children) => {
+                    children.iter().any(|child| child.matches(doc_id))
+                }
+                BooleanQueryAST::Intersection(children) => {
+                    children.iter().all(|child| child.matches(doc_id))
+                }
+            }
+        }
+
+        fn matches_field(doc_id: DocId, field_idx: usize) -> bool {
+            ((doc_id as usize) >> field_idx) & 1 == 1
+        }
+
+        fn to_query(&self, fields: &[Field], range_field: Field) -> Box<dyn Query> {
+            match self {
+                BooleanQueryAST::All => Box::new(AllQuery),
+                BooleanQueryAST::RangeAll => {
+                    // Range query that matches all docs (all have value >= 0)
+                    // This returns bare AllScorer, triggering the bug we fixed
+                    Box::new(RangeQuery::new(
+                        Bound::Included(Term::from_field_i64(range_field, 0)),
+                        Bound::Unbounded,
+                    ))
+                }
+                BooleanQueryAST::Leaf { field_idx } => Box::new(TermQuery::new(
+                    Term::from_field_text(fields[*field_idx], "true"),
+                    crate::schema::IndexRecordOption::Basic,
+                )),
+                BooleanQueryAST::Union(children) => {
+                    let sub_queries = children
+                        .iter()
+                        .map(|child| (Occur::Should, child.to_query(fields, range_field)))
+                        .collect();
+                    Box::new(BooleanQuery::new(sub_queries))
+                }
+                BooleanQueryAST::Intersection(children) => {
+                    let sub_queries = children
+                        .iter()
+                        .map(|child| (Occur::Must, child.to_query(fields, range_field)))
+                        .collect();
+                    Box::new(BooleanQuery::new(sub_queries))
+                }
+            }
+        }
+    }
+
+    fn doc_ids(num_docs: usize, num_fields: usize) -> Range<DocId> {
+        let permutations = 1 << num_fields;
+        let copies = (num_docs as f32 / permutations as f32).ceil() as u32;
+        0..(permutations * copies)
+    }
+
+    fn create_index_with_boolean_permutations(
+        num_docs: usize,
+        num_fields: usize,
+    ) -> (Index, Vec<Field>, Field) {
+        let mut schema_builder = Schema::builder();
+        let fields: Vec<Field> = (0..num_fields)
+            .map(|i| schema_builder.add_text_field(&format!("field_{}", i), TEXT))
+            .collect();
+        // Add a numeric field for RangeQuery tests - all docs have value = doc_id
+        let range_field = schema_builder.add_i64_field(
+            "range_field",
+            NumericOptions::default().set_fast().set_indexed(),
+        );
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        let mut writer = index.writer_for_tests().unwrap();
+
+        for doc_id in doc_ids(num_docs, num_fields) {
+            let mut doc: BTreeMap<_, OwnedValue> = BTreeMap::default();
+            for (field_idx, &field) in fields.iter().enumerate() {
+                if (doc_id >> field_idx) & 1 == 1 {
+                    doc.insert(field, "true".into());
+                }
+            }
+            // All docs have non-negative values, so RangeQuery(>=0) matches all
+            doc.insert(range_field, (doc_id as i64).into());
+            writer.add_document(doc).unwrap();
+        }
+        writer.commit().unwrap();
+        (index, fields, range_field)
+    }
+
+    fn arb_boolean_query_ast(num_fields: usize) -> impl Strategy<Value = BooleanQueryAST> {
+        // Leaf strategies: term queries, AllQuery, and RangeQuery matching all docs
+        let leaf = prop_oneof![
+            (0..num_fields).prop_map(|field_idx| BooleanQueryAST::Leaf { field_idx }),
+            Just(BooleanQueryAST::All),
+            Just(BooleanQueryAST::RangeAll),
+        ];
+        leaf.prop_recursive(
+            8,   // 8 levels of recursion
+            256, // 256 nodes max
+            10,  // 10 items per collection
+            |inner| {
+                prop_oneof![
+                    vec(inner.clone(), 1..10).prop_map(BooleanQueryAST::Union),
+                    vec(inner, 1..10).prop_map(BooleanQueryAST::Intersection),
+                ]
+            },
+        )
+    }
+
+    #[test]
+    fn proptest_boolean_query() {
+        // In the presence of optimizations around buffering, it can take large numbers of
+        // documents to uncover some issues.
+        let num_fields = 8;
+        let num_docs = 1 << num_fields;
+        let (index, fields, range_field) =
+            create_index_with_boolean_permutations(num_docs, num_fields);
+        let searcher = index.reader().unwrap().searcher();
+        proptest!(|(ast in arb_boolean_query_ast(num_fields))| {
+            let query = ast.to_query(&fields, range_field);
+
+            let mut matching_docs = HashSet::new();
+            for doc_id in doc_ids(num_docs, num_fields) {
+                if ast.matches(doc_id as DocId) {
+                    matching_docs.insert(doc_id as DocId);
+                }
+            }
+
+            let doc_addresses = searcher.search(&*query, &DocSetCollector).unwrap();
+            let result_docs: HashSet<DocId> =
+                doc_addresses.into_iter().map(|doc_address| doc_address.doc_id).collect();
+            prop_assert_eq!(result_docs, matching_docs);
+        });
+    }
 }

src/query/boost_query.rs

@@ -104,6 +104,9 @@ impl<S: Scorer> DocSet for BoostScorer<S> {
     fn seek(&mut self, target: DocId) -> DocId {
         self.underlying.seek(target)
     }
+    fn seek_into_the_danger_zone(&mut self, target: DocId) -> bool {
+        self.underlying.seek_into_the_danger_zone(target)
+    }
 
     fn fill_buffer(&mut self, buffer: &mut [DocId; COLLECT_BLOCK_BUFFER_LEN]) -> usize {
         self.underlying.fill_buffer(buffer)
@@ -131,6 +134,7 @@ impl<S: Scorer> DocSet for BoostScorer<S> {
 }
 
 impl<S: Scorer> Scorer for BoostScorer<S> {
+    #[inline]
     fn score(&mut self) -> Score {
         self.underlying.score() * self.boost
     }

src/query/const_score_query.rs

@@ -137,6 +137,7 @@ impl<TDocSet: DocSet> DocSet for ConstScorer<TDocSet> {
 }
 
 impl<TDocSet: DocSet + 'static> Scorer for ConstScorer<TDocSet> {
+    #[inline]
     fn score(&mut self) -> Score {
         self.score
     }

src/query/disjunction.rs

@@ -62,6 +62,16 @@ impl<T: Scorer> DocSet for ScorerWrapper<T> {
         self.current_doc = doc_id;
         doc_id
     }
+    fn seek(&mut self, target: DocId) -> DocId {
+        let doc_id = self.scorer.seek(target);
+        self.current_doc = doc_id;
+        doc_id
+    }
+    fn seek_into_the_danger_zone(&mut self, target: DocId) -> bool {
+        let found = self.scorer.seek_into_the_danger_zone(target);
+        self.current_doc = self.scorer.doc();
+        found
+    }
 
     fn doc(&self) -> DocId {
         self.current_doc
@@ -163,6 +173,7 @@ impl<TScorer: Scorer, TScoreCombiner: ScoreCombiner> DocSet
 impl<TScorer: Scorer, TScoreCombiner: ScoreCombiner> Scorer
     for Disjunction<TScorer, TScoreCombiner>
 {
+    #[inline]
     fn score(&mut self) -> Score {
         self.current_score
     }
@@ -297,6 +308,7 @@ mod tests {
     }
 
     impl Scorer for DummyScorer {
+        #[inline]
         fn score(&mut self) -> Score {
             self.foo.get(self.cursor).map(|x| x.1).unwrap_or(0.0)
         }

src/query/empty_query.rs

@@ -55,6 +55,7 @@ impl DocSet for EmptyScorer {
 }
 
 impl Scorer for EmptyScorer {
+    #[inline]
     fn score(&mut self) -> Score {
         0.0
     }

src/query/exclude.rs

@@ -84,6 +84,7 @@ where
     TScorer: Scorer,
     TDocSetExclude: DocSet + 'static,
 {
+    #[inline]
     fn score(&mut self) -> Score {
         self.underlying_docset.score()
     }

src/query/intersection.rs

@@ -1,5 +1,5 @@
+use super::size_hint::estimate_intersection;
 use crate::docset::{DocSet, TERMINATED};
-use crate::query::size_hint::estimate_intersection;
 use crate::query::term_query::TermScorer;
 use crate::query::{EmptyScorer, Scorer};
 use crate::{DocId, Score};
@@ -12,6 +12,9 @@ use crate::{DocId, Score};
 /// For better performance, the function uses a
 /// specialized implementation if the two
 /// shortest scorers are `TermScorer`s.
+///
+/// num_docs_segment is the number of documents in the segment. It is used for estimating the
+/// `size_hint` of the intersection.
 pub fn intersect_scorers(
     mut scorers: Vec<Box<dyn Scorer>>,
     num_docs_segment: u32,
@@ -102,35 +105,48 @@ impl<TDocSet: DocSet> Intersection<TDocSet, TDocSet> {
 }
 
 impl<TDocSet: DocSet, TOtherDocSet: DocSet> DocSet for Intersection<TDocSet, TOtherDocSet> {
+    #[inline]
     fn advance(&mut self) -> DocId {
         let (left, right) = (&mut self.left, &mut self.right);
         let mut candidate = left.advance();
+        if candidate == TERMINATED {
+            return TERMINATED;
+        }
 
-        'outer: loop {
+        loop {
             // In the first part we look for a document in the intersection
             // of the two rarest `DocSet` in the intersection.
 
             loop {
-                let right_doc = right.seek(candidate);
-                candidate = left.seek(right_doc);
-                if candidate == right_doc {
+                if right.seek_into_the_danger_zone(candidate) {
                     break;
                 }
+                let right_doc = right.doc();
+                // TODO: Think about which value would make sense here
+                // It depends on the DocSet implementation, when a seek would outweigh an advance.
+                if right_doc > candidate.wrapping_add(100) {
+                    candidate = left.seek(right_doc);
+                } else {
+                    candidate = left.advance();
+                }
+                if candidate == TERMINATED {
+                    return TERMINATED;
+                }
             }
 
             debug_assert_eq!(left.doc(), right.doc());
-            // test the remaining scorers;
-            for docset in self.others.iter_mut() {
-                let seek_doc = docset.seek(candidate);
-                if seek_doc > candidate {
-                    candidate = left.seek(seek_doc);
-                    continue 'outer;
-                }
+            // test the remaining scorers
+            if self
+                .others
+                .iter_mut()
+                .all(|docset| docset.seek_into_the_danger_zone(candidate))
+            {
+                debug_assert_eq!(candidate, self.left.doc());
+                debug_assert_eq!(candidate, self.right.doc());
+                debug_assert!(self.others.iter().all(|docset| docset.doc() == candidate));
+                return candidate;
             }
-            debug_assert_eq!(candidate, self.left.doc());
-            debug_assert_eq!(candidate, self.right.doc());
-            debug_assert!(self.others.iter().all(|docset| docset.doc() == candidate));
-            return candidate;
+            candidate = left.advance();
         }
     }
 
@@ -146,6 +162,20 @@ impl<TDocSet: DocSet, TOtherDocSet: DocSet> DocSet for Intersection<TDocSet, TOt
         doc
     }
 
+    /// Seeks to the target if necessary and checks if the target is an exact match.
+    ///
+    /// Some implementations may choose to advance past the target if beneficial for performance.
+    /// The return value is `true` if the target is in the docset, and `false` otherwise.
+    fn seek_into_the_danger_zone(&mut self, target: DocId) -> bool {
+        self.left.seek_into_the_danger_zone(target)
+            && self.right.seek_into_the_danger_zone(target)
+            && self
+                .others
+                .iter_mut()
+                .all(|docset| docset.seek_into_the_danger_zone(target))
+    }
+
+    #[inline]
     fn doc(&self) -> DocId {
         self.left.doc()
     }
@@ -172,6 +202,7 @@ where
     TScorer: Scorer,
     TOtherScorer: Scorer,
 {
+    #[inline]
     fn score(&mut self) -> Score {
         self.left.score()
             + self.right.score()
@@ -181,6 +212,8 @@ where
 
 #[cfg(test)]
 mod tests {
+    use proptest::prelude::*;
+
     use super::Intersection;
     use crate::docset::{DocSet, TERMINATED};
     use crate::postings::tests::test_skip_against_unoptimized;
@@ -270,4 +303,38 @@ mod tests {
         let intersection = Intersection::new(vec![a, b, c], 10);
         assert_eq!(intersection.doc(), TERMINATED);
     }
+
+    // Strategy to generate sorted and deduplicated vectors of u32 document IDs
+    fn sorted_deduped_vec(max_val: u32, max_size: usize) -> impl Strategy<Value = Vec<u32>> {
+        prop::collection::vec(0..max_val, 0..max_size).prop_map(|mut vec| {
+            vec.sort();
+            vec.dedup();
+            vec
+        })
+    }
+
+    proptest! {
+        #[test]
+        fn prop_test_intersection_consistency(
+            a in sorted_deduped_vec(100, 10),
+            b in sorted_deduped_vec(100, 10),
+            num_docs in 100u32..500u32
+        ) {
+            let left = VecDocSet::from(a.clone());
+            let right = VecDocSet::from(b.clone());
+            let mut intersection = Intersection::new(vec![left, right], num_docs);
+
+            let expected: Vec<u32> = a.iter()
+                .cloned()
+                .filter(|doc| b.contains(doc))
+                .collect();
+
+            for expected_doc in expected {
+                assert_eq!(intersection.doc(), expected_doc);
+                intersection.advance();
+            }
+            assert_eq!(intersection.doc(), TERMINATED);
+        }
+
+    }
 }

src/query/mod.rs

@@ -70,9 +70,83 @@ pub use self::weight::Weight;
 
 #[cfg(test)]
 mod tests {
+    use crate::collector::TopDocs;
+    use crate::query::phrase_query::tests::create_index;
     use crate::query::QueryParser;
     use crate::schema::{Schema, TEXT};
-    use crate::{Index, Term};
+    use crate::{DocAddress, Index, Term};
+
+    #[test]
+    pub fn test_mixed_intersection_and_union() -> crate::Result<()> {
+        let index = create_index(&["a b", "a c", "a b c", "b"])?;
+        let schema = index.schema();
+        let text_field = schema.get_field("text").unwrap();
+        let searcher = index.reader()?.searcher();
+
+        let do_search = |term: &str| {
+            let query = QueryParser::for_index(&index, vec![text_field])
+                .parse_query(term)
+                .unwrap();
+            let top_docs: Vec<(f32, DocAddress)> = searcher
+                .search(&query, &TopDocs::with_limit(10).order_by_score())
+                .unwrap();
+
+            top_docs.iter().map(|el| el.1.doc_id).collect::<Vec<_>>()
+        };
+
+        assert_eq!(do_search("a AND b"), vec![0, 2]);
+        assert_eq!(do_search("(a OR b) AND C"), vec![2, 1]);
+        // The intersection code has special code for more than 2 intersections
+        // left, right + others
+        // The will place the union in the "others" insersection to that seek_into_the_danger_zone
+        // is called
+        assert_eq!(
+            do_search("(a OR b) AND (c OR a) AND (b OR c)"),
+            vec![2, 1, 0]
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    pub fn test_mixed_intersection_and_union_with_skip() -> crate::Result<()> {
+        // Test 4096 skip in BufferedUnionScorer
+        let mut data: Vec<&str> = Vec::new();
+        data.push("a b");
+        let zz_data = vec!["z z"; 5000];
+        data.extend_from_slice(&zz_data);
+        data.extend_from_slice(&["a c"]);
+        data.extend_from_slice(&zz_data);
+        data.extend_from_slice(&["a b c", "b"]);
+        let index = create_index(&data)?;
+        let schema = index.schema();
+        let text_field = schema.get_field("text").unwrap();
+        let searcher = index.reader()?.searcher();
+
+        let do_search = |term: &str| {
+            let query = QueryParser::for_index(&index, vec![text_field])
+                .parse_query(term)
+                .unwrap();
+            let top_docs: Vec<(f32, DocAddress)> = searcher
+                .search(&query, &TopDocs::with_limit(10).order_by_score())
+                .unwrap();
+
+            top_docs.iter().map(|el| el.1.doc_id).collect::<Vec<_>>()
+        };
+
+        assert_eq!(do_search("a AND b"), vec![0, 10002]);
+        assert_eq!(do_search("(a OR b) AND C"), vec![10002, 5001]);
+        // The intersection code has special code for more than 2 intersections
+        // left, right + others
+        // The will place the union in the "others" insersection to that seek_into_the_danger_zone
+        // is called
+        assert_eq!(
+            do_search("(a OR b) AND (c OR a) AND (b OR c)"),
+            vec![10002, 5001, 0]
+        );
+
+        Ok(())
+    }
 
     #[test]
     fn test_query_terms() {

src/query/phrase_prefix_query/phrase_prefix_scorer.rs

@@ -81,6 +81,7 @@ impl<TPostings: Postings> DocSet for PhraseKind<TPostings> {
 }
 
 impl<TPostings: Postings> Scorer for PhraseKind<TPostings> {
+    #[inline]
     fn score(&mut self) -> Score {
         match self {
             PhraseKind::SinglePrefix { positions, .. } => {
@@ -193,6 +194,14 @@ impl<TPostings: Postings> DocSet for PhrasePrefixScorer<TPostings> {
         self.advance()
     }
 
+    fn seek_into_the_danger_zone(&mut self, target: DocId) -> bool {
+        if self.phrase_scorer.seek_into_the_danger_zone(target) {
+            self.matches_prefix()
+        } else {
+            false
+        }
+    }
+
     fn doc(&self) -> DocId {
         self.phrase_scorer.doc()
     }
@@ -207,6 +216,7 @@ impl<TPostings: Postings> DocSet for PhrasePrefixScorer<TPostings> {
 }
 
 impl<TPostings: Postings> Scorer for PhrasePrefixScorer<TPostings> {
+    #[inline]
     fn score(&mut self) -> Score {
         // TODO modify score??
         self.phrase_scorer.score()

src/query/phrase_query/phrase_scorer.rs

@@ -382,8 +382,9 @@ impl<TPostings: Postings> PhraseScorer<TPostings> {
                 PostingsWithOffset::new(postings, (max_offset - offset) as u32)
             })
             .collect::<Vec<_>>();
+        let intersection_docset = Intersection::new(postings_with_offsets, num_docs);
         let mut scorer = PhraseScorer {
-            intersection_docset: Intersection::new(postings_with_offsets, num_docs),
+            intersection_docset,
             num_terms: num_docsets,
             left_positions: Vec::with_capacity(100),
             right_positions: Vec::with_capacity(100),
@@ -529,25 +530,40 @@ impl<TPostings: Postings> DocSet for PhraseScorer<TPostings> {
         self.advance()
     }
 
+    fn seek_into_the_danger_zone(&mut self, target: DocId) -> bool {
+        debug_assert!(target >= self.doc());
+        if self.intersection_docset.seek_into_the_danger_zone(target) && self.phrase_match() {
+            return true;
+        }
+        false
+    }
+
     fn doc(&self) -> DocId {
         self.intersection_docset.doc()
     }
 
     fn size_hint(&self) -> u32 {
-        self.intersection_docset.size_hint()
+        // We adjust the intersection estimate, since actual phrase hits are much lower than where
+        // the all appear.
+        // The estimate should depend on average field length, e.g. if the field is really short
+        // a phrase hit is more likely
+        self.intersection_docset.size_hint() / (10 * self.num_terms as u32)
     }
 
     /// Returns a best-effort hint of the
     /// cost to drive the docset.
     fn cost(&self) -> u64 {
-        // Evaluating phrase matches is generally more expensive than simple term matches,
-        // as it requires loading and comparing positions. Use a conservative multiplier
-        // based on the number of terms.
+        // While determing a potential hit is cheap for phrases, evaluating an actual hit is
+        // expensive since it requires to load positions for a doc and check if they are next to
+        // each other.
+        // So the cost estimation would be the number of times we need to check if a doc is a hit *
+        // 10 * self.num_terms.
         self.intersection_docset.size_hint() as u64 * 10 * self.num_terms as u64
     }
 }
 
 impl<TPostings: Postings> Scorer for PhraseScorer<TPostings> {
+    #[inline]
     fn score(&mut self) -> Score {
         let doc = self.doc();
         let fieldnorm_id = self.fieldnorm_reader.fieldnorm_id(doc);

src/query/range_query/fast_field_range_doc_set.rs

@@ -62,6 +62,17 @@ pub(crate) struct RangeDocSet<T> {
 const DEFAULT_FETCH_HORIZON: u32 = 128;
 impl<T: Send + Sync + PartialOrd + Copy + Debug + 'static> RangeDocSet<T> {
     pub(crate) fn new(value_range: RangeInclusive<T>, column: Column<T>) -> Self {
+        if *value_range.start() > column.max_value() || *value_range.end() < column.min_value() {
+            return Self {
+                value_range,
+                column,
+                loaded_docs: VecCursor::new(),
+                next_fetch_start: TERMINATED,
+                fetch_horizon: DEFAULT_FETCH_HORIZON,
+                last_seek_pos_opt: None,
+            };
+        }
+
         let mut range_docset = Self {
             value_range,
             column,
@@ -81,6 +92,9 @@ impl<T: Send + Sync + PartialOrd + Copy + Debug + 'static> RangeDocSet<T> {
 
     /// Returns true if more data could be fetched
     fn fetch_block(&mut self) {
+        if self.next_fetch_start >= self.column.num_docs() {
+            return;
+        }
         const MAX_HORIZON: u32 = 100_000;
         while self.loaded_docs.is_empty() {
             let finished_to_end = self.fetch_horizon(self.fetch_horizon);
@@ -105,10 +119,10 @@ impl<T: Send + Sync + PartialOrd + Copy + Debug + 'static> RangeDocSet<T> {
     fn fetch_horizon(&mut self, horizon: u32) -> bool {
         let mut finished_to_end = false;
 
-        let limit = self.column.num_docs();
-        let mut end = self.next_fetch_start + horizon;
-        if end >= limit {
-            end = limit;
+        let num_docs = self.column.num_docs();
+        let mut fetch_end = self.next_fetch_start + horizon;
+        if fetch_end >= num_docs {
+            fetch_end = num_docs;
             finished_to_end = true;
         }
 
@@ -116,7 +130,7 @@ impl<T: Send + Sync + PartialOrd + Copy + Debug + 'static> RangeDocSet<T> {
         let doc_buffer: &mut Vec<DocId> = self.loaded_docs.get_cleared_data();
         self.column.get_docids_for_value_range(
             self.value_range.clone(),
-            self.next_fetch_start..end,
+            self.next_fetch_start..fetch_end,
             doc_buffer,
         );
         if let Some(last_doc) = last_doc {
@@ -124,7 +138,7 @@ impl<T: Send + Sync + PartialOrd + Copy + Debug + 'static> RangeDocSet<T> {
                 self.loaded_docs.next();
             }
         }
-        self.next_fetch_start = end;
+        self.next_fetch_start = fetch_end;
 
         finished_to_end
     }
@@ -136,9 +150,6 @@ impl<T: Send + Sync + PartialOrd + Copy + Debug + 'static> DocSet for RangeDocSe
         if let Some(docid) = self.loaded_docs.next() {
             return docid;
         }
-        if self.next_fetch_start >= self.column.num_docs() {
-            return TERMINATED;
-        }
         self.fetch_block();
         self.loaded_docs.current().unwrap_or(TERMINATED)
     }
@@ -174,15 +185,25 @@ impl<T: Send + Sync + PartialOrd + Copy + Debug + 'static> DocSet for RangeDocSe
     }
 
     fn size_hint(&self) -> u32 {
-        self.column.num_docs()
+        // TODO: Implement a better size hint
+        self.column.num_docs() / 10
     }
 
     /// Returns a best-effort hint of the
     /// cost to drive the docset.
     fn cost(&self) -> u64 {
-        // Advancing the docset is relatively expensive since it scans the column.
-        // Keep cost relative to a term query driver; use num_docs as baseline.
-        self.column.num_docs() as u64
+        // Advancing the docset is pretty expensive since it scans the whole column, there is no
+        // index currently (will change with an kd-tree)
+        // Since we use SIMD to scan the fast field range query we lower the cost a little bit,
+        // assuming that we hit 10% of the docs like in size_hint.
+        //
+        // If we would return a cost higher than num_docs, we would never choose ff range query as
+        // the driver in a DocSet, when intersecting a term query with a fast field. But
+        // it's the faster choice when the term query has a lot of docids and the range
+        // query has not.
+        //
+        // Ideally this would take the fast field codec into account
+        (self.column.num_docs() as f64 * 0.8) as u64
     }
 }
 
@@ -236,4 +257,52 @@ mod tests {
         let count = searcher.search(&query, &Count).unwrap();
         assert_eq!(count, 500);
     }
+
+    #[test]
+    fn range_query_no_overlap_optimization() {
+        let mut schema_builder = schema::SchemaBuilder::new();
+        let id_field = schema_builder.add_text_field("id", schema::STRING);
+        let value_field = schema_builder.add_u64_field("value", schema::FAST | schema::INDEXED);
+
+        let dir = RamDirectory::default();
+        let index = IndexBuilder::new()
+            .schema(schema_builder.build())
+            .open_or_create(dir)
+            .unwrap();
+
+        {
+            let mut writer = index.writer(15_000_000).unwrap();
+
+            // Add documents with values in the range [10, 20]
+            for i in 0..100 {
+                let mut doc = TantivyDocument::new();
+                doc.add_text(id_field, format!("doc{i}"));
+                doc.add_u64(value_field, 10 + (i % 11) as u64); // values in range 10-20
+
+                writer.add_document(doc).unwrap();
+            }
+            writer.commit().unwrap();
+        }
+
+        let reader = index.reader().unwrap();
+        let searcher = reader.searcher();
+
+        // Test a range query [100, 200] that has no overlap with data range [10, 20]
+        let query = RangeQuery::new(
+            Bound::Included(Term::from_field_u64(value_field, 100)),
+            Bound::Included(Term::from_field_u64(value_field, 200)),
+        );
+
+        let count = searcher.search(&query, &Count).unwrap();
+        assert_eq!(count, 0); // should return 0 results since there's no overlap
+
+        // Test another non-overlapping range: [0, 5] while data range is [10, 20]
+        let query2 = RangeQuery::new(
+            Bound::Included(Term::from_field_u64(value_field, 0)),
+            Bound::Included(Term::from_field_u64(value_field, 5)),
+        );
+
+        let count2 = searcher.search(&query2, &Count).unwrap();
+        assert_eq!(count2, 0); // should return 0 results since there's no overlap
+    }
 }

src/query/range_query/range_query_fastfield.rs

@@ -1598,449 +1598,3 @@ pub(crate) mod ip_range_tests {
         Ok(())
     }
 }
-
-#[cfg(all(test, feature = "unstable"))]
-mod bench {
-
-    use rand::rngs::StdRng;
-    use rand::{Rng, SeedableRng};
-    use test::Bencher;
-
-    use super::tests::*;
-    use super::*;
-    use crate::collector::Count;
-    use crate::query::QueryParser;
-    use crate::Index;
-
-    fn get_index_0_to_100() -> Index {
-        let mut rng = StdRng::from_seed([1u8; 32]);
-        let num_vals = 100_000;
-        let docs: Vec<_> = (0..num_vals)
-            .map(|_i| {
-                let id_name = if rng.gen_bool(0.01) {
-                    "veryfew".to_string() // 1%
-                } else if rng.gen_bool(0.1) {
-                    "few".to_string() // 9%
-                } else {
-                    "many".to_string() // 90%
-                };
-                Doc {
-                    id_name,
-                    id: rng.gen_range(0..100),
-                }
-            })
-            .collect();
-
-        create_index_from_docs(&docs, false)
-    }
-
-    fn get_90_percent() -> RangeInclusive<u64> {
-        0..=90
-    }
-
-    fn get_10_percent() -> RangeInclusive<u64> {
-        0..=10
-    }
-
-    fn get_1_percent() -> RangeInclusive<u64> {
-        10..=10
-    }
-
-    fn execute_query(
-        field: &str,
-        id_range: RangeInclusive<u64>,
-        suffix: &str,
-        index: &Index,
-    ) -> usize {
-        let gen_query_inclusive = |from: &u64, to: &u64| {
-            format!(
-                "{}:[{} TO {}] {}",
-                field,
-                &from.to_string(),
-                &to.to_string(),
-                suffix
-            )
-        };
-
-        let query = gen_query_inclusive(id_range.start(), id_range.end());
-        let query_from_text = |text: &str| {
-            QueryParser::for_index(index, vec![])
-                .parse_query(text)
-                .unwrap()
-        };
-        let query = query_from_text(&query);
-        let reader = index.reader().unwrap();
-        let searcher = reader.searcher();
-        searcher.search(&query, &(Count)).unwrap()
-    }
-
-    #[bench]
-    fn bench_id_range_hit_90_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_90_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_10_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_10_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_1_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_1_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_10_percent_intersect_with_10_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_10_percent(), "AND id_name:few", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_1_percent_intersect_with_10_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_1_percent(), "AND id_name:few", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_1_percent_intersect_with_90_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_1_percent(), "AND id_name:many", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_1_percent_intersect_with_1_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_1_percent(), "AND id_name:veryfew", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_10_percent_intersect_with_90_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_10_percent(), "AND id_name:many", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_90_percent_intersect_with_90_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_90_percent(), "AND id_name:many", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_90_percent_intersect_with_10_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_90_percent(), "AND id_name:few", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_90_percent_intersect_with_1_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("id", get_90_percent(), "AND id_name:veryfew", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_90_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_90_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_10_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_10_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_1_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_1_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_10_percent_intersect_with_10_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_10_percent(), "AND id_name:few", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_1_percent_intersect_with_10_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_1_percent(), "AND id_name:few", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_1_percent_intersect_with_90_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_1_percent(), "AND id_name:many", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_1_percent_intersect_with_1_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_1_percent(), "AND id_name:veryfew", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_10_percent_intersect_with_90_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_10_percent(), "AND id_name:many", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_90_percent_intersect_with_90_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_90_percent(), "AND id_name:many", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_90_percent_intersect_with_10_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_90_percent(), "AND id_name:few", &index));
-    }
-
-    #[bench]
-    fn bench_id_range_hit_90_percent_intersect_with_1_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ids", get_90_percent(), "AND id_name:veryfew", &index));
-    }
-}
-
-#[cfg(all(test, feature = "unstable"))]
-mod bench_ip {
-
-    use rand::rngs::StdRng;
-    use rand::{Rng, SeedableRng};
-    use test::Bencher;
-
-    use super::ip_range_tests::*;
-    use super::*;
-    use crate::collector::Count;
-    use crate::query::QueryParser;
-    use crate::Index;
-
-    fn get_index_0_to_100() -> Index {
-        let mut rng = StdRng::from_seed([1u8; 32]);
-        let num_vals = 100_000;
-        let docs: Vec<_> = (0..num_vals)
-            .map(|_i| {
-                let id = if rng.gen_bool(0.01) {
-                    "veryfew".to_string() // 1%
-                } else if rng.gen_bool(0.1) {
-                    "few".to_string() // 9%
-                } else {
-                    "many".to_string() // 90%
-                };
-                Doc {
-                    id,
-                    // Multiply by 1000, so that we create many buckets in the compact space
-                    // The benches depend on this range to select n-percent of elements with the
-                    // methods below.
-                    ip: Ipv6Addr::from_u128(rng.gen_range(0..100) * 1000),
-                }
-            })
-            .collect();
-
-        create_index_from_ip_docs(&docs)
-    }
-
-    fn get_90_percent() -> RangeInclusive<Ipv6Addr> {
-        let start = Ipv6Addr::from_u128(0);
-        let end = Ipv6Addr::from_u128(90 * 1000);
-        start..=end
-    }
-
-    fn get_10_percent() -> RangeInclusive<Ipv6Addr> {
-        let start = Ipv6Addr::from_u128(0);
-        let end = Ipv6Addr::from_u128(10 * 1000);
-        start..=end
-    }
-
-    fn get_1_percent() -> RangeInclusive<Ipv6Addr> {
-        let start = Ipv6Addr::from_u128(10 * 1000);
-        let end = Ipv6Addr::from_u128(10 * 1000);
-        start..=end
-    }
-
-    fn execute_query(
-        field: &str,
-        ip_range: RangeInclusive<Ipv6Addr>,
-        suffix: &str,
-        index: &Index,
-    ) -> usize {
-        let gen_query_inclusive = |from: &Ipv6Addr, to: &Ipv6Addr| {
-            format!(
-                "{}:[{} TO {}] {}",
-                field,
-                &from.to_string(),
-                &to.to_string(),
-                suffix
-            )
-        };
-
-        let query = gen_query_inclusive(ip_range.start(), ip_range.end());
-        let query_from_text = |text: &str| {
-            QueryParser::for_index(index, vec![])
-                .parse_query(text)
-                .unwrap()
-        };
-        let query = query_from_text(&query);
-        let reader = index.reader().unwrap();
-        let searcher = reader.searcher();
-        searcher.search(&query, &(Count)).unwrap()
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_90_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_90_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_10_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_10_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_1_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_1_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_10_percent_intersect_with_10_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_10_percent(), "AND id:few", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_1_percent_intersect_with_10_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_1_percent(), "AND id:few", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_1_percent_intersect_with_90_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_1_percent(), "AND id:many", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_1_percent_intersect_with_1_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_1_percent(), "AND id:veryfew", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_10_percent_intersect_with_90_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_10_percent(), "AND id:many", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_90_percent_intersect_with_90_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_90_percent(), "AND id:many", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_90_percent_intersect_with_10_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_90_percent(), "AND id:few", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_90_percent_intersect_with_1_percent(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ip", get_90_percent(), "AND id:veryfew", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_90_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ips", get_90_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_10_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ips", get_10_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_1_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ips", get_1_percent(), "", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_10_percent_intersect_with_10_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ips", get_10_percent(), "AND id:few", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_1_percent_intersect_with_10_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ips", get_1_percent(), "AND id:few", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_1_percent_intersect_with_90_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-        bench.iter(|| execute_query("ips", get_1_percent(), "AND id:many", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_1_percent_intersect_with_1_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ips", get_1_percent(), "AND id:veryfew", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_10_percent_intersect_with_90_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ips", get_10_percent(), "AND id:many", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_90_percent_intersect_with_90_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ips", get_90_percent(), "AND id:many", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_90_percent_intersect_with_10_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ips", get_90_percent(), "AND id:few", &index));
-    }
-
-    #[bench]
-    fn bench_ip_range_hit_90_percent_intersect_with_1_percent_multi(bench: &mut Bencher) {
-        let index = get_index_0_to_100();
-
-        bench.iter(|| execute_query("ips", get_90_percent(), "AND id:veryfew", &index));
-    }
-}

src/query/reqopt_scorer.rs

@@ -56,6 +56,11 @@ where
         self.req_scorer.seek(target)
     }
 
+    fn seek_into_the_danger_zone(&mut self, target: DocId) -> bool {
+        self.score_cache = None;
+        self.req_scorer.seek_into_the_danger_zone(target)
+    }
+
     fn doc(&self) -> DocId {
         self.req_scorer.doc()
     }
@@ -76,6 +81,7 @@ where
     TOptScorer: Scorer,
     TScoreCombiner: ScoreCombiner,
 {
+    #[inline]
     fn score(&mut self) -> Score {
         if let Some(score) = self.score_cache {
             return score;

src/query/score_combiner.rs

@@ -29,6 +29,7 @@ impl ScoreCombiner for DoNothingCombiner {
 
     fn clear(&mut self) {}
 
+    #[inline]
     fn score(&self) -> Score {
         1.0
     }
@@ -49,6 +50,7 @@ impl ScoreCombiner for SumCombiner {
         self.score = 0.0;
     }
 
+    #[inline]
     fn score(&self) -> Score {
         self.score
     }
@@ -86,6 +88,7 @@ impl ScoreCombiner for DisjunctionMaxCombiner {
         self.sum = 0.0;
     }
 
+    #[inline]
     fn score(&self) -> Score {
         self.max + (self.sum - self.max) * self.tie_breaker
     }

src/query/scorer.rs

@@ -18,6 +18,7 @@ pub trait Scorer: downcast_rs::Downcast + DocSet + 'static {
 impl_downcast!(Scorer);
 
 impl Scorer for Box<dyn Scorer> {
+    #[inline]
     fn score(&mut self) -> Score {
         self.deref_mut().score()
     }

src/query/term_query/term_scorer.rs

@@ -98,14 +98,17 @@ impl TermScorer {
 }
 
 impl DocSet for TermScorer {
+    #[inline]
     fn advance(&mut self) -> DocId {
         self.postings.advance()
     }
 
+    #[inline]
     fn seek(&mut self, target: DocId) -> DocId {
         self.postings.seek(target)
     }
 
+    #[inline]
     fn doc(&self) -> DocId {
         self.postings.doc()
     }
@@ -116,6 +119,7 @@ impl DocSet for TermScorer {
 }
 
 impl Scorer for TermScorer {
+    #[inline]
     fn score(&mut self) -> Score {
         let fieldnorm_id = self.fieldnorm_id();
         let term_freq = self.term_freq();

src/query/union/buffered_union.rs

@@ -15,7 +15,7 @@ const HORIZON: u32 = 64u32 * 64u32;
 // This function is similar except that it does is not unstable, and
 // it does not keep the original vector ordering.
 //
-// Also, it does not "yield" any elements.
+// Elements are dropped and not yielded.
 fn unordered_drain_filter<T, P>(v: &mut Vec<T>, mut predicate: P)
 where P: FnMut(&mut T) -> bool {
     let mut i = 0;
@@ -128,6 +128,7 @@ impl<TScorer: Scorer, TScoreCombiner: ScoreCombiner> BufferedUnionScorer<TScorer
         }
     }
 
+    #[inline]
     fn advance_buffered(&mut self) -> bool {
         while self.bucket_idx < HORIZON_NUM_TINYBITSETS {
             if let Some(val) = self.bitsets[self.bucket_idx].pop_lowest() {
@@ -143,6 +144,12 @@ impl<TScorer: Scorer, TScoreCombiner: ScoreCombiner> BufferedUnionScorer<TScorer
         }
         false
     }
+
+    fn is_in_horizon(&self, target: DocId) -> bool {
+        // wrapping_sub, because target may be < window_start_doc
+        let gap = target.wrapping_sub(self.window_start_doc);
+        gap < HORIZON
+    }
 }
 
 impl<TScorer, TScoreCombiner> DocSet for BufferedUnionScorer<TScorer, TScoreCombiner>
@@ -150,6 +157,7 @@ where
     TScorer: Scorer,
     TScoreCombiner: ScoreCombiner,
 {
+    #[inline]
     fn advance(&mut self) -> DocId {
         if self.advance_buffered() {
             return self.doc;
@@ -217,8 +225,29 @@ where
         }
     }
 
-    // TODO Also implement `count` with deletes efficiently.
+    fn seek_into_the_danger_zone(&mut self, target: DocId) -> bool {
+        if self.is_in_horizon(target) {
+            // Our value is within the buffered horizon and the docset may already have been
+            // processed and removed, so we need to use seek, which uses the regular advance.
+            self.seek(target) == target
+        } else {
+            // The docsets are not in the buffered range, so we can use seek_into_the_danger_zone
+            // of the underlying docsets
+            let is_hit = self
+                .docsets
+                .iter_mut()
+                .any(|docset| docset.seek_into_the_danger_zone(target));
 
+            // The API requires the DocSet to be in a valid state when `seek_into_the_danger_zone`
+            // returns true.
+            if is_hit {
+                self.seek(target);
+            }
+            is_hit
+        }
+    }
+
+    #[inline]
     fn doc(&self) -> DocId {
         self.doc
     }
@@ -231,6 +260,7 @@ where
         self.docsets.iter().map(|docset| docset.cost()).sum()
     }
 
+    // TODO Also implement `count` with deletes efficiently.
     fn count_including_deleted(&mut self) -> u32 {
         if self.doc == TERMINATED {
             return 0;
@@ -259,6 +289,7 @@ where
     TScoreCombiner: ScoreCombiner,
     TScorer: Scorer,
 {
+    #[inline]
     fn score(&mut self) -> Score {
         self.score
     }

src/query/union/simple_union.rs

@@ -92,6 +92,7 @@ impl<TDocSet: DocSet> DocSet for SimpleUnion<TDocSet> {
     }
 
     fn size_hint(&self) -> u32 {
+        // TODO: use estimate_union
         self.docsets
             .iter()
             .map(|docset| docset.size_hint())

src/schema/document/owned_value.rs

@@ -58,6 +58,31 @@ impl AsRef<OwnedValue> for OwnedValue {
     }
 }
 
+impl OwnedValue {
+    /// Returns a u8 discriminant value for the `OwnedValue` variant.
+    ///
+    /// This can be used to sort `OwnedValue` instances by their type.
+    pub fn discriminant_value(&self) -> u8 {
+        match self {
+            OwnedValue::Null => 0,
+            OwnedValue::Str(_) => 1,
+            OwnedValue::PreTokStr(_) => 2,
+            // It is key to make sure U64, I64, F64 are grouped together in there, otherwise we
+            // might be breaking transivity.
+            OwnedValue::U64(_) => 3,
+            OwnedValue::I64(_) => 4,
+            OwnedValue::F64(_) => 5,
+            OwnedValue::Bool(_) => 6,
+            OwnedValue::Date(_) => 7,
+            OwnedValue::Facet(_) => 8,
+            OwnedValue::Bytes(_) => 9,
+            OwnedValue::Array(_) => 10,
+            OwnedValue::Object(_) => 11,
+            OwnedValue::IpAddr(_) => 12,
+        }
+    }
+}
+
 impl<'a> Value<'a> for &'a OwnedValue {
     type ArrayIter = std::slice::Iter<'a, OwnedValue>;
     type ObjectIter = ObjectMapIter<'a>;

src/schema/mod.rs

@@ -98,6 +98,10 @@
 //! make it possible to access the value given the doc id rapidly. This is useful if the value
 //! of the field is required during scoring or collection for instance.
 //!
+//! Some queries may leverage Fast fields when run on a field that is not indexed. This can be
+//! handy if that kind of request is infrequent, however note that searching on a Fast field is
+//! generally much slower than searching in an index.
+//!
 //! ```
 //! use tantivy::schema::*;
 //! let mut schema_builder = Schema::builder();

src/snippet/mod.rs

@@ -483,7 +483,7 @@ mod tests {
 
     use super::{collapse_overlapped_ranges, search_fragments, select_best_fragment_combination};
     use crate::query::QueryParser;
-    use crate::schema::{IndexRecordOption, Schema, TextFieldIndexing, TextOptions, TEXT};
+    use crate::schema::{Schema, TEXT};
     use crate::snippet::SnippetGenerator;
     use crate::tokenizer::{NgramTokenizer, SimpleTokenizer};
     use crate::Index;
@@ -727,8 +727,10 @@ Survey in 2016, 2017, and 2018."#;
         Ok(())
     }
 
+    #[cfg(feature = "stemmer")]
     #[test]
     fn test_snippet_generator() -> crate::Result<()> {
+        use crate::schema::{IndexRecordOption, TextFieldIndexing, TextOptions};
         let mut schema_builder = Schema::builder();
         let text_options = TextOptions::default().set_indexing_options(
             TextFieldIndexing::default()

src/store/mod.rs

@@ -102,6 +102,7 @@ pub(crate) mod tests {
     }
 
     const NUM_DOCS: usize = 1_000;
+
     #[test]
     fn test_doc_store_iter_with_delete_bug_1077() -> crate::Result<()> {
         // this will cover deletion of the first element in a checkpoint
@@ -113,7 +114,7 @@ pub(crate) mod tests {
         let directory = RamDirectory::create();
         let store_wrt = directory.open_write(path)?;
         let schema =
-            write_lorem_ipsum_store(store_wrt, NUM_DOCS, Compressor::Lz4, BLOCK_SIZE, true);
+            write_lorem_ipsum_store(store_wrt, NUM_DOCS, Compressor::default(), BLOCK_SIZE, true);
         let field_title = schema.get_field("title").unwrap();
         let store_file = directory.open_read(path)?;
         let store = StoreReader::open(store_file, 10)?;

src/store/reader.rs

@@ -465,7 +465,7 @@ mod tests {
         let directory = RamDirectory::create();
         let path = Path::new("store");
         let writer = directory.open_write(path)?;
-        let schema = write_lorem_ipsum_store(writer, 500, Compressor::default(), BLOCK_SIZE, true);
+        let schema = write_lorem_ipsum_store(writer, 500, Compressor::None, BLOCK_SIZE, true);
         let title = schema.get_field("title").unwrap();
         let store_file = directory.open_read(path)?;
         let store = StoreReader::open(store_file, DOCSTORE_CACHE_CAPACITY)?;
@@ -499,7 +499,7 @@ mod tests {
         assert_eq!(store.cache_stats().cache_hits, 1);
         assert_eq!(store.cache_stats().cache_misses, 2);
 
-        assert_eq!(store.cache.peek_lru(), Some(11207));
+        assert_eq!(store.cache.peek_lru(), Some(232206));
 
         Ok(())
     }

src/tokenizer/mod.rs

@@ -132,13 +132,14 @@ mod regex_tokenizer;
 mod remove_long;
 mod simple_tokenizer;
 mod split_compound_words;
-mod stemmer;
 mod stop_word_filter;
 mod tokenized_string;
 mod tokenizer;
 mod tokenizer_manager;
 mod whitespace_tokenizer;
 
+#[cfg(feature = "stemmer")]
+mod stemmer;
 pub use tokenizer_api::{BoxTokenStream, Token, TokenFilter, TokenStream, Tokenizer};
 
 pub use self::alphanum_only::AlphaNumOnlyFilter;
@@ -151,6 +152,7 @@ pub use self::regex_tokenizer::RegexTokenizer;
 pub use self::remove_long::RemoveLongFilter;
 pub use self::simple_tokenizer::{SimpleTokenStream, SimpleTokenizer};
 pub use self::split_compound_words::SplitCompoundWords;
+#[cfg(feature = "stemmer")]
 pub use self::stemmer::{Language, Stemmer};
 pub use self::stop_word_filter::StopWordFilter;
 pub use self::tokenized_string::{PreTokenizedStream, PreTokenizedString};
@@ -167,10 +169,7 @@ pub const MAX_TOKEN_LEN: usize = u16::MAX as usize - 5;
 
 #[cfg(test)]
 pub(crate) mod tests {
-    use super::{
-        Language, LowerCaser, RemoveLongFilter, SimpleTokenizer, Stemmer, Token, TokenizerManager,
-    };
-    use crate::tokenizer::TextAnalyzer;
+    use super::{Token, TokenizerManager};
 
     /// This is a function that can be used in tests and doc tests
     /// to assert a token's correctness.
@@ -205,59 +204,15 @@ pub(crate) mod tests {
     }
 
     #[test]
-    fn test_en_tokenizer() {
+    fn test_tokenizer_does_not_exist() {
         let tokenizer_manager = TokenizerManager::default();
         assert!(tokenizer_manager.get("en_doesnotexist").is_none());
-        let mut en_tokenizer = tokenizer_manager.get("en_stem").unwrap();
-        let mut tokens: Vec<Token> = vec![];
-        {
-            let mut add_token = |token: &Token| {
-                tokens.push(token.clone());
-            };
-            en_tokenizer
-                .token_stream("Hello, happy tax payer!")
-                .process(&mut add_token);
-        }
-
-        assert_eq!(tokens.len(), 4);
-        assert_token(&tokens[0], 0, "hello", 0, 5);
-        assert_token(&tokens[1], 1, "happi", 7, 12);
-        assert_token(&tokens[2], 2, "tax", 13, 16);
-        assert_token(&tokens[3], 3, "payer", 17, 22);
-    }
-
-    #[test]
-    fn test_non_en_tokenizer() {
-        let tokenizer_manager = TokenizerManager::default();
-        tokenizer_manager.register(
-            "el_stem",
-            TextAnalyzer::builder(SimpleTokenizer::default())
-                .filter(RemoveLongFilter::limit(40))
-                .filter(LowerCaser)
-                .filter(Stemmer::new(Language::Greek))
-                .build(),
-        );
-        let mut en_tokenizer = tokenizer_manager.get("el_stem").unwrap();
-        let mut tokens: Vec<Token> = vec![];
-        {
-            let mut add_token = |token: &Token| {
-                tokens.push(token.clone());
-            };
-            en_tokenizer
-                .token_stream("Καλημέρα, χαρούμενε φορολογούμενε!")
-                .process(&mut add_token);
-        }
-
-        assert_eq!(tokens.len(), 3);
-        assert_token(&tokens[0], 0, "καλημερ", 0, 16);
-        assert_token(&tokens[1], 1, "χαρουμεν", 18, 36);
-        assert_token(&tokens[2], 2, "φορολογουμεν", 37, 63);
     }
 
     #[test]
     fn test_tokenizer_empty() {
         let tokenizer_manager = TokenizerManager::default();
-        let mut en_tokenizer = tokenizer_manager.get("en_stem").unwrap();
+        let mut en_tokenizer = tokenizer_manager.get("default").unwrap();
         {
             let mut tokens: Vec<Token> = vec![];
             {

src/tokenizer/stemmer.rs

@@ -142,3 +142,60 @@ impl<T: TokenStream> TokenStream for StemmerTokenStream<T> {
         self.tail.token_mut()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use tokenizer_api::Token;
+
+    use super::*;
+    use crate::tokenizer::tests::assert_token;
+    use crate::tokenizer::{LowerCaser, SimpleTokenizer, TextAnalyzer, TokenizerManager};
+
+    #[test]
+    fn test_en_stem() {
+        let tokenizer_manager = TokenizerManager::default();
+        let mut en_tokenizer = tokenizer_manager.get("en_stem").unwrap();
+        let mut tokens: Vec<Token> = vec![];
+        {
+            let mut add_token = |token: &Token| {
+                tokens.push(token.clone());
+            };
+            en_tokenizer
+                .token_stream("Dogs are the bests!")
+                .process(&mut add_token);
+        }
+
+        assert_eq!(tokens.len(), 4);
+        assert_token(&tokens[0], 0, "dog", 0, 4);
+        assert_token(&tokens[1], 1, "are", 5, 8);
+        assert_token(&tokens[2], 2, "the", 9, 12);
+        assert_token(&tokens[3], 3, "best", 13, 18);
+    }
+
+    #[test]
+    fn test_non_en_stem() {
+        let tokenizer_manager = TokenizerManager::default();
+        tokenizer_manager.register(
+            "el_stem",
+            TextAnalyzer::builder(SimpleTokenizer::default())
+                .filter(LowerCaser)
+                .filter(Stemmer::new(Language::Greek))
+                .build(),
+        );
+        let mut el_tokenizer = tokenizer_manager.get("el_stem").unwrap();
+        let mut tokens: Vec<Token> = vec![];
+        {
+            let mut add_token = |token: &Token| {
+                tokens.push(token.clone());
+            };
+            el_tokenizer
+                .token_stream("Καλημέρα, χαρούμενε φορολογούμενε!")
+                .process(&mut add_token);
+        }
+
+        assert_eq!(tokens.len(), 3);
+        assert_token(&tokens[0], 0, "καλημερ", 0, 16);
+        assert_token(&tokens[1], 1, "χαρουμεν", 18, 36);
+        assert_token(&tokens[2], 2, "φορολογουμεν", 37, 63);
+    }
+}

src/tokenizer/tokenizer_manager.rs

@@ -1,10 +1,9 @@
 use std::collections::HashMap;
 use std::sync::{Arc, RwLock};
 
-use crate::tokenizer::stemmer::Language;
 use crate::tokenizer::tokenizer::TextAnalyzer;
 use crate::tokenizer::{
-    LowerCaser, RawTokenizer, RemoveLongFilter, SimpleTokenizer, Stemmer, WhitespaceTokenizer,
+    LowerCaser, RawTokenizer, RemoveLongFilter, SimpleTokenizer, WhitespaceTokenizer,
 };
 
 /// The tokenizer manager serves as a store for
@@ -64,14 +63,18 @@ impl Default for TokenizerManager {
                 .filter(LowerCaser)
                 .build(),
         );
-        manager.register(
-            "en_stem",
-            TextAnalyzer::builder(SimpleTokenizer::default())
-                .filter(RemoveLongFilter::limit(40))
-                .filter(LowerCaser)
-                .filter(Stemmer::new(Language::English))
-                .build(),
-        );
+        #[cfg(feature = "stemmer")]
+        {
+            use crate::tokenizer::stemmer::{Language, Stemmer};
+            manager.register(
+                "en_stem",
+                TextAnalyzer::builder(SimpleTokenizer::default())
+                    .filter(RemoveLongFilter::limit(40))
+                    .filter(LowerCaser) // The stemmer does not lowercase
+                    .filter(Stemmer::new(Language::English))
+                    .build(),
+            );
+        }
         manager.register("whitespace", WhitespaceTokenizer::default());
         manager
     }

stacker/Cargo.toml

@@ -11,7 +11,6 @@ description = "term hashmap used for indexing"
 murmurhash32 = "0.3"
 common = { version = "0.10", path = "../common/", package = "tantivy-common" }
 ahash = { version = "0.8.11", default-features = false, optional = true }
-rand_distr = "0.4.3"
 
 
 [[bench]]
@@ -29,6 +28,7 @@ zipf = "7.0.0"
 rustc-hash = "2.1.0"
 proptest = "1.2.0"
 binggan = { version = "0.14.0" }
+rand_distr = "0.4.3"
 
 [features]
 compare_hash_only = ["ahash"] # Compare hash only, not the key in the Hashmap

stacker/src/expull.rs

@@ -5,7 +5,7 @@ use common::serialize_vint_u32;
 use crate::fastcpy::fast_short_slice_copy;
 use crate::{Addr, MemoryArena};
 
-const FIRST_BLOCK_NUM: u16 = 2;
+const FIRST_BLOCK_NUM: u32 = 2;
 
 /// An exponential unrolled link.
 ///
@@ -33,8 +33,8 @@ pub struct ExpUnrolledLinkedList {
     // u16, since the max size of each block is (1<<next_cap_pow_2)
     // Limited to 15, so we don't overflow remaining_cap.
     remaining_cap: u16,
-    // To get the current number of blocks: block_num - FIRST_BLOCK_NUM
-    block_num: u16,
+    // Tracks the number of blocks allocated: block_num - FIRST_BLOCK_NUM
+    block_num: u32,
     head: Addr,
     tail: Addr,
 }
@@ -110,16 +110,27 @@ impl ExpUnrolledLinkedListWriter<'_> {
     }
 }
 
-// The block size is 2^block_num + 2, but max 2^15= 32k
-// Initial size is 8, for the first block => block_num == 1
+// The block size is 2^block_num, but max 2^15 = 32KB
+// Initial size is 8 bytes (2^3), for the first block => block_num == 2
+// Block size caps at 32KB (2^15) regardless of how high block_num goes
 #[inline]
-fn get_block_size(block_num: u16) -> u16 {
-    1 << block_num.min(15)
+fn get_block_size(block_num: u32) -> u16 {
+    // Cap at 15 to prevent block sizes > 32KB
+    // block_num can now be much larger than 15, but block size maxes out
+    let exp: u32 = block_num.min(15u32);
+    (1u32 << exp) as u16
 }
 
 impl ExpUnrolledLinkedList {
+    #[inline(always)]
     pub fn increment_num_blocks(&mut self) {
-        self.block_num += 1;
+        // Add overflow check as a safety measure
+        // With u32, we can handle up to ~4 billion blocks before overflow
+        // At 32KB per block (max size), that's 128 TB of data
+        self.block_num = self
+            .block_num
+            .checked_add(1)
+            .expect("ExpUnrolledLinkedList block count overflow - exceeded 4 billion blocks");
     }
 
     #[inline]
@@ -132,9 +143,26 @@ impl ExpUnrolledLinkedList {
         if addr.is_null() {
             return;
         }
-        let last_block_len = get_block_size(self.block_num) as usize - self.remaining_cap as usize;
 
-        // Full Blocks
+        // Calculate last block length with bounds checking to prevent underflow
+        let block_size = get_block_size(self.block_num) as usize;
+        let last_block_len = block_size.saturating_sub(self.remaining_cap as usize);
+
+        // Safety check: if remaining_cap > block_size, the metadata is corrupted
+        assert!(
+            self.remaining_cap as usize <= block_size,
+            "ExpUnrolledLinkedList metadata corruption detected: remaining_cap ({}) > block_size \
+             ({}). This indicates a serious bug, please report! (block_num={}, head={:?}, \
+             tail={:?})",
+            self.remaining_cap,
+            block_size,
+            self.block_num,
+            self.head,
+            self.tail
+        );
+
+        // Full Blocks (iterate through all blocks except the last one)
+        // Note: Blocks are numbered starting from FIRST_BLOCK_NUM+1 (=3) after first allocation
         for block_num in FIRST_BLOCK_NUM + 1..self.block_num {
             let cap = get_block_size(block_num) as usize;
             let data = arena.slice(addr, cap);
@@ -259,6 +287,180 @@ mod tests {
         assert_eq!(&vec1[..], &res1[..]);
         assert_eq!(&vec2[..], &res2[..]);
     }
+
+    // Tests for u32 block_num fix (issue with large arrays)
+
+    #[test]
+    fn test_block_num_exceeds_u16_max() {
+        // Test that we can handle more than 65,535 blocks (old u16 limit)
+        let mut eull = ExpUnrolledLinkedList::default();
+
+        // Simulate allocating 70,000 blocks (exceeds u16::MAX of 65,535)
+        for _ in 0..70_000 {
+            eull.increment_num_blocks();
+        }
+
+        // Verify block_num is correct
+        assert_eq!(eull.block_num, FIRST_BLOCK_NUM + 70_000);
+
+        // Verify we can still get block size (should be capped at 32KB)
+        let block_size = get_block_size(eull.block_num);
+        assert_eq!(block_size, 1 << 15); // 32KB max
+    }
+
+    #[test]
+    #[allow(clippy::needless_range_loop)]
+    fn test_large_dataset_simulation() {
+        // Simulate the scenario: large arrays requiring many blocks
+        // We write enough data to require thousands of blocks
+        let mut arena = MemoryArena::default();
+        let mut eull = ExpUnrolledLinkedList::default();
+
+        // Write 100 MB of data (this will require ~3,200 blocks at 32KB each)
+        // This is enough to validate the system works with large datasets
+        // but not so much that the test is slow
+        let bytes_per_write = 10_000;
+        let num_writes = 10_000; // 10k * 10k = 100 MB
+
+        let data: Vec<u8> = (0..bytes_per_write).map(|i| (i % 256) as u8).collect();
+        for _ in 0..num_writes {
+            eull.writer(&mut arena).extend_from_slice(&data);
+        }
+
+        // Verify we allocated many blocks (should be in the thousands)
+        assert!(
+            eull.block_num > 1000,
+            "block_num ({}) should be > 1000 for this much data",
+            eull.block_num
+        );
+
+        // Verify we can read back correctly
+        let mut buffer = Vec::new();
+        eull.read_to_end(&arena, &mut buffer);
+        assert_eq!(buffer.len(), bytes_per_write * num_writes);
+
+        // Verify data integrity on a sample
+        for i in 0..bytes_per_write {
+            assert_eq!(buffer[i], (i % 256) as u8);
+        }
+    }
+
+    #[test]
+    fn test_get_block_size_with_large_block_num() {
+        // Test that get_block_size handles large u32 values correctly
+
+        // Small block numbers (under 15)
+        assert_eq!(get_block_size(2), 4); // 2^2 = 4
+        assert_eq!(get_block_size(3), 8); // 2^3 = 8
+        assert_eq!(get_block_size(10), 1024); // 2^10 = 1KB
+
+        // At the cap (15)
+        assert_eq!(get_block_size(15), 32768); // 2^15 = 32KB
+
+        // Beyond the cap (should stay at 32KB)
+        assert_eq!(get_block_size(16), 32768);
+        assert_eq!(get_block_size(100), 32768);
+        assert_eq!(get_block_size(65_536), 32768); // Old u16::MAX + 1
+        assert_eq!(get_block_size(100_000), 32768);
+        assert_eq!(get_block_size(1_000_000), 32768);
+    }
+
+    #[test]
+    fn test_increment_blocks_near_u16_boundary() {
+        // Test incrementing around the old u16::MAX boundary
+        let mut eull = ExpUnrolledLinkedList::default();
+
+        // Set to just before old limit
+        for _ in 0..65_533 {
+            eull.increment_num_blocks();
+        }
+        assert_eq!(eull.block_num, FIRST_BLOCK_NUM + 65_533);
+
+        // Cross the old u16::MAX boundary (this would have overflowed before)
+        eull.increment_num_blocks(); // 65,534
+        eull.increment_num_blocks(); // 65,535 (old max)
+        eull.increment_num_blocks(); // 65,536 (would overflow u16)
+        eull.increment_num_blocks(); // 65,537
+
+        // Verify we're past the old limit
+        assert_eq!(eull.block_num, FIRST_BLOCK_NUM + 65_537);
+    }
+
+    #[test]
+    fn test_write_and_read_with_many_blocks() {
+        // Test that write/read works correctly with many blocks
+        let mut arena = MemoryArena::default();
+        let mut eull = ExpUnrolledLinkedList::default();
+
+        // Write data that will span many blocks
+        let test_data: Vec<u8> = (0..50_000).map(|i| (i % 256) as u8).collect();
+        eull.writer(&mut arena).extend_from_slice(&test_data);
+
+        // Read it back
+        let mut buffer = Vec::new();
+        eull.read_to_end(&arena, &mut buffer);
+
+        // Verify data integrity
+        assert_eq!(buffer.len(), test_data.len());
+        assert_eq!(&buffer[..], &test_data[..]);
+    }
+
+    #[test]
+    fn test_multiple_eull_with_large_block_counts() {
+        // Test multiple ExpUnrolledLinkedLists with high block counts
+        // (simulates parallel columnar writes)
+        let mut arena = MemoryArena::default();
+        let mut eull1 = ExpUnrolledLinkedList::default();
+        let mut eull2 = ExpUnrolledLinkedList::default();
+
+        // Write different data to each
+        for i in 0..10_000u32 {
+            eull1.writer(&mut arena).write_u32_vint(i);
+            eull2.writer(&mut arena).write_u32_vint(i * 2);
+        }
+
+        // Read back and verify
+        let mut buf1 = Vec::new();
+        let mut buf2 = Vec::new();
+        eull1.read_to_end(&arena, &mut buf1);
+        eull2.read_to_end(&arena, &mut buf2);
+
+        // Deserialize and check
+        let mut cursor1 = &buf1[..];
+        let mut cursor2 = &buf2[..];
+        for i in 0..10_000u32 {
+            assert_eq!(read_u32_vint(&mut cursor1), i);
+            assert_eq!(read_u32_vint(&mut cursor2), i * 2);
+        }
+    }
+
+    #[test]
+    fn test_block_size_stays_capped() {
+        // Verify that even with massive block numbers, size stays at 32KB
+        let mut eull = ExpUnrolledLinkedList::default();
+
+        // Increment to a very large number
+        for _ in 0..200_000 {
+            eull.increment_num_blocks();
+        }
+
+        let block_size = get_block_size(eull.block_num);
+        assert_eq!(block_size, 32768, "Block size should be capped at 32KB");
+    }
+
+    #[test]
+    #[should_panic(expected = "ExpUnrolledLinkedList block count overflow")]
+    fn test_increment_overflow_protection() {
+        // Test that we panic gracefully if we somehow hit u32::MAX
+        // This is extremely unlikely in practice (would require 128TB of data)
+        let mut eull = ExpUnrolledLinkedList {
+            block_num: u32::MAX,
+            ..Default::default()
+        };
+
+        // This should panic with our custom error message
+        eull.increment_num_blocks();
+    }
 }
 
 #[cfg(all(test, feature = "unstable"))]