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 token_buffer: String,
276
277 numeric_buffer: String,
279
280 dense_vectors: FxHashMap<u32, DenseVectorBuilder>,
283
284 binary_dense_vectors: FxHashMap<u32, BinaryDenseVectorBuilder>,
286
287 sparse_vectors: FxHashMap<u32, SparseVectorBuilder>,
290
291 position_index: HashMap<TermKey, PositionPostingListBuilder>,
294
295 position_enabled_fields: FxHashMap<u32, Option<crate::dsl::PositionMode>>,
297
298 current_element_ordinal: FxHashMap<u32, u32>,
300
301 chunk_maps: FxHashMap<u32, super::chunk_map::ChunkMapBuilder>,
304
305 position_saturation_warned: bool,
308
309 estimated_memory: usize,
311
312 doc_serialize_buffer: Vec<u8>,
314
315 fast_fields: FxHashMap<u32, crate::structures::fast_field::FastFieldWriter>,
317}
318
319impl SegmentBuilder {
320 pub fn new(schema: Arc<Schema>, config: SegmentBuilderConfig) -> Result<Self> {
322 #[cfg(feature = "native")]
323 let (store_file, store_path, spill_path) = {
324 let segment_id = uuid::Uuid::new_v4();
325 let store_path = config
326 .temp_dir
327 .join(format!("hermes_store_{}.tmp", segment_id));
328 let store_file = BufWriter::with_capacity(
329 STORE_BUFFER_SIZE,
330 OpenOptions::new()
331 .create(true)
332 .write(true)
333 .truncate(true)
334 .open(&store_path)?,
335 );
336 let spill_path = config
337 .temp_dir
338 .join(format!("hermes_spill_{}.tmp", segment_id));
339 (store_file, store_path, spill_path)
340 };
341
342 let registry = crate::tokenizer::TokenizerRegistry::new();
344 let mut num_indexed_fields = 0;
345 let mut field_to_slot = FxHashMap::default();
346 let mut position_enabled_fields = FxHashMap::default();
347 let mut tokenizers = FxHashMap::default();
348 let mut tokenizer_hint_fields = FxHashMap::default();
349 for (field, entry) in schema.fields() {
350 if entry.indexed && matches!(entry.field_type, FieldType::Text) {
351 field_to_slot.insert(field.0, num_indexed_fields);
352 num_indexed_fields += 1;
353 if entry.positions.is_some() {
354 position_enabled_fields.insert(field.0, entry.positions);
355 }
356 if let Some(ref tok_name) = entry.tokenizer
357 && let Some(tokenizer) = registry.get(tok_name)
358 {
359 tokenizers.insert(field, tokenizer);
360 }
361 if let Some(hint_field) = schema.tokenizer_hint_field(field) {
362 tokenizer_hint_fields.insert(field, hint_field);
363 }
364 }
365 }
366
367 use crate::structures::fast_field::{FastFieldColumnType, FastFieldWriter};
369 let mut fast_fields = FxHashMap::default();
370 for (field, entry) in schema.fields() {
371 if entry.fast {
372 let writer = if entry.multi {
373 match entry.field_type {
374 FieldType::U64 => {
375 FastFieldWriter::new_numeric_multi(FastFieldColumnType::U64)
376 }
377 FieldType::I64 => {
378 FastFieldWriter::new_numeric_multi(FastFieldColumnType::I64)
379 }
380 FieldType::F64 => {
381 FastFieldWriter::new_numeric_multi(FastFieldColumnType::F64)
382 }
383 FieldType::Text => FastFieldWriter::new_text_multi(),
384 _ => continue,
385 }
386 } else {
387 match entry.field_type {
388 FieldType::U64 => FastFieldWriter::new_numeric(FastFieldColumnType::U64),
389 FieldType::I64 => FastFieldWriter::new_numeric(FastFieldColumnType::I64),
390 FieldType::F64 => FastFieldWriter::new_numeric(FastFieldColumnType::F64),
391 FieldType::Text => FastFieldWriter::new_text(),
392 _ => continue,
393 }
394 };
395 fast_fields.insert(field.0, writer);
396 }
397 }
398
399 Ok(Self {
400 schema,
401 tokenizers,
402 tokenizer_hint_fields,
403 tokenizer_hint_buffer: String::new(),
404 unhinted_dynamic_docs: 0,
405 term_interner: Rodeo::new(),
406 inverted_index: HashMap::with_capacity(config.posting_map_capacity),
407 #[cfg(feature = "native")]
408 posting_spill_file: None,
409 #[cfg(feature = "native")]
410 posting_spill_path: spill_path,
411 #[cfg(feature = "native")]
412 posting_spill_index: HashMap::new(),
413 #[cfg(feature = "native")]
414 posting_spill_offset: 0,
415 #[cfg(feature = "native")]
416 store_file,
417 #[cfg(feature = "native")]
418 store_path,
419 #[cfg(not(feature = "native"))]
420 store_buffer: Vec::with_capacity(STORE_BUFFER_SIZE),
421 next_doc_id: 0,
422 field_stats: FxHashMap::default(),
423 chunk_maps: FxHashMap::default(),
424 doc_field_lengths: Vec::new(),
425 num_indexed_fields,
426 field_to_slot,
427 local_tf_buffer: FxHashMap::default(),
428 local_positions: FxHashMap::default(),
429 token_buffer: String::with_capacity(64),
430 numeric_buffer: String::with_capacity(32),
431 config,
432 dense_vectors: FxHashMap::default(),
433 binary_dense_vectors: FxHashMap::default(),
434 sparse_vectors: FxHashMap::default(),
435 position_index: HashMap::new(),
436 position_enabled_fields,
437 current_element_ordinal: FxHashMap::default(),
438 position_saturation_warned: false,
439 estimated_memory: 0,
440 doc_serialize_buffer: Vec::with_capacity(256),
441 fast_fields,
442 })
443 }
444
445 pub fn set_tokenizer(&mut self, field: Field, tokenizer: BoxedTokenizer) {
446 self.tokenizers.insert(field, tokenizer);
447 }
448
449 pub fn unhinted_dynamic_docs(&self) -> u64 {
451 self.unhinted_dynamic_docs
452 }
453
454 fn resolve_tokenizer_hint(
460 &mut self,
461 field: Field,
462 doc: &Document,
463 element_ordinal: u32,
464 ) -> bool {
465 let Some(&hint_field) = self.tokenizer_hint_fields.get(&field) else {
466 return false;
467 };
468 self.tokenizer_hint_buffer.clear();
469 for value in doc.get_all(hint_field) {
470 if let FieldValue::Text(hint) = value {
471 let hint = hint.trim();
472 if hint.is_empty() {
473 continue;
474 }
475 if !self.tokenizer_hint_buffer.is_empty() {
476 self.tokenizer_hint_buffer.push(',');
477 }
478 for c in hint.chars().flat_map(char::to_lowercase) {
479 self.tokenizer_hint_buffer.push(c);
480 }
481 }
482 }
483 if self.tokenizer_hint_buffer.is_empty() && element_ordinal == 0 {
484 self.unhinted_dynamic_docs += 1;
485 }
486 true
487 }
488
489 fn next_element_ordinal(&mut self, field_id: u32) -> u32 {
492 let ordinal = *self.current_element_ordinal.get(&field_id).unwrap_or(&0);
493 *self.current_element_ordinal.entry(field_id).or_insert(0) += 1;
494 ordinal
495 }
496
497 fn next_vector_ordinal(&mut self, field_id: u32) -> Result<u16> {
498 let ordinal = self.next_element_ordinal(field_id);
499 u16::try_from(ordinal).map_err(|_| {
500 crate::Error::Document(format!(
501 "field {field_id} has more than {} vector values in one document",
502 u16::MAX as usize + 1
503 ))
504 })
505 }
506
507 pub fn num_docs(&self) -> u32 {
508 self.next_doc_id
509 }
510
511 #[inline]
513 pub fn estimated_memory_bytes(&self) -> usize {
514 self.estimated_memory
515 }
516
517 pub fn sparse_dim_count(&self) -> usize {
519 self.sparse_vectors.values().map(|b| b.postings.len()).sum()
520 }
521
522 pub fn stats(&self) -> SegmentBuilderStats {
524 use std::mem::size_of;
525
526 let postings_in_memory: usize =
527 self.inverted_index.values().map(|p| p.postings.len()).sum();
528
529 let compact_posting_size = size_of::<CompactPosting>();
531 let vec_overhead = size_of::<Vec<u8>>(); let term_key_size = size_of::<TermKey>();
533 let posting_builder_size = size_of::<PostingListBuilder>();
534 let spur_size = size_of::<Spur>();
535 let sparse_entry_size = size_of::<(DocId, u16, f32)>();
536
537 let hashmap_entry_base_overhead = 8usize;
540
541 let fxhashmap_entry_overhead = hashmap_entry_base_overhead;
543
544 let postings_bytes: usize = self
546 .inverted_index
547 .values()
548 .map(|p| p.postings.capacity() * compact_posting_size + vec_overhead)
549 .sum();
550
551 let index_overhead_bytes = self.inverted_index.len()
553 * (term_key_size + posting_builder_size + hashmap_entry_base_overhead);
554
555 let interner_arena_overhead = 2 * size_of::<usize>();
558 let avg_term_len = 8; let interner_bytes =
560 self.term_interner.len() * (avg_term_len + spur_size + interner_arena_overhead);
561
562 let field_lengths_bytes =
564 self.doc_field_lengths.capacity() * size_of::<u32>() + vec_overhead;
565
566 let mut dense_vectors_bytes: usize = 0;
568 let mut dense_vector_count: usize = 0;
569 let doc_id_ordinal_size = size_of::<(DocId, u16)>();
570 for b in self.dense_vectors.values() {
571 dense_vectors_bytes += b.vectors.capacity() * size_of::<f32>()
572 + b.doc_ids.capacity() * doc_id_ordinal_size
573 + 2 * vec_overhead; dense_vector_count += b.doc_ids.len();
575 }
576 for b in self.binary_dense_vectors.values() {
578 dense_vectors_bytes += b.vectors.capacity()
579 + b.doc_ids.capacity() * doc_id_ordinal_size
580 + 2 * vec_overhead;
581 dense_vector_count += b.doc_ids.len();
582 }
583
584 let local_tf_entry_size = spur_size + size_of::<u32>() + fxhashmap_entry_overhead;
586 let local_tf_buffer_bytes = self.local_tf_buffer.capacity() * local_tf_entry_size;
587
588 let mut sparse_vectors_bytes: usize = 0;
590 for builder in self.sparse_vectors.values() {
591 for postings in builder.postings.values() {
592 sparse_vectors_bytes += postings.capacity() * sparse_entry_size + vec_overhead;
593 }
594 let inner_entry_size = size_of::<u32>() + vec_overhead + fxhashmap_entry_overhead;
596 sparse_vectors_bytes += builder.postings.len() * inner_entry_size;
597 }
598 let outer_sparse_entry_size =
600 size_of::<u32>() + size_of::<SparseVectorBuilder>() + fxhashmap_entry_overhead;
601 sparse_vectors_bytes += self.sparse_vectors.len() * outer_sparse_entry_size;
602
603 let mut position_index_bytes: usize = 0;
605 for pos_builder in self.position_index.values() {
606 for (_, positions) in &pos_builder.postings {
607 position_index_bytes += positions.capacity() * size_of::<u32>() + vec_overhead;
608 }
609 let pos_entry_size = size_of::<DocId>() + vec_overhead;
611 position_index_bytes += pos_builder.postings.capacity() * pos_entry_size;
612 }
613 let pos_index_entry_size =
615 term_key_size + size_of::<PositionPostingListBuilder>() + hashmap_entry_base_overhead;
616 position_index_bytes += self.position_index.len() * pos_index_entry_size;
617
618 let estimated_memory_bytes = postings_bytes
619 + index_overhead_bytes
620 + interner_bytes
621 + field_lengths_bytes
622 + dense_vectors_bytes
623 + local_tf_buffer_bytes
624 + sparse_vectors_bytes
625 + position_index_bytes;
626
627 let memory_breakdown = MemoryBreakdown {
628 postings_bytes,
629 index_overhead_bytes,
630 interner_bytes,
631 field_lengths_bytes,
632 dense_vectors_bytes,
633 dense_vector_count,
634 sparse_vectors_bytes,
635 position_index_bytes,
636 };
637
638 SegmentBuilderStats {
639 num_docs: self.next_doc_id,
640 unique_terms: self.inverted_index.len(),
641 postings_in_memory,
642 interned_strings: self.term_interner.len(),
643 doc_field_lengths_size: self.doc_field_lengths.len(),
644 estimated_memory_bytes,
645 memory_breakdown,
646 unhinted_dynamic_docs: self.unhinted_dynamic_docs,
647 }
648 }
649
650 fn validate_document_against_schema(&self, doc: &Document) -> Result<()> {
665 validate_vector_value_counts(doc, &self.schema)?;
666
667 for (field, value) in doc.field_values() {
668 let Some(entry) = self.schema.get_field_entry(*field) else {
669 continue;
670 };
671
672 if !matches!(
675 &entry.field_type,
676 FieldType::DenseVector | FieldType::BinaryDenseVector
677 ) && !entry.indexed
678 && !entry.fast
679 {
680 continue;
681 }
682
683 match (&entry.field_type, value) {
684 (FieldType::SparseVector, FieldValue::SparseVector(entries)) => {
685 if let Some(config) = entry.sparse_vector_config.as_ref()
686 && config.format == crate::structures::SparseFormat::Bmp
687 {
688 let dims = config.dims.unwrap_or(DEFAULT_BMP_SPARSE_DIMS);
689 if let Some(&(dim_id, _)) =
690 entries.iter().find(|&&(dim_id, _)| dim_id >= dims)
691 {
692 return Err(crate::Error::Schema(format!(
693 "sparse vector for field '{}' contains dim_id {} out of \
694 range for the configured BMP dims={}: dimensions >= dims \
695 are never written to the block-max grid and can never \
696 match a query; raise `dims` in the field's sparse_vector \
697 config or fix the embedding model",
698 entry.name, dim_id, dims
699 )));
700 }
701 }
702 }
703 (FieldType::Text, FieldValue::Text(_))
705 | (FieldType::U64, FieldValue::U64(_))
706 | (FieldType::I64, FieldValue::I64(_))
707 | (FieldType::F64, FieldValue::F64(_))
708 | (FieldType::DenseVector, FieldValue::DenseVector(_))
709 | (FieldType::BinaryDenseVector, FieldValue::BinaryDenseVector(_))
710 | (FieldType::Bytes, FieldValue::Bytes(_))
712 | (FieldType::Json, FieldValue::Json(_)) => {}
713 (expected, got) => {
714 return Err(crate::Error::Schema(format!(
715 "type mismatch for field '{}': schema expects a {} value, got {}; \
716 the value would be stored but never indexed, so queries on this \
717 field could never match the document — fix the document or the \
718 schema",
719 entry.name,
720 field_type_name(expected),
721 field_value_type_name(got),
722 )));
723 }
724 }
725 }
726 Ok(())
727 }
728
729 pub fn add_document(&mut self, doc: Document) -> Result<DocId> {
731 self.validate_document_against_schema(&doc)?;
733
734 let doc_id = self.next_doc_id;
735 self.next_doc_id += 1;
736
737 let base_idx = self.doc_field_lengths.len();
739 self.doc_field_lengths
740 .resize(base_idx + self.num_indexed_fields, 0);
741 self.estimated_memory += self.num_indexed_fields * std::mem::size_of::<u32>();
742
743 self.current_element_ordinal.clear();
745
746 for (field, value) in doc.field_values() {
747 let Some(entry) = self.schema.get_field_entry(*field) else {
748 continue;
749 };
750
751 if !matches!(
754 &entry.field_type,
755 FieldType::DenseVector | FieldType::BinaryDenseVector
756 ) && !entry.indexed
757 && !entry.fast
758 {
759 continue;
760 }
761
762 match (&entry.field_type, value) {
763 (FieldType::Text, FieldValue::Text(text)) => {
764 if entry.indexed && entry.chunked {
765 let field_id = field.0;
769 let element_ordinal = self.next_element_ordinal(field_id);
770 let ordinal = u16::try_from(element_ordinal).map_err(|_| {
771 crate::Error::Document(format!(
772 "chunked field {field_id} has more than {} chunk values in one document",
773 u16::MAX as usize + 1
774 ))
775 })?;
776 let hinted = self.resolve_tokenizer_hint(*field, &doc, element_ordinal);
777 let vid = self
778 .chunk_maps
779 .get(&field.0)
780 .map_or(0usize, |map| map.len());
781 let vid = u32::try_from(vid).map_err(|_| {
782 crate::Error::Document(format!(
783 "chunked field {field_id} exceeds u32::MAX chunks in one segment"
784 ))
785 })?;
786 let token_count = self.index_text_field(*field, vid, text, 0, hinted)?;
790 self.chunk_maps.entry(field.0).or_default().push(
791 doc_id,
792 ordinal,
793 token_count,
794 )?;
795 self.estimated_memory += 8;
796
797 let stats = self.field_stats.entry(field.0).or_default();
800 stats.total_tokens += token_count as u64;
801 stats.doc_count += 1;
802 } else if entry.indexed {
803 let element_ordinal = self.next_element_ordinal(field.0);
804 let hinted = self.resolve_tokenizer_hint(*field, &doc, element_ordinal);
805 let token_count =
806 self.index_text_field(*field, doc_id, text, element_ordinal, hinted)?;
807
808 let stats = self.field_stats.entry(field.0).or_default();
809 stats.total_tokens += token_count as u64;
810 if element_ordinal == 0 {
811 stats.doc_count += 1;
812 }
813
814 if let Some(&slot) = self.field_to_slot.get(&field.0) {
815 let len = &mut self.doc_field_lengths[base_idx + slot];
818 *len = len.saturating_add(token_count);
819 }
820 }
821
822 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
824 ff.add_text(doc_id, text);
825 }
826 }
827 (FieldType::U64, FieldValue::U64(v)) => {
828 if entry.indexed {
829 self.index_numeric_field(*field, doc_id, *v)?;
830 }
831 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
832 ff.add_u64(doc_id, *v);
833 }
834 }
835 (FieldType::I64, FieldValue::I64(v)) => {
836 if entry.indexed {
837 self.index_numeric_field(*field, doc_id, *v as u64)?;
838 }
839 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
840 ff.add_i64(doc_id, *v);
841 }
842 }
843 (FieldType::F64, FieldValue::F64(v)) => {
844 if entry.indexed {
845 self.index_numeric_field(*field, doc_id, v.to_bits())?;
846 }
847 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
848 ff.add_f64(doc_id, *v);
849 }
850 }
851 (FieldType::DenseVector, FieldValue::DenseVector(vec))
852 if entry.indexed || entry.stored =>
853 {
854 let ordinal = self.next_vector_ordinal(field.0)?;
855 self.index_dense_vector_field(*field, doc_id, ordinal, vec)?;
856 }
857 (FieldType::BinaryDenseVector, FieldValue::BinaryDenseVector(bytes))
858 if entry.indexed || entry.stored =>
859 {
860 let ordinal = self.next_vector_ordinal(field.0)?;
861 self.index_binary_dense_vector_field(*field, doc_id, ordinal, bytes)?;
862 }
863 (FieldType::SparseVector, FieldValue::SparseVector(entries)) => {
864 let ordinal = self.next_vector_ordinal(field.0)?;
865 self.index_sparse_vector_field(*field, doc_id, ordinal, entries)?;
866 }
867 _ => {}
872 }
873 }
874
875 self.write_document_to_store(&doc)?;
877
878 Ok(doc_id)
879 }
880
881 fn index_text_field(
890 &mut self,
891 field: Field,
892 doc_id: DocId,
893 text: &str,
894 element_ordinal: u32,
895 hinted: bool,
896 ) -> Result<u32> {
897 use crate::dsl::PositionMode;
898
899 let field_id = field.0;
900 let position_mode = self
901 .position_enabled_fields
902 .get(&field_id)
903 .copied()
904 .flatten();
905
906 let encoded_ordinal = if position_mode.is_some_and(|m| m.tracks_ordinal())
909 && element_ordinal > MAX_POSITION_ELEMENT_ORDINAL
910 {
911 self.warn_position_saturation(
912 "element ordinal",
913 element_ordinal,
914 MAX_POSITION_ELEMENT_ORDINAL,
915 );
916 MAX_POSITION_ELEMENT_ORDINAL
917 } else {
918 element_ordinal
919 };
920
921 self.local_tf_buffer.clear();
925 for v in self.local_positions.values_mut() {
927 v.clear();
928 }
929
930 let mut token_position = 0u32;
931
932 let custom_tokens = self.tokenizers.get(&field).map(|t| {
936 if hinted {
937 let hint = (!self.tokenizer_hint_buffer.is_empty())
938 .then_some(self.tokenizer_hint_buffer.as_str());
939 t.tokenize_hinted(text, hint)
940 } else {
941 t.tokenize(text)
942 }
943 });
944
945 if let Some(tokens) = custom_tokens {
946 for token in &tokens {
948 let term_spur = if let Some(spur) = self.term_interner.get(&token.text) {
949 spur
950 } else {
951 let spur = self.term_interner.get_or_intern(&token.text);
952 self.estimated_memory += token.text.len() + INTERN_OVERHEAD;
953 spur
954 };
955 *self.local_tf_buffer.entry(term_spur).or_insert(0) += 1;
956
957 if let Some(mode) = position_mode {
958 let encoded_pos = match mode {
959 PositionMode::Ordinal => encoded_ordinal << 20,
960 PositionMode::TokenPosition => token.position,
961 PositionMode::Full => {
962 (encoded_ordinal << 20) | self.saturate_token_position(token.position)
963 }
964 };
965 self.local_positions
966 .entry(term_spur)
967 .or_default()
968 .push(encoded_pos);
969 }
970 }
971 token_position = tokens.len() as u32;
972 } else {
973 for word in text.split_whitespace() {
975 self.token_buffer.clear();
976 for c in word.chars() {
977 if c.is_alphanumeric() {
978 for lc in c.to_lowercase() {
979 self.token_buffer.push(lc);
980 }
981 }
982 }
983
984 if self.token_buffer.is_empty() {
985 continue;
986 }
987
988 let term_spur = if let Some(spur) = self.term_interner.get(&self.token_buffer) {
989 spur
990 } else {
991 let spur = self.term_interner.get_or_intern(&self.token_buffer);
992 self.estimated_memory += self.token_buffer.len() + INTERN_OVERHEAD;
993 spur
994 };
995 *self.local_tf_buffer.entry(term_spur).or_insert(0) += 1;
996
997 if let Some(mode) = position_mode {
998 let encoded_pos = match mode {
999 PositionMode::Ordinal => encoded_ordinal << 20,
1000 PositionMode::TokenPosition => token_position,
1001 PositionMode::Full => {
1002 (encoded_ordinal << 20) | self.saturate_token_position(token_position)
1003 }
1004 };
1005 self.local_positions
1006 .entry(term_spur)
1007 .or_default()
1008 .push(encoded_pos);
1009 }
1010
1011 token_position += 1;
1012 }
1013 }
1014
1015 for (&term_spur, &tf) in &self.local_tf_buffer {
1018 let term_key = TermKey {
1019 field: field_id,
1020 term: term_spur,
1021 };
1022
1023 match self.inverted_index.entry(term_key) {
1024 hashbrown::hash_map::Entry::Occupied(mut o) => {
1025 o.get_mut().add(doc_id, tf);
1026 self.estimated_memory += size_of::<CompactPosting>();
1027 #[cfg(feature = "native")]
1029 if o.get().should_spill() {
1030 use byteorder::{LittleEndian, WriteBytesExt};
1031
1032 let builder = o.get_mut();
1033 let count = builder.postings.len() as u32;
1034 let offset = self.posting_spill_offset;
1035
1036 let spill_file = if let Some(ref mut f) = self.posting_spill_file {
1038 f
1039 } else {
1040 self.posting_spill_file = Some(BufWriter::with_capacity(
1041 256 * 1024,
1042 OpenOptions::new()
1043 .create(true)
1044 .write(true)
1045 .truncate(true)
1046 .open(&self.posting_spill_path)?,
1047 ));
1048 self.posting_spill_file.as_mut().unwrap()
1049 };
1050 for p in &builder.postings {
1051 spill_file.write_u32::<LittleEndian>(p.doc_id)?;
1052 spill_file.write_u16::<LittleEndian>(p.term_freq)?;
1053 }
1054 self.posting_spill_offset += count as u64 * 6;
1055 self.posting_spill_index
1056 .entry(term_key)
1057 .or_default()
1058 .push((offset, count));
1059
1060 let freed = builder.postings.len() * size_of::<CompactPosting>();
1061 builder.spilled_count += count;
1062 builder.postings.clear();
1063 self.estimated_memory -= freed;
1064 }
1065 }
1066 hashbrown::hash_map::Entry::Vacant(v) => {
1067 let mut posting = PostingListBuilder::new();
1068 posting.add(doc_id, tf);
1069 v.insert(posting);
1070 self.estimated_memory += size_of::<CompactPosting>() + NEW_TERM_OVERHEAD;
1071 }
1072 }
1073
1074 if position_mode.is_some()
1075 && let Some(positions) = self.local_positions.get(&term_spur)
1076 {
1077 match self.position_index.entry(term_key) {
1078 hashbrown::hash_map::Entry::Occupied(mut o) => {
1079 for &pos in positions {
1080 o.get_mut().add_position(doc_id, pos);
1081 }
1082 self.estimated_memory += positions.len() * size_of::<u32>();
1083 }
1084 hashbrown::hash_map::Entry::Vacant(v) => {
1085 let mut pos_posting = PositionPostingListBuilder::new();
1086 for &pos in positions {
1087 pos_posting.add_position(doc_id, pos);
1088 }
1089 self.estimated_memory +=
1090 positions.len() * size_of::<u32>() + NEW_POS_TERM_OVERHEAD;
1091 v.insert(pos_posting);
1092 }
1093 }
1094 }
1095 }
1096
1097 Ok(token_position)
1098 }
1099
1100 #[inline]
1103 fn saturate_token_position(&mut self, token_position: u32) -> u32 {
1104 if token_position > MAX_TOKEN_POSITION {
1105 self.warn_position_saturation("token position", token_position, MAX_TOKEN_POSITION);
1106 MAX_TOKEN_POSITION
1107 } else {
1108 token_position
1109 }
1110 }
1111
1112 #[cold]
1114 fn warn_position_saturation(&mut self, what: &str, value: u32, max: u32) {
1115 if !self.position_saturation_warned {
1116 self.position_saturation_warned = true;
1117 log::warn!(
1118 "[segment_builder] index={} {what} {value} exceeds the position-encoding limit {max}; \
1119 saturating — phrase/ordinal matching degrades for the overflowing \
1120 elements/tokens instead of corrupting other documents' matches \
1121 (further occurrences in this segment are not logged)",
1122 self.schema.index_label()
1123 );
1124 }
1125 }
1126
1127 fn index_numeric_field(&mut self, field: Field, doc_id: DocId, value: u64) -> Result<()> {
1128 use std::fmt::Write;
1129
1130 self.numeric_buffer.clear();
1131 write!(self.numeric_buffer, "__num_{}", value).unwrap();
1132 let term_spur = if let Some(spur) = self.term_interner.get(&self.numeric_buffer) {
1133 spur
1134 } else {
1135 let spur = self.term_interner.get_or_intern(&self.numeric_buffer);
1136 self.estimated_memory += self.numeric_buffer.len() + INTERN_OVERHEAD;
1137 spur
1138 };
1139
1140 let term_key = TermKey {
1141 field: field.0,
1142 term: term_spur,
1143 };
1144
1145 match self.inverted_index.entry(term_key) {
1146 hashbrown::hash_map::Entry::Occupied(mut o) => {
1147 o.get_mut().add(doc_id, 1);
1148 self.estimated_memory += size_of::<CompactPosting>();
1149 }
1150 hashbrown::hash_map::Entry::Vacant(v) => {
1151 let mut posting = PostingListBuilder::new();
1152 posting.add(doc_id, 1);
1153 v.insert(posting);
1154 self.estimated_memory += size_of::<CompactPosting>() + NEW_TERM_OVERHEAD;
1155 }
1156 }
1157
1158 Ok(())
1159 }
1160
1161 fn index_dense_vector_field(
1163 &mut self,
1164 field: Field,
1165 doc_id: DocId,
1166 ordinal: u16,
1167 vector: &[f32],
1168 ) -> Result<()> {
1169 let dim = vector.len();
1170 let expected_dim = self
1171 .schema
1172 .get_field_entry(field)
1173 .and_then(|entry| entry.dense_vector_config.as_ref())
1174 .map(|config| config.dim)
1175 .ok_or_else(|| crate::Error::Schema("DenseVector field missing config".to_string()))?;
1176 if dim != expected_dim {
1177 return Err(crate::Error::Schema(format!(
1178 "Dense vector dimension mismatch: schema expects {}, got {}",
1179 expected_dim, dim
1180 )));
1181 }
1182 if let Some((index, value)) = vector
1183 .iter()
1184 .enumerate()
1185 .find(|(_, value)| !value.is_finite())
1186 {
1187 return Err(crate::Error::Document(format!(
1188 "dense vector contains non-finite value {value} at index {index}"
1189 )));
1190 }
1191
1192 let builder = self
1193 .dense_vectors
1194 .entry(field.0)
1195 .or_insert_with(|| DenseVectorBuilder::new(dim));
1196
1197 if builder.dim != dim && builder.len() > 0 {
1199 return Err(crate::Error::Schema(format!(
1200 "Dense vector dimension mismatch: expected {}, got {}",
1201 builder.dim, dim
1202 )));
1203 }
1204
1205 builder.add(doc_id, ordinal, vector);
1206
1207 self.estimated_memory += std::mem::size_of_val(vector) + size_of::<(DocId, u16)>();
1208
1209 Ok(())
1210 }
1211
1212 fn index_binary_dense_vector_field(
1214 &mut self,
1215 field: Field,
1216 doc_id: DocId,
1217 ordinal: u16,
1218 bytes: &[u8],
1219 ) -> Result<()> {
1220 let dim_bits = self
1221 .schema
1222 .get_field_entry(field)
1223 .and_then(|e| e.binary_dense_vector_config.as_ref())
1224 .map(|c| c.dim)
1225 .ok_or_else(|| {
1226 crate::Error::Schema("BinaryDenseVector field missing config".to_string())
1227 })?;
1228
1229 let expected_byte_len = dim_bits.div_ceil(8);
1230 if dim_bits == 0 || !dim_bits.is_multiple_of(8) {
1231 return Err(crate::Error::Schema(format!(
1232 "Binary vector dimension must be a positive multiple of 8, got {dim_bits}"
1233 )));
1234 }
1235 if bytes.len() != expected_byte_len {
1236 return Err(crate::Error::Schema(format!(
1237 "Binary vector byte length mismatch: expected {} (dim={}), got {}",
1238 expected_byte_len,
1239 dim_bits,
1240 bytes.len()
1241 )));
1242 }
1243
1244 let builder = self
1245 .binary_dense_vectors
1246 .entry(field.0)
1247 .or_insert_with(|| BinaryDenseVectorBuilder::new(dim_bits));
1248
1249 builder.add(doc_id, ordinal, bytes);
1250 self.estimated_memory += bytes.len() + size_of::<(DocId, u16)>();
1251
1252 Ok(())
1253 }
1254
1255 fn index_sparse_vector_field(
1265 &mut self,
1266 field: Field,
1267 doc_id: DocId,
1268 ordinal: u16,
1269 entries: &[(u32, f32)],
1270 ) -> Result<()> {
1271 if let Some((index, (_, weight))) = entries
1272 .iter()
1273 .enumerate()
1274 .find(|(_, (_, weight))| !weight.is_finite())
1275 {
1276 return Err(crate::Error::Document(format!(
1277 "sparse vector contains non-finite weight {weight} at index {index}"
1278 )));
1279 }
1280 let (weight_threshold, doc_mass, min_terms) = self
1281 .schema
1282 .get_field_entry(field)
1283 .and_then(|entry| entry.sparse_vector_config.as_ref())
1284 .map(|config| (config.weight_threshold, config.doc_mass, config.min_terms))
1285 .unwrap_or((0.0, None, 0));
1286
1287 let builder = self
1288 .sparse_vectors
1289 .entry(field.0)
1290 .or_insert_with(SparseVectorBuilder::new);
1291
1292 builder.inc_vector_count();
1293
1294 let mass_cutoff = match doc_mass {
1298 Some(mass) if mass < 1.0 && entries.len() > min_terms => {
1299 let mut weights: Vec<f32> = entries
1300 .iter()
1301 .map(|&(_, w)| w.abs())
1302 .filter(|w| *w >= weight_threshold)
1303 .collect();
1304 weights.sort_unstable_by(|a, b| b.total_cmp(a));
1305 let total: f64 = weights.iter().map(|&w| w as f64).sum();
1306 let target = total * mass as f64;
1307 let mut cumulative = 0.0f64;
1308 let mut cutoff = 0.0f32;
1309 for &w in &weights {
1310 if cumulative >= target {
1311 break;
1312 }
1313 cumulative += w as f64;
1314 cutoff = w;
1315 }
1316 cutoff
1317 }
1318 _ => 0.0,
1319 };
1320
1321 for &(dim_id, weight) in entries {
1322 if weight.abs() < weight_threshold || weight.abs() < mass_cutoff {
1324 continue;
1325 }
1326
1327 let is_new_dim = !builder.postings.contains_key(&dim_id);
1328 builder.add(dim_id, doc_id, ordinal, weight);
1329 self.estimated_memory += size_of::<(DocId, u16, f32)>();
1330 if is_new_dim {
1331 self.estimated_memory += size_of::<u32>() + size_of::<Vec<(DocId, u16, f32)>>() + 8; }
1334 }
1335
1336 Ok(())
1337 }
1338
1339 fn write_document_to_store(&mut self, doc: &Document) -> Result<()> {
1341 use byteorder::{LittleEndian, WriteBytesExt};
1342
1343 super::store::serialize_document_into(doc, &self.schema, &mut self.doc_serialize_buffer)?;
1344
1345 #[cfg(feature = "native")]
1346 {
1347 self.store_file
1348 .write_u32::<LittleEndian>(self.doc_serialize_buffer.len() as u32)?;
1349 self.store_file.write_all(&self.doc_serialize_buffer)?;
1350 }
1351 #[cfg(not(feature = "native"))]
1352 {
1353 self.store_buffer
1354 .write_u32::<LittleEndian>(self.doc_serialize_buffer.len() as u32)?;
1355 self.store_buffer.write_all(&self.doc_serialize_buffer)?;
1356 self.estimated_memory += size_of::<u32>() + self.doc_serialize_buffer.len();
1360 }
1361
1362 Ok(())
1363 }
1364
1365 pub async fn build<D: Directory + DirectoryWriter>(
1371 mut self,
1372 dir: &D,
1373 segment_id: SegmentId,
1374 trained: Option<&super::TrainedVectorStructures>,
1375 ) -> Result<SegmentMeta> {
1376 #[cfg(feature = "native")]
1378 self.store_file.flush()?;
1379
1380 let files = SegmentFiles::new(segment_id.0);
1381
1382 let position_index = std::mem::take(&mut self.position_index);
1384 let position_offsets = if !position_index.is_empty() {
1385 let mut pos_writer = dir.streaming_writer(&files.positions).await?;
1386 let offsets = postings::build_positions_streaming(
1387 position_index,
1388 &self.term_interner,
1389 &mut *pos_writer,
1390 )?;
1391 pos_writer.finish()?;
1392 offsets
1393 } else {
1394 FxHashMap::default()
1395 };
1396
1397 let chunk_maps = std::mem::take(&mut self.chunk_maps);
1400 {
1401 let mut fields: Vec<(u32, &super::chunk_map::ChunkMapBuilder)> = chunk_maps
1402 .iter()
1403 .filter(|(_, map)| !map.is_empty())
1404 .map(|(field_id, map)| (*field_id, map))
1405 .collect();
1406 fields.sort_by_key(|(field_id, _)| *field_id);
1407 let num_docs = self.next_doc_id as usize;
1408 let mut columns: Vec<(u32, Vec<u16>, u64)> = Vec::new();
1409 for (&field_id, &slot) in &self.field_to_slot {
1410 if self
1411 .schema
1412 .get_field_entry(crate::dsl::Field(field_id))
1413 .is_some_and(|entry| entry.chunked)
1414 {
1415 continue;
1416 }
1417 let mut total = 0u64;
1418 let lengths: Vec<u16> = (0..num_docs)
1419 .map(|doc| {
1420 let len = self.doc_field_lengths[doc * self.num_indexed_fields + slot];
1421 total += u64::from(len);
1422 len.min(super::chunk_map::MAX_CHUNK_LENGTH) as u16
1423 })
1424 .collect();
1425 if total > 0 {
1426 columns.push((field_id, lengths, total));
1427 }
1428 }
1429 columns.sort_by_key(|(field_id, _, _)| *field_id);
1430 let norms: Vec<super::chunk_map::DocLengthsColumn<'_>> = columns
1431 .iter()
1432 .map(
1433 |(field_id, lengths, total)| super::chunk_map::DocLengthsColumn {
1434 field_id: *field_id,
1435 lengths,
1436 total_tokens: *total,
1437 },
1438 )
1439 .collect();
1440 if !fields.is_empty() || !norms.is_empty() {
1441 let mut writer = dir.streaming_writer(&files.chunks).await?;
1442 super::chunk_map::write_chunk_maps(&mut *writer, &fields, &norms)?;
1443 writer.finish()?;
1444 }
1445 }
1446 let length_lookup = postings::LengthLookup {
1447 doc_lengths: &self.doc_field_lengths,
1448 num_indexed_fields: self.num_indexed_fields,
1449 field_to_slot: &self.field_to_slot,
1450 chunk_maps: &chunk_maps,
1451 };
1452
1453 let inverted_index = std::mem::take(&mut self.inverted_index);
1456 let term_interner = std::mem::replace(&mut self.term_interner, Rodeo::new());
1457 #[cfg(feature = "native")]
1458 let store_path = self.store_path.clone();
1459 #[cfg(feature = "native")]
1460 let num_compression_threads = self.config.num_compression_threads;
1461 let compression_level = self.config.compression_level;
1462 let dense_vectors = std::mem::take(&mut self.dense_vectors);
1463 let binary_dense_vectors = std::mem::take(&mut self.binary_dense_vectors);
1464 let mut sparse_vectors = std::mem::take(&mut self.sparse_vectors);
1465 let schema = &self.schema;
1466
1467 let mut term_dict_writer =
1470 super::OffsetWriter::new(dir.streaming_writer(&files.term_dict).await?);
1471 let mut postings_writer =
1472 super::OffsetWriter::new(dir.streaming_writer(&files.postings).await?);
1473 let mut store_writer = super::OffsetWriter::new(dir.streaming_writer(&files.store).await?);
1474 let mut vectors_writer = if !dense_vectors.is_empty() || !binary_dense_vectors.is_empty() {
1475 Some(super::OffsetWriter::new(
1476 dir.streaming_writer(&files.vectors).await?,
1477 ))
1478 } else {
1479 None
1480 };
1481 let mut sparse_writer = if !sparse_vectors.is_empty() {
1482 Some(super::OffsetWriter::new(
1483 dir.streaming_writer(&files.sparse).await?,
1484 ))
1485 } else {
1486 None
1487 };
1488 let mut fast_fields = std::mem::take(&mut self.fast_fields);
1489 let num_docs = self.next_doc_id;
1490 let mut fast_writer = if !fast_fields.is_empty() {
1491 Some(super::OffsetWriter::new(
1492 dir.streaming_writer(&files.fast).await?,
1493 ))
1494 } else {
1495 None
1496 };
1497
1498 #[cfg(feature = "native")]
1499 {
1500 if let Some(ref mut f) = self.posting_spill_file {
1501 f.flush()?;
1502 }
1503 let posting_spill_index = std::mem::take(&mut self.posting_spill_index);
1504 let mut spill_reader_opt = if !posting_spill_index.is_empty() {
1505 let spill_file = std::fs::File::open(&self.posting_spill_path)?;
1506 Some((std::io::BufReader::new(spill_file), posting_spill_index))
1507 } else {
1508 None
1509 };
1510
1511 let ((postings_result, store_result), ((vectors_result, sparse_result), fast_result)) =
1512 rayon::join(
1513 || {
1514 rayon::join(
1515 || {
1516 let spill_arg = spill_reader_opt.as_mut().map(|(r, idx)| {
1517 (
1518 r as &mut std::io::BufReader<std::fs::File>,
1519 idx as &postings::SpillIndex,
1520 )
1521 });
1522 postings::build_postings_streaming(
1523 inverted_index,
1524 term_interner,
1525 &position_offsets,
1526 &length_lookup,
1527 &mut term_dict_writer,
1528 &mut postings_writer,
1529 spill_arg,
1530 )
1531 },
1532 || {
1533 store::build_store_streaming(
1534 &store_path,
1535 num_compression_threads,
1536 compression_level,
1537 &mut store_writer,
1538 num_docs,
1539 )
1540 },
1541 )
1542 },
1543 || {
1544 rayon::join(
1545 || {
1546 rayon::join(
1547 || -> Result<()> {
1548 if let Some(ref mut w) = vectors_writer {
1549 dense::build_vectors_streaming(
1550 dense_vectors,
1551 binary_dense_vectors,
1552 schema,
1553 trained,
1554 w,
1555 )?;
1556 }
1557 Ok(())
1558 },
1559 || -> Result<()> {
1560 if let Some(ref mut w) = sparse_writer {
1561 sparse::build_sparse_streaming(
1562 &mut sparse_vectors,
1563 schema,
1564 w,
1565 )?;
1566 }
1567 Ok(())
1568 },
1569 )
1570 },
1571 || -> Result<()> {
1572 if let Some(ref mut w) = fast_writer {
1573 build_fast_fields_streaming(&mut fast_fields, num_docs, w)?;
1574 }
1575 Ok(())
1576 },
1577 )
1578 },
1579 );
1580 postings_result?;
1581 store_result?;
1582 vectors_result?;
1583 sparse_result?;
1584 fast_result?;
1585 }
1586
1587 #[cfg(not(feature = "native"))]
1588 {
1589 postings::build_postings_streaming(
1590 inverted_index,
1591 term_interner,
1592 &position_offsets,
1593 &length_lookup,
1594 &mut term_dict_writer,
1595 &mut postings_writer,
1596 )?;
1597 store::build_store_streaming_from_buffer(
1598 &self.store_buffer,
1599 compression_level,
1600 &mut store_writer,
1601 num_docs,
1602 )?;
1603 if let Some(ref mut w) = vectors_writer {
1604 dense::build_vectors_streaming(
1605 dense_vectors,
1606 binary_dense_vectors,
1607 schema,
1608 trained,
1609 w,
1610 )?;
1611 }
1612 if let Some(ref mut w) = sparse_writer {
1613 sparse::build_sparse_streaming(&mut sparse_vectors, schema, w)?;
1614 }
1615 if let Some(ref mut w) = fast_writer {
1616 build_fast_fields_streaming(&mut fast_fields, num_docs, w)?;
1617 }
1618 }
1619
1620 let term_dict_bytes = term_dict_writer.offset() as usize;
1621 let postings_bytes = postings_writer.offset() as usize;
1622 let store_bytes = store_writer.offset() as usize;
1623 let vectors_bytes = vectors_writer.as_ref().map_or(0, |w| w.offset() as usize);
1624 let sparse_bytes = sparse_writer.as_ref().map_or(0, |w| w.offset() as usize);
1625 let fast_bytes = fast_writer.as_ref().map_or(0, |w| w.offset() as usize);
1626
1627 term_dict_writer.finish()?;
1628 postings_writer.finish()?;
1629 store_writer.finish()?;
1630 if let Some(w) = vectors_writer {
1631 w.finish()?;
1632 }
1633 if let Some(w) = sparse_writer {
1634 w.finish()?;
1635 }
1636 if let Some(w) = fast_writer {
1637 w.finish()?;
1638 }
1639 drop(position_offsets);
1640 drop(sparse_vectors);
1641
1642 log::info!(
1643 "[segment_build] index={} docs={}: term_dict={}, postings={}, store={}, dense_vectors={}, sparse_vectors={}, fast_fields={}",
1644 self.schema.index_label(),
1645 num_docs,
1646 crate::format_bytes(term_dict_bytes as u64),
1647 crate::format_bytes(postings_bytes as u64),
1648 crate::format_bytes(store_bytes as u64),
1649 crate::format_bytes(vectors_bytes as u64),
1650 crate::format_bytes(sparse_bytes as u64),
1651 crate::format_bytes(fast_bytes as u64),
1652 );
1653
1654 let meta = SegmentMeta {
1655 id: segment_id.0,
1656 num_docs: self.next_doc_id,
1657 field_stats: self.field_stats.clone(),
1658 };
1659
1660 dir.write_durable(&files.meta, &meta.serialize()?).await?;
1665
1666 #[cfg(feature = "native")]
1668 {
1669 let _ = std::fs::remove_file(&self.store_path);
1670 }
1671
1672 Ok(meta)
1673 }
1674}
1675
1676fn build_fast_fields_streaming(
1678 fast_fields: &mut FxHashMap<u32, crate::structures::fast_field::FastFieldWriter>,
1679 num_docs: u32,
1680 writer: &mut dyn Write,
1681) -> Result<()> {
1682 use crate::structures::fast_field::{FastFieldTocEntry, write_fast_field_toc_and_footer};
1683
1684 if fast_fields.is_empty() {
1685 return Ok(());
1686 }
1687
1688 let mut field_ids: Vec<u32> = fast_fields.keys().copied().collect();
1690 field_ids.sort_unstable();
1691
1692 let mut toc_entries: Vec<FastFieldTocEntry> = Vec::with_capacity(field_ids.len());
1693 let mut current_offset = 0u64;
1694
1695 for &field_id in &field_ids {
1696 let ff = fast_fields.get_mut(&field_id).unwrap();
1697 ff.pad_to(num_docs);
1698
1699 let (mut toc, bytes_written) = ff.serialize(writer, current_offset)?;
1700 toc.field_id = field_id;
1701 current_offset += bytes_written;
1702 toc_entries.push(toc);
1703 }
1704
1705 let toc_offset = current_offset;
1707 write_fast_field_toc_and_footer(writer, toc_offset, &toc_entries)?;
1708
1709 Ok(())
1710}
1711
1712#[cfg(feature = "native")]
1713impl Drop for SegmentBuilder {
1714 fn drop(&mut self) {
1715 let _ = std::fs::remove_file(&self.store_path);
1716 if self.posting_spill_file.is_some() {
1717 let _ = std::fs::remove_file(&self.posting_spill_path);
1718 }
1719 }
1720}
1721
1722#[cfg(test)]
1723impl SegmentBuilder {
1724 fn positions_for_term(&self, field: Field, term: &str) -> Vec<u32> {
1726 let Some(spur) = self.term_interner.get(term) else {
1727 return Vec::new();
1728 };
1729 let key = TermKey {
1730 field: field.0,
1731 term: spur,
1732 };
1733 self.position_index
1734 .get(&key)
1735 .map(|b| {
1736 b.postings
1737 .iter()
1738 .flat_map(|(_, ps)| ps.iter().copied())
1739 .collect()
1740 })
1741 .unwrap_or_default()
1742 }
1743}
1744
1745#[cfg(test)]
1746mod tests {
1747 use super::*;
1748 use crate::dsl::SchemaBuilder;
1749
1750 fn builder_for(schema: Schema) -> SegmentBuilder {
1751 SegmentBuilder::new(Arc::new(schema), SegmentBuilderConfig::default()).unwrap()
1752 }
1753
1754 #[test]
1760 fn test_add_document_rejects_type_mismatched_field_value() {
1761 let mut sb = SchemaBuilder::default();
1762 let views = sb.add_u64_field("views", true, true);
1763 let mut builder = builder_for(sb.build());
1764
1765 let mut doc = Document::new();
1766 doc.add_text(views, "123");
1767 let err = builder
1768 .add_document(doc)
1769 .expect_err("schema-mismatched value must be rejected loudly, not silently unindexed");
1770 let msg = err.to_string();
1771 assert!(msg.contains("views"), "error must name the field: {msg}");
1772 assert!(
1773 msg.contains("u64"),
1774 "error must name the expected type: {msg}"
1775 );
1776 assert!(msg.contains("text"), "error must name the got type: {msg}");
1777
1778 assert_eq!(builder.num_docs(), 0);
1780
1781 let mut doc = Document::new();
1783 doc.add_u64(views, 123);
1784 builder.add_document(doc).unwrap();
1785 assert_eq!(builder.num_docs(), 1);
1786 }
1787
1788 #[test]
1794 fn test_add_document_rejects_bmp_sparse_dim_out_of_range() {
1795 use crate::structures::{SparseFormat, SparseVectorConfig};
1796
1797 let mut sb = SchemaBuilder::default();
1798 let config = SparseVectorConfig {
1799 format: SparseFormat::Bmp,
1800 dims: Some(100),
1801 ..Default::default()
1802 };
1803 let spv = sb.add_sparse_vector_field_with_config("spv", true, false, config);
1804 let mut builder = builder_for(sb.build());
1805
1806 let mut doc = Document::new();
1808 doc.add_sparse_vector(spv, vec![(50, 1.0)]);
1809 builder.add_document(doc).unwrap();
1810
1811 let mut doc = Document::new();
1813 doc.add_sparse_vector(spv, vec![(50, 1.0), (150, 2.0)]);
1814 let err = builder
1815 .add_document(doc)
1816 .expect_err("out-of-range BMP dim must be rejected, not silently unsearchable");
1817 let msg = err.to_string();
1818 assert!(msg.contains("spv"), "error must name the field: {msg}");
1819 assert!(msg.contains("150"), "error must name the dim_id: {msg}");
1820 assert!(
1821 msg.contains("100"),
1822 "error must name the configured dims: {msg}"
1823 );
1824 assert_eq!(
1825 builder.num_docs(),
1826 1,
1827 "rejected doc must not consume a doc id"
1828 );
1829 }
1830
1831 #[test]
1832 fn test_add_document_maxscore_sparse_dims_unbounded() {
1833 let mut sb = SchemaBuilder::default();
1836 let spv = sb.add_sparse_vector_field("spv", true, false);
1837 let mut builder = builder_for(sb.build());
1838
1839 let mut doc = Document::new();
1840 doc.add_sparse_vector(spv, vec![(3_000_000, 1.0)]);
1841 builder.add_document(doc).unwrap();
1842 }
1843
1844 #[test]
1851 fn test_position_element_ordinal_overflow_saturates_instead_of_wrapping() {
1852 use crate::dsl::PositionMode;
1853
1854 let mut sb = SchemaBuilder::default();
1855 let body = sb.add_text_field("body", true, false);
1856 sb.set_positions(body, PositionMode::Full);
1857 let mut builder = builder_for(sb.build());
1858
1859 let mut doc = Document::new();
1862 doc.add_text(body, "anchor");
1863 for _ in 0..4095 {
1864 doc.add_text(body, "filler");
1865 }
1866 doc.add_text(body, "needle");
1867 builder.add_document(doc).unwrap();
1868
1869 let positions = builder.positions_for_term(body, "needle");
1870 assert_eq!(positions.len(), 1);
1871 let encoded = positions[0];
1872 assert_ne!(
1873 encoded >> 20,
1874 0,
1875 "element ordinal 4096 must not alias element 0"
1876 );
1877 assert_eq!(
1878 encoded >> 20,
1879 4095,
1880 "overflowing element ordinal must saturate at 4095"
1881 );
1882 }
1883
1884 #[test]
1885 fn test_position_token_position_overflow_saturates_instead_of_bleeding() {
1886 use crate::dsl::PositionMode;
1887
1888 let mut sb = SchemaBuilder::default();
1889 let body = sb.add_text_field("body", true, false);
1890 sb.set_positions(body, PositionMode::Full);
1891 let mut builder = builder_for(sb.build());
1892
1893 let mut text = "w ".repeat(1 << 20);
1896 text.push_str("needle");
1897 let mut doc = Document::new();
1898 doc.add_text(body, text);
1899 builder.add_document(doc).unwrap();
1900
1901 let positions = builder.positions_for_term(body, "needle");
1902 assert_eq!(positions.len(), 1);
1903 let encoded = positions[0];
1904 assert_eq!(
1905 encoded >> 20,
1906 0,
1907 "token position overflow must not decode as a different element ordinal"
1908 );
1909 assert_eq!(
1910 encoded & 0xFFFFF,
1911 0xFFFFF,
1912 "overflowing token position must saturate at 2^20 - 1"
1913 );
1914 }
1915
1916 #[cfg(feature = "native")]
1923 #[tokio::test]
1924 async fn test_spill_mid_document_does_not_duplicate_postings() {
1925 use crate::directories::RamDirectory;
1926 use crate::structures::TERMINATED;
1927
1928 let mut sb = SchemaBuilder::default();
1929 let body = sb.add_text_field("body", true, false);
1930 let schema = Arc::new(sb.build());
1931 let mut builder =
1932 SegmentBuilder::new(Arc::clone(&schema), SegmentBuilderConfig::default()).unwrap();
1933
1934 for _ in 0..16383 {
1937 let mut doc = Document::new();
1938 doc.add_text(body, "hot");
1939 builder.add_document(doc).unwrap();
1940 }
1941
1942 let mut doc = Document::new();
1947 doc.add_text(body, "hot");
1948 doc.add_text(body, "hot");
1949 let boundary_doc = builder.add_document(doc).unwrap();
1950 assert_eq!(boundary_doc, 16383);
1951
1952 let dir = RamDirectory::new();
1953 let segment_id = crate::segment::SegmentId::new();
1954 builder.build(&dir, segment_id, None).await.unwrap();
1955
1956 let reader = crate::segment::SegmentReader::open(&dir, segment_id, schema, 16)
1957 .await
1958 .unwrap();
1959 let postings = reader
1960 .get_postings(body, b"hot")
1961 .await
1962 .unwrap()
1963 .expect("postings for 'hot'");
1964 assert_eq!(
1965 postings.doc_count(),
1966 16384,
1967 "each document must appear exactly once per term (spill-boundary duplicate)"
1968 );
1969
1970 let mut it = postings.iterator();
1973 let mut prev: Option<DocId> = None;
1974 let mut boundary_tf = 0u32;
1975 let mut d = it.doc();
1976 while d != TERMINATED {
1977 if let Some(p) = prev {
1978 assert!(p < d, "duplicate/unordered doc id {d} after {p}");
1979 }
1980 if d == boundary_doc {
1981 boundary_tf = it.term_freq();
1982 }
1983 prev = Some(d);
1984 d = it.advance();
1985 }
1986 assert_eq!(prev, Some(boundary_doc));
1987 assert_eq!(
1988 boundary_tf, 2,
1989 "boundary doc's term frequency must combine both values"
1990 );
1991 }
1992}