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
136fn field_type_name(field_type: &FieldType) -> &'static str {
138 match field_type {
139 FieldType::Text => "text",
140 FieldType::U64 => "u64",
141 FieldType::I64 => "i64",
142 FieldType::F64 => "f64",
143 FieldType::Bytes => "bytes",
144 FieldType::SparseVector => "sparse_vector",
145 FieldType::DenseVector => "dense_vector",
146 FieldType::Json => "json",
147 FieldType::BinaryDenseVector => "binary_dense_vector",
148 }
149}
150
151fn field_value_type_name(value: &FieldValue) -> &'static str {
153 match value {
154 FieldValue::Text(_) => "text",
155 FieldValue::U64(_) => "u64",
156 FieldValue::I64(_) => "i64",
157 FieldValue::F64(_) => "f64",
158 FieldValue::Bytes(_) => "bytes",
159 FieldValue::SparseVector(_) => "sparse_vector",
160 FieldValue::DenseVector(_) => "dense_vector",
161 FieldValue::Json(_) => "json",
162 FieldValue::BinaryDenseVector(_) => "binary_dense_vector",
163 }
164}
165
166pub(crate) fn validate_vector_value_counts(doc: &Document, schema: &Schema) -> Result<()> {
173 let mut vector_values_per_field: FxHashMap<u32, usize> = FxHashMap::default();
174
175 for (field, value) in doc.field_values() {
176 let Some(entry) = schema.get_field_entry(*field) else {
177 continue;
178 };
179
180 let consumes_vector_ordinal = match (&entry.field_type, value) {
181 (FieldType::DenseVector, FieldValue::DenseVector(_))
182 | (FieldType::BinaryDenseVector, FieldValue::BinaryDenseVector(_)) => {
183 entry.indexed || entry.stored
184 }
185 (FieldType::SparseVector, FieldValue::SparseVector(_)) => entry.indexed || entry.fast,
186 _ => false,
187 };
188 if !consumes_vector_ordinal {
189 continue;
190 }
191
192 let count = vector_values_per_field.entry(field.0).or_insert(0);
193 *count += 1;
194 if *count > MAX_VECTOR_VALUES_PER_FIELD {
195 return Err(crate::Error::Document(format!(
196 "field '{}' (id {}) has more than {} vector values in one document",
197 entry.name, field.0, MAX_VECTOR_VALUES_PER_FIELD
198 )));
199 }
200 }
201 Ok(())
202}
203
204pub struct SegmentBuilder {
211 schema: Arc<Schema>,
212 config: SegmentBuilderConfig,
213 tokenizers: FxHashMap<Field, BoxedTokenizer>,
214 tokenizer_hint_fields: FxHashMap<Field, Field>,
217 tokenizer_hint_buffer: String,
219 unhinted_dynamic_docs: u64,
222
223 term_interner: Rodeo,
225
226 inverted_index: HashMap<TermKey, PostingListBuilder>,
228
229 #[cfg(feature = "native")]
231 posting_spill_file: Option<BufWriter<File>>,
232 #[cfg(feature = "native")]
233 posting_spill_path: PathBuf,
234 #[cfg(feature = "native")]
236 posting_spill_index: HashMap<TermKey, Vec<(u64, u32)>>,
237 #[cfg(feature = "native")]
238 posting_spill_offset: u64,
239
240 #[cfg(feature = "native")]
242 store_file: BufWriter<File>,
243 #[cfg(feature = "native")]
244 store_path: PathBuf,
245 #[cfg(not(feature = "native"))]
246 store_buffer: Vec<u8>,
247
248 next_doc_id: DocId,
250
251 field_stats: FxHashMap<u32, FieldStats>,
253
254 doc_field_lengths: Vec<u32>,
258 row_stat_lengths: Vec<u64>,
260 num_indexed_fields: usize,
261 field_to_slot: FxHashMap<u32, usize>,
262
263 local_tf_buffer: FxHashMap<Spur, u32>,
266
267 local_positions: FxHashMap<Spur, Vec<u32>>,
270
271 local_position_terms: Vec<Spur>,
274
275 token_buffer: String,
277
278 numeric_buffer: String,
280
281 dense_vectors: FxHashMap<u32, DenseVectorBuilder>,
284
285 binary_dense_vectors: FxHashMap<u32, BinaryDenseVectorBuilder>,
287
288 sparse_vectors: FxHashMap<u32, SparseVectorBuilder>,
291
292 position_index: HashMap<TermKey, PositionPostingListBuilder>,
295
296 position_enabled_fields: FxHashMap<u32, Option<crate::dsl::PositionMode>>,
298
299 current_element_ordinal: FxHashMap<u32, u32>,
301
302 chunk_maps: FxHashMap<u32, super::chunk_map::ChunkMapBuilder>,
305
306 position_saturation_warned: bool,
309
310 estimated_memory: usize,
312
313 doc_serialize_buffer: Vec<u8>,
315
316 fast_fields: FxHashMap<u32, crate::structures::fast_field::FastFieldWriter>,
318}
319
320impl SegmentBuilder {
321 pub fn new(schema: Arc<Schema>, config: SegmentBuilderConfig) -> Result<Self> {
323 #[cfg(feature = "native")]
324 let (store_file, store_path, spill_path) = {
325 let segment_id = uuid::Uuid::new_v4();
326 let store_path = config
327 .temp_dir
328 .join(format!("summa_store_{}.tmp", segment_id));
329 let store_file = BufWriter::with_capacity(
330 STORE_BUFFER_SIZE,
331 OpenOptions::new()
332 .create(true)
333 .write(true)
334 .truncate(true)
335 .open(&store_path)?,
336 );
337 let spill_path = config
338 .temp_dir
339 .join(format!("summa_spill_{}.tmp", segment_id));
340 (store_file, store_path, spill_path)
341 };
342
343 let registry = crate::tokenizer::TokenizerRegistry::new();
345 let mut num_indexed_fields = 0;
346 let mut field_to_slot = FxHashMap::default();
347 let mut position_enabled_fields = FxHashMap::default();
348 let mut tokenizers = FxHashMap::default();
349 let mut tokenizer_hint_fields = FxHashMap::default();
350 for (field, entry) in schema.fields() {
351 if (entry.indexed && entry.field_type == FieldType::Text)
352 || entry.field_type == FieldType::SparseVector
353 {
354 field_to_slot.insert(field.0, num_indexed_fields);
355 num_indexed_fields += 1;
356 if entry.positions.is_some() {
357 position_enabled_fields.insert(field.0, entry.positions);
358 }
359 if let Some(ref tok_name) = entry.tokenizer
360 && let Some(tokenizer) = registry.get(tok_name)
361 {
362 tokenizers.insert(field, tokenizer);
363 }
364 if let Some(hint_field) = schema.tokenizer_hint_field(field) {
365 tokenizer_hint_fields.insert(field, hint_field);
366 }
367 }
368 }
369
370 use crate::structures::fast_field::{FastFieldColumnType, FastFieldWriter};
372 let mut fast_fields = FxHashMap::default();
373 for (field, entry) in schema.fields() {
374 if entry.fast {
375 let writer = if entry.multi {
376 match entry.field_type {
377 FieldType::U64 => {
378 FastFieldWriter::new_numeric_multi(FastFieldColumnType::U64)
379 }
380 FieldType::I64 => {
381 FastFieldWriter::new_numeric_multi(FastFieldColumnType::I64)
382 }
383 FieldType::F64 => {
384 FastFieldWriter::new_numeric_multi(FastFieldColumnType::F64)
385 }
386 FieldType::Text => FastFieldWriter::new_text_multi(),
387 _ => continue,
388 }
389 } else {
390 match entry.field_type {
391 FieldType::U64 => FastFieldWriter::new_numeric(FastFieldColumnType::U64),
392 FieldType::I64 => FastFieldWriter::new_numeric(FastFieldColumnType::I64),
393 FieldType::F64 => FastFieldWriter::new_numeric(FastFieldColumnType::F64),
394 FieldType::Text => FastFieldWriter::new_text(),
395 _ => continue,
396 }
397 };
398 fast_fields.insert(field.0, writer);
399 }
400 }
401
402 Ok(Self {
403 schema,
404 tokenizers,
405 tokenizer_hint_fields,
406 tokenizer_hint_buffer: String::new(),
407 unhinted_dynamic_docs: 0,
408 term_interner: Rodeo::new(),
409 inverted_index: HashMap::with_capacity(config.posting_map_capacity),
410 #[cfg(feature = "native")]
411 posting_spill_file: None,
412 #[cfg(feature = "native")]
413 posting_spill_path: spill_path,
414 #[cfg(feature = "native")]
415 posting_spill_index: HashMap::new(),
416 #[cfg(feature = "native")]
417 posting_spill_offset: 0,
418 #[cfg(feature = "native")]
419 store_file,
420 #[cfg(feature = "native")]
421 store_path,
422 #[cfg(not(feature = "native"))]
423 store_buffer: Vec::with_capacity(STORE_BUFFER_SIZE),
424 next_doc_id: 0,
425 field_stats: FxHashMap::default(),
426 chunk_maps: FxHashMap::default(),
427 doc_field_lengths: Vec::new(),
428 row_stat_lengths: Vec::new(),
429 num_indexed_fields,
430 field_to_slot,
431 local_tf_buffer: FxHashMap::default(),
432 local_positions: FxHashMap::default(),
433 local_position_terms: Vec::new(),
434 token_buffer: String::with_capacity(64),
435 numeric_buffer: String::with_capacity(32),
436 config,
437 dense_vectors: FxHashMap::default(),
438 binary_dense_vectors: FxHashMap::default(),
439 sparse_vectors: FxHashMap::default(),
440 position_index: HashMap::new(),
441 position_enabled_fields,
442 current_element_ordinal: FxHashMap::default(),
443 position_saturation_warned: false,
444 estimated_memory: 0,
445 doc_serialize_buffer: Vec::with_capacity(256),
446 fast_fields,
447 })
448 }
449
450 pub fn set_tokenizer(&mut self, field: Field, tokenizer: BoxedTokenizer) {
451 self.tokenizers.insert(field, tokenizer);
452 }
453
454 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 sparse_vectors_bytes += builder.keys.capacity() * std::mem::size_of::<(DocId, u16)>();
597 for postings in builder.postings.values() {
598 sparse_vectors_bytes += postings.capacity() * sparse_entry_size + vec_overhead;
599 }
600 let inner_entry_size = size_of::<u32>() + vec_overhead + fxhashmap_entry_overhead;
602 sparse_vectors_bytes += builder.postings.len() * inner_entry_size;
603 }
604 let outer_sparse_entry_size =
606 size_of::<u32>() + size_of::<SparseVectorBuilder>() + fxhashmap_entry_overhead;
607 sparse_vectors_bytes += self.sparse_vectors.len() * outer_sparse_entry_size;
608
609 let mut position_index_bytes: usize = 0;
611 for pos_builder in self.position_index.values() {
612 for (_, positions) in &pos_builder.postings {
613 position_index_bytes += positions.capacity() * size_of::<u32>() + vec_overhead;
614 }
615 let pos_entry_size = size_of::<DocId>() + vec_overhead;
617 position_index_bytes += pos_builder.postings.capacity() * pos_entry_size;
618 }
619 let pos_index_entry_size =
621 term_key_size + size_of::<PositionPostingListBuilder>() + hashmap_entry_base_overhead;
622 position_index_bytes += self.position_index.len() * pos_index_entry_size;
623
624 let estimated_memory_bytes = postings_bytes
625 + index_overhead_bytes
626 + interner_bytes
627 + field_lengths_bytes
628 + dense_vectors_bytes
629 + local_tf_buffer_bytes
630 + sparse_vectors_bytes
631 + position_index_bytes;
632
633 let memory_breakdown = MemoryBreakdown {
634 postings_bytes,
635 index_overhead_bytes,
636 interner_bytes,
637 field_lengths_bytes,
638 dense_vectors_bytes,
639 dense_vector_count,
640 sparse_vectors_bytes,
641 position_index_bytes,
642 };
643
644 SegmentBuilderStats {
645 num_docs: self.next_doc_id,
646 unique_terms: self.inverted_index.len(),
647 postings_in_memory,
648 interned_strings: self.term_interner.len(),
649 doc_field_lengths_size: self.doc_field_lengths.len(),
650 estimated_memory_bytes,
651 memory_breakdown,
652 unhinted_dynamic_docs: self.unhinted_dynamic_docs,
653 }
654 }
655
656 fn validate_document_against_schema(&self, doc: &Document) -> Result<()> {
669 validate_vector_value_counts(doc, &self.schema)?;
670
671 for (field, value) in doc.field_values() {
672 let Some(entry) = self.schema.get_field_entry(*field) else {
673 continue;
674 };
675
676 if !matches!(
679 &entry.field_type,
680 FieldType::DenseVector | FieldType::BinaryDenseVector
681 ) && !entry.indexed
682 && !entry.fast
683 {
684 continue;
685 }
686
687 match (&entry.field_type, value) {
688 (FieldType::SparseVector, FieldValue::SparseVector(entries)) => {
689 if let Some(config) = entry.sparse_vector_config.as_ref() {
690 let dims = config.dims.or_else(|| {
691 (config.format == crate::structures::SparseFormat::Bmp).then_some(105879)
692 });
693 if let Some(&(dim_id, _)) = entries.iter().find(|&&(dim_id, _)| {
694 dims.is_some_and(|dims| dim_id >= dims)
695 || dim_id > config.index_size.max_value()
696 }) {
697 let bound = match dims {
698 Some(dims) => format!("dims {dims} (exclusive), maximum input ID {}", config.index_size.max_value()),
699 None => format!("maximum input ID {}", config.index_size.max_value()),
700 };
701 return Err(crate::Error::Schema(format!(
702 "sparse vector for field '{}' contains dim_id {} outside configured dimension bounds: {}",
703 entry.name, dim_id, bound
704 )));
705 }
706 }
707 }
708 (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.row_stat_lengths
747 .resize(base_idx + self.num_indexed_fields, 0);
748 self.estimated_memory +=
749 self.num_indexed_fields * (std::mem::size_of::<u32>() + std::mem::size_of::<u64>());
750
751 self.current_element_ordinal.clear();
753
754 for (field, value) in doc.field_values() {
755 let Some(entry) = self.schema.get_field_entry(*field) else {
756 continue;
757 };
758
759 if !matches!(
762 &entry.field_type,
763 FieldType::DenseVector | FieldType::BinaryDenseVector
764 ) && !entry.indexed
765 && !entry.fast
766 {
767 continue;
768 }
769
770 match (&entry.field_type, value) {
771 (FieldType::Text, FieldValue::Text(text)) => {
772 if entry.indexed && entry.chunked {
773 let field_id = field.0;
777 let element_ordinal = self.next_element_ordinal(field_id);
778 let ordinal = u16::try_from(element_ordinal).map_err(|_| {
779 crate::Error::Document(format!(
780 "chunked field {field_id} has more than {} chunk values in one document",
781 u16::MAX as usize + 1
782 ))
783 })?;
784 let hinted = self.resolve_tokenizer_hint(*field, &doc, element_ordinal);
785 let vid = self
786 .chunk_maps
787 .get(&field.0)
788 .map_or(0usize, |map| map.len());
789 let vid = u32::try_from(vid).map_err(|_| {
790 crate::Error::Document(format!(
791 "chunked field {field_id} exceeds u32::MAX chunks in one segment"
792 ))
793 })?;
794 let token_count = self.index_text_field(*field, vid, text, 0, hinted)?;
798 self.chunk_maps.entry(field.0).or_default().push(
799 doc_id,
800 ordinal,
801 token_count,
802 )?;
803 self.estimated_memory += 8;
804
805 let stats = self.field_stats.entry(field.0).or_default();
808 stats.total_tokens += token_count as u64;
809 stats.doc_count += 1;
810 let slot = self.field_to_slot[&field.0];
811 let exact = &mut self.row_stat_lengths[base_idx + slot];
812 *exact = (*exact)
813 .max(1)
814 .checked_add(u64::from(token_count))
815 .ok_or_else(|| {
816 crate::Error::Document("per-row token count overflow".into())
817 })?;
818 } else if entry.indexed {
819 let element_ordinal = self.next_element_ordinal(field.0);
820 let hinted = self.resolve_tokenizer_hint(*field, &doc, element_ordinal);
821 let token_count =
822 self.index_text_field(*field, doc_id, text, element_ordinal, hinted)?;
823
824 let stats = self.field_stats.entry(field.0).or_default();
825 stats.total_tokens += token_count as u64;
826 if element_ordinal == 0 {
827 stats.doc_count += 1;
828 }
829
830 if let Some(&slot) = self.field_to_slot.get(&field.0) {
831 let len = &mut self.doc_field_lengths[base_idx + slot];
834 *len = len.saturating_add(token_count);
835 let exact = &mut self.row_stat_lengths[base_idx + slot];
836 *exact = (*exact)
837 .max(1)
838 .checked_add(u64::from(token_count))
839 .ok_or_else(|| {
840 crate::Error::Document("per-row token count overflow".into())
841 })?;
842 }
843 }
844
845 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
847 ff.add_text(doc_id, text);
848 }
849 }
850 (FieldType::U64, FieldValue::U64(v)) => {
851 if entry.indexed {
852 self.index_numeric_field(*field, doc_id, *v)?;
853 }
854 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
855 ff.add_u64(doc_id, *v);
856 }
857 }
858 (FieldType::I64, FieldValue::I64(v)) => {
859 if entry.indexed {
860 self.index_numeric_field(*field, doc_id, *v as u64)?;
861 }
862 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
863 ff.add_i64(doc_id, *v);
864 }
865 }
866 (FieldType::F64, FieldValue::F64(v)) => {
867 if entry.indexed {
868 self.index_numeric_field(*field, doc_id, v.to_bits())?;
869 }
870 if let Some(ff) = self.fast_fields.get_mut(&field.0) {
871 ff.add_f64(doc_id, *v);
872 }
873 }
874 (FieldType::DenseVector, FieldValue::DenseVector(vec))
875 if entry.indexed || entry.stored =>
876 {
877 let ordinal = self.next_vector_ordinal(field.0)?;
878 self.index_dense_vector_field(*field, doc_id, ordinal, vec)?;
879 }
880 (FieldType::BinaryDenseVector, FieldValue::BinaryDenseVector(bytes))
881 if entry.indexed || entry.stored =>
882 {
883 let ordinal = self.next_vector_ordinal(field.0)?;
884 self.index_binary_dense_vector_field(*field, doc_id, ordinal, bytes)?;
885 }
886 (FieldType::SparseVector, FieldValue::SparseVector(entries)) => {
887 let ordinal = self.next_vector_ordinal(field.0)?;
888 if let Some(&slot) = self.field_to_slot.get(&field.0) {
889 self.row_stat_lengths[base_idx + slot] += 1;
890 }
891 self.index_sparse_vector_field(*field, doc_id, ordinal, entries)?;
892 }
893 _ => {}
898 }
899 }
900
901 self.write_document_to_store(&doc)?;
903
904 Ok(doc_id)
905 }
906
907 fn index_text_field(
916 &mut self,
917 field: Field,
918 doc_id: DocId,
919 text: &str,
920 element_ordinal: u32,
921 hinted: bool,
922 ) -> Result<u32> {
923 use crate::dsl::PositionMode;
924
925 let field_id = field.0;
926 let position_mode = self
927 .position_enabled_fields
928 .get(&field_id)
929 .copied()
930 .flatten();
931
932 let encoded_ordinal = if position_mode.is_some_and(|m| m.tracks_ordinal())
935 && element_ordinal > MAX_POSITION_ELEMENT_ORDINAL
936 {
937 self.warn_position_saturation(
938 "element ordinal",
939 element_ordinal,
940 MAX_POSITION_ELEMENT_ORDINAL,
941 );
942 MAX_POSITION_ELEMENT_ORDINAL
943 } else {
944 element_ordinal
945 };
946
947 self.local_tf_buffer.clear();
951 for term in self.local_position_terms.drain(..) {
955 self.local_positions.get_mut(&term).unwrap().clear();
956 }
957
958 let mut token_position = 0u32;
959
960 let custom_tokens = self.tokenizers.get(&field).map(|t| {
964 if hinted {
965 let hint = (!self.tokenizer_hint_buffer.is_empty())
966 .then_some(self.tokenizer_hint_buffer.as_str());
967 t.tokenize_with(text, hint, crate::tokenizer::Purpose::Index)
968 } else {
969 t.tokenize(text)
970 }
971 });
972
973 if let Some(tokens) = custom_tokens {
974 for token in &tokens {
976 let term_spur = if let Some(spur) = self.term_interner.get(&token.text) {
977 spur
978 } else {
979 let spur = self.term_interner.get_or_intern(&token.text);
980 self.estimated_memory += token.text.len() + INTERN_OVERHEAD;
981 spur
982 };
983 *self.local_tf_buffer.entry(term_spur).or_insert(0) += 1;
984
985 if let Some(mode) = position_mode {
986 let encoded_pos = match mode {
987 PositionMode::Ordinal => encoded_ordinal << 20,
988 PositionMode::TokenPosition => token.position,
989 PositionMode::Full => {
990 (encoded_ordinal << 20) | self.saturate_token_position(token.position)
991 }
992 };
993 let positions = self.local_positions.entry(term_spur).or_default();
994 if positions.is_empty() {
995 self.local_position_terms.push(term_spur);
996 }
997 positions.push(encoded_pos);
998 }
999 }
1000 token_position = tokens.iter().filter(|t| !t.variant).count() as u32;
1003 } else {
1004 for word in text.split_whitespace() {
1006 self.token_buffer.clear();
1007 for c in word.chars() {
1008 if c.is_alphanumeric() {
1009 for lc in c.to_lowercase() {
1010 self.token_buffer.push(lc);
1011 }
1012 }
1013 }
1014
1015 if self.token_buffer.is_empty() {
1016 continue;
1017 }
1018
1019 let term_spur = if let Some(spur) = self.term_interner.get(&self.token_buffer) {
1020 spur
1021 } else {
1022 let spur = self.term_interner.get_or_intern(&self.token_buffer);
1023 self.estimated_memory += self.token_buffer.len() + INTERN_OVERHEAD;
1024 spur
1025 };
1026 *self.local_tf_buffer.entry(term_spur).or_insert(0) += 1;
1027
1028 if let Some(mode) = position_mode {
1029 let encoded_pos = match mode {
1030 PositionMode::Ordinal => encoded_ordinal << 20,
1031 PositionMode::TokenPosition => token_position,
1032 PositionMode::Full => {
1033 (encoded_ordinal << 20) | self.saturate_token_position(token_position)
1034 }
1035 };
1036 let positions = self.local_positions.entry(term_spur).or_default();
1037 if positions.is_empty() {
1038 self.local_position_terms.push(term_spur);
1039 }
1040 positions.push(encoded_pos);
1041 }
1042
1043 token_position += 1;
1044 }
1045 }
1046
1047 for (&term_spur, &tf) in &self.local_tf_buffer {
1050 let term_key = TermKey {
1051 field: field_id,
1052 term: term_spur,
1053 };
1054
1055 match self.inverted_index.entry(term_key) {
1056 hashbrown::hash_map::Entry::Occupied(mut o) => {
1057 o.get_mut().add(doc_id, tf);
1058 self.estimated_memory += size_of::<CompactPosting>();
1059 #[cfg(feature = "native")]
1061 if o.get().should_spill() {
1062 use byteorder::{LittleEndian, WriteBytesExt};
1063
1064 let builder = o.get_mut();
1065 let count = builder.postings.len() as u32;
1066 let offset = self.posting_spill_offset;
1067
1068 let spill_file = if let Some(ref mut f) = self.posting_spill_file {
1070 f
1071 } else {
1072 self.posting_spill_file = Some(BufWriter::with_capacity(
1073 256 * 1024,
1074 OpenOptions::new()
1075 .create(true)
1076 .write(true)
1077 .truncate(true)
1078 .open(&self.posting_spill_path)?,
1079 ));
1080 self.posting_spill_file.as_mut().unwrap()
1081 };
1082 for p in &builder.postings {
1083 spill_file.write_u32::<LittleEndian>(p.doc_id)?;
1084 spill_file.write_u16::<LittleEndian>(p.term_freq)?;
1085 }
1086 self.posting_spill_offset += count as u64 * 6;
1087 self.posting_spill_index
1088 .entry(term_key)
1089 .or_default()
1090 .push((offset, count));
1091
1092 let freed = builder.postings.len() * size_of::<CompactPosting>();
1093 builder.spilled_count += count;
1094 builder.postings.clear();
1095 self.estimated_memory -= freed;
1096 }
1097 }
1098 hashbrown::hash_map::Entry::Vacant(v) => {
1099 let mut posting = PostingListBuilder::new();
1100 posting.add(doc_id, tf);
1101 v.insert(posting);
1102 self.estimated_memory += size_of::<CompactPosting>() + NEW_TERM_OVERHEAD;
1103 }
1104 }
1105
1106 if position_mode.is_some()
1107 && let Some(positions) = self.local_positions.get(&term_spur)
1108 {
1109 match self.position_index.entry(term_key) {
1110 hashbrown::hash_map::Entry::Occupied(mut o) => {
1111 for &pos in positions {
1112 o.get_mut().add_position(doc_id, pos);
1113 }
1114 self.estimated_memory += positions.len() * size_of::<u32>();
1115 }
1116 hashbrown::hash_map::Entry::Vacant(v) => {
1117 let mut pos_posting = PositionPostingListBuilder::new();
1118 for &pos in positions {
1119 pos_posting.add_position(doc_id, pos);
1120 }
1121 self.estimated_memory +=
1122 positions.len() * size_of::<u32>() + NEW_POS_TERM_OVERHEAD;
1123 v.insert(pos_posting);
1124 }
1125 }
1126 }
1127 }
1128
1129 Ok(token_position)
1130 }
1131
1132 #[inline]
1135 fn saturate_token_position(&mut self, token_position: u32) -> u32 {
1136 if token_position > MAX_TOKEN_POSITION {
1137 self.warn_position_saturation("token position", token_position, MAX_TOKEN_POSITION);
1138 MAX_TOKEN_POSITION
1139 } else {
1140 token_position
1141 }
1142 }
1143
1144 #[cold]
1146 fn warn_position_saturation(&mut self, what: &str, value: u32, max: u32) {
1147 if !self.position_saturation_warned {
1148 self.position_saturation_warned = true;
1149 log::warn!(
1150 "[segment_builder] index={} {what} {value} exceeds the position-encoding limit {max}; \
1151 saturating — phrase/ordinal matching degrades for the overflowing \
1152 elements/tokens instead of corrupting other documents' matches \
1153 (further occurrences in this segment are not logged)",
1154 self.schema.index_label()
1155 );
1156 }
1157 }
1158
1159 fn index_numeric_field(&mut self, field: Field, doc_id: DocId, value: u64) -> Result<()> {
1160 use std::fmt::Write;
1161
1162 self.numeric_buffer.clear();
1163 write!(self.numeric_buffer, "__num_{}", value).unwrap();
1164 let term_spur = if let Some(spur) = self.term_interner.get(&self.numeric_buffer) {
1165 spur
1166 } else {
1167 let spur = self.term_interner.get_or_intern(&self.numeric_buffer);
1168 self.estimated_memory += self.numeric_buffer.len() + INTERN_OVERHEAD;
1169 spur
1170 };
1171
1172 let term_key = TermKey {
1173 field: field.0,
1174 term: term_spur,
1175 };
1176
1177 match self.inverted_index.entry(term_key) {
1178 hashbrown::hash_map::Entry::Occupied(mut o) => {
1179 o.get_mut().add(doc_id, 1);
1180 self.estimated_memory += size_of::<CompactPosting>();
1181 }
1182 hashbrown::hash_map::Entry::Vacant(v) => {
1183 let mut posting = PostingListBuilder::new();
1184 posting.add(doc_id, 1);
1185 v.insert(posting);
1186 self.estimated_memory += size_of::<CompactPosting>() + NEW_TERM_OVERHEAD;
1187 }
1188 }
1189
1190 Ok(())
1191 }
1192
1193 fn index_dense_vector_field(
1195 &mut self,
1196 field: Field,
1197 doc_id: DocId,
1198 ordinal: u16,
1199 vector: &[f32],
1200 ) -> Result<()> {
1201 let dim = vector.len();
1202 let expected_dim = self
1203 .schema
1204 .get_field_entry(field)
1205 .and_then(|entry| entry.dense_vector_config.as_ref())
1206 .map(|config| config.dim)
1207 .ok_or_else(|| crate::Error::Schema("DenseVector field missing config".to_string()))?;
1208 if dim != expected_dim {
1209 return Err(crate::Error::Schema(format!(
1210 "Dense vector dimension mismatch: schema expects {}, got {}",
1211 expected_dim, dim
1212 )));
1213 }
1214 if let Some((index, value)) = vector
1215 .iter()
1216 .enumerate()
1217 .find(|(_, value)| !value.is_finite())
1218 {
1219 return Err(crate::Error::Document(format!(
1220 "dense vector contains non-finite value {value} at index {index}"
1221 )));
1222 }
1223
1224 let builder = self
1225 .dense_vectors
1226 .entry(field.0)
1227 .or_insert_with(|| DenseVectorBuilder::new(dim));
1228
1229 if builder.dim != dim && builder.len() > 0 {
1231 return Err(crate::Error::Schema(format!(
1232 "Dense vector dimension mismatch: expected {}, got {}",
1233 builder.dim, dim
1234 )));
1235 }
1236
1237 builder.add(doc_id, ordinal, vector);
1238
1239 self.estimated_memory += std::mem::size_of_val(vector) + size_of::<(DocId, u16)>();
1240
1241 Ok(())
1242 }
1243
1244 fn index_binary_dense_vector_field(
1246 &mut self,
1247 field: Field,
1248 doc_id: DocId,
1249 ordinal: u16,
1250 bytes: &[u8],
1251 ) -> Result<()> {
1252 let dim_bits = self
1253 .schema
1254 .get_field_entry(field)
1255 .and_then(|e| e.binary_dense_vector_config.as_ref())
1256 .map(|c| c.dim)
1257 .ok_or_else(|| {
1258 crate::Error::Schema("BinaryDenseVector field missing config".to_string())
1259 })?;
1260
1261 let expected_byte_len = dim_bits.div_ceil(8);
1262 if dim_bits == 0 || !dim_bits.is_multiple_of(8) {
1263 return Err(crate::Error::Schema(format!(
1264 "Binary vector dimension must be a positive multiple of 8, got {dim_bits}"
1265 )));
1266 }
1267 if bytes.len() != expected_byte_len {
1268 return Err(crate::Error::Schema(format!(
1269 "Binary vector byte length mismatch: expected {} (dim={}), got {}",
1270 expected_byte_len,
1271 dim_bits,
1272 bytes.len()
1273 )));
1274 }
1275
1276 let builder = self
1277 .binary_dense_vectors
1278 .entry(field.0)
1279 .or_insert_with(|| BinaryDenseVectorBuilder::new(dim_bits));
1280
1281 builder.add(doc_id, ordinal, bytes);
1282 self.estimated_memory += bytes.len() + size_of::<(DocId, u16)>();
1283
1284 Ok(())
1285 }
1286
1287 fn index_sparse_vector_field(
1297 &mut self,
1298 field: Field,
1299 doc_id: DocId,
1300 ordinal: u16,
1301 entries: &[(u32, f32)],
1302 ) -> Result<()> {
1303 if let Some((index, (_, weight))) = entries
1304 .iter()
1305 .enumerate()
1306 .find(|(_, (_, weight))| !weight.is_finite())
1307 {
1308 return Err(crate::Error::Document(format!(
1309 "sparse vector contains non-finite weight {weight} at index {index}"
1310 )));
1311 }
1312 let (weight_threshold, doc_mass, min_terms) = self
1313 .schema
1314 .get_field_entry(field)
1315 .and_then(|entry| entry.sparse_vector_config.as_ref())
1316 .map(|config| (config.weight_threshold, config.doc_mass, config.min_terms))
1317 .unwrap_or((0.0, None, 0));
1318
1319 let builder = self
1320 .sparse_vectors
1321 .entry(field.0)
1322 .or_insert_with(SparseVectorBuilder::new);
1323
1324 builder.inc_vector_count(doc_id, ordinal);
1325
1326 let mass_cutoff = match doc_mass {
1330 Some(mass) if mass < 1.0 && entries.len() > min_terms => {
1331 let mut weights: Vec<f32> = entries
1332 .iter()
1333 .map(|&(_, w)| w.abs())
1334 .filter(|w| *w >= weight_threshold)
1335 .collect();
1336 weights.sort_unstable_by(|a, b| b.total_cmp(a));
1337 let total: f64 = weights.iter().map(|&w| w as f64).sum();
1338 let target = total * mass as f64;
1339 let mut cumulative = 0.0f64;
1340 let mut cutoff = 0.0f32;
1341 for &w in &weights {
1342 if cumulative >= target {
1343 break;
1344 }
1345 cumulative += w as f64;
1346 cutoff = w;
1347 }
1348 cutoff
1349 }
1350 _ => 0.0,
1351 };
1352
1353 for &(dim_id, weight) in entries {
1354 if weight.abs() < weight_threshold || weight.abs() < mass_cutoff {
1356 continue;
1357 }
1358
1359 let is_new_dim = !builder.postings.contains_key(&dim_id);
1360 builder.add(dim_id, doc_id, ordinal, weight);
1361 self.estimated_memory += size_of::<(DocId, u16, f32)>();
1362 if is_new_dim {
1363 self.estimated_memory += size_of::<u32>() + size_of::<Vec<(DocId, u16, f32)>>() + 8; }
1366 }
1367
1368 Ok(())
1369 }
1370
1371 fn write_document_to_store(&mut self, doc: &Document) -> Result<()> {
1373 use byteorder::{LittleEndian, WriteBytesExt};
1374
1375 super::store::serialize_document_into(doc, &self.schema, &mut self.doc_serialize_buffer)?;
1376
1377 #[cfg(feature = "native")]
1378 {
1379 self.store_file
1380 .write_u32::<LittleEndian>(self.doc_serialize_buffer.len() as u32)?;
1381 self.store_file.write_all(&self.doc_serialize_buffer)?;
1382 }
1383 #[cfg(not(feature = "native"))]
1384 {
1385 self.store_buffer
1386 .write_u32::<LittleEndian>(self.doc_serialize_buffer.len() as u32)?;
1387 self.store_buffer.write_all(&self.doc_serialize_buffer)?;
1388 self.estimated_memory += size_of::<u32>() + self.doc_serialize_buffer.len();
1392 }
1393
1394 Ok(())
1395 }
1396
1397 pub async fn build<D: Directory + DirectoryWriter>(
1403 mut self,
1404 dir: &D,
1405 segment_id: SegmentId,
1406 trained: Option<&super::TrainedVectorStructures>,
1407 ) -> Result<SegmentMeta> {
1408 #[cfg(feature = "native")]
1410 self.store_file.flush()?;
1411
1412 let files = SegmentFiles::new(segment_id.0);
1413
1414 if self.schema.fields().any(|(_, e)| {
1417 (e.indexed && e.field_type == FieldType::Text)
1418 || e.field_type == FieldType::SparseVector
1419 }) {
1420 use crate::structures::fast_field::{
1421 FastFieldColumnType, FastFieldWriter, write_fast_field_toc_and_footer,
1422 };
1423 let mut writer =
1424 super::OffsetWriter::new(dir.streaming_writer(&files.row_stats).await?);
1425 let mut entries = Vec::new();
1426 for (field, entry) in self.schema.fields() {
1427 if !((entry.indexed && entry.field_type == FieldType::Text)
1428 || entry.field_type == FieldType::SparseVector)
1429 {
1430 continue;
1431 }
1432 let slot = self.field_to_slot[&field.0];
1433 let mut column = FastFieldWriter::new_numeric(FastFieldColumnType::U64);
1434 for doc in 0..self.next_doc_id {
1435 column.add_u64(
1436 doc,
1437 self.row_stat_lengths[doc as usize * self.num_indexed_fields + slot],
1438 );
1439 }
1440 let offset = writer.offset();
1441 let (mut toc, _) = column.serialize(&mut writer, offset)?;
1442 toc.field_id = field.0;
1443 entries.push(toc);
1444 }
1445 let offset = writer.offset();
1446 write_fast_field_toc_and_footer(&mut writer, offset, &entries)?;
1447 writer.finish()?;
1448 }
1449 self.row_stat_lengths.clear();
1450 self.row_stat_lengths.shrink_to_fit();
1451
1452 let position_index = std::mem::take(&mut self.position_index);
1454 let position_offsets = if !position_index.is_empty() {
1455 let mut pos_writer = dir.streaming_writer(&files.positions).await?;
1456 let offsets = postings::build_positions_streaming(
1457 position_index,
1458 &self.term_interner,
1459 &mut *pos_writer,
1460 self.config.posting_codec,
1461 self.config.compact_text,
1462 )?;
1463 pos_writer.finish()?;
1464 offsets
1465 } else {
1466 FxHashMap::default()
1467 };
1468
1469 let mut chunk_maps = std::mem::take(&mut self.chunk_maps);
1472 for (field, entry) in self.schema.fields() {
1476 if entry.field_type != FieldType::Text
1477 || !entry.indexed
1478 || !entry.reorder
1479 || entry.chunked
1480 {
1481 continue;
1482 }
1483 let slot = self.field_to_slot[&field.0];
1484 let mut map = super::chunk_map::ChunkMapBuilder::default();
1485 map.set_document_units(true);
1486 for doc in 0..self.next_doc_id {
1487 let length = self.doc_field_lengths[doc as usize * self.num_indexed_fields + slot];
1488 map.push(doc, 0, length)?;
1489 }
1490 chunk_maps.insert(field.0, map);
1491 }
1492 {
1493 let mut fields: Vec<(u32, &super::chunk_map::ChunkMapBuilder)> = chunk_maps
1494 .iter()
1495 .filter(|(_, map)| !map.is_empty())
1496 .map(|(field_id, map)| (*field_id, map))
1497 .collect();
1498 fields.sort_by_key(|(field_id, _)| *field_id);
1499 let num_docs = self.next_doc_id as usize;
1500 let mut columns: Vec<(u32, Vec<u16>, u64)> = Vec::new();
1501 for (&field_id, &slot) in &self.field_to_slot {
1502 if self
1503 .schema
1504 .get_field_entry(crate::dsl::Field(field_id))
1505 .is_some_and(|entry| entry.chunked || entry.reorder)
1506 {
1507 continue;
1508 }
1509 let mut total = 0u64;
1510 let lengths: Vec<u16> = (0..num_docs)
1511 .map(|doc| {
1512 let len = self.doc_field_lengths[doc * self.num_indexed_fields + slot];
1513 total += u64::from(len);
1514 len.min(super::chunk_map::MAX_CHUNK_LENGTH) as u16
1515 })
1516 .collect();
1517 if total > 0 {
1518 columns.push((field_id, lengths, total));
1519 }
1520 }
1521 columns.sort_by_key(|(field_id, _, _)| *field_id);
1522 let norms: Vec<super::chunk_map::DocLengthsColumn<'_>> = columns
1523 .iter()
1524 .map(
1525 |(field_id, lengths, total)| super::chunk_map::DocLengthsColumn {
1526 field_id: *field_id,
1527 lengths,
1528 total_tokens: *total,
1529 },
1530 )
1531 .collect();
1532 if !fields.is_empty() || !norms.is_empty() {
1533 let mut writer = dir.streaming_writer(&files.chunks).await?;
1534 super::chunk_map::write_chunk_maps_with_norms(
1535 &mut *writer,
1536 &fields,
1537 &norms,
1538 self.config.quantized_norms,
1539 )?;
1540 writer.finish()?;
1541 }
1542 }
1543 let length_lookup = postings::LengthLookup {
1544 quantized_norms: self.config.quantized_norms,
1545 doc_lengths: &self.doc_field_lengths,
1546 num_indexed_fields: self.num_indexed_fields,
1547 field_to_slot: &self.field_to_slot,
1548 chunk_maps: &chunk_maps,
1549 };
1550
1551 let inverted_index = std::mem::take(&mut self.inverted_index);
1554 let term_interner = std::mem::replace(&mut self.term_interner, Rodeo::new());
1555 #[cfg(feature = "native")]
1556 let store_path = self.store_path.clone();
1557 #[cfg(feature = "native")]
1558 let num_compression_threads = self.config.num_compression_threads;
1559 let compression_level = self.config.compression_level;
1560 let posting_config = &self.config;
1561 let dense_vectors = std::mem::take(&mut self.dense_vectors);
1562 let binary_dense_vectors = std::mem::take(&mut self.binary_dense_vectors);
1563 let mut sparse_vectors = std::mem::take(&mut self.sparse_vectors);
1564 let schema = &self.schema;
1565
1566 let mut term_dict_writer =
1569 super::OffsetWriter::new(dir.streaming_writer(&files.term_dict).await?);
1570 let mut postings_writer =
1571 super::OffsetWriter::new(dir.streaming_writer(&files.postings).await?);
1572 let mut store_writer = super::OffsetWriter::new(dir.streaming_writer(&files.store).await?);
1573 let mut vectors_writer = if !dense_vectors.is_empty() || !binary_dense_vectors.is_empty() {
1574 Some(super::OffsetWriter::new(
1575 dir.streaming_writer(&files.vectors).await?,
1576 ))
1577 } else {
1578 None
1579 };
1580 let mut sparse_writer = if !sparse_vectors.is_empty() {
1581 Some(super::OffsetWriter::new(
1582 dir.streaming_writer(&files.sparse).await?,
1583 ))
1584 } else {
1585 None
1586 };
1587 let has_seismic = sparse_vectors.iter().any(|(&field, builder)| {
1588 !builder.is_empty()
1589 && schema
1590 .get_field_entry(crate::dsl::Field(field))
1591 .and_then(|entry| entry.sparse_vector_config.as_ref())
1592 .is_some_and(|config| config.format == crate::structures::SparseFormat::Seismic)
1593 });
1594 let mut sparse_partitions = if has_seismic {
1595 Some(
1596 super::sparse_partitions::SparsePartitionWriters::create(dir, &files, |_| true)
1597 .await?,
1598 )
1599 } else {
1600 None
1601 };
1602 let mut fast_fields = std::mem::take(&mut self.fast_fields);
1603 let num_docs = self.next_doc_id;
1604 let mut fast_writer = if !fast_fields.is_empty() {
1605 Some(super::OffsetWriter::new(
1606 dir.streaming_writer(&files.fast).await?,
1607 ))
1608 } else {
1609 None
1610 };
1611
1612 #[cfg(feature = "native")]
1613 {
1614 if let Some(ref mut f) = self.posting_spill_file {
1615 f.flush()?;
1616 }
1617 let posting_spill_index = std::mem::take(&mut self.posting_spill_index);
1618 let mut spill_reader_opt = if !posting_spill_index.is_empty() {
1619 let spill_file = std::fs::File::open(&self.posting_spill_path)?;
1620 Some((std::io::BufReader::new(spill_file), posting_spill_index))
1621 } else {
1622 None
1623 };
1624
1625 let ((postings_result, store_result), ((vectors_result, sparse_result), fast_result)) =
1626 rayon::join(
1627 || {
1628 rayon::join(
1629 || {
1630 let spill_arg = spill_reader_opt.as_mut().map(|(r, idx)| {
1631 (
1632 r as &mut std::io::BufReader<std::fs::File>,
1633 idx as &postings::SpillIndex,
1634 )
1635 });
1636 postings::build_postings_streaming(
1637 inverted_index,
1638 term_interner,
1639 &position_offsets,
1640 &length_lookup,
1641 &mut term_dict_writer,
1642 &mut postings_writer,
1643 posting_config,
1644 spill_arg,
1645 )
1646 },
1647 || {
1648 store::build_store_streaming(
1649 &store_path,
1650 num_compression_threads,
1651 compression_level,
1652 &mut store_writer,
1653 num_docs,
1654 )
1655 },
1656 )
1657 },
1658 || {
1659 rayon::join(
1660 || {
1661 rayon::join(
1662 || -> Result<()> {
1663 if let Some(ref mut w) = vectors_writer {
1664 dense::build_vectors_streaming(
1665 dense_vectors,
1666 binary_dense_vectors,
1667 schema,
1668 trained,
1669 w,
1670 )?;
1671 }
1672 Ok(())
1673 },
1674 || -> Result<()> {
1675 if let Some(ref mut w) = sparse_writer {
1676 sparse::build_sparse_streaming(
1677 &mut sparse_vectors,
1678 schema,
1679 w,
1680 sparse_partitions.as_mut(),
1681 )?;
1682 }
1683 Ok(())
1684 },
1685 )
1686 },
1687 || -> Result<()> {
1688 if let Some(ref mut w) = fast_writer {
1689 build_fast_fields_streaming(&mut fast_fields, num_docs, w)?;
1690 }
1691 Ok(())
1692 },
1693 )
1694 },
1695 );
1696 postings_result?;
1697 store_result?;
1698 vectors_result?;
1699 sparse_result?;
1700 fast_result?;
1701 }
1702
1703 #[cfg(not(feature = "native"))]
1704 {
1705 postings::build_postings_streaming(
1706 inverted_index,
1707 term_interner,
1708 &position_offsets,
1709 &length_lookup,
1710 &mut term_dict_writer,
1711 &mut postings_writer,
1712 posting_config,
1713 )?;
1714 store::build_store_streaming_from_buffer(
1715 &self.store_buffer,
1716 compression_level,
1717 &mut store_writer,
1718 num_docs,
1719 )?;
1720 if let Some(ref mut w) = vectors_writer {
1721 dense::build_vectors_streaming(
1722 dense_vectors,
1723 binary_dense_vectors,
1724 schema,
1725 trained,
1726 w,
1727 )?;
1728 }
1729 if let Some(ref mut w) = sparse_writer {
1730 sparse::build_sparse_streaming(
1731 &mut sparse_vectors,
1732 schema,
1733 w,
1734 sparse_partitions.as_mut(),
1735 )?;
1736 }
1737 if let Some(ref mut w) = fast_writer {
1738 build_fast_fields_streaming(&mut fast_fields, num_docs, w)?;
1739 }
1740 }
1741
1742 let term_dict_bytes = term_dict_writer.offset() as usize;
1743 let postings_bytes = postings_writer.offset() as usize;
1744 let store_bytes = store_writer.offset() as usize;
1745 let vectors_bytes = vectors_writer.as_ref().map_or(0, |w| w.offset() as usize);
1746 let sparse_partition_bytes = match sparse_partitions {
1747 Some(partitions) => partitions.finish()?,
1748 None => 0,
1749 };
1750 let sparse_bytes =
1751 sparse_writer.as_ref().map_or(0, |w| w.offset() as usize) + sparse_partition_bytes;
1752 let fast_bytes = fast_writer.as_ref().map_or(0, |w| w.offset() as usize);
1753
1754 term_dict_writer.finish()?;
1755 postings_writer.finish()?;
1756 store_writer.finish()?;
1757 if let Some(w) = vectors_writer {
1758 w.finish()?;
1759 }
1760 if let Some(w) = sparse_writer {
1761 w.finish()?;
1762 }
1763 if let Some(w) = fast_writer {
1764 w.finish()?;
1765 }
1766 drop(position_offsets);
1767 drop(sparse_vectors);
1768
1769 log::info!(
1770 "[segment_build] index={} docs={}: term_dict={}, postings={}, store={}, dense_vectors={}, sparse_vectors={}, fast_fields={}",
1771 self.schema.index_label(),
1772 num_docs,
1773 crate::format_bytes(term_dict_bytes as u64),
1774 crate::format_bytes(postings_bytes as u64),
1775 crate::format_bytes(store_bytes as u64),
1776 crate::format_bytes(vectors_bytes as u64),
1777 crate::format_bytes(sparse_bytes as u64),
1778 crate::format_bytes(fast_bytes as u64),
1779 );
1780
1781 let meta = SegmentMeta {
1782 id: segment_id.0,
1783 num_docs: self.next_doc_id,
1784 field_stats: self.field_stats.clone(),
1785 };
1786
1787 dir.write_durable(&files.meta, &meta.serialize()?).await?;
1792
1793 #[cfg(feature = "native")]
1795 {
1796 let _ = std::fs::remove_file(&self.store_path);
1797 }
1798
1799 Ok(meta)
1800 }
1801}
1802
1803fn build_fast_fields_streaming(
1805 fast_fields: &mut FxHashMap<u32, crate::structures::fast_field::FastFieldWriter>,
1806 num_docs: u32,
1807 writer: &mut dyn Write,
1808) -> Result<()> {
1809 use crate::structures::fast_field::{FastFieldTocEntry, write_fast_field_toc_and_footer};
1810
1811 if fast_fields.is_empty() {
1812 return Ok(());
1813 }
1814
1815 let mut field_ids: Vec<u32> = fast_fields.keys().copied().collect();
1817 field_ids.sort_unstable();
1818
1819 let mut toc_entries: Vec<FastFieldTocEntry> = Vec::with_capacity(field_ids.len());
1820 let mut current_offset = 0u64;
1821
1822 for &field_id in &field_ids {
1823 let ff = fast_fields.get_mut(&field_id).unwrap();
1824 ff.pad_to(num_docs);
1825
1826 let (mut toc, bytes_written) = ff.serialize(writer, current_offset)?;
1827 toc.field_id = field_id;
1828 current_offset += bytes_written;
1829 toc_entries.push(toc);
1830 }
1831
1832 let toc_offset = current_offset;
1834 write_fast_field_toc_and_footer(writer, toc_offset, &toc_entries)?;
1835
1836 Ok(())
1837}
1838
1839#[cfg(feature = "native")]
1840impl Drop for SegmentBuilder {
1841 fn drop(&mut self) {
1842 let _ = std::fs::remove_file(&self.store_path);
1843 if self.posting_spill_file.is_some() {
1844 let _ = std::fs::remove_file(&self.posting_spill_path);
1845 }
1846 }
1847}
1848
1849#[cfg(test)]
1850impl SegmentBuilder {
1851 fn positions_for_term(&self, field: Field, term: &str) -> Vec<u32> {
1853 let Some(spur) = self.term_interner.get(term) else {
1854 return Vec::new();
1855 };
1856 let key = TermKey {
1857 field: field.0,
1858 term: spur,
1859 };
1860 self.position_index
1861 .get(&key)
1862 .map(|b| {
1863 b.postings
1864 .iter()
1865 .flat_map(|(_, ps)| ps.iter().copied())
1866 .collect()
1867 })
1868 .unwrap_or_default()
1869 }
1870}
1871
1872#[cfg(test)]
1873mod tests {
1874 use super::*;
1875 use crate::dsl::SchemaBuilder;
1876
1877 fn builder_for(schema: Schema) -> SegmentBuilder {
1878 SegmentBuilder::new(Arc::new(schema), SegmentBuilderConfig::default()).unwrap()
1879 }
1880
1881 #[test]
1882 fn position_scratch_resets_only_touched_terms_across_fields_and_chunks() {
1883 use crate::dsl::PositionMode;
1884 use crate::tokenizer::SimpleTokenizer;
1885
1886 for custom in [false, true] {
1887 for mode in [
1888 PositionMode::Ordinal,
1889 PositionMode::TokenPosition,
1890 PositionMode::Full,
1891 ] {
1892 let mut schema = SchemaBuilder::default();
1893 let body = schema.add_text_field("body", true, false);
1894 schema.set_positions(body, mode);
1895 let other = schema.add_text_field("other", true, false);
1896 schema.set_positions(other, PositionMode::TokenPosition);
1897 let no_positions = schema.add_text_field("no_positions", true, false);
1898 let mut builder = builder_for(schema.build());
1899 if custom {
1900 builder.set_tokenizer(body, Box::new(SimpleTokenizer));
1901 }
1902
1903 for doc in 0..2_000 {
1906 builder
1907 .index_text_field(
1908 body,
1909 doc,
1910 &format!("unique{doc} anchor anchor"),
1911 0,
1912 false,
1913 )
1914 .unwrap();
1915 assert_eq!(builder.local_position_terms.len(), 2);
1916 }
1917 assert_eq!(builder.local_positions.len(), 2_001);
1918 let anchor = builder.term_interner.get("anchor").unwrap();
1919 let capacity = builder.local_positions[&anchor].capacity();
1920
1921 builder
1922 .index_text_field(no_positions, 2_000, "anchor plain", 0, false)
1923 .unwrap();
1924 assert!(builder.local_position_terms.is_empty());
1925 assert!(builder.local_positions[&anchor].is_empty());
1926 assert_eq!(builder.local_positions[&anchor].capacity(), capacity);
1927 builder
1928 .index_text_field(other, 2_000, "anchor", 0, false)
1929 .unwrap();
1930 builder.index_text_field(body, 2_000, "", 0, false).unwrap();
1931 assert!(builder.local_position_terms.is_empty());
1932 builder
1933 .index_text_field(body, 2_000, "anchor", 1, false)
1934 .unwrap();
1935 builder
1936 .index_text_field(body, 2_001, "anchor", 2, false)
1937 .unwrap();
1938
1939 assert_eq!(builder.positions_for_term(other, "anchor"), vec![0]);
1940 let positions = builder.positions_for_term(body, "anchor");
1941 assert_eq!(
1942 positions.len(),
1943 4_002,
1944 "positions leaked between chunks or fields"
1945 );
1946 let expected = match mode {
1947 PositionMode::TokenPosition => [0, 0],
1948 PositionMode::Ordinal | PositionMode::Full => [1 << 20, 2 << 20],
1949 };
1950 assert_eq!(&positions[4_000..], &expected);
1951 }
1952 }
1953 }
1954
1955 #[test]
1961 fn test_add_document_rejects_type_mismatched_field_value() {
1962 let mut sb = SchemaBuilder::default();
1963 let views = sb.add_u64_field("views", true, true);
1964 let mut builder = builder_for(sb.build());
1965
1966 let mut doc = Document::new();
1967 doc.add_text(views, "123");
1968 let err = builder
1969 .add_document(doc)
1970 .expect_err("schema-mismatched value must be rejected loudly, not silently unindexed");
1971 let msg = err.to_string();
1972 assert!(msg.contains("views"), "error must name the field: {msg}");
1973 assert!(
1974 msg.contains("u64"),
1975 "error must name the expected type: {msg}"
1976 );
1977 assert!(msg.contains("text"), "error must name the got type: {msg}");
1978
1979 assert_eq!(builder.num_docs(), 0);
1981
1982 let mut doc = Document::new();
1984 doc.add_u64(views, 123);
1985 builder.add_document(doc).unwrap();
1986 assert_eq!(builder.num_docs(), 1);
1987 }
1988
1989 #[test]
1994 fn test_add_document_rejects_sparse_dimension_outside_declared_bound() {
1995 use crate::structures::{SparseFormat, SparseVectorConfig};
1996
1997 let mut sb = SchemaBuilder::default();
1998 let config = SparseVectorConfig {
1999 format: SparseFormat::Seismic,
2000 dims: Some(100),
2001 ..Default::default()
2002 };
2003 let spv = sb.add_sparse_vector_field_with_config("spv", true, false, config);
2004 let mut builder = builder_for(sb.build());
2005
2006 let mut doc = Document::new();
2008 doc.add_sparse_vector(spv, vec![(50, 1.0)]);
2009 builder.add_document(doc).unwrap();
2010
2011 let mut doc = Document::new();
2013 doc.add_sparse_vector(spv, vec![(50, 1.0), (150, 2.0)]);
2014 let err = builder
2015 .add_document(doc)
2016 .expect_err("out-of-range sparse dimension must be rejected");
2017 let msg = err.to_string();
2018 assert!(msg.contains("spv"), "error must name the field: {msg}");
2019 assert!(msg.contains("150"), "error must name the dim_id: {msg}");
2020 assert!(
2021 msg.contains("100"),
2022 "error must name the configured dims: {msg}"
2023 );
2024 assert_eq!(
2025 builder.num_docs(),
2026 1,
2027 "rejected doc must not consume a doc id"
2028 );
2029 }
2030
2031 #[test]
2032 fn test_add_document_sparse_dims_without_declared_bound() {
2033 let mut sb = SchemaBuilder::default();
2036 let spv = sb.add_sparse_vector_field_with_config(
2037 "spv",
2038 true,
2039 false,
2040 crate::structures::SparseVectorConfig {
2041 format: crate::structures::SparseFormat::MaxScore,
2042 index_size: crate::structures::IndexSize::U32,
2043 ..Default::default()
2044 },
2045 );
2046 let mut builder = builder_for(sb.build());
2047
2048 let mut doc = Document::new();
2049 doc.add_sparse_vector(spv, vec![(3_000_000, 1.0)]);
2050 builder.add_document(doc).unwrap();
2051 }
2052
2053 #[test]
2060 fn test_position_element_ordinal_overflow_saturates_instead_of_wrapping() {
2061 use crate::dsl::PositionMode;
2062
2063 let mut sb = SchemaBuilder::default();
2064 let body = sb.add_text_field("body", true, false);
2065 sb.set_positions(body, PositionMode::Full);
2066 let mut builder = builder_for(sb.build());
2067
2068 let mut doc = Document::new();
2071 doc.add_text(body, "anchor");
2072 for _ in 0..4095 {
2073 doc.add_text(body, "filler");
2074 }
2075 doc.add_text(body, "needle");
2076 builder.add_document(doc).unwrap();
2077
2078 let positions = builder.positions_for_term(body, "needle");
2079 assert_eq!(positions.len(), 1);
2080 let encoded = positions[0];
2081 assert_ne!(
2082 encoded >> 20,
2083 0,
2084 "element ordinal 4096 must not alias element 0"
2085 );
2086 assert_eq!(
2087 encoded >> 20,
2088 4095,
2089 "overflowing element ordinal must saturate at 4095"
2090 );
2091 }
2092
2093 #[test]
2094 fn test_position_token_position_overflow_saturates_instead_of_bleeding() {
2095 use crate::dsl::PositionMode;
2096
2097 let mut sb = SchemaBuilder::default();
2098 let body = sb.add_text_field("body", true, false);
2099 sb.set_positions(body, PositionMode::Full);
2100 let mut builder = builder_for(sb.build());
2101
2102 let mut text = "w ".repeat(1 << 20);
2105 text.push_str("needle");
2106 let mut doc = Document::new();
2107 doc.add_text(body, text);
2108 builder.add_document(doc).unwrap();
2109
2110 let positions = builder.positions_for_term(body, "needle");
2111 assert_eq!(positions.len(), 1);
2112 let encoded = positions[0];
2113 assert_eq!(
2114 encoded >> 20,
2115 0,
2116 "token position overflow must not decode as a different element ordinal"
2117 );
2118 assert_eq!(
2119 encoded & 0xFFFFF,
2120 0xFFFFF,
2121 "overflowing token position must saturate at 2^20 - 1"
2122 );
2123 }
2124
2125 #[cfg(feature = "native")]
2132 #[tokio::test]
2133 async fn test_spill_mid_document_does_not_duplicate_postings() {
2134 use crate::directories::RamDirectory;
2135 use crate::structures::TERMINATED;
2136
2137 let mut sb = SchemaBuilder::default();
2138 let body = sb.add_text_field("body", true, false);
2139 let schema = Arc::new(sb.build());
2140 let mut builder =
2141 SegmentBuilder::new(Arc::clone(&schema), SegmentBuilderConfig::default()).unwrap();
2142
2143 for _ in 0..16383 {
2146 let mut doc = Document::new();
2147 doc.add_text(body, "hot");
2148 builder.add_document(doc).unwrap();
2149 }
2150
2151 let mut doc = Document::new();
2156 doc.add_text(body, "hot");
2157 doc.add_text(body, "hot");
2158 let boundary_doc = builder.add_document(doc).unwrap();
2159 assert_eq!(boundary_doc, 16383);
2160
2161 let dir = RamDirectory::new();
2162 let segment_id = crate::segment::SegmentId::new();
2163 builder.build(&dir, segment_id, None).await.unwrap();
2164
2165 let reader = crate::segment::SegmentReader::open(&dir, segment_id, schema, 16)
2166 .await
2167 .unwrap();
2168 let postings = reader
2169 .get_postings(body, b"hot")
2170 .await
2171 .unwrap()
2172 .expect("postings for 'hot'");
2173 assert_eq!(
2174 postings.doc_count(),
2175 16384,
2176 "each document must appear exactly once per term (spill-boundary duplicate)"
2177 );
2178
2179 let mut it = postings.iterator();
2182 let mut prev: Option<DocId> = None;
2183 let mut boundary_tf = 0u32;
2184 let mut d = it.doc();
2185 while d != TERMINATED {
2186 if let Some(p) = prev {
2187 assert!(p < d, "duplicate/unordered doc id {d} after {p}");
2188 }
2189 if d == boundary_doc {
2190 boundary_tf = it.term_freq();
2191 }
2192 prev = Some(d);
2193 d = it.advance();
2194 }
2195 assert_eq!(prev, Some(boundary_doc));
2196 assert_eq!(
2197 boundary_tf, 2,
2198 "boundary doc's term frequency must combine both values"
2199 );
2200 }
2201}