Skip to main content

hermes_core/segment/builder/
mod.rs

1//! Streaming segment builder with optimized memory usage
2//!
3//! Key optimizations:
4//! - **String interning**: Terms are interned using `lasso` to avoid repeated allocations
5//! - **hashbrown HashMap**: O(1) average insertion instead of BTreeMap's O(log n)
6//! - **Streaming document store**: Documents written to disk immediately
7//! - **Zero-copy store build**: Pre-serialized doc bytes passed directly to compressor
8//! - **Parallel posting serialization**: Rayon parallel sort + serialize
9//! - **Inline posting fast path**: Small terms skip PostingList/BlockPostingList entirely
10
11#[cfg_attr(not(feature = "native"), allow(dead_code))]
12pub(crate) mod bmp;
13mod config;
14mod dense;
15#[cfg(feature = "diagnostics")]
16mod diagnostics;
17#[cfg_attr(not(feature = "native"), allow(dead_code))]
18pub(crate) mod graph_bisection;
19pub use graph_bisection::BpBudget;
20mod postings;
21mod sparse;
22mod store;
23
24pub use config::{MemoryBreakdown, SegmentBuilderConfig, SegmentBuilderStats};
25
26#[cfg(feature = "native")]
27use std::fs::{File, OpenOptions};
28#[cfg(feature = "native")]
29use std::io::BufWriter;
30use std::io::Write;
31use std::mem::size_of;
32#[cfg(feature = "native")]
33use std::path::PathBuf;
34
35use hashbrown::HashMap;
36use rustc_hash::FxHashMap;
37
38// String interning: lasso on native (fast arena), HashMap on WASM (no C deps)
39#[cfg(feature = "native")]
40use lasso::{Rodeo, Spur};
41
42#[cfg(not(feature = "native"))]
43pub(crate) mod simple_interner {
44    use hashbrown::HashMap;
45
46    #[derive(Clone, Copy, PartialEq, Eq, Hash)]
47    pub struct Spur(u32);
48
49    /// Simple string interner for WASM (replaces lasso::Rodeo).
50    /// Stores each string once in a Vec; HashMap maps &str → index.
51    pub struct Rodeo {
52        /// Canonical storage — each string lives here exactly once.
53        strings: Vec<Box<str>>,
54        /// Maps borrowed string slices (pointing into `strings`) to their index.
55        /// Safety: entries are never removed and Box<str> has a stable address.
56        map: HashMap<&'static str, u32>,
57    }
58
59    impl Rodeo {
60        pub fn new() -> Self {
61            Self {
62                strings: Vec::new(),
63                map: HashMap::new(),
64            }
65        }
66
67        pub fn get(&self, key: &str) -> Option<Spur> {
68            self.map.get(key).map(|&id| Spur(id))
69        }
70
71        pub fn get_or_intern(&mut self, key: &str) -> Spur {
72            if let Some(&id) = self.map.get(key) {
73                return Spur(id);
74            }
75            let id = self.strings.len() as u32;
76            let boxed: Box<str> = key.into();
77            // Safety: the Box<str> is stored in self.strings (append-only Vec)
78            // and never moved or freed while the Rodeo is alive.
79            let static_ref: &'static str = unsafe { &*(boxed.as_ref() as *const str) };
80            self.strings.push(boxed);
81            self.map.insert(static_ref, id);
82            Spur(id)
83        }
84
85        pub fn resolve(&self, spur: &Spur) -> &str {
86            &self.strings[spur.0 as usize]
87        }
88
89        pub fn len(&self) -> usize {
90            self.strings.len()
91        }
92    }
93}
94
95#[cfg(not(feature = "native"))]
96use simple_interner::{Rodeo, Spur};
97
98use super::types::{FieldStats, SegmentFiles, SegmentId, SegmentMeta};
99use std::sync::Arc;
100
101use crate::directories::{Directory, DirectoryWriter};
102use crate::dsl::{Document, Field, FieldType, FieldValue, Schema};
103use crate::tokenizer::BoxedTokenizer;
104use crate::{DocId, Result};
105
106use dense::{BinaryDenseVectorBuilder, DenseVectorBuilder};
107use postings::{CompactPosting, PositionPostingListBuilder, PostingListBuilder, TermKey};
108use sparse::SparseVectorBuilder;
109
110/// Size of the document store buffer before writing to disk
111const STORE_BUFFER_SIZE: usize = 16 * 1024 * 1024; // 16MB
112
113/// Memory overhead per new term in the inverted index:
114/// HashMap entry control byte + padding + TermKey + PostingListBuilder + Vec header
115const NEW_TERM_OVERHEAD: usize = size_of::<TermKey>() + size_of::<PostingListBuilder>() + 24;
116
117/// Memory overhead per newly interned string: Spur + arena pointers (2 × usize)
118const INTERN_OVERHEAD: usize = size_of::<Spur>() + 2 * size_of::<usize>();
119
120/// Memory overhead per new term in the position index
121const NEW_POS_TERM_OVERHEAD: usize =
122    size_of::<TermKey>() + size_of::<PositionPostingListBuilder>() + 24;
123
124/// Packed position encoding is `(element_ordinal << 20) | token_position`:
125/// 12 bits of element ordinal, 20 bits of token position. Values beyond these
126/// maxima must saturate — a plain shift/or silently corrupts the neighboring
127/// bit field (ordinal 4096 wraps to 0 and aliases element 0; token positions
128/// >= 2^20 bleed into the ordinal bits).
129const MAX_POSITION_ELEMENT_ORDINAL: u32 = (1 << 12) - 1;
130const MAX_TOKEN_POSITION: u32 = (1 << 20) - 1;
131
132/// Vector ordinals are zero-based `u16`s, so all 65,536 ordinal values are
133/// available to one vector field in one document.
134pub(crate) const MAX_VECTOR_VALUES_PER_FIELD: usize = u16::MAX as usize + 1;
135
136/// Default BMP vocabulary size when `dims` is unset in the sparse vector
137/// config (SPLADE unigram vocabulary). Must match the build-time defaults in
138/// `builder/sparse.rs` and `merger/sparse.rs`.
139const DEFAULT_BMP_SPARSE_DIMS: u32 = 105879;
140
141/// Human-readable name of a schema field type (matches the SDL/serde names).
142fn field_type_name(field_type: &FieldType) -> &'static str {
143    match field_type {
144        FieldType::Text => "text",
145        FieldType::U64 => "u64",
146        FieldType::I64 => "i64",
147        FieldType::F64 => "f64",
148        FieldType::Bytes => "bytes",
149        FieldType::SparseVector => "sparse_vector",
150        FieldType::DenseVector => "dense_vector",
151        FieldType::Json => "json",
152        FieldType::BinaryDenseVector => "binary_dense_vector",
153    }
154}
155
156/// Human-readable name of a document field value's type (matches SDL names).
157fn field_value_type_name(value: &FieldValue) -> &'static str {
158    match value {
159        FieldValue::Text(_) => "text",
160        FieldValue::U64(_) => "u64",
161        FieldValue::I64(_) => "i64",
162        FieldValue::F64(_) => "f64",
163        FieldValue::Bytes(_) => "bytes",
164        FieldValue::SparseVector(_) => "sparse_vector",
165        FieldValue::DenseVector(_) => "dense_vector",
166        FieldValue::Json(_) => "json",
167        FieldValue::BinaryDenseVector(_) => "binary_dense_vector",
168    }
169}
170
171/// Reject vector lists that cannot be represented by the ordinal wire format.
172///
173/// This validation is intentionally pure and runs before a document enters a
174/// native worker queue or mutates an inline/WASM segment builder. Otherwise a
175/// single oversized document is discovered only after sibling documents have
176/// been indexed, forcing the entire commit generation to be discarded.
177pub(crate) fn validate_vector_value_counts(doc: &Document, schema: &Schema) -> Result<()> {
178    let mut vector_values_per_field: FxHashMap<u32, usize> = FxHashMap::default();
179
180    for (field, value) in doc.field_values() {
181        let Some(entry) = schema.get_field_entry(*field) else {
182            continue;
183        };
184
185        let consumes_vector_ordinal = match (&entry.field_type, value) {
186            (FieldType::DenseVector, FieldValue::DenseVector(_))
187            | (FieldType::BinaryDenseVector, FieldValue::BinaryDenseVector(_)) => {
188                entry.indexed || entry.stored
189            }
190            (FieldType::SparseVector, FieldValue::SparseVector(_)) => entry.indexed || entry.fast,
191            _ => false,
192        };
193        if !consumes_vector_ordinal {
194            continue;
195        }
196
197        let count = vector_values_per_field.entry(field.0).or_insert(0);
198        *count += 1;
199        if *count > MAX_VECTOR_VALUES_PER_FIELD {
200            return Err(crate::Error::Document(format!(
201                "field '{}' (id {}) has more than {} vector values in one document",
202                entry.name, field.0, MAX_VECTOR_VALUES_PER_FIELD
203            )));
204        }
205    }
206    Ok(())
207}
208
209/// Segment builder with optimized memory usage
210///
211/// Features:
212/// - Streams documents to disk immediately (no in-memory document storage)
213/// - Uses string interning for terms (reduced allocations)
214/// - Uses hashbrown HashMap (faster than BTreeMap)
215pub struct SegmentBuilder {
216    schema: Arc<Schema>,
217    config: SegmentBuilderConfig,
218    tokenizers: FxHashMap<Field, BoxedTokenizer>,
219    /// Text field → sibling field whose values hint its dynamic tokenizer
220    /// (`text<lex(by: languages, ...)>`).
221    tokenizer_hint_fields: FxHashMap<Field, Field>,
222    /// Reusable buffer holding the comma-joined hint of the current document.
223    tokenizer_hint_buffer: String,
224    /// Documents indexed into a hinted field without any hint value present
225    /// (fell back to the tokenizer's default). Reported in builder stats.
226    unhinted_dynamic_docs: u64,
227
228    /// String interner for terms - O(1) lookup and deduplication
229    term_interner: Rodeo,
230
231    /// Inverted index: term key -> posting list
232    inverted_index: HashMap<TermKey, PostingListBuilder>,
233
234    /// Spill file for high-frequency posting lists (lazily created on first spill).
235    #[cfg(feature = "native")]
236    posting_spill_file: Option<BufWriter<File>>,
237    #[cfg(feature = "native")]
238    posting_spill_path: PathBuf,
239    /// Tracks spilled ranges per term key: (file_offset, posting_count).
240    #[cfg(feature = "native")]
241    posting_spill_index: HashMap<TermKey, Vec<(u64, u32)>>,
242    #[cfg(feature = "native")]
243    posting_spill_offset: u64,
244
245    /// Streaming document store writer (native: temp file on disk, WASM: in-memory buffer)
246    #[cfg(feature = "native")]
247    store_file: BufWriter<File>,
248    #[cfg(feature = "native")]
249    store_path: PathBuf,
250    #[cfg(not(feature = "native"))]
251    store_buffer: Vec<u8>,
252
253    /// Document count
254    next_doc_id: DocId,
255
256    /// Per-field statistics for BM25F
257    field_stats: FxHashMap<u32, FieldStats>,
258
259    /// Per-document field lengths stored compactly
260    /// Uses a flat `Vec` instead of `Vec<HashMap>` for better cache locality
261    /// Layout: [doc0_field0_len, doc0_field1_len, ..., doc1_field0_len, ...]
262    doc_field_lengths: Vec<u32>,
263    num_indexed_fields: usize,
264    field_to_slot: FxHashMap<u32, usize>,
265
266    /// Reusable buffer for per-document term frequency aggregation
267    /// Avoids allocating a new hashmap for each document
268    local_tf_buffer: FxHashMap<Spur, u32>,
269
270    /// Reusable buffer for per-document position tracking (when positions enabled)
271    /// Avoids allocating a new hashmap for each text field per document
272    local_positions: FxHashMap<Spur, Vec<u32>>,
273
274    /// Terms with nonempty position scratch from the previous field/chunk.
275    /// Reset only these, never the vocabulary accumulated by the whole segment.
276    local_position_terms: Vec<Spur>,
277
278    /// Reusable buffer for tokenization to avoid per-token String allocations
279    token_buffer: String,
280
281    /// Reusable buffer for numeric field term encoding (avoids format!() alloc per call)
282    numeric_buffer: String,
283
284    /// Dense vector storage per field: field -> (doc_ids, vectors)
285    /// Vectors are stored as flat f32 arrays for global IVF-PQ indexing.
286    dense_vectors: FxHashMap<u32, DenseVectorBuilder>,
287
288    /// Binary dense vector storage per field: field -> packed-bit vectors
289    binary_dense_vectors: FxHashMap<u32, BinaryDenseVectorBuilder>,
290
291    /// Sparse vector storage per field: field -> SparseVectorBuilder
292    /// Uses proper BlockSparsePostingList with configurable quantization
293    sparse_vectors: FxHashMap<u32, SparseVectorBuilder>,
294
295    /// Position index for fields with positions enabled
296    /// term key -> position posting list
297    position_index: HashMap<TermKey, PositionPostingListBuilder>,
298
299    /// Fields that have position tracking enabled, with their mode
300    position_enabled_fields: FxHashMap<u32, Option<crate::dsl::PositionMode>>,
301
302    /// Current element ordinal for multi-valued fields (reset per document)
303    current_element_ordinal: FxHashMap<u32, u32>,
304
305    /// Virtual-id maps of chunked text fields: field -> (doc, ordinal, length)
306    /// per chunk. Postings of these fields are keyed by the virtual id.
307    chunk_maps: FxHashMap<u32, super::chunk_map::ChunkMapBuilder>,
308
309    /// Whether the once-per-segment position-encoding saturation warning
310    /// has already been emitted (see MAX_POSITION_ELEMENT_ORDINAL).
311    position_saturation_warned: bool,
312
313    /// Incrementally tracked memory estimate (avoids expensive stats() calls)
314    estimated_memory: usize,
315
316    /// Reusable buffer for document serialization (avoids per-document allocation)
317    doc_serialize_buffer: Vec<u8>,
318
319    /// Fast-field columnar writers per field_id (only for fields with fast=true)
320    fast_fields: FxHashMap<u32, crate::structures::fast_field::FastFieldWriter>,
321}
322
323impl SegmentBuilder {
324    /// Create a new segment builder
325    pub fn new(schema: Arc<Schema>, config: SegmentBuilderConfig) -> Result<Self> {
326        #[cfg(feature = "native")]
327        let (store_file, store_path, spill_path) = {
328            let segment_id = uuid::Uuid::new_v4();
329            let store_path = config
330                .temp_dir
331                .join(format!("hermes_store_{}.tmp", segment_id));
332            let store_file = BufWriter::with_capacity(
333                STORE_BUFFER_SIZE,
334                OpenOptions::new()
335                    .create(true)
336                    .write(true)
337                    .truncate(true)
338                    .open(&store_path)?,
339            );
340            let spill_path = config
341                .temp_dir
342                .join(format!("hermes_spill_{}.tmp", segment_id));
343            (store_file, store_path, spill_path)
344        };
345
346        // Count indexed fields, track positions, and auto-configure tokenizers
347        let registry = crate::tokenizer::TokenizerRegistry::new();
348        let mut num_indexed_fields = 0;
349        let mut field_to_slot = FxHashMap::default();
350        let mut position_enabled_fields = FxHashMap::default();
351        let mut tokenizers = FxHashMap::default();
352        let mut tokenizer_hint_fields = FxHashMap::default();
353        for (field, entry) in schema.fields() {
354            if entry.indexed && matches!(entry.field_type, FieldType::Text) {
355                field_to_slot.insert(field.0, num_indexed_fields);
356                num_indexed_fields += 1;
357                if entry.positions.is_some() {
358                    position_enabled_fields.insert(field.0, entry.positions);
359                }
360                if let Some(ref tok_name) = entry.tokenizer
361                    && let Some(tokenizer) = registry.get(tok_name)
362                {
363                    tokenizers.insert(field, tokenizer);
364                }
365                if let Some(hint_field) = schema.tokenizer_hint_field(field) {
366                    tokenizer_hint_fields.insert(field, hint_field);
367                }
368            }
369        }
370
371        // Initialize fast-field writers for fields with fast=true
372        use crate::structures::fast_field::{FastFieldColumnType, FastFieldWriter};
373        let mut fast_fields = FxHashMap::default();
374        for (field, entry) in schema.fields() {
375            if entry.fast {
376                let writer = if entry.multi {
377                    match entry.field_type {
378                        FieldType::U64 => {
379                            FastFieldWriter::new_numeric_multi(FastFieldColumnType::U64)
380                        }
381                        FieldType::I64 => {
382                            FastFieldWriter::new_numeric_multi(FastFieldColumnType::I64)
383                        }
384                        FieldType::F64 => {
385                            FastFieldWriter::new_numeric_multi(FastFieldColumnType::F64)
386                        }
387                        FieldType::Text => FastFieldWriter::new_text_multi(),
388                        _ => continue,
389                    }
390                } else {
391                    match entry.field_type {
392                        FieldType::U64 => FastFieldWriter::new_numeric(FastFieldColumnType::U64),
393                        FieldType::I64 => FastFieldWriter::new_numeric(FastFieldColumnType::I64),
394                        FieldType::F64 => FastFieldWriter::new_numeric(FastFieldColumnType::F64),
395                        FieldType::Text => FastFieldWriter::new_text(),
396                        _ => continue,
397                    }
398                };
399                fast_fields.insert(field.0, writer);
400            }
401        }
402
403        Ok(Self {
404            schema,
405            tokenizers,
406            tokenizer_hint_fields,
407            tokenizer_hint_buffer: String::new(),
408            unhinted_dynamic_docs: 0,
409            term_interner: Rodeo::new(),
410            inverted_index: HashMap::with_capacity(config.posting_map_capacity),
411            #[cfg(feature = "native")]
412            posting_spill_file: None,
413            #[cfg(feature = "native")]
414            posting_spill_path: spill_path,
415            #[cfg(feature = "native")]
416            posting_spill_index: HashMap::new(),
417            #[cfg(feature = "native")]
418            posting_spill_offset: 0,
419            #[cfg(feature = "native")]
420            store_file,
421            #[cfg(feature = "native")]
422            store_path,
423            #[cfg(not(feature = "native"))]
424            store_buffer: Vec::with_capacity(STORE_BUFFER_SIZE),
425            next_doc_id: 0,
426            field_stats: FxHashMap::default(),
427            chunk_maps: FxHashMap::default(),
428            doc_field_lengths: Vec::new(),
429            num_indexed_fields,
430            field_to_slot,
431            local_tf_buffer: FxHashMap::default(),
432            local_positions: FxHashMap::default(),
433            local_position_terms: Vec::new(),
434            token_buffer: String::with_capacity(64),
435            numeric_buffer: String::with_capacity(32),
436            config,
437            dense_vectors: FxHashMap::default(),
438            binary_dense_vectors: FxHashMap::default(),
439            sparse_vectors: FxHashMap::default(),
440            position_index: HashMap::new(),
441            position_enabled_fields,
442            current_element_ordinal: FxHashMap::default(),
443            position_saturation_warned: false,
444            estimated_memory: 0,
445            doc_serialize_buffer: Vec::with_capacity(256),
446            fast_fields,
447        })
448    }
449
450    pub fn set_tokenizer(&mut self, field: Field, tokenizer: BoxedTokenizer) {
451        self.tokenizers.insert(field, tokenizer);
452    }
453
454    /// Documents that hit a dynamically tokenized field without a hint value.
455    pub fn unhinted_dynamic_docs(&self) -> u64 {
456        self.unhinted_dynamic_docs
457    }
458
459    /// Resolve the tokenizer hint for `field` from the document's hint field.
460    ///
461    /// Returns `true` when `field` is dynamically tokenized; the hint text
462    /// (all values of the hint field, trimmed, lowercased, comma-joined) is
463    /// left in `tokenizer_hint_buffer`, empty when the document carries none.
464    fn resolve_tokenizer_hint(
465        &mut self,
466        field: Field,
467        doc: &Document,
468        element_ordinal: u32,
469    ) -> bool {
470        let Some(&hint_field) = self.tokenizer_hint_fields.get(&field) else {
471            return false;
472        };
473        self.tokenizer_hint_buffer.clear();
474        for value in doc.get_all(hint_field) {
475            if let FieldValue::Text(hint) = value {
476                let hint = hint.trim();
477                if hint.is_empty() {
478                    continue;
479                }
480                if !self.tokenizer_hint_buffer.is_empty() {
481                    self.tokenizer_hint_buffer.push(',');
482                }
483                for c in hint.chars().flat_map(char::to_lowercase) {
484                    self.tokenizer_hint_buffer.push(c);
485                }
486            }
487        }
488        if self.tokenizer_hint_buffer.is_empty() && element_ordinal == 0 {
489            self.unhinted_dynamic_docs += 1;
490        }
491        true
492    }
493
494    /// Get the current element ordinal for a field and increment it.
495    /// Used for multi-valued fields (text, dense_vector, sparse_vector).
496    fn next_element_ordinal(&mut self, field_id: u32) -> u32 {
497        let ordinal = *self.current_element_ordinal.get(&field_id).unwrap_or(&0);
498        *self.current_element_ordinal.entry(field_id).or_insert(0) += 1;
499        ordinal
500    }
501
502    fn next_vector_ordinal(&mut self, field_id: u32) -> Result<u16> {
503        let ordinal = self.next_element_ordinal(field_id);
504        u16::try_from(ordinal).map_err(|_| {
505            crate::Error::Document(format!(
506                "field {field_id} has more than {} vector values in one document",
507                u16::MAX as usize + 1
508            ))
509        })
510    }
511
512    pub fn num_docs(&self) -> u32 {
513        self.next_doc_id
514    }
515
516    /// Fast O(1) memory estimate - updated incrementally during indexing
517    #[inline]
518    pub fn estimated_memory_bytes(&self) -> usize {
519        self.estimated_memory
520    }
521
522    /// Count total unique sparse dimensions across all fields
523    pub fn sparse_dim_count(&self) -> usize {
524        self.sparse_vectors.values().map(|b| b.postings.len()).sum()
525    }
526
527    /// Get current statistics for debugging performance (expensive - iterates all data)
528    pub fn stats(&self) -> SegmentBuilderStats {
529        use std::mem::size_of;
530
531        let postings_in_memory: usize =
532            self.inverted_index.values().map(|p| p.postings.len()).sum();
533
534        // Size constants computed from actual types
535        let compact_posting_size = size_of::<CompactPosting>();
536        let vec_overhead = size_of::<Vec<u8>>(); // Vec header: ptr + len + cap = 24 bytes on 64-bit
537        let term_key_size = size_of::<TermKey>();
538        let posting_builder_size = size_of::<PostingListBuilder>();
539        let spur_size = size_of::<Spur>();
540        let sparse_entry_size = size_of::<(DocId, u16, f32)>();
541
542        // hashbrown HashMap entry overhead: key + value + 1 byte control + padding
543        // Measured: ~(key_size + value_size + 8) per entry on average
544        let hashmap_entry_base_overhead = 8usize;
545
546        // FxHashMap uses same layout as hashbrown
547        let fxhashmap_entry_overhead = hashmap_entry_base_overhead;
548
549        // Postings memory
550        let postings_bytes: usize = self
551            .inverted_index
552            .values()
553            .map(|p| p.postings.capacity() * compact_posting_size + vec_overhead)
554            .sum();
555
556        // Inverted index overhead
557        let index_overhead_bytes = self.inverted_index.len()
558            * (term_key_size + posting_builder_size + hashmap_entry_base_overhead);
559
560        // Term interner: Rodeo stores strings + metadata
561        // Rodeo internal: string bytes + Spur + arena overhead (~2 pointers per string)
562        let interner_arena_overhead = 2 * size_of::<usize>();
563        let avg_term_len = 8; // Estimated average term length
564        let interner_bytes =
565            self.term_interner.len() * (avg_term_len + spur_size + interner_arena_overhead);
566
567        // Doc field lengths
568        let field_lengths_bytes =
569            self.doc_field_lengths.capacity() * size_of::<u32>() + vec_overhead;
570
571        // Dense vectors
572        let mut dense_vectors_bytes: usize = 0;
573        let mut dense_vector_count: usize = 0;
574        let doc_id_ordinal_size = size_of::<(DocId, u16)>();
575        for b in self.dense_vectors.values() {
576            dense_vectors_bytes += b.vectors.capacity() * size_of::<f32>()
577                + b.doc_ids.capacity() * doc_id_ordinal_size
578                + 2 * vec_overhead; // Two Vecs
579            dense_vector_count += b.doc_ids.len();
580        }
581        // Binary dense vectors
582        for b in self.binary_dense_vectors.values() {
583            dense_vectors_bytes += b.vectors.capacity()
584                + b.doc_ids.capacity() * doc_id_ordinal_size
585                + 2 * vec_overhead;
586            dense_vector_count += b.doc_ids.len();
587        }
588
589        // Local buffers
590        let local_tf_entry_size = spur_size + size_of::<u32>() + fxhashmap_entry_overhead;
591        let local_tf_buffer_bytes = self.local_tf_buffer.capacity() * local_tf_entry_size;
592
593        // Sparse vectors
594        let mut sparse_vectors_bytes: usize = 0;
595        for builder in self.sparse_vectors.values() {
596            for postings in builder.postings.values() {
597                sparse_vectors_bytes += postings.capacity() * sparse_entry_size + vec_overhead;
598            }
599            // Inner FxHashMap overhead: u32 key + Vec value ptr + overhead
600            let inner_entry_size = size_of::<u32>() + vec_overhead + fxhashmap_entry_overhead;
601            sparse_vectors_bytes += builder.postings.len() * inner_entry_size;
602        }
603        // Outer FxHashMap overhead
604        let outer_sparse_entry_size =
605            size_of::<u32>() + size_of::<SparseVectorBuilder>() + fxhashmap_entry_overhead;
606        sparse_vectors_bytes += self.sparse_vectors.len() * outer_sparse_entry_size;
607
608        // Position index
609        let mut position_index_bytes: usize = 0;
610        for pos_builder in self.position_index.values() {
611            for (_, positions) in &pos_builder.postings {
612                position_index_bytes += positions.capacity() * size_of::<u32>() + vec_overhead;
613            }
614            // Vec<(DocId, Vec<u32>)> entry size
615            let pos_entry_size = size_of::<DocId>() + vec_overhead;
616            position_index_bytes += pos_builder.postings.capacity() * pos_entry_size;
617        }
618        // HashMap overhead for position_index
619        let pos_index_entry_size =
620            term_key_size + size_of::<PositionPostingListBuilder>() + hashmap_entry_base_overhead;
621        position_index_bytes += self.position_index.len() * pos_index_entry_size;
622
623        let estimated_memory_bytes = postings_bytes
624            + index_overhead_bytes
625            + interner_bytes
626            + field_lengths_bytes
627            + dense_vectors_bytes
628            + local_tf_buffer_bytes
629            + sparse_vectors_bytes
630            + position_index_bytes;
631
632        let memory_breakdown = MemoryBreakdown {
633            postings_bytes,
634            index_overhead_bytes,
635            interner_bytes,
636            field_lengths_bytes,
637            dense_vectors_bytes,
638            dense_vector_count,
639            sparse_vectors_bytes,
640            position_index_bytes,
641        };
642
643        SegmentBuilderStats {
644            num_docs: self.next_doc_id,
645            unique_terms: self.inverted_index.len(),
646            postings_in_memory,
647            interned_strings: self.term_interner.len(),
648            doc_field_lengths_size: self.doc_field_lengths.len(),
649            estimated_memory_bytes,
650            memory_breakdown,
651            unhinted_dynamic_docs: self.unhinted_dynamic_docs,
652        }
653    }
654
655    /// Fail-loud pre-validation of a document's field values against the
656    /// schema. Runs BEFORE any builder state is mutated, so a rejected
657    /// document never poisons the builder (doc id advanced, postings written,
658    /// store write skipped).
659    ///
660    /// - A value whose runtime type does not match the schema field type
661    ///   would previously fall through `add_document`'s match silently: the
662    ///   value was stored but never indexed, so queries on the field could
663    ///   never match the document. Reject it loudly instead.
664    /// - Sparse entries destined for a BMP-format field must fit the
665    ///   configured `dims`: the block-max grid only has rows for
666    ///   `dim_id < dims`, so out-of-range entries would be silently dropped
667    ///   from the grid and silently filtered from queries — permanently
668    ///   unsearchable.
669    fn validate_document_against_schema(&self, doc: &Document) -> Result<()> {
670        validate_vector_value_counts(doc, &self.schema)?;
671
672        for (field, value) in doc.field_values() {
673            let Some(entry) = self.schema.get_field_entry(*field) else {
674                continue;
675            };
676
677            // Mirror the indexing skip below: values that are neither indexed
678            // nor fast (and are not vector types) are only stored verbatim.
679            if !matches!(
680                &entry.field_type,
681                FieldType::DenseVector | FieldType::BinaryDenseVector
682            ) && !entry.indexed
683                && !entry.fast
684            {
685                continue;
686            }
687
688            match (&entry.field_type, value) {
689                (FieldType::SparseVector, FieldValue::SparseVector(entries)) => {
690                    if let Some(config) = entry.sparse_vector_config.as_ref()
691                        && config.format == crate::structures::SparseFormat::Bmp
692                    {
693                        let dims = config.dims.unwrap_or(DEFAULT_BMP_SPARSE_DIMS);
694                        if let Some(&(dim_id, _)) =
695                            entries.iter().find(|&&(dim_id, _)| dim_id >= dims)
696                        {
697                            return Err(crate::Error::Schema(format!(
698                                "sparse vector for field '{}' contains dim_id {} out of \
699                                 range for the configured BMP dims={}: dimensions >= dims \
700                                 are never written to the block-max grid and can never \
701                                 match a query; raise `dims` in the field's sparse_vector \
702                                 config or fix the embedding model",
703                                entry.name, dim_id, dims
704                            )));
705                        }
706                    }
707                }
708                // Matching (type, value) pairs — indexed by `add_document`.
709                (FieldType::Text, FieldValue::Text(_))
710                | (FieldType::U64, FieldValue::U64(_))
711                | (FieldType::I64, FieldValue::I64(_))
712                | (FieldType::F64, FieldValue::F64(_))
713                | (FieldType::DenseVector, FieldValue::DenseVector(_))
714                | (FieldType::BinaryDenseVector, FieldValue::BinaryDenseVector(_))
715                // Stored-only types: no indexing support, value stored verbatim.
716                | (FieldType::Bytes, FieldValue::Bytes(_))
717                | (FieldType::Json, FieldValue::Json(_)) => {}
718                (expected, got) => {
719                    return Err(crate::Error::Schema(format!(
720                        "type mismatch for field '{}': schema expects a {} value, got {}; \
721                         the value would be stored but never indexed, so queries on this \
722                         field could never match the document — fix the document or the \
723                         schema",
724                        entry.name,
725                        field_type_name(expected),
726                        field_value_type_name(got),
727                    )));
728                }
729            }
730        }
731        Ok(())
732    }
733
734    /// Add a document - streams to disk immediately
735    pub fn add_document(&mut self, doc: Document) -> Result<DocId> {
736        // Reject schema-mismatched values before mutating any builder state.
737        self.validate_document_against_schema(&doc)?;
738
739        let doc_id = self.next_doc_id;
740        self.next_doc_id += 1;
741
742        // Initialize field lengths for this document
743        let base_idx = self.doc_field_lengths.len();
744        self.doc_field_lengths
745            .resize(base_idx + self.num_indexed_fields, 0);
746        self.estimated_memory += self.num_indexed_fields * std::mem::size_of::<u32>();
747
748        // Reset element ordinals for this document (for multi-valued fields)
749        self.current_element_ordinal.clear();
750
751        for (field, value) in doc.field_values() {
752            let Some(entry) = self.schema.get_field_entry(*field) else {
753                continue;
754            };
755
756            // Dense/binary vectors are written to .vectors when indexed || stored
757            // Other field types require indexed or fast
758            if !matches!(
759                &entry.field_type,
760                FieldType::DenseVector | FieldType::BinaryDenseVector
761            ) && !entry.indexed
762                && !entry.fast
763            {
764                continue;
765            }
766
767            match (&entry.field_type, value) {
768                (FieldType::Text, FieldValue::Text(text)) => {
769                    if entry.indexed && entry.chunked {
770                        // Chunked field: this value is its own scoring unit.
771                        // Postings and positions are keyed by the virtual id;
772                        // the chunk map resolves it back to (doc, ordinal).
773                        let field_id = field.0;
774                        let element_ordinal = self.next_element_ordinal(field_id);
775                        let ordinal = u16::try_from(element_ordinal).map_err(|_| {
776                            crate::Error::Document(format!(
777                                "chunked field {field_id} has more than {} chunk values in one document",
778                                u16::MAX as usize + 1
779                            ))
780                        })?;
781                        let hinted = self.resolve_tokenizer_hint(*field, &doc, element_ordinal);
782                        let vid = self
783                            .chunk_maps
784                            .get(&field.0)
785                            .map_or(0usize, |map| map.len());
786                        let vid = u32::try_from(vid).map_err(|_| {
787                            crate::Error::Document(format!(
788                                "chunked field {field_id} exceeds u32::MAX chunks in one segment"
789                            ))
790                        })?;
791                        // Positions restart at 0 in every chunk (ordinal 0 in
792                        // the encoded position; the schema forbids ordinal
793                        // tracking modes on chunked fields).
794                        let token_count = self.index_text_field(*field, vid, text, 0, hinted)?;
795                        self.chunk_maps.entry(field.0).or_default().push(
796                            doc_id,
797                            ordinal,
798                            token_count,
799                        )?;
800                        self.estimated_memory += 8;
801
802                        // Chunk statistics: `doc_count` counts chunks so
803                        // `avg_field_len` is the average chunk length.
804                        let stats = self.field_stats.entry(field.0).or_default();
805                        stats.total_tokens += token_count as u64;
806                        stats.doc_count += 1;
807                    } else if entry.indexed {
808                        let element_ordinal = self.next_element_ordinal(field.0);
809                        let hinted = self.resolve_tokenizer_hint(*field, &doc, element_ordinal);
810                        let token_count =
811                            self.index_text_field(*field, doc_id, text, element_ordinal, hinted)?;
812
813                        let stats = self.field_stats.entry(field.0).or_default();
814                        stats.total_tokens += token_count as u64;
815                        if element_ordinal == 0 {
816                            stats.doc_count += 1;
817                        }
818
819                        if let Some(&slot) = self.field_to_slot.get(&field.0) {
820                            // Multi-valued fields: the document's length is
821                            // the sum over its values.
822                            let len = &mut self.doc_field_lengths[base_idx + slot];
823                            *len = len.saturating_add(token_count);
824                        }
825                    }
826
827                    // Fast-field: store raw text for text ordinal column
828                    if let Some(ff) = self.fast_fields.get_mut(&field.0) {
829                        ff.add_text(doc_id, text);
830                    }
831                }
832                (FieldType::U64, FieldValue::U64(v)) => {
833                    if entry.indexed {
834                        self.index_numeric_field(*field, doc_id, *v)?;
835                    }
836                    if let Some(ff) = self.fast_fields.get_mut(&field.0) {
837                        ff.add_u64(doc_id, *v);
838                    }
839                }
840                (FieldType::I64, FieldValue::I64(v)) => {
841                    if entry.indexed {
842                        self.index_numeric_field(*field, doc_id, *v as u64)?;
843                    }
844                    if let Some(ff) = self.fast_fields.get_mut(&field.0) {
845                        ff.add_i64(doc_id, *v);
846                    }
847                }
848                (FieldType::F64, FieldValue::F64(v)) => {
849                    if entry.indexed {
850                        self.index_numeric_field(*field, doc_id, v.to_bits())?;
851                    }
852                    if let Some(ff) = self.fast_fields.get_mut(&field.0) {
853                        ff.add_f64(doc_id, *v);
854                    }
855                }
856                (FieldType::DenseVector, FieldValue::DenseVector(vec))
857                    if entry.indexed || entry.stored =>
858                {
859                    let ordinal = self.next_vector_ordinal(field.0)?;
860                    self.index_dense_vector_field(*field, doc_id, ordinal, vec)?;
861                }
862                (FieldType::BinaryDenseVector, FieldValue::BinaryDenseVector(bytes))
863                    if entry.indexed || entry.stored =>
864                {
865                    let ordinal = self.next_vector_ordinal(field.0)?;
866                    self.index_binary_dense_vector_field(*field, doc_id, ordinal, bytes)?;
867                }
868                (FieldType::SparseVector, FieldValue::SparseVector(entries)) => {
869                    let ordinal = self.next_vector_ordinal(field.0)?;
870                    self.index_sparse_vector_field(*field, doc_id, ordinal, entries)?;
871                }
872                // Only reachable for stored-only types (bytes/json) and for
873                // vector values on fields that are neither indexed nor
874                // stored: type-mismatched values are rejected loudly by
875                // `validate_document_against_schema` before this loop.
876                _ => {}
877            }
878        }
879
880        // Stream document to disk immediately
881        self.write_document_to_store(&doc)?;
882
883        Ok(doc_id)
884    }
885
886    /// Index a text field using interned terms
887    ///
888    /// Uses a custom tokenizer when set for the field (via `set_tokenizer`),
889    /// otherwise falls back to an inline zero-allocation path (split_whitespace
890    /// + lowercase + strip non-alphanumeric).
891    ///
892    /// If position recording is enabled for this field, also records token positions
893    /// encoded as (element_ordinal << 20) | token_position.
894    fn index_text_field(
895        &mut self,
896        field: Field,
897        doc_id: DocId,
898        text: &str,
899        element_ordinal: u32,
900        hinted: bool,
901    ) -> Result<u32> {
902        use crate::dsl::PositionMode;
903
904        let field_id = field.0;
905        let position_mode = self
906            .position_enabled_fields
907            .get(&field_id)
908            .copied()
909            .flatten();
910
911        // Saturate the packed 12-bit ordinal field instead of letting the
912        // shift silently wrap (ordinal 4096 << 20 == 0, aliasing element 0).
913        let encoded_ordinal = if position_mode.is_some_and(|m| m.tracks_ordinal())
914            && element_ordinal > MAX_POSITION_ELEMENT_ORDINAL
915        {
916            self.warn_position_saturation(
917                "element ordinal",
918                element_ordinal,
919                MAX_POSITION_ELEMENT_ORDINAL,
920            );
921            MAX_POSITION_ELEMENT_ORDINAL
922        } else {
923            element_ordinal
924        };
925
926        // Phase 1: Aggregate term frequencies within this document
927        // Also collect positions if enabled
928        // Reuse buffers to avoid allocations
929        self.local_tf_buffer.clear();
930        // Retain per-term allocations, but visit only the preceding field's
931        // terms. Scanning local_positions costs O(segment vocabulary) for
932        // EVERY field/chunk, including fields that do not record positions.
933        for term in self.local_position_terms.drain(..) {
934            self.local_positions.get_mut(&term).unwrap().clear();
935        }
936
937        let mut token_position = 0u32;
938
939        // Tokenize: use custom tokenizer if set, else inline zero-alloc path.
940        // The owned Vec<Token> is computed first so the immutable borrow of
941        // self.tokenizers ends before we mutate other fields.
942        let custom_tokens = self.tokenizers.get(&field).map(|t| {
943            if hinted {
944                let hint = (!self.tokenizer_hint_buffer.is_empty())
945                    .then_some(self.tokenizer_hint_buffer.as_str());
946                t.tokenize_with(text, hint, crate::tokenizer::Purpose::Index)
947            } else {
948                t.tokenize(text)
949            }
950        });
951
952        if let Some(tokens) = custom_tokens {
953            // Custom tokenizer path
954            for token in &tokens {
955                let term_spur = if let Some(spur) = self.term_interner.get(&token.text) {
956                    spur
957                } else {
958                    let spur = self.term_interner.get_or_intern(&token.text);
959                    self.estimated_memory += token.text.len() + INTERN_OVERHEAD;
960                    spur
961                };
962                *self.local_tf_buffer.entry(term_spur).or_insert(0) += 1;
963
964                if let Some(mode) = position_mode {
965                    let encoded_pos = match mode {
966                        PositionMode::Ordinal => encoded_ordinal << 20,
967                        PositionMode::TokenPosition => token.position,
968                        PositionMode::Full => {
969                            (encoded_ordinal << 20) | self.saturate_token_position(token.position)
970                        }
971                    };
972                    let positions = self.local_positions.entry(term_spur).or_default();
973                    if positions.is_empty() {
974                        self.local_position_terms.push(term_spur);
975                    }
976                    positions.push(encoded_pos);
977                }
978            }
979            // Variants share a position with their original and are not
980            // tokens of the document for length normalisation.
981            token_position = tokens.iter().filter(|t| !t.variant).count() as u32;
982        } else {
983            // Inline zero-allocation path: split_whitespace + lowercase + strip non-alphanumeric
984            for word in text.split_whitespace() {
985                self.token_buffer.clear();
986                for c in word.chars() {
987                    if c.is_alphanumeric() {
988                        for lc in c.to_lowercase() {
989                            self.token_buffer.push(lc);
990                        }
991                    }
992                }
993
994                if self.token_buffer.is_empty() {
995                    continue;
996                }
997
998                let term_spur = if let Some(spur) = self.term_interner.get(&self.token_buffer) {
999                    spur
1000                } else {
1001                    let spur = self.term_interner.get_or_intern(&self.token_buffer);
1002                    self.estimated_memory += self.token_buffer.len() + INTERN_OVERHEAD;
1003                    spur
1004                };
1005                *self.local_tf_buffer.entry(term_spur).or_insert(0) += 1;
1006
1007                if let Some(mode) = position_mode {
1008                    let encoded_pos = match mode {
1009                        PositionMode::Ordinal => encoded_ordinal << 20,
1010                        PositionMode::TokenPosition => token_position,
1011                        PositionMode::Full => {
1012                            (encoded_ordinal << 20) | self.saturate_token_position(token_position)
1013                        }
1014                    };
1015                    let positions = self.local_positions.entry(term_spur).or_default();
1016                    if positions.is_empty() {
1017                        self.local_position_terms.push(term_spur);
1018                    }
1019                    positions.push(encoded_pos);
1020                }
1021
1022                token_position += 1;
1023            }
1024        }
1025
1026        // Phase 2: Insert aggregated terms into inverted index
1027        // Now we only do one inverted_index lookup per unique term in doc
1028        for (&term_spur, &tf) in &self.local_tf_buffer {
1029            let term_key = TermKey {
1030                field: field_id,
1031                term: term_spur,
1032            };
1033
1034            match self.inverted_index.entry(term_key) {
1035                hashbrown::hash_map::Entry::Occupied(mut o) => {
1036                    o.get_mut().add(doc_id, tf);
1037                    self.estimated_memory += size_of::<CompactPosting>();
1038                    // Spill large posting lists to disk to reduce peak memory
1039                    #[cfg(feature = "native")]
1040                    if o.get().should_spill() {
1041                        use byteorder::{LittleEndian, WriteBytesExt};
1042
1043                        let builder = o.get_mut();
1044                        let count = builder.postings.len() as u32;
1045                        let offset = self.posting_spill_offset;
1046
1047                        // Lazily create the spill file on first spill
1048                        let spill_file = if let Some(ref mut f) = self.posting_spill_file {
1049                            f
1050                        } else {
1051                            self.posting_spill_file = Some(BufWriter::with_capacity(
1052                                256 * 1024,
1053                                OpenOptions::new()
1054                                    .create(true)
1055                                    .write(true)
1056                                    .truncate(true)
1057                                    .open(&self.posting_spill_path)?,
1058                            ));
1059                            self.posting_spill_file.as_mut().unwrap()
1060                        };
1061                        for p in &builder.postings {
1062                            spill_file.write_u32::<LittleEndian>(p.doc_id)?;
1063                            spill_file.write_u16::<LittleEndian>(p.term_freq)?;
1064                        }
1065                        self.posting_spill_offset += count as u64 * 6;
1066                        self.posting_spill_index
1067                            .entry(term_key)
1068                            .or_default()
1069                            .push((offset, count));
1070
1071                        let freed = builder.postings.len() * size_of::<CompactPosting>();
1072                        builder.spilled_count += count;
1073                        builder.postings.clear();
1074                        self.estimated_memory -= freed;
1075                    }
1076                }
1077                hashbrown::hash_map::Entry::Vacant(v) => {
1078                    let mut posting = PostingListBuilder::new();
1079                    posting.add(doc_id, tf);
1080                    v.insert(posting);
1081                    self.estimated_memory += size_of::<CompactPosting>() + NEW_TERM_OVERHEAD;
1082                }
1083            }
1084
1085            if position_mode.is_some()
1086                && let Some(positions) = self.local_positions.get(&term_spur)
1087            {
1088                match self.position_index.entry(term_key) {
1089                    hashbrown::hash_map::Entry::Occupied(mut o) => {
1090                        for &pos in positions {
1091                            o.get_mut().add_position(doc_id, pos);
1092                        }
1093                        self.estimated_memory += positions.len() * size_of::<u32>();
1094                    }
1095                    hashbrown::hash_map::Entry::Vacant(v) => {
1096                        let mut pos_posting = PositionPostingListBuilder::new();
1097                        for &pos in positions {
1098                            pos_posting.add_position(doc_id, pos);
1099                        }
1100                        self.estimated_memory +=
1101                            positions.len() * size_of::<u32>() + NEW_POS_TERM_OVERHEAD;
1102                        v.insert(pos_posting);
1103                    }
1104                }
1105            }
1106        }
1107
1108        Ok(token_position)
1109    }
1110
1111    /// Saturate a token position at the 20-bit packed-encoding maximum so it
1112    /// cannot bleed into the element-ordinal bits.
1113    #[inline]
1114    fn saturate_token_position(&mut self, token_position: u32) -> u32 {
1115        if token_position > MAX_TOKEN_POSITION {
1116            self.warn_position_saturation("token position", token_position, MAX_TOKEN_POSITION);
1117            MAX_TOKEN_POSITION
1118        } else {
1119            token_position
1120        }
1121    }
1122
1123    /// Warn once per segment when the packed position encoding saturates.
1124    #[cold]
1125    fn warn_position_saturation(&mut self, what: &str, value: u32, max: u32) {
1126        if !self.position_saturation_warned {
1127            self.position_saturation_warned = true;
1128            log::warn!(
1129                "[segment_builder] index={} {what} {value} exceeds the position-encoding limit {max}; \
1130                 saturating — phrase/ordinal matching degrades for the overflowing \
1131                 elements/tokens instead of corrupting other documents' matches \
1132                 (further occurrences in this segment are not logged)",
1133                self.schema.index_label()
1134            );
1135        }
1136    }
1137
1138    fn index_numeric_field(&mut self, field: Field, doc_id: DocId, value: u64) -> Result<()> {
1139        use std::fmt::Write;
1140
1141        self.numeric_buffer.clear();
1142        write!(self.numeric_buffer, "__num_{}", value).unwrap();
1143        let term_spur = if let Some(spur) = self.term_interner.get(&self.numeric_buffer) {
1144            spur
1145        } else {
1146            let spur = self.term_interner.get_or_intern(&self.numeric_buffer);
1147            self.estimated_memory += self.numeric_buffer.len() + INTERN_OVERHEAD;
1148            spur
1149        };
1150
1151        let term_key = TermKey {
1152            field: field.0,
1153            term: term_spur,
1154        };
1155
1156        match self.inverted_index.entry(term_key) {
1157            hashbrown::hash_map::Entry::Occupied(mut o) => {
1158                o.get_mut().add(doc_id, 1);
1159                self.estimated_memory += size_of::<CompactPosting>();
1160            }
1161            hashbrown::hash_map::Entry::Vacant(v) => {
1162                let mut posting = PostingListBuilder::new();
1163                posting.add(doc_id, 1);
1164                v.insert(posting);
1165                self.estimated_memory += size_of::<CompactPosting>() + NEW_TERM_OVERHEAD;
1166            }
1167        }
1168
1169        Ok(())
1170    }
1171
1172    /// Index a dense vector field with ordinal tracking
1173    fn index_dense_vector_field(
1174        &mut self,
1175        field: Field,
1176        doc_id: DocId,
1177        ordinal: u16,
1178        vector: &[f32],
1179    ) -> Result<()> {
1180        let dim = vector.len();
1181        let expected_dim = self
1182            .schema
1183            .get_field_entry(field)
1184            .and_then(|entry| entry.dense_vector_config.as_ref())
1185            .map(|config| config.dim)
1186            .ok_or_else(|| crate::Error::Schema("DenseVector field missing config".to_string()))?;
1187        if dim != expected_dim {
1188            return Err(crate::Error::Schema(format!(
1189                "Dense vector dimension mismatch: schema expects {}, got {}",
1190                expected_dim, dim
1191            )));
1192        }
1193        if let Some((index, value)) = vector
1194            .iter()
1195            .enumerate()
1196            .find(|(_, value)| !value.is_finite())
1197        {
1198            return Err(crate::Error::Document(format!(
1199                "dense vector contains non-finite value {value} at index {index}"
1200            )));
1201        }
1202
1203        let builder = self
1204            .dense_vectors
1205            .entry(field.0)
1206            .or_insert_with(|| DenseVectorBuilder::new(dim));
1207
1208        // Verify dimension consistency
1209        if builder.dim != dim && builder.len() > 0 {
1210            return Err(crate::Error::Schema(format!(
1211                "Dense vector dimension mismatch: expected {}, got {}",
1212                builder.dim, dim
1213            )));
1214        }
1215
1216        builder.add(doc_id, ordinal, vector);
1217
1218        self.estimated_memory += std::mem::size_of_val(vector) + size_of::<(DocId, u16)>();
1219
1220        Ok(())
1221    }
1222
1223    /// Index a binary dense vector field with ordinal tracking
1224    fn index_binary_dense_vector_field(
1225        &mut self,
1226        field: Field,
1227        doc_id: DocId,
1228        ordinal: u16,
1229        bytes: &[u8],
1230    ) -> Result<()> {
1231        let dim_bits = self
1232            .schema
1233            .get_field_entry(field)
1234            .and_then(|e| e.binary_dense_vector_config.as_ref())
1235            .map(|c| c.dim)
1236            .ok_or_else(|| {
1237                crate::Error::Schema("BinaryDenseVector field missing config".to_string())
1238            })?;
1239
1240        let expected_byte_len = dim_bits.div_ceil(8);
1241        if dim_bits == 0 || !dim_bits.is_multiple_of(8) {
1242            return Err(crate::Error::Schema(format!(
1243                "Binary vector dimension must be a positive multiple of 8, got {dim_bits}"
1244            )));
1245        }
1246        if bytes.len() != expected_byte_len {
1247            return Err(crate::Error::Schema(format!(
1248                "Binary vector byte length mismatch: expected {} (dim={}), got {}",
1249                expected_byte_len,
1250                dim_bits,
1251                bytes.len()
1252            )));
1253        }
1254
1255        let builder = self
1256            .binary_dense_vectors
1257            .entry(field.0)
1258            .or_insert_with(|| BinaryDenseVectorBuilder::new(dim_bits));
1259
1260        builder.add(doc_id, ordinal, bytes);
1261        self.estimated_memory += bytes.len() + size_of::<(DocId, u16)>();
1262
1263        Ok(())
1264    }
1265
1266    /// Index a sparse vector field using dedicated sparse posting lists
1267    ///
1268    /// Collects (doc_id, ordinal, weight) postings per dimension. During commit, these are
1269    /// converted to BlockSparsePostingList with proper quantization from SparseVectorConfig.
1270    ///
1271    /// Weights below the configured `weight_threshold` are not indexed. When
1272    /// `doc_mass` is configured, only the top-|weight| entries covering that
1273    /// fraction of the vector's total |weight| mass are kept (the excessive
1274    /// tail of SPLADE-style vectors is cropped).
1275    fn index_sparse_vector_field(
1276        &mut self,
1277        field: Field,
1278        doc_id: DocId,
1279        ordinal: u16,
1280        entries: &[(u32, f32)],
1281    ) -> Result<()> {
1282        if let Some((index, (_, weight))) = entries
1283            .iter()
1284            .enumerate()
1285            .find(|(_, (_, weight))| !weight.is_finite())
1286        {
1287            return Err(crate::Error::Document(format!(
1288                "sparse vector contains non-finite weight {weight} at index {index}"
1289            )));
1290        }
1291        let (weight_threshold, doc_mass, min_terms) = self
1292            .schema
1293            .get_field_entry(field)
1294            .and_then(|entry| entry.sparse_vector_config.as_ref())
1295            .map(|config| (config.weight_threshold, config.doc_mass, config.min_terms))
1296            .unwrap_or((0.0, None, 0));
1297
1298        let builder = self
1299            .sparse_vectors
1300            .entry(field.0)
1301            .or_insert_with(SparseVectorBuilder::new);
1302
1303        builder.inc_vector_count();
1304
1305        // Document-side mass cropping: determine the per-vector weight cutoff
1306        // below which entries fall outside the doc_mass fraction of total mass.
1307        // Short vectors (<= min_terms entries) are never cropped.
1308        let mass_cutoff = match doc_mass {
1309            Some(mass) if mass < 1.0 && entries.len() > min_terms => {
1310                let mut weights: Vec<f32> = entries
1311                    .iter()
1312                    .map(|&(_, w)| w.abs())
1313                    .filter(|w| *w >= weight_threshold)
1314                    .collect();
1315                weights.sort_unstable_by(|a, b| b.total_cmp(a));
1316                let total: f64 = weights.iter().map(|&w| w as f64).sum();
1317                let target = total * mass as f64;
1318                let mut cumulative = 0.0f64;
1319                let mut cutoff = 0.0f32;
1320                for &w in &weights {
1321                    if cumulative >= target {
1322                        break;
1323                    }
1324                    cumulative += w as f64;
1325                    cutoff = w;
1326                }
1327                cutoff
1328            }
1329            _ => 0.0,
1330        };
1331
1332        for &(dim_id, weight) in entries {
1333            // Skip weights below threshold or outside the doc_mass prefix
1334            if weight.abs() < weight_threshold || weight.abs() < mass_cutoff {
1335                continue;
1336            }
1337
1338            let is_new_dim = !builder.postings.contains_key(&dim_id);
1339            builder.add(dim_id, doc_id, ordinal, weight);
1340            self.estimated_memory += size_of::<(DocId, u16, f32)>();
1341            if is_new_dim {
1342                // HashMap entry overhead + Vec header
1343                self.estimated_memory += size_of::<u32>() + size_of::<Vec<(DocId, u16, f32)>>() + 8; // 8 = hashmap control byte + padding
1344            }
1345        }
1346
1347        Ok(())
1348    }
1349
1350    /// Write document to streaming store (reuses internal buffer to avoid per-doc allocation)
1351    fn write_document_to_store(&mut self, doc: &Document) -> Result<()> {
1352        use byteorder::{LittleEndian, WriteBytesExt};
1353
1354        super::store::serialize_document_into(doc, &self.schema, &mut self.doc_serialize_buffer)?;
1355
1356        #[cfg(feature = "native")]
1357        {
1358            self.store_file
1359                .write_u32::<LittleEndian>(self.doc_serialize_buffer.len() as u32)?;
1360            self.store_file.write_all(&self.doc_serialize_buffer)?;
1361        }
1362        #[cfg(not(feature = "native"))]
1363        {
1364            self.store_buffer
1365                .write_u32::<LittleEndian>(self.doc_serialize_buffer.len() as u32)?;
1366            self.store_buffer.write_all(&self.doc_serialize_buffer)?;
1367            // The in-memory store buffer is often the largest allocation on
1368            // the wasm branch (native streams docs to a temp file instead).
1369            // Count it so the memory-budget flush check can see it.
1370            self.estimated_memory += size_of::<u32>() + self.doc_serialize_buffer.len();
1371        }
1372
1373        Ok(())
1374    }
1375
1376    /// Build the final segment
1377    ///
1378    /// Streams all data directly to disk via StreamingWriter to avoid buffering
1379    /// entire serialized outputs in memory. Each phase consumes and drops its
1380    /// source data before the next phase begins.
1381    pub async fn build<D: Directory + DirectoryWriter>(
1382        mut self,
1383        dir: &D,
1384        segment_id: SegmentId,
1385        trained: Option<&super::TrainedVectorStructures>,
1386    ) -> Result<SegmentMeta> {
1387        // Flush any buffered data
1388        #[cfg(feature = "native")]
1389        self.store_file.flush()?;
1390
1391        let files = SegmentFiles::new(segment_id.0);
1392
1393        // Phase 1: Stream positions directly to disk (consumes position_index)
1394        let position_index = std::mem::take(&mut self.position_index);
1395        let position_offsets = if !position_index.is_empty() {
1396            let mut pos_writer = dir.streaming_writer(&files.positions).await?;
1397            let offsets = postings::build_positions_streaming(
1398                position_index,
1399                &self.term_interner,
1400                &mut *pos_writer,
1401            )?;
1402            pos_writer.finish()?;
1403            offsets
1404        } else {
1405            FxHashMap::default()
1406        };
1407
1408        // Phase 1b: chunk maps of chunked text fields (8 bytes per chunk) and
1409        // per-document length columns of plain text fields (2 bytes per doc).
1410        let chunk_maps = std::mem::take(&mut self.chunk_maps);
1411        {
1412            let mut fields: Vec<(u32, &super::chunk_map::ChunkMapBuilder)> = chunk_maps
1413                .iter()
1414                .filter(|(_, map)| !map.is_empty())
1415                .map(|(field_id, map)| (*field_id, map))
1416                .collect();
1417            fields.sort_by_key(|(field_id, _)| *field_id);
1418            let num_docs = self.next_doc_id as usize;
1419            let mut columns: Vec<(u32, Vec<u16>, u64)> = Vec::new();
1420            for (&field_id, &slot) in &self.field_to_slot {
1421                if self
1422                    .schema
1423                    .get_field_entry(crate::dsl::Field(field_id))
1424                    .is_some_and(|entry| entry.chunked)
1425                {
1426                    continue;
1427                }
1428                let mut total = 0u64;
1429                let lengths: Vec<u16> = (0..num_docs)
1430                    .map(|doc| {
1431                        let len = self.doc_field_lengths[doc * self.num_indexed_fields + slot];
1432                        total += u64::from(len);
1433                        len.min(super::chunk_map::MAX_CHUNK_LENGTH) as u16
1434                    })
1435                    .collect();
1436                if total > 0 {
1437                    columns.push((field_id, lengths, total));
1438                }
1439            }
1440            columns.sort_by_key(|(field_id, _, _)| *field_id);
1441            let norms: Vec<super::chunk_map::DocLengthsColumn<'_>> = columns
1442                .iter()
1443                .map(
1444                    |(field_id, lengths, total)| super::chunk_map::DocLengthsColumn {
1445                        field_id: *field_id,
1446                        lengths,
1447                        total_tokens: *total,
1448                    },
1449                )
1450                .collect();
1451            if !fields.is_empty() || !norms.is_empty() {
1452                let mut writer = dir.streaming_writer(&files.chunks).await?;
1453                super::chunk_map::write_chunk_maps(&mut *writer, &fields, &norms)?;
1454                writer.finish()?;
1455            }
1456        }
1457        let length_lookup = postings::LengthLookup {
1458            doc_lengths: &self.doc_field_lengths,
1459            num_indexed_fields: self.num_indexed_fields,
1460            field_to_slot: &self.field_to_slot,
1461            chunk_maps: &chunk_maps,
1462        };
1463
1464        // Phase 2: 4-way parallel build — postings, store, dense vectors, sparse vectors
1465        // These are fully independent: different source data, different output files.
1466        let inverted_index = std::mem::take(&mut self.inverted_index);
1467        let term_interner = std::mem::replace(&mut self.term_interner, Rodeo::new());
1468        #[cfg(feature = "native")]
1469        let store_path = self.store_path.clone();
1470        #[cfg(feature = "native")]
1471        let num_compression_threads = self.config.num_compression_threads;
1472        let compression_level = self.config.compression_level;
1473        let optimization = self.config.optimization;
1474        let posting_codec = self.config.posting_codec;
1475        let dense_vectors = std::mem::take(&mut self.dense_vectors);
1476        let binary_dense_vectors = std::mem::take(&mut self.binary_dense_vectors);
1477        let mut sparse_vectors = std::mem::take(&mut self.sparse_vectors);
1478        let schema = &self.schema;
1479
1480        // Pre-create all streaming writers (async) before entering sync rayon scope
1481        // Wrapped in OffsetWriter to track bytes written per phase.
1482        let mut term_dict_writer =
1483            super::OffsetWriter::new(dir.streaming_writer(&files.term_dict).await?);
1484        let mut postings_writer =
1485            super::OffsetWriter::new(dir.streaming_writer(&files.postings).await?);
1486        let mut store_writer = super::OffsetWriter::new(dir.streaming_writer(&files.store).await?);
1487        let mut vectors_writer = if !dense_vectors.is_empty() || !binary_dense_vectors.is_empty() {
1488            Some(super::OffsetWriter::new(
1489                dir.streaming_writer(&files.vectors).await?,
1490            ))
1491        } else {
1492            None
1493        };
1494        let mut sparse_writer = if !sparse_vectors.is_empty() {
1495            Some(super::OffsetWriter::new(
1496                dir.streaming_writer(&files.sparse).await?,
1497            ))
1498        } else {
1499            None
1500        };
1501        let mut fast_fields = std::mem::take(&mut self.fast_fields);
1502        let num_docs = self.next_doc_id;
1503        let mut fast_writer = if !fast_fields.is_empty() {
1504            Some(super::OffsetWriter::new(
1505                dir.streaming_writer(&files.fast).await?,
1506            ))
1507        } else {
1508            None
1509        };
1510
1511        #[cfg(feature = "native")]
1512        {
1513            if let Some(ref mut f) = self.posting_spill_file {
1514                f.flush()?;
1515            }
1516            let posting_spill_index = std::mem::take(&mut self.posting_spill_index);
1517            let mut spill_reader_opt = if !posting_spill_index.is_empty() {
1518                let spill_file = std::fs::File::open(&self.posting_spill_path)?;
1519                Some((std::io::BufReader::new(spill_file), posting_spill_index))
1520            } else {
1521                None
1522            };
1523
1524            let ((postings_result, store_result), ((vectors_result, sparse_result), fast_result)) =
1525                rayon::join(
1526                    || {
1527                        rayon::join(
1528                            || {
1529                                let spill_arg = spill_reader_opt.as_mut().map(|(r, idx)| {
1530                                    (
1531                                        r as &mut std::io::BufReader<std::fs::File>,
1532                                        idx as &postings::SpillIndex,
1533                                    )
1534                                });
1535                                postings::build_postings_streaming(
1536                                    inverted_index,
1537                                    term_interner,
1538                                    &position_offsets,
1539                                    &length_lookup,
1540                                    &mut term_dict_writer,
1541                                    &mut postings_writer,
1542                                    (optimization, posting_codec),
1543                                    spill_arg,
1544                                )
1545                            },
1546                            || {
1547                                store::build_store_streaming(
1548                                    &store_path,
1549                                    num_compression_threads,
1550                                    compression_level,
1551                                    &mut store_writer,
1552                                    num_docs,
1553                                )
1554                            },
1555                        )
1556                    },
1557                    || {
1558                        rayon::join(
1559                            || {
1560                                rayon::join(
1561                                    || -> Result<()> {
1562                                        if let Some(ref mut w) = vectors_writer {
1563                                            dense::build_vectors_streaming(
1564                                                dense_vectors,
1565                                                binary_dense_vectors,
1566                                                schema,
1567                                                trained,
1568                                                w,
1569                                            )?;
1570                                        }
1571                                        Ok(())
1572                                    },
1573                                    || -> Result<()> {
1574                                        if let Some(ref mut w) = sparse_writer {
1575                                            sparse::build_sparse_streaming(
1576                                                &mut sparse_vectors,
1577                                                schema,
1578                                                w,
1579                                            )?;
1580                                        }
1581                                        Ok(())
1582                                    },
1583                                )
1584                            },
1585                            || -> Result<()> {
1586                                if let Some(ref mut w) = fast_writer {
1587                                    build_fast_fields_streaming(&mut fast_fields, num_docs, w)?;
1588                                }
1589                                Ok(())
1590                            },
1591                        )
1592                    },
1593                );
1594            postings_result?;
1595            store_result?;
1596            vectors_result?;
1597            sparse_result?;
1598            fast_result?;
1599        }
1600
1601        #[cfg(not(feature = "native"))]
1602        {
1603            postings::build_postings_streaming(
1604                inverted_index,
1605                term_interner,
1606                &position_offsets,
1607                &length_lookup,
1608                &mut term_dict_writer,
1609                &mut postings_writer,
1610                (optimization, posting_codec),
1611            )?;
1612            store::build_store_streaming_from_buffer(
1613                &self.store_buffer,
1614                compression_level,
1615                &mut store_writer,
1616                num_docs,
1617            )?;
1618            if let Some(ref mut w) = vectors_writer {
1619                dense::build_vectors_streaming(
1620                    dense_vectors,
1621                    binary_dense_vectors,
1622                    schema,
1623                    trained,
1624                    w,
1625                )?;
1626            }
1627            if let Some(ref mut w) = sparse_writer {
1628                sparse::build_sparse_streaming(&mut sparse_vectors, schema, w)?;
1629            }
1630            if let Some(ref mut w) = fast_writer {
1631                build_fast_fields_streaming(&mut fast_fields, num_docs, w)?;
1632            }
1633        }
1634
1635        let term_dict_bytes = term_dict_writer.offset() as usize;
1636        let postings_bytes = postings_writer.offset() as usize;
1637        let store_bytes = store_writer.offset() as usize;
1638        let vectors_bytes = vectors_writer.as_ref().map_or(0, |w| w.offset() as usize);
1639        let sparse_bytes = sparse_writer.as_ref().map_or(0, |w| w.offset() as usize);
1640        let fast_bytes = fast_writer.as_ref().map_or(0, |w| w.offset() as usize);
1641
1642        term_dict_writer.finish()?;
1643        postings_writer.finish()?;
1644        store_writer.finish()?;
1645        if let Some(w) = vectors_writer {
1646            w.finish()?;
1647        }
1648        if let Some(w) = sparse_writer {
1649            w.finish()?;
1650        }
1651        if let Some(w) = fast_writer {
1652            w.finish()?;
1653        }
1654        drop(position_offsets);
1655        drop(sparse_vectors);
1656
1657        log::info!(
1658            "[segment_build] index={} docs={}: term_dict={}, postings={}, store={}, dense_vectors={}, sparse_vectors={}, fast_fields={}",
1659            self.schema.index_label(),
1660            num_docs,
1661            crate::format_bytes(term_dict_bytes as u64),
1662            crate::format_bytes(postings_bytes as u64),
1663            crate::format_bytes(store_bytes as u64),
1664            crate::format_bytes(vectors_bytes as u64),
1665            crate::format_bytes(sparse_bytes as u64),
1666            crate::format_bytes(fast_bytes as u64),
1667        );
1668
1669        let meta = SegmentMeta {
1670            id: segment_id.0,
1671            num_docs: self.next_doc_id,
1672            field_stats: self.field_stats.clone(),
1673        };
1674
1675        // Durable: committed metadata.json will reference this segment, so a
1676        // torn/unsynced .meta after power loss would make the commit
1677        // unreadable (every other segment file is fsynced by its streaming
1678        // writer's finish()).
1679        dir.write_durable(&files.meta, &meta.serialize()?).await?;
1680
1681        // Cleanup temp files
1682        #[cfg(feature = "native")]
1683        {
1684            let _ = std::fs::remove_file(&self.store_path);
1685        }
1686
1687        Ok(meta)
1688    }
1689}
1690
1691/// Serialize all fast-field columns to a `.fast` file.
1692fn build_fast_fields_streaming(
1693    fast_fields: &mut FxHashMap<u32, crate::structures::fast_field::FastFieldWriter>,
1694    num_docs: u32,
1695    writer: &mut dyn Write,
1696) -> Result<()> {
1697    use crate::structures::fast_field::{FastFieldTocEntry, write_fast_field_toc_and_footer};
1698
1699    if fast_fields.is_empty() {
1700        return Ok(());
1701    }
1702
1703    // Sort fields by id for deterministic output
1704    let mut field_ids: Vec<u32> = fast_fields.keys().copied().collect();
1705    field_ids.sort_unstable();
1706
1707    let mut toc_entries: Vec<FastFieldTocEntry> = Vec::with_capacity(field_ids.len());
1708    let mut current_offset = 0u64;
1709
1710    for &field_id in &field_ids {
1711        let ff = fast_fields.get_mut(&field_id).unwrap();
1712        ff.pad_to(num_docs);
1713
1714        let (mut toc, bytes_written) = ff.serialize(writer, current_offset)?;
1715        toc.field_id = field_id;
1716        current_offset += bytes_written;
1717        toc_entries.push(toc);
1718    }
1719
1720    // Write TOC + footer
1721    let toc_offset = current_offset;
1722    write_fast_field_toc_and_footer(writer, toc_offset, &toc_entries)?;
1723
1724    Ok(())
1725}
1726
1727#[cfg(feature = "native")]
1728impl Drop for SegmentBuilder {
1729    fn drop(&mut self) {
1730        let _ = std::fs::remove_file(&self.store_path);
1731        if self.posting_spill_file.is_some() {
1732            let _ = std::fs::remove_file(&self.posting_spill_path);
1733        }
1734    }
1735}
1736
1737#[cfg(test)]
1738impl SegmentBuilder {
1739    /// Test helper: all encoded positions recorded for `(field, term)`.
1740    fn positions_for_term(&self, field: Field, term: &str) -> Vec<u32> {
1741        let Some(spur) = self.term_interner.get(term) else {
1742            return Vec::new();
1743        };
1744        let key = TermKey {
1745            field: field.0,
1746            term: spur,
1747        };
1748        self.position_index
1749            .get(&key)
1750            .map(|b| {
1751                b.postings
1752                    .iter()
1753                    .flat_map(|(_, ps)| ps.iter().copied())
1754                    .collect()
1755            })
1756            .unwrap_or_default()
1757    }
1758}
1759
1760#[cfg(test)]
1761mod tests {
1762    use super::*;
1763    use crate::dsl::SchemaBuilder;
1764
1765    fn builder_for(schema: Schema) -> SegmentBuilder {
1766        SegmentBuilder::new(Arc::new(schema), SegmentBuilderConfig::default()).unwrap()
1767    }
1768
1769    #[test]
1770    fn position_scratch_resets_only_touched_terms_across_fields_and_chunks() {
1771        use crate::dsl::PositionMode;
1772        use crate::tokenizer::SimpleTokenizer;
1773
1774        for custom in [false, true] {
1775            for mode in [
1776                PositionMode::Ordinal,
1777                PositionMode::TokenPosition,
1778                PositionMode::Full,
1779            ] {
1780                let mut schema = SchemaBuilder::default();
1781                let body = schema.add_text_field("body", true, false);
1782                schema.set_positions(body, mode);
1783                let other = schema.add_text_field("other", true, false);
1784                schema.set_positions(other, PositionMode::TokenPosition);
1785                let no_positions = schema.add_text_field("no_positions", true, false);
1786                let mut builder = builder_for(schema.build());
1787                if custom {
1788                    builder.set_tokenizer(body, Box::new(SimpleTokenizer));
1789                }
1790
1791                // Accumulate segment vocabulary while each field/chunk has
1792                // just two unique terms. Duplicate tokens must be reset once.
1793                for doc in 0..2_000 {
1794                    builder
1795                        .index_text_field(
1796                            body,
1797                            doc,
1798                            &format!("unique{doc} anchor anchor"),
1799                            0,
1800                            false,
1801                        )
1802                        .unwrap();
1803                    assert_eq!(builder.local_position_terms.len(), 2);
1804                }
1805                assert_eq!(builder.local_positions.len(), 2_001);
1806                let anchor = builder.term_interner.get("anchor").unwrap();
1807                let capacity = builder.local_positions[&anchor].capacity();
1808
1809                builder
1810                    .index_text_field(no_positions, 2_000, "anchor plain", 0, false)
1811                    .unwrap();
1812                assert!(builder.local_position_terms.is_empty());
1813                assert!(builder.local_positions[&anchor].is_empty());
1814                assert_eq!(builder.local_positions[&anchor].capacity(), capacity);
1815                builder
1816                    .index_text_field(other, 2_000, "anchor", 0, false)
1817                    .unwrap();
1818                builder.index_text_field(body, 2_000, "", 0, false).unwrap();
1819                assert!(builder.local_position_terms.is_empty());
1820                builder
1821                    .index_text_field(body, 2_000, "anchor", 1, false)
1822                    .unwrap();
1823                builder
1824                    .index_text_field(body, 2_001, "anchor", 2, false)
1825                    .unwrap();
1826
1827                assert_eq!(builder.positions_for_term(other, "anchor"), vec![0]);
1828                let positions = builder.positions_for_term(body, "anchor");
1829                assert_eq!(
1830                    positions.len(),
1831                    4_002,
1832                    "positions leaked between chunks or fields"
1833                );
1834                let expected = match mode {
1835                    PositionMode::TokenPosition => [0, 0],
1836                    PositionMode::Ordinal | PositionMode::Full => [1 << 20, 2 << 20],
1837                };
1838                assert_eq!(&positions[4_000..], &expected);
1839            }
1840        }
1841    }
1842
1843    // ------------------------------------------------------------------
1844    // Finding: field values whose runtime type does not match the schema
1845    // field type fell into `_ => {}` and were silently not indexed while
1846    // still being stored — queries could never match the document.
1847    // ------------------------------------------------------------------
1848    #[test]
1849    fn test_add_document_rejects_type_mismatched_field_value() {
1850        let mut sb = SchemaBuilder::default();
1851        let views = sb.add_u64_field("views", true, true);
1852        let mut builder = builder_for(sb.build());
1853
1854        let mut doc = Document::new();
1855        doc.add_text(views, "123");
1856        let err = builder
1857            .add_document(doc)
1858            .expect_err("schema-mismatched value must be rejected loudly, not silently unindexed");
1859        let msg = err.to_string();
1860        assert!(msg.contains("views"), "error must name the field: {msg}");
1861        assert!(
1862            msg.contains("u64"),
1863            "error must name the expected type: {msg}"
1864        );
1865        assert!(msg.contains("text"), "error must name the got type: {msg}");
1866
1867        // The rejected document must not have consumed a doc id (no poisoning).
1868        assert_eq!(builder.num_docs(), 0);
1869
1870        // A well-typed document still indexes fine afterwards.
1871        let mut doc = Document::new();
1872        doc.add_u64(views, 123);
1873        builder.add_document(doc).unwrap();
1874        assert_eq!(builder.num_docs(), 1);
1875    }
1876
1877    // ------------------------------------------------------------------
1878    // Finding: sparse entries with dim_id >= the configured BMP `dims`
1879    // were accepted at index time but silently dropped from the BMP grid
1880    // and silently filtered from queries — permanently unsearchable.
1881    // ------------------------------------------------------------------
1882    #[test]
1883    fn test_add_document_rejects_bmp_sparse_dim_out_of_range() {
1884        use crate::structures::{SparseFormat, SparseVectorConfig};
1885
1886        let mut sb = SchemaBuilder::default();
1887        let config = SparseVectorConfig {
1888            format: SparseFormat::Bmp,
1889            dims: Some(100),
1890            ..Default::default()
1891        };
1892        let spv = sb.add_sparse_vector_field_with_config("spv", true, false, config);
1893        let mut builder = builder_for(sb.build());
1894
1895        // In-range dims are accepted.
1896        let mut doc = Document::new();
1897        doc.add_sparse_vector(spv, vec![(50, 1.0)]);
1898        builder.add_document(doc).unwrap();
1899
1900        // dim_id >= dims must be rejected with an actionable error.
1901        let mut doc = Document::new();
1902        doc.add_sparse_vector(spv, vec![(50, 1.0), (150, 2.0)]);
1903        let err = builder
1904            .add_document(doc)
1905            .expect_err("out-of-range BMP dim must be rejected, not silently unsearchable");
1906        let msg = err.to_string();
1907        assert!(msg.contains("spv"), "error must name the field: {msg}");
1908        assert!(msg.contains("150"), "error must name the dim_id: {msg}");
1909        assert!(
1910            msg.contains("100"),
1911            "error must name the configured dims: {msg}"
1912        );
1913        assert_eq!(
1914            builder.num_docs(),
1915            1,
1916            "rejected doc must not consume a doc id"
1917        );
1918    }
1919
1920    #[test]
1921    fn test_add_document_maxscore_sparse_dims_unbounded() {
1922        // MaxScore-format sparse fields have no dims bound — large dim ids
1923        // stay legal (the per-dim TOC addresses any u32 dimension).
1924        let mut sb = SchemaBuilder::default();
1925        let spv = sb.add_sparse_vector_field("spv", true, false);
1926        let mut builder = builder_for(sb.build());
1927
1928        let mut doc = Document::new();
1929        doc.add_sparse_vector(spv, vec![(3_000_000, 1.0)]);
1930        builder.add_document(doc).unwrap();
1931    }
1932
1933    // ------------------------------------------------------------------
1934    // Finding: `(element_ordinal << 20) | token_position` silently
1935    // corrupted when element_ordinal >= 4096 (shifted out of the u32,
1936    // aliasing element 0) or token_position >= 2^20 (bleeding into the
1937    // ordinal bits). Both must saturate at their field maxima.
1938    // ------------------------------------------------------------------
1939    #[test]
1940    fn test_position_element_ordinal_overflow_saturates_instead_of_wrapping() {
1941        use crate::dsl::PositionMode;
1942
1943        let mut sb = SchemaBuilder::default();
1944        let body = sb.add_text_field("body", true, false);
1945        sb.set_positions(body, PositionMode::Full);
1946        let mut builder = builder_for(sb.build());
1947
1948        // 4097 values: element ordinal 4096 does not fit the 12-bit ordinal
1949        // field ((4096u32 << 20) wraps to 0, colliding with element 0).
1950        let mut doc = Document::new();
1951        doc.add_text(body, "anchor");
1952        for _ in 0..4095 {
1953            doc.add_text(body, "filler");
1954        }
1955        doc.add_text(body, "needle");
1956        builder.add_document(doc).unwrap();
1957
1958        let positions = builder.positions_for_term(body, "needle");
1959        assert_eq!(positions.len(), 1);
1960        let encoded = positions[0];
1961        assert_ne!(
1962            encoded >> 20,
1963            0,
1964            "element ordinal 4096 must not alias element 0"
1965        );
1966        assert_eq!(
1967            encoded >> 20,
1968            4095,
1969            "overflowing element ordinal must saturate at 4095"
1970        );
1971    }
1972
1973    #[test]
1974    fn test_position_token_position_overflow_saturates_instead_of_bleeding() {
1975        use crate::dsl::PositionMode;
1976
1977        let mut sb = SchemaBuilder::default();
1978        let body = sb.add_text_field("body", true, false);
1979        sb.set_positions(body, PositionMode::Full);
1980        let mut builder = builder_for(sb.build());
1981
1982        // One value with 2^20 + 1 tokens: the last token's position does not
1983        // fit the 20-bit position field and would bleed into ordinal bit 0.
1984        let mut text = "w ".repeat(1 << 20);
1985        text.push_str("needle");
1986        let mut doc = Document::new();
1987        doc.add_text(body, text);
1988        builder.add_document(doc).unwrap();
1989
1990        let positions = builder.positions_for_term(body, "needle");
1991        assert_eq!(positions.len(), 1);
1992        let encoded = positions[0];
1993        assert_eq!(
1994            encoded >> 20,
1995            0,
1996            "token position overflow must not decode as a different element ordinal"
1997        );
1998        assert_eq!(
1999            encoded & 0xFFFFF,
2000            0xFFFFF,
2001            "overflowing token position must saturate at 2^20 - 1"
2002        );
2003    }
2004
2005    // ------------------------------------------------------------------
2006    // Finding: a posting-list spill firing between two values of the same
2007    // document split that document's postings across the spilled range and
2008    // the in-memory tail; the build-time merge concatenated them without
2009    // deduplication (inflated doc_freq, doc visited twice, split tf).
2010    // ------------------------------------------------------------------
2011    #[cfg(feature = "native")]
2012    #[tokio::test]
2013    async fn test_spill_mid_document_does_not_duplicate_postings() {
2014        use crate::directories::RamDirectory;
2015        use crate::structures::TERMINATED;
2016
2017        let mut sb = SchemaBuilder::default();
2018        let body = sb.add_text_field("body", true, false);
2019        let schema = Arc::new(sb.build());
2020        let mut builder =
2021            SegmentBuilder::new(Arc::clone(&schema), SegmentBuilderConfig::default()).unwrap();
2022
2023        // Docs 0..16382 each contribute one posting for "hot", leaving the
2024        // in-memory posting list one entry short of SPILL_THRESHOLD (16384).
2025        for _ in 0..16383 {
2026            let mut doc = Document::new();
2027            doc.add_text(body, "hot");
2028            builder.add_document(doc).unwrap();
2029        }
2030
2031        // Doc 16383 has TWO values containing "hot": indexing the first value
2032        // reaches the spill threshold and spills the list INCLUDING this doc's
2033        // entry; the second value then re-adds the same doc to the now-empty
2034        // in-memory tail.
2035        let mut doc = Document::new();
2036        doc.add_text(body, "hot");
2037        doc.add_text(body, "hot");
2038        let boundary_doc = builder.add_document(doc).unwrap();
2039        assert_eq!(boundary_doc, 16383);
2040
2041        let dir = RamDirectory::new();
2042        let segment_id = crate::segment::SegmentId::new();
2043        builder.build(&dir, segment_id, None).await.unwrap();
2044
2045        let reader = crate::segment::SegmentReader::open(&dir, segment_id, schema, 16)
2046            .await
2047            .unwrap();
2048        let postings = reader
2049            .get_postings(body, b"hot")
2050            .await
2051            .unwrap()
2052            .expect("postings for 'hot'");
2053        assert_eq!(
2054            postings.doc_count(),
2055            16384,
2056            "each document must appear exactly once per term (spill-boundary duplicate)"
2057        );
2058
2059        // Doc ids must be strictly increasing and the boundary document's
2060        // split term frequency must be merged into a single posting.
2061        let mut it = postings.iterator();
2062        let mut prev: Option<DocId> = None;
2063        let mut boundary_tf = 0u32;
2064        let mut d = it.doc();
2065        while d != TERMINATED {
2066            if let Some(p) = prev {
2067                assert!(p < d, "duplicate/unordered doc id {d} after {p}");
2068            }
2069            if d == boundary_doc {
2070                boundary_tf = it.term_freq();
2071            }
2072            prev = Some(d);
2073            d = it.advance();
2074        }
2075        assert_eq!(prev, Some(boundary_doc));
2076        assert_eq!(
2077            boundary_tf, 2,
2078            "boundary doc's term frequency must combine both values"
2079        );
2080    }
2081}