1#[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#[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 pub struct Rodeo {
52 strings: Vec<Box<str>>,
54 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 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
110const STORE_BUFFER_SIZE: usize = 16 * 1024 * 1024; const NEW_TERM_OVERHEAD: usize = size_of::<TermKey>() + size_of::<PostingListBuilder>() + 24;
116
117const INTERN_OVERHEAD: usize = size_of::<Spur>() + 2 * size_of::<usize>();
119
120const NEW_POS_TERM_OVERHEAD: usize =
122 size_of::<TermKey>() + size_of::<PositionPostingListBuilder>() + 24;
123
124const MAX_POSITION_ELEMENT_ORDINAL: u32 = (1 << 12) - 1;
130const MAX_TOKEN_POSITION: u32 = (1 << 20) - 1;
131
132pub(crate) const MAX_VECTOR_VALUES_PER_FIELD: usize = u16::MAX as usize + 1;
135
136const DEFAULT_BMP_SPARSE_DIMS: u32 = 105879;
140
141fn field_type_name(field_type: &FieldType) -> &'static str {
143 match field_type {
144 FieldType::Text => "text",
145 FieldType::U64 => "u64",
146 FieldType::I64 => "i64",
147 FieldType::F64 => "f64",
148 FieldType::Bytes => "bytes",
149 FieldType::SparseVector => "sparse_vector",
150 FieldType::DenseVector => "dense_vector",
151 FieldType::Json => "json",
152 FieldType::BinaryDenseVector => "binary_dense_vector",
153 }
154}
155
156fn field_value_type_name(value: &FieldValue) -> &'static str {
158 match value {
159 FieldValue::Text(_) => "text",
160 FieldValue::U64(_) => "u64",
161 FieldValue::I64(_) => "i64",
162 FieldValue::F64(_) => "f64",
163 FieldValue::Bytes(_) => "bytes",
164 FieldValue::SparseVector(_) => "sparse_vector",
165 FieldValue::DenseVector(_) => "dense_vector",
166 FieldValue::Json(_) => "json",
167 FieldValue::BinaryDenseVector(_) => "binary_dense_vector",
168 }
169}
170
171pub(crate) fn validate_vector_value_counts(doc: &Document, schema: &Schema) -> Result<()> {
178 let mut vector_values_per_field: FxHashMap<u32, usize> = FxHashMap::default();
179
180 for (field, value) in doc.field_values() {
181 let Some(entry) = schema.get_field_entry(*field) else {
182 continue;
183 };
184
185 let consumes_vector_ordinal = match (&entry.field_type, value) {
186 (FieldType::DenseVector, FieldValue::DenseVector(_))
187 | (FieldType::BinaryDenseVector, FieldValue::BinaryDenseVector(_)) => {
188 entry.indexed || entry.stored
189 }
190 (FieldType::SparseVector, FieldValue::SparseVector(_)) => entry.indexed || entry.fast,
191 _ => false,
192 };
193 if !consumes_vector_ordinal {
194 continue;
195 }
196
197 let count = vector_values_per_field.entry(field.0).or_insert(0);
198 *count += 1;
199 if *count > MAX_VECTOR_VALUES_PER_FIELD {
200 return Err(crate::Error::Document(format!(
201 "field '{}' (id {}) has more than {} vector values in one document",
202 entry.name, field.0, MAX_VECTOR_VALUES_PER_FIELD
203 )));
204 }
205 }
206 Ok(())
207}
208
209pub struct SegmentBuilder {
216 schema: Arc<Schema>,
217 config: SegmentBuilderConfig,
218 tokenizers: FxHashMap<Field, BoxedTokenizer>,
219 tokenizer_hint_fields: FxHashMap<Field, Field>,
222 tokenizer_hint_buffer: String,
224 unhinted_dynamic_docs: u64,
227
228 term_interner: Rodeo,
230
231 inverted_index: HashMap<TermKey, PostingListBuilder>,
233
234 #[cfg(feature = "native")]
236 posting_spill_file: Option<BufWriter<File>>,
237 #[cfg(feature = "native")]
238 posting_spill_path: PathBuf,
239 #[cfg(feature = "native")]
241 posting_spill_index: HashMap<TermKey, Vec<(u64, u32)>>,
242 #[cfg(feature = "native")]
243 posting_spill_offset: u64,
244
245 #[cfg(feature = "native")]
247 store_file: BufWriter<File>,
248 #[cfg(feature = "native")]
249 store_path: PathBuf,
250 #[cfg(not(feature = "native"))]
251 store_buffer: Vec<u8>,
252
253 next_doc_id: DocId,
255
256 field_stats: FxHashMap<u32, FieldStats>,
258
259 doc_field_lengths: Vec<u32>,
263 num_indexed_fields: usize,
264 field_to_slot: FxHashMap<u32, usize>,
265
266 local_tf_buffer: FxHashMap<Spur, u32>,
269
270 local_positions: FxHashMap<Spur, Vec<u32>>,
273
274 local_position_terms: Vec<Spur>,
277
278 token_buffer: String,
280
281 numeric_buffer: String,
283
284 dense_vectors: FxHashMap<u32, DenseVectorBuilder>,
287
288 binary_dense_vectors: FxHashMap<u32, BinaryDenseVectorBuilder>,
290
291 sparse_vectors: FxHashMap<u32, SparseVectorBuilder>,
294
295 position_index: HashMap<TermKey, PositionPostingListBuilder>,
298
299 position_enabled_fields: FxHashMap<u32, Option<crate::dsl::PositionMode>>,
301
302 current_element_ordinal: FxHashMap<u32, u32>,
304
305 chunk_maps: FxHashMap<u32, super::chunk_map::ChunkMapBuilder>,
308
309 position_saturation_warned: bool,
312
313 estimated_memory: usize,
315
316 doc_serialize_buffer: Vec<u8>,
318
319 fast_fields: FxHashMap<u32, crate::structures::fast_field::FastFieldWriter>,
321}
322
323impl SegmentBuilder {
324 pub fn new(schema: Arc<Schema>, config: SegmentBuilderConfig) -> Result<Self> {
326 #[cfg(feature = "native")]
327 let (store_file, store_path, spill_path) = {
328 let segment_id = uuid::Uuid::new_v4();
329 let store_path = config
330 .temp_dir
331 .join(format!("hermes_store_{}.tmp", segment_id));
332 let store_file = BufWriter::with_capacity(
333 STORE_BUFFER_SIZE,
334 OpenOptions::new()
335 .create(true)
336 .write(true)
337 .truncate(true)
338 .open(&store_path)?,
339 );
340 let spill_path = config
341 .temp_dir
342 .join(format!("hermes_spill_{}.tmp", segment_id));
343 (store_file, store_path, spill_path)
344 };
345
346 let registry = crate::tokenizer::TokenizerRegistry::new();
348 let mut num_indexed_fields = 0;
349 let mut field_to_slot = FxHashMap::default();
350 let mut position_enabled_fields = FxHashMap::default();
351 let mut tokenizers = FxHashMap::default();
352 let mut tokenizer_hint_fields = FxHashMap::default();
353 for (field, entry) in schema.fields() {
354 if entry.indexed && matches!(entry.field_type, FieldType::Text) {
355 field_to_slot.insert(field.0, num_indexed_fields);
356 num_indexed_fields += 1;
357 if entry.positions.is_some() {
358 position_enabled_fields.insert(field.0, entry.positions);
359 }
360 if let Some(ref tok_name) = entry.tokenizer
361 && let Some(tokenizer) = registry.get(tok_name)
362 {
363 tokenizers.insert(field, tokenizer);
364 }
365 if let Some(hint_field) = schema.tokenizer_hint_field(field) {
366 tokenizer_hint_fields.insert(field, hint_field);
367 }
368 }
369 }
370
371 use crate::structures::fast_field::{FastFieldColumnType, FastFieldWriter};
373 let mut fast_fields = FxHashMap::default();
374 for (field, entry) in schema.fields() {
375 if entry.fast {
376 let writer = if entry.multi {
377 match entry.field_type {
378 FieldType::U64 => {
379 FastFieldWriter::new_numeric_multi(FastFieldColumnType::U64)
380 }
381 FieldType::I64 => {
382 FastFieldWriter::new_numeric_multi(FastFieldColumnType::I64)
383 }
384 FieldType::F64 => {
385 FastFieldWriter::new_numeric_multi(FastFieldColumnType::F64)
386 }
387 FieldType::Text => FastFieldWriter::new_text_multi(),
388 _ => continue,
389 }
390 } else {
391 match entry.field_type {
392 FieldType::U64 => FastFieldWriter::new_numeric(FastFieldColumnType::U64),
393 FieldType::I64 => FastFieldWriter::new_numeric(FastFieldColumnType::I64),
394 FieldType::F64 => FastFieldWriter::new_numeric(FastFieldColumnType::F64),
395 FieldType::Text => FastFieldWriter::new_text(),
396 _ => continue,
397 }
398 };
399 fast_fields.insert(field.0, writer);
400 }
401 }
402
403 Ok(Self {
404 schema,
405 tokenizers,
406 tokenizer_hint_fields,
407 tokenizer_hint_buffer: String::new(),
408 unhinted_dynamic_docs: 0,
409 term_interner: Rodeo::new(),
410 inverted_index: HashMap::with_capacity(config.posting_map_capacity),
411 #[cfg(feature = "native")]
412 posting_spill_file: None,
413 #[cfg(feature = "native")]
414 posting_spill_path: spill_path,
415 #[cfg(feature = "native")]
416 posting_spill_index: HashMap::new(),
417 #[cfg(feature = "native")]
418 posting_spill_offset: 0,
419 #[cfg(feature = "native")]
420 store_file,
421 #[cfg(feature = "native")]
422 store_path,
423 #[cfg(not(feature = "native"))]
424 store_buffer: Vec::with_capacity(STORE_BUFFER_SIZE),
425 next_doc_id: 0,
426 field_stats: FxHashMap::default(),
427 chunk_maps: FxHashMap::default(),
428 doc_field_lengths: Vec::new(),
429 num_indexed_fields,
430 field_to_slot,
431 local_tf_buffer: FxHashMap::default(),
432 local_positions: FxHashMap::default(),
433 local_position_terms: Vec::new(),
434 token_buffer: String::with_capacity(64),
435 numeric_buffer: String::with_capacity(32),
436 config,
437 dense_vectors: FxHashMap::default(),
438 binary_dense_vectors: FxHashMap::default(),
439 sparse_vectors: FxHashMap::default(),
440 position_index: HashMap::new(),
441 position_enabled_fields,
442 current_element_ordinal: FxHashMap::default(),
443 position_saturation_warned: false,
444 estimated_memory: 0,
445 doc_serialize_buffer: Vec::with_capacity(256),
446 fast_fields,
447 })
448 }
449
450 pub fn set_tokenizer(&mut self, field: Field, tokenizer: BoxedTokenizer) {
451 self.tokenizers.insert(field, tokenizer);
452 }
453
454 pub fn unhinted_dynamic_docs(&self) -> u64 {
456 self.unhinted_dynamic_docs
457 }
458
459 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 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 #[inline]
518 pub fn estimated_memory_bytes(&self) -> usize {
519 self.estimated_memory
520 }
521
522 pub fn sparse_dim_count(&self) -> usize {
524 self.sparse_vectors.values().map(|b| b.postings.len()).sum()
525 }
526
527 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 let compact_posting_size = size_of::<CompactPosting>();
536 let vec_overhead = size_of::<Vec<u8>>(); 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 let hashmap_entry_base_overhead = 8usize;
545
546 let fxhashmap_entry_overhead = hashmap_entry_base_overhead;
548
549 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 let index_overhead_bytes = self.inverted_index.len()
558 * (term_key_size + posting_builder_size + hashmap_entry_base_overhead);
559
560 let interner_arena_overhead = 2 * size_of::<usize>();
563 let avg_term_len = 8; let interner_bytes =
565 self.term_interner.len() * (avg_term_len + spur_size + interner_arena_overhead);
566
567 let field_lengths_bytes =
569 self.doc_field_lengths.capacity() * size_of::<u32>() + vec_overhead;
570
571 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; dense_vector_count += b.doc_ids.len();
580 }
581 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 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 let mut sparse_vectors_bytes: usize = 0;
595 for builder in self.sparse_vectors.values() {
596 for postings in builder.postings.values() {
597 sparse_vectors_bytes += postings.capacity() * sparse_entry_size + vec_overhead;
598 }
599 let inner_entry_size = size_of::<u32>() + vec_overhead + fxhashmap_entry_overhead;
601 sparse_vectors_bytes += builder.postings.len() * inner_entry_size;
602 }
603 let outer_sparse_entry_size =
605 size_of::<u32>() + size_of::<SparseVectorBuilder>() + fxhashmap_entry_overhead;
606 sparse_vectors_bytes += self.sparse_vectors.len() * outer_sparse_entry_size;
607
608 let mut position_index_bytes: usize = 0;
610 for pos_builder in self.position_index.values() {
611 for (_, positions) in &pos_builder.postings {
612 position_index_bytes += positions.capacity() * size_of::<u32>() + vec_overhead;
613 }
614 let pos_entry_size = size_of::<DocId>() + vec_overhead;
616 position_index_bytes += pos_builder.postings.capacity() * pos_entry_size;
617 }
618 let pos_index_entry_size =
620 term_key_size + size_of::<PositionPostingListBuilder>() + hashmap_entry_base_overhead;
621 position_index_bytes += self.position_index.len() * pos_index_entry_size;
622
623 let estimated_memory_bytes = postings_bytes
624 + index_overhead_bytes
625 + interner_bytes
626 + field_lengths_bytes
627 + dense_vectors_bytes
628 + local_tf_buffer_bytes
629 + sparse_vectors_bytes
630 + position_index_bytes;
631
632 let memory_breakdown = MemoryBreakdown {
633 postings_bytes,
634 index_overhead_bytes,
635 interner_bytes,
636 field_lengths_bytes,
637 dense_vectors_bytes,
638 dense_vector_count,
639 sparse_vectors_bytes,
640 position_index_bytes,
641 };
642
643 SegmentBuilderStats {
644 num_docs: self.next_doc_id,
645 unique_terms: self.inverted_index.len(),
646 postings_in_memory,
647 interned_strings: self.term_interner.len(),
648 doc_field_lengths_size: self.doc_field_lengths.len(),
649 estimated_memory_bytes,
650 memory_breakdown,
651 unhinted_dynamic_docs: self.unhinted_dynamic_docs,
652 }
653 }
654
655 fn validate_document_against_schema(&self, doc: &Document) -> Result<()> {
670 validate_vector_value_counts(doc, &self.schema)?;
671
672 for (field, value) in doc.field_values() {
673 let Some(entry) = self.schema.get_field_entry(*field) else {
674 continue;
675 };
676
677 if !matches!(
680 &entry.field_type,
681 FieldType::DenseVector | FieldType::BinaryDenseVector
682 ) && !entry.indexed
683 && !entry.fast
684 {
685 continue;
686 }
687
688 match (&entry.field_type, value) {
689 (FieldType::SparseVector, FieldValue::SparseVector(entries)) => {
690 if let Some(config) = entry.sparse_vector_config.as_ref()
691 && config.format == crate::structures::SparseFormat::Bmp
692 {
693 let dims = config.dims.unwrap_or(DEFAULT_BMP_SPARSE_DIMS);
694 if let Some(&(dim_id, _)) =
695 entries.iter().find(|&&(dim_id, _)| dim_id >= dims)
696 {
697 return Err(crate::Error::Schema(format!(
698 "sparse vector for field '{}' contains dim_id {} out of \
699 range for the configured BMP dims={}: dimensions >= dims \
700 are never written to the block-max grid and can never \
701 match a query; raise `dims` in the field's sparse_vector \
702 config or fix the embedding model",
703 entry.name, dim_id, dims
704 )));
705 }
706 }
707 }
708 (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 | (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 pub fn add_document(&mut self, doc: Document) -> Result<DocId> {
736 self.validate_document_against_schema(&doc)?;
738
739 let doc_id = self.next_doc_id;
740 self.next_doc_id += 1;
741
742 let base_idx = self.doc_field_lengths.len();
744 self.doc_field_lengths
745 .resize(base_idx + self.num_indexed_fields, 0);
746 self.estimated_memory += self.num_indexed_fields * std::mem::size_of::<u32>();
747
748 self.current_element_ordinal.clear();
750
751 for (field, value) in doc.field_values() {
752 let Some(entry) = self.schema.get_field_entry(*field) else {
753 continue;
754 };
755
756 if !matches!(
759 &entry.field_type,
760 FieldType::DenseVector | FieldType::BinaryDenseVector
761 ) && !entry.indexed
762 && !entry.fast
763 {
764 continue;
765 }
766
767 match (&entry.field_type, value) {
768 (FieldType::Text, FieldValue::Text(text)) => {
769 if entry.indexed && entry.chunked {
770 let field_id = field.0;
774 let element_ordinal = self.next_element_ordinal(field_id);
775 let ordinal = u16::try_from(element_ordinal).map_err(|_| {
776 crate::Error::Document(format!(
777 "chunked field {field_id} has more than {} chunk values in one document",
778 u16::MAX as usize + 1
779 ))
780 })?;
781 let hinted = self.resolve_tokenizer_hint(*field, &doc, element_ordinal);
782 let vid = self
783 .chunk_maps
784 .get(&field.0)
785 .map_or(0usize, |map| map.len());
786 let vid = u32::try_from(vid).map_err(|_| {
787 crate::Error::Document(format!(
788 "chunked field {field_id} exceeds u32::MAX chunks in one segment"
789 ))
790 })?;
791 let token_count = self.index_text_field(*field, vid, text, 0, hinted)?;
795 self.chunk_maps.entry(field.0).or_default().push(
796 doc_id,
797 ordinal,
798 token_count,
799 )?;
800 self.estimated_memory += 8;
801
802 let stats = self.field_stats.entry(field.0).or_default();
805 stats.total_tokens += token_count as u64;
806 stats.doc_count += 1;
807 } else if entry.indexed {
808 let element_ordinal = self.next_element_ordinal(field.0);
809 let hinted = self.resolve_tokenizer_hint(*field, &doc, element_ordinal);
810 let token_count =
811 self.index_text_field(*field, doc_id, text, element_ordinal, hinted)?;
812
813 let stats = self.field_stats.entry(field.0).or_default();
814 stats.total_tokens += token_count as u64;
815 if element_ordinal == 0 {
816 stats.doc_count += 1;
817 }
818
819 if let Some(&slot) = self.field_to_slot.get(&field.0) {
820 let len = &mut self.doc_field_lengths[base_idx + slot];
823 *len = len.saturating_add(token_count);
824 }
825 }
826
827 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
829 ff.add_text(doc_id, text);
830 }
831 }
832 (FieldType::U64, FieldValue::U64(v)) => {
833 if entry.indexed {
834 self.index_numeric_field(*field, doc_id, *v)?;
835 }
836 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
837 ff.add_u64(doc_id, *v);
838 }
839 }
840 (FieldType::I64, FieldValue::I64(v)) => {
841 if entry.indexed {
842 self.index_numeric_field(*field, doc_id, *v as u64)?;
843 }
844 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
845 ff.add_i64(doc_id, *v);
846 }
847 }
848 (FieldType::F64, FieldValue::F64(v)) => {
849 if entry.indexed {
850 self.index_numeric_field(*field, doc_id, v.to_bits())?;
851 }
852 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
853 ff.add_f64(doc_id, *v);
854 }
855 }
856 (FieldType::DenseVector, FieldValue::DenseVector(vec))
857 if entry.indexed || entry.stored =>
858 {
859 let ordinal = self.next_vector_ordinal(field.0)?;
860 self.index_dense_vector_field(*field, doc_id, ordinal, vec)?;
861 }
862 (FieldType::BinaryDenseVector, FieldValue::BinaryDenseVector(bytes))
863 if entry.indexed || entry.stored =>
864 {
865 let ordinal = self.next_vector_ordinal(field.0)?;
866 self.index_binary_dense_vector_field(*field, doc_id, ordinal, bytes)?;
867 }
868 (FieldType::SparseVector, FieldValue::SparseVector(entries)) => {
869 let ordinal = self.next_vector_ordinal(field.0)?;
870 self.index_sparse_vector_field(*field, doc_id, ordinal, entries)?;
871 }
872 _ => {}
877 }
878 }
879
880 self.write_document_to_store(&doc)?;
882
883 Ok(doc_id)
884 }
885
886 fn index_text_field(
895 &mut self,
896 field: Field,
897 doc_id: DocId,
898 text: &str,
899 element_ordinal: u32,
900 hinted: bool,
901 ) -> Result<u32> {
902 use crate::dsl::PositionMode;
903
904 let field_id = field.0;
905 let position_mode = self
906 .position_enabled_fields
907 .get(&field_id)
908 .copied()
909 .flatten();
910
911 let encoded_ordinal = if position_mode.is_some_and(|m| m.tracks_ordinal())
914 && element_ordinal > MAX_POSITION_ELEMENT_ORDINAL
915 {
916 self.warn_position_saturation(
917 "element ordinal",
918 element_ordinal,
919 MAX_POSITION_ELEMENT_ORDINAL,
920 );
921 MAX_POSITION_ELEMENT_ORDINAL
922 } else {
923 element_ordinal
924 };
925
926 self.local_tf_buffer.clear();
930 for term in self.local_position_terms.drain(..) {
934 self.local_positions.get_mut(&term).unwrap().clear();
935 }
936
937 let mut token_position = 0u32;
938
939 let custom_tokens = self.tokenizers.get(&field).map(|t| {
943 if hinted {
944 let hint = (!self.tokenizer_hint_buffer.is_empty())
945 .then_some(self.tokenizer_hint_buffer.as_str());
946 t.tokenize_with(text, hint, crate::tokenizer::Purpose::Index)
947 } else {
948 t.tokenize(text)
949 }
950 });
951
952 if let Some(tokens) = custom_tokens {
953 for token in &tokens {
955 let term_spur = if let Some(spur) = self.term_interner.get(&token.text) {
956 spur
957 } else {
958 let spur = self.term_interner.get_or_intern(&token.text);
959 self.estimated_memory += token.text.len() + INTERN_OVERHEAD;
960 spur
961 };
962 *self.local_tf_buffer.entry(term_spur).or_insert(0) += 1;
963
964 if let Some(mode) = position_mode {
965 let encoded_pos = match mode {
966 PositionMode::Ordinal => encoded_ordinal << 20,
967 PositionMode::TokenPosition => token.position,
968 PositionMode::Full => {
969 (encoded_ordinal << 20) | self.saturate_token_position(token.position)
970 }
971 };
972 let positions = self.local_positions.entry(term_spur).or_default();
973 if positions.is_empty() {
974 self.local_position_terms.push(term_spur);
975 }
976 positions.push(encoded_pos);
977 }
978 }
979 token_position = tokens.iter().filter(|t| !t.variant).count() as u32;
982 } else {
983 for word in text.split_whitespace() {
985 self.token_buffer.clear();
986 for c in word.chars() {
987 if c.is_alphanumeric() {
988 for lc in c.to_lowercase() {
989 self.token_buffer.push(lc);
990 }
991 }
992 }
993
994 if self.token_buffer.is_empty() {
995 continue;
996 }
997
998 let term_spur = if let Some(spur) = self.term_interner.get(&self.token_buffer) {
999 spur
1000 } else {
1001 let spur = self.term_interner.get_or_intern(&self.token_buffer);
1002 self.estimated_memory += self.token_buffer.len() + INTERN_OVERHEAD;
1003 spur
1004 };
1005 *self.local_tf_buffer.entry(term_spur).or_insert(0) += 1;
1006
1007 if let Some(mode) = position_mode {
1008 let encoded_pos = match mode {
1009 PositionMode::Ordinal => encoded_ordinal << 20,
1010 PositionMode::TokenPosition => token_position,
1011 PositionMode::Full => {
1012 (encoded_ordinal << 20) | self.saturate_token_position(token_position)
1013 }
1014 };
1015 let positions = self.local_positions.entry(term_spur).or_default();
1016 if positions.is_empty() {
1017 self.local_position_terms.push(term_spur);
1018 }
1019 positions.push(encoded_pos);
1020 }
1021
1022 token_position += 1;
1023 }
1024 }
1025
1026 for (&term_spur, &tf) in &self.local_tf_buffer {
1029 let term_key = TermKey {
1030 field: field_id,
1031 term: term_spur,
1032 };
1033
1034 match self.inverted_index.entry(term_key) {
1035 hashbrown::hash_map::Entry::Occupied(mut o) => {
1036 o.get_mut().add(doc_id, tf);
1037 self.estimated_memory += size_of::<CompactPosting>();
1038 #[cfg(feature = "native")]
1040 if o.get().should_spill() {
1041 use byteorder::{LittleEndian, WriteBytesExt};
1042
1043 let builder = o.get_mut();
1044 let count = builder.postings.len() as u32;
1045 let offset = self.posting_spill_offset;
1046
1047 let spill_file = if let Some(ref mut f) = self.posting_spill_file {
1049 f
1050 } else {
1051 self.posting_spill_file = Some(BufWriter::with_capacity(
1052 256 * 1024,
1053 OpenOptions::new()
1054 .create(true)
1055 .write(true)
1056 .truncate(true)
1057 .open(&self.posting_spill_path)?,
1058 ));
1059 self.posting_spill_file.as_mut().unwrap()
1060 };
1061 for p in &builder.postings {
1062 spill_file.write_u32::<LittleEndian>(p.doc_id)?;
1063 spill_file.write_u16::<LittleEndian>(p.term_freq)?;
1064 }
1065 self.posting_spill_offset += count as u64 * 6;
1066 self.posting_spill_index
1067 .entry(term_key)
1068 .or_default()
1069 .push((offset, count));
1070
1071 let freed = builder.postings.len() * size_of::<CompactPosting>();
1072 builder.spilled_count += count;
1073 builder.postings.clear();
1074 self.estimated_memory -= freed;
1075 }
1076 }
1077 hashbrown::hash_map::Entry::Vacant(v) => {
1078 let mut posting = PostingListBuilder::new();
1079 posting.add(doc_id, tf);
1080 v.insert(posting);
1081 self.estimated_memory += size_of::<CompactPosting>() + NEW_TERM_OVERHEAD;
1082 }
1083 }
1084
1085 if position_mode.is_some()
1086 && let Some(positions) = self.local_positions.get(&term_spur)
1087 {
1088 match self.position_index.entry(term_key) {
1089 hashbrown::hash_map::Entry::Occupied(mut o) => {
1090 for &pos in positions {
1091 o.get_mut().add_position(doc_id, pos);
1092 }
1093 self.estimated_memory += positions.len() * size_of::<u32>();
1094 }
1095 hashbrown::hash_map::Entry::Vacant(v) => {
1096 let mut pos_posting = PositionPostingListBuilder::new();
1097 for &pos in positions {
1098 pos_posting.add_position(doc_id, pos);
1099 }
1100 self.estimated_memory +=
1101 positions.len() * size_of::<u32>() + NEW_POS_TERM_OVERHEAD;
1102 v.insert(pos_posting);
1103 }
1104 }
1105 }
1106 }
1107
1108 Ok(token_position)
1109 }
1110
1111 #[inline]
1114 fn saturate_token_position(&mut self, token_position: u32) -> u32 {
1115 if token_position > MAX_TOKEN_POSITION {
1116 self.warn_position_saturation("token position", token_position, MAX_TOKEN_POSITION);
1117 MAX_TOKEN_POSITION
1118 } else {
1119 token_position
1120 }
1121 }
1122
1123 #[cold]
1125 fn warn_position_saturation(&mut self, what: &str, value: u32, max: u32) {
1126 if !self.position_saturation_warned {
1127 self.position_saturation_warned = true;
1128 log::warn!(
1129 "[segment_builder] index={} {what} {value} exceeds the position-encoding limit {max}; \
1130 saturating — phrase/ordinal matching degrades for the overflowing \
1131 elements/tokens instead of corrupting other documents' matches \
1132 (further occurrences in this segment are not logged)",
1133 self.schema.index_label()
1134 );
1135 }
1136 }
1137
1138 fn index_numeric_field(&mut self, field: Field, doc_id: DocId, value: u64) -> Result<()> {
1139 use std::fmt::Write;
1140
1141 self.numeric_buffer.clear();
1142 write!(self.numeric_buffer, "__num_{}", value).unwrap();
1143 let term_spur = if let Some(spur) = self.term_interner.get(&self.numeric_buffer) {
1144 spur
1145 } else {
1146 let spur = self.term_interner.get_or_intern(&self.numeric_buffer);
1147 self.estimated_memory += self.numeric_buffer.len() + INTERN_OVERHEAD;
1148 spur
1149 };
1150
1151 let term_key = TermKey {
1152 field: field.0,
1153 term: term_spur,
1154 };
1155
1156 match self.inverted_index.entry(term_key) {
1157 hashbrown::hash_map::Entry::Occupied(mut o) => {
1158 o.get_mut().add(doc_id, 1);
1159 self.estimated_memory += size_of::<CompactPosting>();
1160 }
1161 hashbrown::hash_map::Entry::Vacant(v) => {
1162 let mut posting = PostingListBuilder::new();
1163 posting.add(doc_id, 1);
1164 v.insert(posting);
1165 self.estimated_memory += size_of::<CompactPosting>() + NEW_TERM_OVERHEAD;
1166 }
1167 }
1168
1169 Ok(())
1170 }
1171
1172 fn index_dense_vector_field(
1174 &mut self,
1175 field: Field,
1176 doc_id: DocId,
1177 ordinal: u16,
1178 vector: &[f32],
1179 ) -> Result<()> {
1180 let dim = vector.len();
1181 let expected_dim = self
1182 .schema
1183 .get_field_entry(field)
1184 .and_then(|entry| entry.dense_vector_config.as_ref())
1185 .map(|config| config.dim)
1186 .ok_or_else(|| crate::Error::Schema("DenseVector field missing config".to_string()))?;
1187 if dim != expected_dim {
1188 return Err(crate::Error::Schema(format!(
1189 "Dense vector dimension mismatch: schema expects {}, got {}",
1190 expected_dim, dim
1191 )));
1192 }
1193 if let Some((index, value)) = vector
1194 .iter()
1195 .enumerate()
1196 .find(|(_, value)| !value.is_finite())
1197 {
1198 return Err(crate::Error::Document(format!(
1199 "dense vector contains non-finite value {value} at index {index}"
1200 )));
1201 }
1202
1203 let builder = self
1204 .dense_vectors
1205 .entry(field.0)
1206 .or_insert_with(|| DenseVectorBuilder::new(dim));
1207
1208 if builder.dim != dim && builder.len() > 0 {
1210 return Err(crate::Error::Schema(format!(
1211 "Dense vector dimension mismatch: expected {}, got {}",
1212 builder.dim, dim
1213 )));
1214 }
1215
1216 builder.add(doc_id, ordinal, vector);
1217
1218 self.estimated_memory += std::mem::size_of_val(vector) + size_of::<(DocId, u16)>();
1219
1220 Ok(())
1221 }
1222
1223 fn index_binary_dense_vector_field(
1225 &mut self,
1226 field: Field,
1227 doc_id: DocId,
1228 ordinal: u16,
1229 bytes: &[u8],
1230 ) -> Result<()> {
1231 let dim_bits = self
1232 .schema
1233 .get_field_entry(field)
1234 .and_then(|e| e.binary_dense_vector_config.as_ref())
1235 .map(|c| c.dim)
1236 .ok_or_else(|| {
1237 crate::Error::Schema("BinaryDenseVector field missing config".to_string())
1238 })?;
1239
1240 let expected_byte_len = dim_bits.div_ceil(8);
1241 if dim_bits == 0 || !dim_bits.is_multiple_of(8) {
1242 return Err(crate::Error::Schema(format!(
1243 "Binary vector dimension must be a positive multiple of 8, got {dim_bits}"
1244 )));
1245 }
1246 if bytes.len() != expected_byte_len {
1247 return Err(crate::Error::Schema(format!(
1248 "Binary vector byte length mismatch: expected {} (dim={}), got {}",
1249 expected_byte_len,
1250 dim_bits,
1251 bytes.len()
1252 )));
1253 }
1254
1255 let builder = self
1256 .binary_dense_vectors
1257 .entry(field.0)
1258 .or_insert_with(|| BinaryDenseVectorBuilder::new(dim_bits));
1259
1260 builder.add(doc_id, ordinal, bytes);
1261 self.estimated_memory += bytes.len() + size_of::<(DocId, u16)>();
1262
1263 Ok(())
1264 }
1265
1266 fn index_sparse_vector_field(
1276 &mut self,
1277 field: Field,
1278 doc_id: DocId,
1279 ordinal: u16,
1280 entries: &[(u32, f32)],
1281 ) -> Result<()> {
1282 if let Some((index, (_, weight))) = entries
1283 .iter()
1284 .enumerate()
1285 .find(|(_, (_, weight))| !weight.is_finite())
1286 {
1287 return Err(crate::Error::Document(format!(
1288 "sparse vector contains non-finite weight {weight} at index {index}"
1289 )));
1290 }
1291 let (weight_threshold, doc_mass, min_terms) = self
1292 .schema
1293 .get_field_entry(field)
1294 .and_then(|entry| entry.sparse_vector_config.as_ref())
1295 .map(|config| (config.weight_threshold, config.doc_mass, config.min_terms))
1296 .unwrap_or((0.0, None, 0));
1297
1298 let builder = self
1299 .sparse_vectors
1300 .entry(field.0)
1301 .or_insert_with(SparseVectorBuilder::new);
1302
1303 builder.inc_vector_count();
1304
1305 let mass_cutoff = match doc_mass {
1309 Some(mass) if mass < 1.0 && entries.len() > min_terms => {
1310 let mut weights: Vec<f32> = entries
1311 .iter()
1312 .map(|&(_, w)| w.abs())
1313 .filter(|w| *w >= weight_threshold)
1314 .collect();
1315 weights.sort_unstable_by(|a, b| b.total_cmp(a));
1316 let total: f64 = weights.iter().map(|&w| w as f64).sum();
1317 let target = total * mass as f64;
1318 let mut cumulative = 0.0f64;
1319 let mut cutoff = 0.0f32;
1320 for &w in &weights {
1321 if cumulative >= target {
1322 break;
1323 }
1324 cumulative += w as f64;
1325 cutoff = w;
1326 }
1327 cutoff
1328 }
1329 _ => 0.0,
1330 };
1331
1332 for &(dim_id, weight) in entries {
1333 if weight.abs() < weight_threshold || weight.abs() < mass_cutoff {
1335 continue;
1336 }
1337
1338 let is_new_dim = !builder.postings.contains_key(&dim_id);
1339 builder.add(dim_id, doc_id, ordinal, weight);
1340 self.estimated_memory += size_of::<(DocId, u16, f32)>();
1341 if is_new_dim {
1342 self.estimated_memory += size_of::<u32>() + size_of::<Vec<(DocId, u16, f32)>>() + 8; }
1345 }
1346
1347 Ok(())
1348 }
1349
1350 fn write_document_to_store(&mut self, doc: &Document) -> Result<()> {
1352 use byteorder::{LittleEndian, WriteBytesExt};
1353
1354 super::store::serialize_document_into(doc, &self.schema, &mut self.doc_serialize_buffer)?;
1355
1356 #[cfg(feature = "native")]
1357 {
1358 self.store_file
1359 .write_u32::<LittleEndian>(self.doc_serialize_buffer.len() as u32)?;
1360 self.store_file.write_all(&self.doc_serialize_buffer)?;
1361 }
1362 #[cfg(not(feature = "native"))]
1363 {
1364 self.store_buffer
1365 .write_u32::<LittleEndian>(self.doc_serialize_buffer.len() as u32)?;
1366 self.store_buffer.write_all(&self.doc_serialize_buffer)?;
1367 self.estimated_memory += size_of::<u32>() + self.doc_serialize_buffer.len();
1371 }
1372
1373 Ok(())
1374 }
1375
1376 pub async fn build<D: Directory + DirectoryWriter>(
1382 mut self,
1383 dir: &D,
1384 segment_id: SegmentId,
1385 trained: Option<&super::TrainedVectorStructures>,
1386 ) -> Result<SegmentMeta> {
1387 #[cfg(feature = "native")]
1389 self.store_file.flush()?;
1390
1391 let files = SegmentFiles::new(segment_id.0);
1392
1393 let position_index = std::mem::take(&mut self.position_index);
1395 let position_offsets = if !position_index.is_empty() {
1396 let mut pos_writer = dir.streaming_writer(&files.positions).await?;
1397 let offsets = postings::build_positions_streaming(
1398 position_index,
1399 &self.term_interner,
1400 &mut *pos_writer,
1401 )?;
1402 pos_writer.finish()?;
1403 offsets
1404 } else {
1405 FxHashMap::default()
1406 };
1407
1408 let chunk_maps = std::mem::take(&mut self.chunk_maps);
1411 {
1412 let mut fields: Vec<(u32, &super::chunk_map::ChunkMapBuilder)> = chunk_maps
1413 .iter()
1414 .filter(|(_, map)| !map.is_empty())
1415 .map(|(field_id, map)| (*field_id, map))
1416 .collect();
1417 fields.sort_by_key(|(field_id, _)| *field_id);
1418 let num_docs = self.next_doc_id as usize;
1419 let mut columns: Vec<(u32, Vec<u16>, u64)> = Vec::new();
1420 for (&field_id, &slot) in &self.field_to_slot {
1421 if self
1422 .schema
1423 .get_field_entry(crate::dsl::Field(field_id))
1424 .is_some_and(|entry| entry.chunked)
1425 {
1426 continue;
1427 }
1428 let mut total = 0u64;
1429 let lengths: Vec<u16> = (0..num_docs)
1430 .map(|doc| {
1431 let len = self.doc_field_lengths[doc * self.num_indexed_fields + slot];
1432 total += u64::from(len);
1433 len.min(super::chunk_map::MAX_CHUNK_LENGTH) as u16
1434 })
1435 .collect();
1436 if total > 0 {
1437 columns.push((field_id, lengths, total));
1438 }
1439 }
1440 columns.sort_by_key(|(field_id, _, _)| *field_id);
1441 let norms: Vec<super::chunk_map::DocLengthsColumn<'_>> = columns
1442 .iter()
1443 .map(
1444 |(field_id, lengths, total)| super::chunk_map::DocLengthsColumn {
1445 field_id: *field_id,
1446 lengths,
1447 total_tokens: *total,
1448 },
1449 )
1450 .collect();
1451 if !fields.is_empty() || !norms.is_empty() {
1452 let mut writer = dir.streaming_writer(&files.chunks).await?;
1453 super::chunk_map::write_chunk_maps(&mut *writer, &fields, &norms)?;
1454 writer.finish()?;
1455 }
1456 }
1457 let length_lookup = postings::LengthLookup {
1458 doc_lengths: &self.doc_field_lengths,
1459 num_indexed_fields: self.num_indexed_fields,
1460 field_to_slot: &self.field_to_slot,
1461 chunk_maps: &chunk_maps,
1462 };
1463
1464 let inverted_index = std::mem::take(&mut self.inverted_index);
1467 let term_interner = std::mem::replace(&mut self.term_interner, Rodeo::new());
1468 #[cfg(feature = "native")]
1469 let store_path = self.store_path.clone();
1470 #[cfg(feature = "native")]
1471 let num_compression_threads = self.config.num_compression_threads;
1472 let compression_level = self.config.compression_level;
1473 let optimization = self.config.optimization;
1474 let posting_codec = self.config.posting_codec;
1475 let dense_vectors = std::mem::take(&mut self.dense_vectors);
1476 let binary_dense_vectors = std::mem::take(&mut self.binary_dense_vectors);
1477 let mut sparse_vectors = std::mem::take(&mut self.sparse_vectors);
1478 let schema = &self.schema;
1479
1480 let mut term_dict_writer =
1483 super::OffsetWriter::new(dir.streaming_writer(&files.term_dict).await?);
1484 let mut postings_writer =
1485 super::OffsetWriter::new(dir.streaming_writer(&files.postings).await?);
1486 let mut store_writer = super::OffsetWriter::new(dir.streaming_writer(&files.store).await?);
1487 let mut vectors_writer = if !dense_vectors.is_empty() || !binary_dense_vectors.is_empty() {
1488 Some(super::OffsetWriter::new(
1489 dir.streaming_writer(&files.vectors).await?,
1490 ))
1491 } else {
1492 None
1493 };
1494 let mut sparse_writer = if !sparse_vectors.is_empty() {
1495 Some(super::OffsetWriter::new(
1496 dir.streaming_writer(&files.sparse).await?,
1497 ))
1498 } else {
1499 None
1500 };
1501 let mut fast_fields = std::mem::take(&mut self.fast_fields);
1502 let num_docs = self.next_doc_id;
1503 let mut fast_writer = if !fast_fields.is_empty() {
1504 Some(super::OffsetWriter::new(
1505 dir.streaming_writer(&files.fast).await?,
1506 ))
1507 } else {
1508 None
1509 };
1510
1511 #[cfg(feature = "native")]
1512 {
1513 if let Some(ref mut f) = self.posting_spill_file {
1514 f.flush()?;
1515 }
1516 let posting_spill_index = std::mem::take(&mut self.posting_spill_index);
1517 let mut spill_reader_opt = if !posting_spill_index.is_empty() {
1518 let spill_file = std::fs::File::open(&self.posting_spill_path)?;
1519 Some((std::io::BufReader::new(spill_file), posting_spill_index))
1520 } else {
1521 None
1522 };
1523
1524 let ((postings_result, store_result), ((vectors_result, sparse_result), fast_result)) =
1525 rayon::join(
1526 || {
1527 rayon::join(
1528 || {
1529 let spill_arg = spill_reader_opt.as_mut().map(|(r, idx)| {
1530 (
1531 r as &mut std::io::BufReader<std::fs::File>,
1532 idx as &postings::SpillIndex,
1533 )
1534 });
1535 postings::build_postings_streaming(
1536 inverted_index,
1537 term_interner,
1538 &position_offsets,
1539 &length_lookup,
1540 &mut term_dict_writer,
1541 &mut postings_writer,
1542 (optimization, posting_codec),
1543 spill_arg,
1544 )
1545 },
1546 || {
1547 store::build_store_streaming(
1548 &store_path,
1549 num_compression_threads,
1550 compression_level,
1551 &mut store_writer,
1552 num_docs,
1553 )
1554 },
1555 )
1556 },
1557 || {
1558 rayon::join(
1559 || {
1560 rayon::join(
1561 || -> Result<()> {
1562 if let Some(ref mut w) = vectors_writer {
1563 dense::build_vectors_streaming(
1564 dense_vectors,
1565 binary_dense_vectors,
1566 schema,
1567 trained,
1568 w,
1569 )?;
1570 }
1571 Ok(())
1572 },
1573 || -> Result<()> {
1574 if let Some(ref mut w) = sparse_writer {
1575 sparse::build_sparse_streaming(
1576 &mut sparse_vectors,
1577 schema,
1578 w,
1579 )?;
1580 }
1581 Ok(())
1582 },
1583 )
1584 },
1585 || -> Result<()> {
1586 if let Some(ref mut w) = fast_writer {
1587 build_fast_fields_streaming(&mut fast_fields, num_docs, w)?;
1588 }
1589 Ok(())
1590 },
1591 )
1592 },
1593 );
1594 postings_result?;
1595 store_result?;
1596 vectors_result?;
1597 sparse_result?;
1598 fast_result?;
1599 }
1600
1601 #[cfg(not(feature = "native"))]
1602 {
1603 postings::build_postings_streaming(
1604 inverted_index,
1605 term_interner,
1606 &position_offsets,
1607 &length_lookup,
1608 &mut term_dict_writer,
1609 &mut postings_writer,
1610 (optimization, posting_codec),
1611 )?;
1612 store::build_store_streaming_from_buffer(
1613 &self.store_buffer,
1614 compression_level,
1615 &mut store_writer,
1616 num_docs,
1617 )?;
1618 if let Some(ref mut w) = vectors_writer {
1619 dense::build_vectors_streaming(
1620 dense_vectors,
1621 binary_dense_vectors,
1622 schema,
1623 trained,
1624 w,
1625 )?;
1626 }
1627 if let Some(ref mut w) = sparse_writer {
1628 sparse::build_sparse_streaming(&mut sparse_vectors, schema, w)?;
1629 }
1630 if let Some(ref mut w) = fast_writer {
1631 build_fast_fields_streaming(&mut fast_fields, num_docs, w)?;
1632 }
1633 }
1634
1635 let term_dict_bytes = term_dict_writer.offset() as usize;
1636 let postings_bytes = postings_writer.offset() as usize;
1637 let store_bytes = store_writer.offset() as usize;
1638 let vectors_bytes = vectors_writer.as_ref().map_or(0, |w| w.offset() as usize);
1639 let sparse_bytes = sparse_writer.as_ref().map_or(0, |w| w.offset() as usize);
1640 let fast_bytes = fast_writer.as_ref().map_or(0, |w| w.offset() as usize);
1641
1642 term_dict_writer.finish()?;
1643 postings_writer.finish()?;
1644 store_writer.finish()?;
1645 if let Some(w) = vectors_writer {
1646 w.finish()?;
1647 }
1648 if let Some(w) = sparse_writer {
1649 w.finish()?;
1650 }
1651 if let Some(w) = fast_writer {
1652 w.finish()?;
1653 }
1654 drop(position_offsets);
1655 drop(sparse_vectors);
1656
1657 log::info!(
1658 "[segment_build] index={} docs={}: term_dict={}, postings={}, store={}, dense_vectors={}, sparse_vectors={}, fast_fields={}",
1659 self.schema.index_label(),
1660 num_docs,
1661 crate::format_bytes(term_dict_bytes as u64),
1662 crate::format_bytes(postings_bytes as u64),
1663 crate::format_bytes(store_bytes as u64),
1664 crate::format_bytes(vectors_bytes as u64),
1665 crate::format_bytes(sparse_bytes as u64),
1666 crate::format_bytes(fast_bytes as u64),
1667 );
1668
1669 let meta = SegmentMeta {
1670 id: segment_id.0,
1671 num_docs: self.next_doc_id,
1672 field_stats: self.field_stats.clone(),
1673 };
1674
1675 dir.write_durable(&files.meta, &meta.serialize()?).await?;
1680
1681 #[cfg(feature = "native")]
1683 {
1684 let _ = std::fs::remove_file(&self.store_path);
1685 }
1686
1687 Ok(meta)
1688 }
1689}
1690
1691fn build_fast_fields_streaming(
1693 fast_fields: &mut FxHashMap<u32, crate::structures::fast_field::FastFieldWriter>,
1694 num_docs: u32,
1695 writer: &mut dyn Write,
1696) -> Result<()> {
1697 use crate::structures::fast_field::{FastFieldTocEntry, write_fast_field_toc_and_footer};
1698
1699 if fast_fields.is_empty() {
1700 return Ok(());
1701 }
1702
1703 let mut field_ids: Vec<u32> = fast_fields.keys().copied().collect();
1705 field_ids.sort_unstable();
1706
1707 let mut toc_entries: Vec<FastFieldTocEntry> = Vec::with_capacity(field_ids.len());
1708 let mut current_offset = 0u64;
1709
1710 for &field_id in &field_ids {
1711 let ff = fast_fields.get_mut(&field_id).unwrap();
1712 ff.pad_to(num_docs);
1713
1714 let (mut toc, bytes_written) = ff.serialize(writer, current_offset)?;
1715 toc.field_id = field_id;
1716 current_offset += bytes_written;
1717 toc_entries.push(toc);
1718 }
1719
1720 let toc_offset = current_offset;
1722 write_fast_field_toc_and_footer(writer, toc_offset, &toc_entries)?;
1723
1724 Ok(())
1725}
1726
1727#[cfg(feature = "native")]
1728impl Drop for SegmentBuilder {
1729 fn drop(&mut self) {
1730 let _ = std::fs::remove_file(&self.store_path);
1731 if self.posting_spill_file.is_some() {
1732 let _ = std::fs::remove_file(&self.posting_spill_path);
1733 }
1734 }
1735}
1736
1737#[cfg(test)]
1738impl SegmentBuilder {
1739 fn positions_for_term(&self, field: Field, term: &str) -> Vec<u32> {
1741 let Some(spur) = self.term_interner.get(term) else {
1742 return Vec::new();
1743 };
1744 let key = TermKey {
1745 field: field.0,
1746 term: spur,
1747 };
1748 self.position_index
1749 .get(&key)
1750 .map(|b| {
1751 b.postings
1752 .iter()
1753 .flat_map(|(_, ps)| ps.iter().copied())
1754 .collect()
1755 })
1756 .unwrap_or_default()
1757 }
1758}
1759
1760#[cfg(test)]
1761mod tests {
1762 use super::*;
1763 use crate::dsl::SchemaBuilder;
1764
1765 fn builder_for(schema: Schema) -> SegmentBuilder {
1766 SegmentBuilder::new(Arc::new(schema), SegmentBuilderConfig::default()).unwrap()
1767 }
1768
1769 #[test]
1770 fn position_scratch_resets_only_touched_terms_across_fields_and_chunks() {
1771 use crate::dsl::PositionMode;
1772 use crate::tokenizer::SimpleTokenizer;
1773
1774 for custom in [false, true] {
1775 for mode in [
1776 PositionMode::Ordinal,
1777 PositionMode::TokenPosition,
1778 PositionMode::Full,
1779 ] {
1780 let mut schema = SchemaBuilder::default();
1781 let body = schema.add_text_field("body", true, false);
1782 schema.set_positions(body, mode);
1783 let other = schema.add_text_field("other", true, false);
1784 schema.set_positions(other, PositionMode::TokenPosition);
1785 let no_positions = schema.add_text_field("no_positions", true, false);
1786 let mut builder = builder_for(schema.build());
1787 if custom {
1788 builder.set_tokenizer(body, Box::new(SimpleTokenizer));
1789 }
1790
1791 for doc in 0..2_000 {
1794 builder
1795 .index_text_field(
1796 body,
1797 doc,
1798 &format!("unique{doc} anchor anchor"),
1799 0,
1800 false,
1801 )
1802 .unwrap();
1803 assert_eq!(builder.local_position_terms.len(), 2);
1804 }
1805 assert_eq!(builder.local_positions.len(), 2_001);
1806 let anchor = builder.term_interner.get("anchor").unwrap();
1807 let capacity = builder.local_positions[&anchor].capacity();
1808
1809 builder
1810 .index_text_field(no_positions, 2_000, "anchor plain", 0, false)
1811 .unwrap();
1812 assert!(builder.local_position_terms.is_empty());
1813 assert!(builder.local_positions[&anchor].is_empty());
1814 assert_eq!(builder.local_positions[&anchor].capacity(), capacity);
1815 builder
1816 .index_text_field(other, 2_000, "anchor", 0, false)
1817 .unwrap();
1818 builder.index_text_field(body, 2_000, "", 0, false).unwrap();
1819 assert!(builder.local_position_terms.is_empty());
1820 builder
1821 .index_text_field(body, 2_000, "anchor", 1, false)
1822 .unwrap();
1823 builder
1824 .index_text_field(body, 2_001, "anchor", 2, false)
1825 .unwrap();
1826
1827 assert_eq!(builder.positions_for_term(other, "anchor"), vec![0]);
1828 let positions = builder.positions_for_term(body, "anchor");
1829 assert_eq!(
1830 positions.len(),
1831 4_002,
1832 "positions leaked between chunks or fields"
1833 );
1834 let expected = match mode {
1835 PositionMode::TokenPosition => [0, 0],
1836 PositionMode::Ordinal | PositionMode::Full => [1 << 20, 2 << 20],
1837 };
1838 assert_eq!(&positions[4_000..], &expected);
1839 }
1840 }
1841 }
1842
1843 #[test]
1849 fn test_add_document_rejects_type_mismatched_field_value() {
1850 let mut sb = SchemaBuilder::default();
1851 let views = sb.add_u64_field("views", true, true);
1852 let mut builder = builder_for(sb.build());
1853
1854 let mut doc = Document::new();
1855 doc.add_text(views, "123");
1856 let err = builder
1857 .add_document(doc)
1858 .expect_err("schema-mismatched value must be rejected loudly, not silently unindexed");
1859 let msg = err.to_string();
1860 assert!(msg.contains("views"), "error must name the field: {msg}");
1861 assert!(
1862 msg.contains("u64"),
1863 "error must name the expected type: {msg}"
1864 );
1865 assert!(msg.contains("text"), "error must name the got type: {msg}");
1866
1867 assert_eq!(builder.num_docs(), 0);
1869
1870 let mut doc = Document::new();
1872 doc.add_u64(views, 123);
1873 builder.add_document(doc).unwrap();
1874 assert_eq!(builder.num_docs(), 1);
1875 }
1876
1877 #[test]
1883 fn test_add_document_rejects_bmp_sparse_dim_out_of_range() {
1884 use crate::structures::{SparseFormat, SparseVectorConfig};
1885
1886 let mut sb = SchemaBuilder::default();
1887 let config = SparseVectorConfig {
1888 format: SparseFormat::Bmp,
1889 dims: Some(100),
1890 ..Default::default()
1891 };
1892 let spv = sb.add_sparse_vector_field_with_config("spv", true, false, config);
1893 let mut builder = builder_for(sb.build());
1894
1895 let mut doc = Document::new();
1897 doc.add_sparse_vector(spv, vec![(50, 1.0)]);
1898 builder.add_document(doc).unwrap();
1899
1900 let mut doc = Document::new();
1902 doc.add_sparse_vector(spv, vec![(50, 1.0), (150, 2.0)]);
1903 let err = builder
1904 .add_document(doc)
1905 .expect_err("out-of-range BMP dim must be rejected, not silently unsearchable");
1906 let msg = err.to_string();
1907 assert!(msg.contains("spv"), "error must name the field: {msg}");
1908 assert!(msg.contains("150"), "error must name the dim_id: {msg}");
1909 assert!(
1910 msg.contains("100"),
1911 "error must name the configured dims: {msg}"
1912 );
1913 assert_eq!(
1914 builder.num_docs(),
1915 1,
1916 "rejected doc must not consume a doc id"
1917 );
1918 }
1919
1920 #[test]
1921 fn test_add_document_maxscore_sparse_dims_unbounded() {
1922 let mut sb = SchemaBuilder::default();
1925 let spv = sb.add_sparse_vector_field("spv", true, false);
1926 let mut builder = builder_for(sb.build());
1927
1928 let mut doc = Document::new();
1929 doc.add_sparse_vector(spv, vec![(3_000_000, 1.0)]);
1930 builder.add_document(doc).unwrap();
1931 }
1932
1933 #[test]
1940 fn test_position_element_ordinal_overflow_saturates_instead_of_wrapping() {
1941 use crate::dsl::PositionMode;
1942
1943 let mut sb = SchemaBuilder::default();
1944 let body = sb.add_text_field("body", true, false);
1945 sb.set_positions(body, PositionMode::Full);
1946 let mut builder = builder_for(sb.build());
1947
1948 let mut doc = Document::new();
1951 doc.add_text(body, "anchor");
1952 for _ in 0..4095 {
1953 doc.add_text(body, "filler");
1954 }
1955 doc.add_text(body, "needle");
1956 builder.add_document(doc).unwrap();
1957
1958 let positions = builder.positions_for_term(body, "needle");
1959 assert_eq!(positions.len(), 1);
1960 let encoded = positions[0];
1961 assert_ne!(
1962 encoded >> 20,
1963 0,
1964 "element ordinal 4096 must not alias element 0"
1965 );
1966 assert_eq!(
1967 encoded >> 20,
1968 4095,
1969 "overflowing element ordinal must saturate at 4095"
1970 );
1971 }
1972
1973 #[test]
1974 fn test_position_token_position_overflow_saturates_instead_of_bleeding() {
1975 use crate::dsl::PositionMode;
1976
1977 let mut sb = SchemaBuilder::default();
1978 let body = sb.add_text_field("body", true, false);
1979 sb.set_positions(body, PositionMode::Full);
1980 let mut builder = builder_for(sb.build());
1981
1982 let mut text = "w ".repeat(1 << 20);
1985 text.push_str("needle");
1986 let mut doc = Document::new();
1987 doc.add_text(body, text);
1988 builder.add_document(doc).unwrap();
1989
1990 let positions = builder.positions_for_term(body, "needle");
1991 assert_eq!(positions.len(), 1);
1992 let encoded = positions[0];
1993 assert_eq!(
1994 encoded >> 20,
1995 0,
1996 "token position overflow must not decode as a different element ordinal"
1997 );
1998 assert_eq!(
1999 encoded & 0xFFFFF,
2000 0xFFFFF,
2001 "overflowing token position must saturate at 2^20 - 1"
2002 );
2003 }
2004
2005 #[cfg(feature = "native")]
2012 #[tokio::test]
2013 async fn test_spill_mid_document_does_not_duplicate_postings() {
2014 use crate::directories::RamDirectory;
2015 use crate::structures::TERMINATED;
2016
2017 let mut sb = SchemaBuilder::default();
2018 let body = sb.add_text_field("body", true, false);
2019 let schema = Arc::new(sb.build());
2020 let mut builder =
2021 SegmentBuilder::new(Arc::clone(&schema), SegmentBuilderConfig::default()).unwrap();
2022
2023 for _ in 0..16383 {
2026 let mut doc = Document::new();
2027 doc.add_text(body, "hot");
2028 builder.add_document(doc).unwrap();
2029 }
2030
2031 let mut doc = Document::new();
2036 doc.add_text(body, "hot");
2037 doc.add_text(body, "hot");
2038 let boundary_doc = builder.add_document(doc).unwrap();
2039 assert_eq!(boundary_doc, 16383);
2040
2041 let dir = RamDirectory::new();
2042 let segment_id = crate::segment::SegmentId::new();
2043 builder.build(&dir, segment_id, None).await.unwrap();
2044
2045 let reader = crate::segment::SegmentReader::open(&dir, segment_id, schema, 16)
2046 .await
2047 .unwrap();
2048 let postings = reader
2049 .get_postings(body, b"hot")
2050 .await
2051 .unwrap()
2052 .expect("postings for 'hot'");
2053 assert_eq!(
2054 postings.doc_count(),
2055 16384,
2056 "each document must appear exactly once per term (spill-boundary duplicate)"
2057 );
2058
2059 let mut it = postings.iterator();
2062 let mut prev: Option<DocId> = None;
2063 let mut boundary_tf = 0u32;
2064 let mut d = it.doc();
2065 while d != TERMINATED {
2066 if let Some(p) = prev {
2067 assert!(p < d, "duplicate/unordered doc id {d} after {p}");
2068 }
2069 if d == boundary_doc {
2070 boundary_tf = it.term_freq();
2071 }
2072 prev = Some(d);
2073 d = it.advance();
2074 }
2075 assert_eq!(prev, Some(boundary_doc));
2076 assert_eq!(
2077 boundary_tf, 2,
2078 "boundary doc's term frequency must combine both values"
2079 );
2080 }
2081}