Skip to main content

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