1use std::io::{self, Write};
34
35use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt};
36use rustc_hash::FxHashMap;
37
38use crate::DocId;
39use crate::directories::OwnedBytes;
40
41const MAGIC: u32 = 0x4B4E_4843; const VERSION: u32 = 5;
43const DOCUMENT_VERSION: u32 = 4;
44const ADDRESSED_VERSION: u32 = 3;
45const HEADER_SIZE: usize = 12;
46const TOC_ENTRY_SIZE_V1: usize = 24;
47const TOC_ENTRY_SIZE: usize = 28;
48const KIND_CHUNK_MAP: u32 = 0;
49const KIND_DOC_LENGTHS: u32 = 1;
50const KIND_ADDRESSED_CHUNK_MAP: u32 = 2;
51const KIND_DOCUMENT_MAP: u32 = 3;
52const KIND_BYTE_NORMS: u32 = 4;
53
54pub const MAX_CHUNK_LENGTH: u32 = u16::MAX as u32;
56
57#[derive(Debug, Default, Clone)]
59pub struct ChunkMapBuilder {
60 doc_ids: Vec<DocId>,
61 ordinals: Vec<u16>,
62 lengths: Vec<u16>,
63 total_tokens: u64,
64 document_units: bool,
65}
66
67impl ChunkMapBuilder {
68 #[cfg(feature = "native")]
69 pub(crate) fn with_capacity(count: usize) -> Self {
70 Self {
71 doc_ids: Vec::with_capacity(count),
72 ordinals: Vec::with_capacity(count),
73 lengths: Vec::with_capacity(count),
74 total_tokens: 0,
75 document_units: false,
76 }
77 }
78
79 #[cfg(any(feature = "native", feature = "wasm", test))]
81 pub(crate) fn set_document_units(&mut self, document_units: bool) {
82 self.document_units = document_units;
83 }
84
85 fn section_kind(&self) -> u32 {
86 if self.document_units {
87 KIND_DOCUMENT_MAP
88 } else if self.logically_ordered() {
89 KIND_CHUNK_MAP
90 } else {
91 KIND_ADDRESSED_CHUNK_MAP
92 }
93 }
94
95 #[cfg(feature = "native")]
96 pub(crate) fn set_total_tokens(&mut self, total: u64) {
97 self.total_tokens = total;
98 }
99
100 pub fn len(&self) -> usize {
102 self.doc_ids.len()
103 }
104
105 pub fn is_empty(&self) -> bool {
106 self.doc_ids.is_empty()
107 }
108
109 pub fn push(&mut self, doc_id: DocId, ordinal: u16, token_count: u32) -> io::Result<u32> {
111 let vid = u32::try_from(self.doc_ids.len()).map_err(|_| {
112 io::Error::new(
113 io::ErrorKind::InvalidData,
114 "chunked text field exceeds u32::MAX chunks in one segment",
115 )
116 })?;
117 self.doc_ids.push(doc_id);
118 self.ordinals.push(ordinal);
119 self.lengths.push(token_count.min(MAX_CHUNK_LENGTH) as u16);
120 self.total_tokens += u64::from(token_count);
121 Ok(vid)
122 }
123
124 pub fn estimated_bytes(&self) -> usize {
126 self.doc_ids.capacity() * 4 + self.ordinals.capacity() * 2 + self.lengths.capacity() * 2
127 }
128
129 fn logically_ordered(&self) -> bool {
130 self.doc_ids.iter().zip(&self.ordinals).is_sorted()
131 && self
132 .doc_ids
133 .iter()
134 .zip(&self.ordinals)
135 .zip(self.doc_ids.iter().zip(&self.ordinals).skip(1))
136 .all(|(a, b)| a != b)
137 }
138
139 fn section_bytes(&self) -> u64 {
140 self.doc_ids.len() as u64
141 * if self.section_kind() == KIND_CHUNK_MAP {
142 8
143 } else {
144 12
145 }
146 }
147
148 pub fn length(&self, vid: u32) -> u32 {
150 self.lengths
151 .get(vid as usize)
152 .map_or(0, |len| u32::from(*len))
153 }
154}
155
156pub struct DocLengthsColumn<'a> {
158 pub field_id: u32,
159 pub lengths: &'a [u16],
161 pub total_tokens: u64,
163}
164
165pub fn write_chunk_maps<W: Write + ?Sized>(
171 writer: &mut W,
172 fields: &[(u32, &ChunkMapBuilder)],
173 norms: &[DocLengthsColumn<'_>],
174) -> io::Result<u64> {
175 write_chunk_maps_with_norms(writer, fields, norms, false)
176}
177
178enum NormColumn<'a> {
181 New(&'a DocLengthsColumn<'a>, bool),
182 #[cfg(feature = "native")]
183 Encoded(u32, &'a DocLengths),
184}
185
186impl NormColumn<'_> {
187 fn field_id(&self) -> u32 {
188 match self {
189 Self::New(column, _) => column.field_id,
190 #[cfg(feature = "native")]
191 Self::Encoded(field, _) => *field,
192 }
193 }
194 fn quantized(&self) -> bool {
195 match self {
196 Self::New(_, quantized) => *quantized,
197 #[cfg(feature = "native")]
198 Self::Encoded(_, lengths) => lengths.is_quantized(),
199 }
200 }
201 fn count(&self) -> usize {
202 match self {
203 Self::New(column, _) => column.lengths.len(),
204 #[cfg(feature = "native")]
205 Self::Encoded(_, lengths) => lengths.num_docs() as usize,
206 }
207 }
208 fn total_tokens(&self) -> u64 {
209 match self {
210 Self::New(column, _) => column.total_tokens,
211 #[cfg(feature = "native")]
212 Self::Encoded(_, lengths) => lengths.total_tokens(),
213 }
214 }
215 fn write(&self, writer: &mut (impl Write + ?Sized)) -> io::Result<()> {
216 match self {
217 #[cfg(feature = "native")]
218 Self::Encoded(_, lengths) => writer.write_all(lengths.length_bytes()),
219 Self::New(column, quantized) => {
220 for &length in column.lengths {
221 if *quantized {
222 writer.write_all(&[super::norms::encode(u32::from(length))])?;
223 } else {
224 writer.write_u16::<LittleEndian>(length)?;
225 }
226 }
227 Ok(())
228 }
229 }
230 }
231}
232
233#[cfg(feature = "native")]
234pub(crate) fn write_chunk_maps_with_copied_norms<W: Write + ?Sized>(
235 writer: &mut W,
236 fields: &[(u32, &ChunkMapBuilder)],
237 norms: &[(u32, &DocLengths)],
238) -> io::Result<u64> {
239 let columns: Vec<_> = norms
240 .iter()
241 .map(|&(field, lengths)| NormColumn::Encoded(field, lengths))
242 .collect();
243 write_chunk_map_columns(
244 writer,
245 fields,
246 &columns,
247 norms.iter().any(|(_, lengths)| lengths.is_quantized()),
248 )
249}
250
251pub(crate) fn write_chunk_maps_with_norms<W: Write + ?Sized>(
252 writer: &mut W,
253 fields: &[(u32, &ChunkMapBuilder)],
254 norms: &[DocLengthsColumn<'_>],
255 quantized: bool,
256) -> io::Result<u64> {
257 let columns: Vec<_> = norms
258 .iter()
259 .map(|column| NormColumn::New(column, quantized))
260 .collect();
261 write_chunk_map_columns(writer, fields, &columns, quantized)
262}
263
264#[cfg(feature = "native")]
267pub(crate) fn write_chunk_maps_with_norm_policy<W: Write + ?Sized>(
268 writer: &mut W,
269 fields: &[(u32, &ChunkMapBuilder)],
270 norms: &[DocLengthsColumn<'_>],
271 quantized: impl Fn(u32) -> bool,
272) -> io::Result<u64> {
273 let columns: Vec<_> = norms
274 .iter()
275 .map(|column| NormColumn::New(column, quantized(column.field_id)))
276 .collect();
277 write_chunk_map_columns(
278 writer,
279 fields,
280 &columns,
281 columns.iter().any(NormColumn::quantized),
282 )
283}
284
285fn write_chunk_map_columns<W: Write + ?Sized>(
286 writer: &mut W,
287 fields: &[(u32, &ChunkMapBuilder)],
288 norms: &[NormColumn<'_>],
289 quantized_version: bool,
290) -> io::Result<u64> {
291 let sections = fields.len() + norms.len();
292 let mut offset = (HEADER_SIZE + TOC_ENTRY_SIZE * sections) as u64;
293 writer.write_u32::<LittleEndian>(MAGIC)?;
294 writer.write_u32::<LittleEndian>(if quantized_version {
295 VERSION
296 } else if fields.iter().any(|(_, m)| m.document_units) {
297 DOCUMENT_VERSION
298 } else {
299 ADDRESSED_VERSION
300 })?;
301 writer.write_u32::<LittleEndian>(sections as u32)?;
302 for (field_id, map) in fields {
303 writer.write_u32::<LittleEndian>(*field_id)?;
304 writer.write_u32::<LittleEndian>(map.section_kind())?;
305 writer.write_u32::<LittleEndian>(map.len() as u32)?;
306 writer.write_u64::<LittleEndian>(map.total_tokens)?;
307 writer.write_u64::<LittleEndian>(offset)?;
308 offset += map.section_bytes();
309 }
310 for column in norms {
311 writer.write_u32::<LittleEndian>(column.field_id())?;
312 writer.write_u32::<LittleEndian>(if column.quantized() {
313 KIND_BYTE_NORMS
314 } else {
315 KIND_DOC_LENGTHS
316 })?;
317 writer.write_u32::<LittleEndian>(column.count() as u32)?;
318 writer.write_u64::<LittleEndian>(column.total_tokens())?;
319 writer.write_u64::<LittleEndian>(offset)?;
320 offset += column.count() as u64 * if column.quantized() { 1 } else { 2 };
321 }
322 for (_, map) in fields {
323 for doc_id in &map.doc_ids {
324 writer.write_u32::<LittleEndian>(*doc_id)?;
325 }
326 for ordinal in &map.ordinals {
327 writer.write_u16::<LittleEndian>(*ordinal)?;
328 }
329 for length in &map.lengths {
330 writer.write_u16::<LittleEndian>(*length)?;
331 }
332 if map.section_kind() != KIND_CHUNK_MAP {
333 let mut slots: Vec<u32> = (0..map.len() as u32).collect();
334 slots.sort_unstable_by_key(|&slot| {
335 (map.doc_ids[slot as usize], map.ordinals[slot as usize])
336 });
337 let mut previous = None;
338 for slot in slots {
339 let key = (map.doc_ids[slot as usize], map.ordinals[slot as usize]);
340 if previous == Some(key) {
341 return Err(io::Error::new(
342 io::ErrorKind::InvalidData,
343 "duplicate logical text chunk",
344 ));
345 }
346 previous = Some(key);
347 writer.write_u32::<LittleEndian>(slot)?;
348 }
349 }
350 }
351 for column in norms {
352 column.write(writer)?;
353 }
354 Ok(offset)
355}
356
357static INVALID_CHUNK_ID_REPORTED: std::sync::atomic::AtomicBool =
361 std::sync::atomic::AtomicBool::new(false);
362
363const INVALID_CHUNK_LENGTH: u32 = 1;
367
368#[cold]
369#[inline(never)]
370fn invalid_chunk_id(vid: u32, num_chunks: usize) -> u32 {
371 if !INVALID_CHUNK_ID_REPORTED.swap(true, std::sync::atomic::Ordering::Relaxed) {
372 log::error!(
373 "chunk map lookup out of range: virtual chunk id {vid} >= {num_chunks} chunks; \
374 the postings and .chunks file disagree (corrupt segment). Scoring that chunk \
375 with length {INVALID_CHUNK_LENGTH}; further occurrences are not logged"
376 );
377 }
378 INVALID_CHUNK_LENGTH
379}
380
381#[cfg(test)]
383mod quantized_norm_tests {
384 use super::*;
385 #[test]
386 fn byte_norm_columns_and_mixed_merges_preserve_their_scoring_lengths() {
387 let values = [0, 1, 41, 999, u16::MAX];
388 let make = |quantized| {
389 let mut bytes = Vec::new();
390 write_chunk_maps_with_norms(
391 &mut bytes,
392 &[],
393 &[DocLengthsColumn {
394 field_id: 0,
395 lengths: &values,
396 total_tokens: 90000,
397 }],
398 quantized,
399 )
400 .unwrap();
401 read_chunk_maps(OwnedBytes::new(bytes))
402 .unwrap()
403 .doc_lengths
404 .remove(&0)
405 .unwrap()
406 };
407 let old = make(false);
408 let quantized = make(true);
409 assert_eq!(quantized.length_bytes().len(), values.len());
410 assert_eq!(old.length_bytes().len(), values.len() * 2);
411 assert_eq!(quantized.total_tokens(), 90000);
412 assert_eq!(old.avg_len(), quantized.avg_len());
413 for (doc, &value) in values.iter().enumerate() {
414 assert_eq!(
415 quantized.length(doc as u32),
416 crate::segment::norms::quantize(u32::from(value))
417 );
418 }
419 for second in [&old, &quantized] {
420 let mut bytes = Vec::new();
421 write_merged_chunk_maps(
422 &mut bytes,
423 &[],
424 &[(
425 0,
426 vec![
427 DocLengthsSource {
428 lengths: Some(&quantized),
429 num_docs: 5,
430 },
431 DocLengthsSource {
432 lengths: None,
433 num_docs: 2,
434 },
435 DocLengthsSource {
436 lengths: Some(second),
437 num_docs: 5,
438 },
439 ],
440 )],
441 )
442 .unwrap();
443 let merged = read_chunk_maps(OwnedBytes::new(bytes)).unwrap();
444 let merged = &merged.doc_lengths[&0];
445 assert_eq!(merged.is_quantized(), second.is_quantized());
446 for i in 0..5 {
447 assert_eq!(merged.length(i), quantized.length(i));
448 assert_eq!(merged.length(i + 7), second.length(i));
449 }
450 assert_eq!(merged.length(5), 0);
451 assert_eq!(merged.length(6), 0);
452 assert_eq!(merged.total_tokens(), 180000);
453 }
454 }
455}
456
457#[cfg(test)]
458fn invalid_chunk_id_reported() -> bool {
459 INVALID_CHUNK_ID_REPORTED.load(std::sync::atomic::Ordering::Relaxed)
460}
461
462fn gather_lengths<const STRICT: bool>(bytes: &[u8], ids: &[u32], out: &mut [u32], floor: u32) {
467 let pairs = bytes.as_chunks::<2>().0;
468 for (&id, slot) in ids.iter().zip(&mut out[..ids.len()]) {
469 let value = match pairs.get(id as usize) {
470 Some(pair) => u32::from(u16::from_le_bytes(*pair)),
471 None if STRICT => invalid_chunk_id(id, pairs.len()),
472 None => 0,
473 };
474 *slot = value.max(floor);
475 }
476}
477
478#[derive(Debug, Clone)]
481pub struct DocLengths {
482 quantized: bool,
483 lengths: OwnedBytes,
484 num_docs: u32,
485 total_tokens: u64,
486}
487
488impl DocLengths {
489 #[cfg(test)]
491 pub(crate) fn from_lengths(lengths: &[u16]) -> Self {
492 let mut bytes = Vec::with_capacity(lengths.len() * 2);
493 for len in lengths {
494 bytes.extend_from_slice(&len.to_le_bytes());
495 }
496 Self {
497 quantized: false,
498 lengths: OwnedBytes::new(bytes),
499 num_docs: lengths.len() as u32,
500 total_tokens: lengths.iter().map(|&l| u64::from(l)).sum(),
501 }
502 }
503
504 pub fn num_docs(&self) -> u32 {
505 self.num_docs
506 }
507
508 pub fn total_tokens(&self) -> u64 {
509 self.total_tokens
510 }
511
512 pub fn avg_len(&self) -> f32 {
514 let with_value = self
515 .lengths
516 .as_slice()
517 .chunks_exact(if self.quantized { 1 } else { 2 })
518 .filter(|b| b.iter().any(|&value| value != 0))
519 .count();
520 if with_value == 0 {
521 1.0
522 } else {
523 (self.total_tokens as f64 / with_value as f64) as f32
524 }
525 }
526
527 #[inline]
531 pub fn length(&self, doc_id: DocId) -> u32 {
532 if self.quantized {
533 return super::norms::decode(self.norm_code(doc_id));
534 }
535 self.lengths
536 .as_slice()
537 .as_chunks::<2>()
538 .0
539 .get(doc_id as usize)
540 .copied()
541 .map_or(0, |b| u32::from(u16::from_le_bytes(b)))
542 }
543
544 pub(crate) fn gather_lengths(&self, ids: &[u32], out: &mut [u32]) {
545 if self.quantized {
546 for (&id, value) in ids.iter().zip(out) {
547 *value = self.length(id);
548 }
549 } else {
550 gather_lengths::<false>(self.length_bytes(), ids, out, 0);
551 }
552 }
553
554 pub(crate) fn is_quantized(&self) -> bool {
555 self.quantized
556 }
557
558 #[inline]
559 pub(crate) fn norm_code(&self, doc: DocId) -> u8 {
560 self.lengths
561 .as_slice()
562 .get(doc as usize)
563 .copied()
564 .unwrap_or(0)
565 }
566
567 pub(crate) fn length_bytes(&self) -> &[u8] {
568 self.lengths.as_slice()
569 }
570}
571
572#[derive(Debug, Default)]
574pub struct ChunkMapFile {
575 pub chunk_maps: FxHashMap<u32, ChunkMap>,
576 pub doc_lengths: FxHashMap<u32, DocLengths>,
577}
578
579#[derive(Debug, Clone)]
581pub struct ChunkMap {
582 doc_ids: OwnedBytes,
583 ordinals: OwnedBytes,
584 lengths: OwnedBytes,
585 num_chunks: u32,
586 total_tokens: u64,
587 length_floor: u32,
591 logically_ordered: bool,
592 logical_slots: Option<OwnedBytes>,
593 doc_ids_monotonic: bool,
595 document_units: bool,
596}
597
598fn nominal_chunk_length(lengths: &[u8]) -> u32 {
600 let n = lengths.len() / 2;
601 if n == 0 {
602 return 0;
603 }
604 let mut histogram = vec![0u32; u16::MAX as usize + 1];
605 for pair in lengths.chunks_exact(2) {
606 histogram[u16::from_le_bytes([pair[0], pair[1]]) as usize] += 1;
607 }
608 let target = (n as u64 * 9).div_ceil(10);
610 let mut seen = 0u64;
611 for (len, &count) in histogram.iter().enumerate() {
612 seen += u64::from(count);
613 if seen >= target {
614 return len as u32;
615 }
616 }
617 u16::MAX as u32
618}
619
620impl ChunkMap {
621 pub(crate) fn is_document_map(&self) -> bool {
622 self.document_units
623 }
624
625 #[cfg(feature = "native")]
629 pub(crate) fn identity_documents(
630 num_docs: u32,
631 lengths: Option<&DocLengths>,
632 ) -> io::Result<Self> {
633 if lengths.is_some_and(|lengths| lengths.num_docs() != num_docs) {
634 return Err(io::Error::new(
635 io::ErrorKind::InvalidData,
636 "document length count mismatch",
637 ));
638 }
639 let mut ids = Vec::with_capacity(num_docs as usize * 4);
640 for doc in 0..num_docs {
641 ids.extend_from_slice(&doc.to_le_bytes());
642 }
643 Ok(Self {
644 doc_ids: OwnedBytes::new(ids),
645 ordinals: OwnedBytes::new(vec![0; num_docs as usize * 2]),
646 lengths: lengths.map_or_else(
647 || OwnedBytes::new(vec![0; num_docs as usize * 2]),
648 |lengths| {
649 if !lengths.quantized {
650 return lengths.lengths.clone();
651 }
652 let mut bytes = Vec::with_capacity(num_docs as usize * 2);
653 for doc in 0..num_docs {
654 bytes.extend_from_slice(&(lengths.length(doc) as u16).to_le_bytes());
655 }
656 OwnedBytes::new(bytes)
657 },
658 ),
659 num_chunks: num_docs,
660 total_tokens: lengths.map_or(0, DocLengths::total_tokens),
661 length_floor: 0,
662 logically_ordered: true,
663 logical_slots: None,
664 doc_ids_monotonic: true,
665 document_units: true,
666 })
667 }
668
669 pub(crate) fn has_logical_addressing(&self) -> bool {
670 self.logically_ordered || self.logical_slots.is_some()
671 }
672
673 fn logical_slot(&self, index: u32) -> u32 {
674 self.logical_slots.as_ref().map_or(index, |slots| {
675 let offset = index as usize * 4;
676 u32::from_le_bytes(slots.as_slice()[offset..offset + 4].try_into().unwrap())
677 })
678 }
679
680 pub(crate) fn document_slot(&self, doc: DocId) -> Option<u32> {
683 (self.document_units && doc < self.num_chunks).then(|| self.logical_slot(doc))
684 }
685
686 pub(crate) fn slots_for_document(&self, doc: DocId) -> impl Iterator<Item = (u16, u32)> + '_ {
687 super::logical_address::ordered_document_slots(self.num_chunks, doc, |index| {
688 let (doc, ordinal) = self.resolve(self.logical_slot(index));
689 Some(super::logical_address::LogicalUnit { doc, ordinal })
690 })
691 .map(|(ordinal, index)| (ordinal, self.logical_slot(index)))
692 }
693
694 pub(crate) fn slot_for_unit(&self, unit: super::logical_address::LogicalUnit) -> Option<u32> {
695 super::logical_address::ordered_slot_for_unit(self.num_chunks, unit, |index| {
696 let (doc, ordinal) = self.resolve(self.logical_slot(index));
697 Some(super::logical_address::LogicalUnit { doc, ordinal })
698 })
699 .map(|index| self.logical_slot(index))
700 }
701
702 pub(crate) fn logically_ordered(&self) -> bool {
703 self.logically_ordered
704 }
705
706 pub(crate) fn is_doc_ordered(&self) -> bool {
707 self.doc_ids_monotonic
708 }
709
710 pub(crate) fn lower_bound_doc(&self, target: DocId) -> u32 {
713 debug_assert!(self.is_doc_ordered());
714 let (mut lo, mut hi) = (0, self.num_chunks);
715 while lo < hi {
716 let mid = lo + (hi - lo) / 2;
717 if self.doc_id(mid) < target {
718 lo = mid + 1;
719 } else {
720 hi = mid;
721 }
722 }
723 lo
724 }
725
726 #[inline]
728 pub fn num_chunks(&self) -> u32 {
729 self.num_chunks
730 }
731
732 pub fn total_tokens(&self) -> u64 {
734 self.total_tokens
735 }
736
737 pub fn avg_len(&self) -> f32 {
739 if self.num_chunks == 0 {
740 1.0
741 } else {
742 (self.total_tokens as f64 / f64::from(self.num_chunks)) as f32
743 }
744 }
745
746 #[inline]
748 pub fn length_floor(&self) -> u32 {
749 self.length_floor
750 }
751
752 #[inline]
755 pub fn bm25_length(&self, vid: u32) -> u32 {
756 self.length(vid).max(self.length_floor)
757 }
758
759 #[inline]
761 pub fn doc_id(&self, vid: u32) -> DocId {
762 let at = vid as usize * 4;
763 let b = &self.doc_ids.as_slice()[at..at + 4];
764 u32::from_le_bytes([b[0], b[1], b[2], b[3]])
765 }
766
767 #[inline]
769 pub fn ordinal(&self, vid: u32) -> u16 {
770 let at = vid as usize * 2;
771 let b = &self.ordinals.as_slice()[at..at + 2];
772 u16::from_le_bytes([b[0], b[1]])
773 }
774
775 #[inline]
779 pub fn length(&self, vid: u32) -> u32 {
780 let pairs = self.lengths.as_slice().as_chunks::<2>().0;
781 match pairs.get(vid as usize) {
782 Some(pair) => u32::from(u16::from_le_bytes(*pair)),
783 None => invalid_chunk_id(vid, pairs.len()),
784 }
785 }
786
787 pub(crate) fn gather_bm25_lengths(&self, ids: &[u32], out: &mut [u32]) {
788 gather_lengths::<true>(self.length_bytes(), ids, out, self.length_floor);
789 }
790
791 #[inline]
793 pub fn resolve(&self, vid: u32) -> (DocId, u16) {
794 (self.doc_id(vid), self.ordinal(vid))
795 }
796
797 pub(crate) fn doc_id_bytes(&self) -> &[u8] {
799 self.doc_ids.as_slice()
800 }
801
802 pub(crate) fn ordinal_bytes(&self) -> &[u8] {
804 self.ordinals.as_slice()
805 }
806
807 pub(crate) fn length_bytes(&self) -> &[u8] {
809 self.lengths.as_slice()
810 }
811}
812
813pub fn read_chunk_maps(bytes: OwnedBytes) -> io::Result<ChunkMapFile> {
815 let data = bytes.as_slice();
816 if data.len() < HEADER_SIZE {
817 return Err(io::Error::new(
818 io::ErrorKind::InvalidData,
819 "chunk map file shorter than its header",
820 ));
821 }
822 let mut cursor = io::Cursor::new(data);
823 let magic = cursor.read_u32::<LittleEndian>()?;
824 if magic != MAGIC {
825 return Err(io::Error::new(
826 io::ErrorKind::InvalidData,
827 format!("chunk map magic mismatch: {magic:#x}"),
828 ));
829 }
830 let version = cursor.read_u32::<LittleEndian>()?;
831 let entry_size = match version {
832 1 => TOC_ENTRY_SIZE_V1,
833 2 | ADDRESSED_VERSION | DOCUMENT_VERSION | VERSION => TOC_ENTRY_SIZE,
834 other => {
835 return Err(io::Error::new(
836 io::ErrorKind::InvalidData,
837 format!("unsupported chunk map version {other} (expected {VERSION})"),
838 ));
839 }
840 };
841 let num_sections = cursor.read_u32::<LittleEndian>()? as usize;
842 let overflow = || io::Error::new(io::ErrorKind::InvalidData, "chunk map size overflow");
843 let toc_end = num_sections
844 .checked_mul(entry_size)
845 .and_then(|n| HEADER_SIZE.checked_add(n))
846 .ok_or_else(overflow)?;
847 if data.len() < toc_end {
848 return Err(io::Error::new(
849 io::ErrorKind::InvalidData,
850 "chunk map table of contents truncated",
851 ));
852 }
853 let mut expected_offset = toc_end;
854 let mut file = ChunkMapFile::default();
855 for _ in 0..num_sections {
856 let field_id = cursor.read_u32::<LittleEndian>()?;
857 let kind = if version == 1 {
858 KIND_CHUNK_MAP
859 } else {
860 cursor.read_u32::<LittleEndian>()?
861 };
862 let count = cursor.read_u32::<LittleEndian>()?;
863 let total_tokens = cursor.read_u64::<LittleEndian>()?;
864 let offset = usize::try_from(cursor.read_u64::<LittleEndian>()?).map_err(|_| overflow())?;
865 if offset != expected_offset
866 || file.chunk_maps.contains_key(&field_id)
867 || file.doc_lengths.contains_key(&field_id)
868 {
869 return Err(io::Error::new(
870 io::ErrorKind::InvalidData,
871 "chunk map has overlapping sections, gaps or duplicate fields",
872 ));
873 }
874 let n = count as usize;
875 let bytes_per_entry = match kind {
876 KIND_CHUNK_MAP => 8,
877 KIND_ADDRESSED_CHUNK_MAP if version >= 3 => 12,
878 KIND_DOCUMENT_MAP if version >= 4 => 12,
879 KIND_DOC_LENGTHS => 2,
880 KIND_BYTE_NORMS if version >= 5 => 1,
881 other => {
882 return Err(io::Error::new(
883 io::ErrorKind::InvalidData,
884 format!("unknown chunk map section kind {other} for field {field_id}"),
885 ));
886 }
887 };
888 let end = offset
889 .checked_add(n.checked_mul(bytes_per_entry).ok_or_else(overflow)?)
890 .ok_or_else(overflow)?;
891 if end > data.len() {
892 return Err(io::Error::new(
893 io::ErrorKind::InvalidData,
894 format!("chunk map section of field {field_id} exceeds file length"),
895 ));
896 }
897 expected_offset = end;
898 match kind {
899 KIND_CHUNK_MAP | KIND_ADDRESSED_CHUNK_MAP | KIND_DOCUMENT_MAP => {
900 let doc_ids = bytes.slice(offset..offset + n * 4);
901 let ordinals = bytes.slice(offset + n * 4..offset + n * 6);
902 let lengths = bytes.slice(offset + n * 6..offset + n * 8);
903 let logical_slots =
904 (kind != KIND_CHUNK_MAP).then(|| bytes.slice(offset + n * 8..end));
905 let document_units = kind == KIND_DOCUMENT_MAP;
906 let length_floor = if document_units {
907 0
908 } else {
909 nominal_chunk_length(lengths.as_slice())
910 };
911 let logically_ordered = super::logical_address::logically_ordered(
912 doc_ids
913 .as_slice()
914 .chunks_exact(4)
915 .zip(ordinals.as_slice().chunks_exact(2))
916 .map(|(doc, ordinal)| {
917 Some(super::logical_address::LogicalUnit {
918 doc: u32::from_le_bytes(doc.try_into().unwrap()),
919 ordinal: u16::from_le_bytes(ordinal.try_into().unwrap()),
920 })
921 }),
922 );
923 let doc_ids_monotonic = doc_ids
924 .as_slice()
925 .chunks_exact(4)
926 .map(|b| u32::from_le_bytes([b[0], b[1], b[2], b[3]]))
927 .is_sorted();
928 let map = ChunkMap {
929 doc_ids,
930 ordinals,
931 lengths,
932 num_chunks: count,
933 total_tokens,
934 length_floor,
935 logically_ordered,
936 logical_slots,
937 doc_ids_monotonic,
938 document_units,
939 };
940 if version >= 3 && kind == KIND_CHUNK_MAP && !map.logically_ordered {
941 return Err(io::Error::new(
942 io::ErrorKind::InvalidData,
943 "V3 chunk map requires logical ordering or an addressed section",
944 ));
945 }
946 if let Some(slots) = &map.logical_slots {
947 let mut previous = None;
948 for raw in slots.as_slice().chunks_exact(4) {
949 let slot = u32::from_le_bytes(raw.try_into().unwrap());
950 if slot >= count {
951 return Err(io::Error::new(
952 io::ErrorKind::InvalidData,
953 "text logical slot out of range",
954 ));
955 }
956 let key = map.resolve(slot);
957 if document_units && (key.0 >= count || key.1 != 0) {
958 return Err(io::Error::new(
959 io::ErrorKind::InvalidData,
960 "document map must cover each document once with ordinal zero",
961 ));
962 }
963 if previous.is_some_and(|p| p >= key) {
964 return Err(io::Error::new(
965 io::ErrorKind::InvalidData,
966 "text logical slots are not a sorted unique permutation",
967 ));
968 }
969 previous = Some(key);
970 }
971 }
972 file.chunk_maps.insert(field_id, map);
973 }
974 _ => {
975 if kind == KIND_BYTE_NORMS
976 && bytes[offset..end]
977 .iter()
978 .any(|&code| super::norms::decode(code) > MAX_CHUNK_LENGTH)
979 {
980 return Err(io::Error::new(
981 io::ErrorKind::InvalidData,
982 "byte norm exceeds saturated document length",
983 ));
984 }
985 file.doc_lengths.insert(
986 field_id,
987 DocLengths {
988 quantized: kind == KIND_BYTE_NORMS,
989 lengths: bytes.slice(offset..end),
990 num_docs: count,
991 total_tokens,
992 },
993 );
994 }
995 }
996 }
997 if expected_offset != data.len() {
998 return Err(io::Error::new(
999 io::ErrorKind::InvalidData,
1000 "trailing chunk map data",
1001 ));
1002 }
1003 Ok(file)
1004}
1005
1006pub struct ChunkMapSource<'a> {
1008 pub map: &'a ChunkMap,
1009 pub doc_offset: u32,
1011}
1012
1013pub struct DocLengthsSource<'a> {
1016 pub lengths: Option<&'a DocLengths>,
1017 pub num_docs: u32,
1018}
1019
1020fn all_quantized(sources: &[DocLengthsSource<'_>]) -> bool {
1021 sources
1022 .iter()
1023 .filter_map(|source| source.lengths)
1024 .all(DocLengths::is_quantized)
1025}
1026
1027pub fn write_merged_chunk_maps<W: Write + ?Sized>(
1035 writer: &mut W,
1036 fields: &[(u32, Vec<ChunkMapSource<'_>>)],
1037 norms: &[(u32, Vec<DocLengthsSource<'_>>)],
1038) -> io::Result<u64> {
1039 write_merged_chunk_maps_ordered(writer, fields, norms, &FxHashMap::default(), || Ok(()))
1040}
1041
1042pub(crate) struct ChunkMapOrder<'a> {
1044 pub order: &'a [u32],
1045 pub inverse: &'a [u32],
1046}
1047
1048pub(crate) fn write_merged_chunk_maps_ordered<W: Write + ?Sized>(
1049 writer: &mut W,
1050 fields: &[(u32, Vec<ChunkMapSource<'_>>)],
1051 norms: &[(u32, Vec<DocLengthsSource<'_>>)],
1052 orders: &FxHashMap<u32, ChunkMapOrder<'_>>,
1053 mut check_cancelled: impl FnMut() -> io::Result<()>,
1054) -> io::Result<u64> {
1055 let live: Vec<&(u32, Vec<ChunkMapSource<'_>>)> = fields
1056 .iter()
1057 .filter(|(_, sources)| sources.iter().any(|s| s.map.num_chunks() > 0))
1058 .collect();
1059 let mut legacy = false;
1062 let mut addressed = false;
1063 let mut document_units = false;
1064 for (field, sources) in &live {
1065 check_cancelled()?;
1066 if let Some(plan) = orders.get(field) {
1067 let count: u64 = sources.iter().map(|s| u64::from(s.map.num_chunks())).sum();
1068 if count != plan.order.len() as u64 || plan.order.len() != plan.inverse.len() {
1069 return Err(io::Error::new(
1070 io::ErrorKind::InvalidData,
1071 "text map permutation has the wrong extent",
1072 ));
1073 }
1074 for (new, &old) in plan.order.iter().enumerate() {
1075 if new.is_multiple_of(4096) {
1076 check_cancelled()?;
1077 }
1078 if plan.inverse.get(old as usize).copied() != Some(new as u32) {
1079 return Err(io::Error::new(
1080 io::ErrorKind::InvalidData,
1081 "text map order and inverse disagree",
1082 ));
1083 }
1084 }
1085 }
1086 let documents = sources.iter().any(|s| s.map.document_units);
1087 if documents && sources.iter().any(|s| !s.map.document_units) {
1088 return Err(io::Error::new(
1089 io::ErrorKind::InvalidData,
1090 "cannot merge document and chunk scoring units for one text field",
1091 ));
1092 }
1093 document_units |= documents;
1094 let needs_migration =
1095 !orders.contains_key(field) && sources.iter().any(|s| !s.map.has_logical_addressing());
1096 if needs_migration
1097 && sources
1098 .iter()
1099 .any(|s| s.map.num_chunks() > 0 && s.map.has_logical_addressing())
1100 {
1101 return Err(io::Error::new(
1102 io::ErrorKind::InvalidData,
1103 "cannot merge prepared and unprepared text chunk maps; explicitly reorder legacy segments first",
1104 ));
1105 }
1106 legacy |= needs_migration;
1107 addressed |= documents
1108 || orders.contains_key(field)
1109 || (!needs_migration && sources.iter().any(|s| !s.map.logically_ordered()));
1110 }
1111 if legacy && addressed {
1112 return Err(io::Error::new(
1113 io::ErrorKind::InvalidData,
1114 "legacy text chunks require explicit reorder before merging with V3 addressed fields",
1115 ));
1116 }
1117 let sections = live.len() + norms.len();
1118 let mut offset = (HEADER_SIZE + TOC_ENTRY_SIZE * sections) as u64;
1119 writer.write_u32::<LittleEndian>(MAGIC)?;
1120 writer.write_u32::<LittleEndian>(
1121 if !legacy && norms.iter().any(|(_, sources)| all_quantized(sources)) {
1122 VERSION
1123 } else if legacy {
1124 2
1125 } else if document_units {
1126 DOCUMENT_VERSION
1127 } else {
1128 ADDRESSED_VERSION
1129 },
1130 )?;
1131 writer.write_u32::<LittleEndian>(sections as u32)?;
1132 for (field_id, sources) in &live {
1133 let mut num_chunks = 0u64;
1134 let mut total_tokens = 0u64;
1135 for source in sources {
1136 num_chunks += u64::from(source.map.num_chunks());
1137 total_tokens += source.map.total_tokens();
1138 }
1139 let num_chunks = u32::try_from(num_chunks).map_err(|_| {
1140 io::Error::new(
1141 io::ErrorKind::InvalidData,
1142 format!("chunked field {field_id} exceeds u32::MAX chunks after merge"),
1143 )
1144 })?;
1145 writer.write_u32::<LittleEndian>(*field_id)?;
1146 writer.write_u32::<LittleEndian>(if sources.iter().any(|s| s.map.document_units) {
1147 KIND_DOCUMENT_MAP
1148 } else if orders.contains_key(field_id)
1149 || (sources.iter().all(|s| s.map.has_logical_addressing())
1150 && sources.iter().any(|s| !s.map.logically_ordered()))
1151 {
1152 KIND_ADDRESSED_CHUNK_MAP
1153 } else {
1154 KIND_CHUNK_MAP
1155 })?;
1156 writer.write_u32::<LittleEndian>(num_chunks)?;
1157 writer.write_u64::<LittleEndian>(total_tokens)?;
1158 writer.write_u64::<LittleEndian>(offset)?;
1159 offset += u64::from(num_chunks)
1160 * if orders.contains_key(field_id)
1161 || sources.iter().any(|s| s.map.document_units)
1162 || (sources.iter().all(|s| s.map.has_logical_addressing())
1163 && sources.iter().any(|s| !s.map.logically_ordered()))
1164 {
1165 12
1166 } else {
1167 8
1168 };
1169 }
1170 for (field_id, sources) in norms {
1171 let num_docs: u64 = sources.iter().map(|s| u64::from(s.num_docs)).sum();
1172 let num_docs = u32::try_from(num_docs).map_err(|_| {
1173 io::Error::new(
1174 io::ErrorKind::InvalidData,
1175 format!("field {field_id} exceeds u32::MAX documents after merge"),
1176 )
1177 })?;
1178 let total_tokens: u64 = sources
1179 .iter()
1180 .filter_map(|s| s.lengths.map(DocLengths::total_tokens))
1181 .sum();
1182 writer.write_u32::<LittleEndian>(*field_id)?;
1183 writer.write_u32::<LittleEndian>(if !legacy && all_quantized(sources) {
1184 KIND_BYTE_NORMS
1185 } else {
1186 KIND_DOC_LENGTHS
1187 })?;
1188 writer.write_u32::<LittleEndian>(num_docs)?;
1189 writer.write_u64::<LittleEndian>(total_tokens)?;
1190 writer.write_u64::<LittleEndian>(offset)?;
1191 offset += u64::from(num_docs)
1192 * if !legacy && all_quantized(sources) {
1193 1
1194 } else {
1195 2
1196 };
1197 }
1198 let mut patched = Vec::with_capacity(64 * 1024);
1199 for (field, sources) in &live {
1200 check_cancelled()?;
1201 if let Some(plan) = orders.get(field) {
1202 let mut bases = Vec::with_capacity(sources.len());
1203 let mut base = 0u32;
1204 for source in sources {
1205 bases.push(base);
1206 base += source.map.num_chunks();
1207 }
1208 for column in 0..3 {
1209 for batch in plan.order.chunks(4096) {
1210 check_cancelled()?;
1211 patched.clear();
1212 for &old in batch {
1213 let index = bases.partition_point(|&base| base <= old) - 1;
1214 let source = &sources[index];
1215 let slot = old - bases[index];
1216 match column {
1217 0 => {
1218 let doc = source
1219 .map
1220 .doc_id(slot)
1221 .checked_add(source.doc_offset)
1222 .ok_or_else(|| {
1223 io::Error::new(
1224 io::ErrorKind::InvalidData,
1225 "reordered document ID overflow",
1226 )
1227 })?;
1228 patched.extend_from_slice(&doc.to_le_bytes());
1229 }
1230 1 => patched.extend_from_slice(&source.map.ordinal(slot).to_le_bytes()),
1231 _ => patched
1232 .extend_from_slice(&(source.map.length(slot) as u16).to_le_bytes()),
1233 }
1234 }
1235 writer.write_all(&patched)?;
1236 }
1237 }
1238 for (source, &base) in sources.iter().zip(&bases) {
1239 for first in (0..source.map.num_chunks()).step_by(4096) {
1240 check_cancelled()?;
1241 patched.clear();
1242 for logical in first..source.map.num_chunks().min(first.saturating_add(4096)) {
1243 let old = source.map.logical_slot(logical) + base;
1244 patched.extend_from_slice(&plan.inverse[old as usize].to_le_bytes());
1245 }
1246 writer.write_all(&patched)?;
1247 }
1248 }
1249 continue;
1250 }
1251 for source in sources {
1252 if source.doc_offset == 0 {
1253 writer.write_all(source.map.doc_id_bytes())?;
1254 continue;
1255 }
1256 for batch in source.map.doc_id_bytes().chunks(64 * 1024) {
1257 patched.clear();
1258 for chunk in batch.chunks_exact(4) {
1259 let doc = u32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
1260 let remapped = doc.checked_add(source.doc_offset).ok_or_else(|| {
1261 io::Error::new(
1262 io::ErrorKind::InvalidData,
1263 "document id overflow while merging chunk maps",
1264 )
1265 })?;
1266 patched.extend_from_slice(&remapped.to_le_bytes());
1267 }
1268 writer.write_all(&patched)?;
1269 }
1270 }
1271 for source in sources {
1272 writer.write_all(source.map.ordinal_bytes())?;
1273 }
1274 for source in sources {
1275 writer.write_all(source.map.length_bytes())?;
1276 }
1277 let addressed = sources.iter().any(|s| s.map.document_units)
1278 || (sources.iter().all(|s| s.map.has_logical_addressing())
1279 && sources.iter().any(|s| !s.map.logically_ordered()));
1280 if addressed {
1281 let mut base = 0u32;
1282 for source in sources {
1283 for i in 0..source.map.num_chunks() {
1284 writer.write_u32::<LittleEndian>(source.map.logical_slot(i) + base)?;
1285 }
1286 base += source.map.num_chunks();
1287 }
1288 }
1289 }
1290 let zeros = [0u8; 2 * 1024];
1291 for (_, sources) in norms {
1292 let quantized = !legacy && all_quantized(sources);
1293 for source in sources {
1294 match source.lengths {
1295 Some(lengths) if lengths.num_docs() == source.num_docs => {
1296 if lengths.quantized && !quantized {
1297 for batch in lengths.length_bytes().chunks(32 * 1024) {
1298 patched.clear();
1299 for &code in batch {
1300 patched.extend_from_slice(
1301 &(super::norms::decode(code) as u16).to_le_bytes(),
1302 );
1303 }
1304 writer.write_all(&patched)?;
1305 }
1306 } else {
1307 writer.write_all(lengths.length_bytes())?;
1308 }
1309 }
1310 Some(lengths) => {
1311 return Err(io::Error::new(
1312 io::ErrorKind::InvalidData,
1313 format!(
1314 "length column covers {} documents, segment has {}",
1315 lengths.num_docs(),
1316 source.num_docs
1317 ),
1318 ));
1319 }
1320 None => {
1321 let mut remaining = source.num_docs as usize * if quantized { 1 } else { 2 };
1322 while remaining > 0 {
1323 let take = remaining.min(zeros.len());
1324 writer.write_all(&zeros[..take])?;
1325 remaining -= take;
1326 }
1327 }
1328 }
1329 }
1330 }
1331 Ok(offset)
1332}
1333
1334#[cfg(test)]
1335mod tests {
1336 use super::*;
1337
1338 #[cfg(feature = "native")]
1339 #[test]
1340 fn ordered_merge_map_validates_permutations_and_cancels_during_column_output() {
1341 let map = ChunkMap::identity_documents(8192, None).unwrap();
1342 let order: Vec<u32> = (0..8192).rev().collect();
1343 let fields = [(
1344 1,
1345 vec![ChunkMapSource {
1346 map: &map,
1347 doc_offset: 0,
1348 }],
1349 )];
1350 let plans = FxHashMap::from_iter([(
1351 1,
1352 ChunkMapOrder {
1353 order: &order,
1354 inverse: &order,
1355 },
1356 )]);
1357 let mut bytes = Vec::new();
1358 let mut checks = 0;
1359 let error = write_merged_chunk_maps_ordered(&mut bytes, &fields, &[], &plans, || {
1360 checks += 1;
1361 if checks == 6 {
1362 Err(io::Error::new(io::ErrorKind::Interrupted, "cancelled"))
1363 } else {
1364 Ok(())
1365 }
1366 })
1367 .unwrap_err();
1368 assert_eq!(error.kind(), io::ErrorKind::Interrupted);
1369 assert!(
1370 bytes.len() > HEADER_SIZE + TOC_ENTRY_SIZE,
1371 "cancellation must interrupt an in-progress column"
1372 );
1373 let invalid = vec![0; order.len()];
1374 let plans = FxHashMap::from_iter([(
1375 1,
1376 ChunkMapOrder {
1377 order: &order,
1378 inverse: &invalid,
1379 },
1380 )]);
1381 bytes.clear();
1382 assert!(
1383 write_merged_chunk_maps_ordered(&mut bytes, &fields, &[], &plans, || Ok(())).is_err()
1384 );
1385 assert!(
1386 bytes.is_empty(),
1387 "reject inconsistent permutations before writing a header"
1388 );
1389 }
1390
1391 #[test]
1392 fn document_maps_preserve_plain_lengths_and_require_the_new_version() {
1393 let mut builder = ChunkMapBuilder::default();
1394 for (doc, length) in [(2, 10), (0, 20), (1, 100)] {
1395 builder.push(doc, 0, length).unwrap();
1396 }
1397 let mut bytes = Vec::new();
1398 write_chunk_maps(&mut bytes, &[(0, &builder)], &[]).unwrap();
1399 assert_eq!(u32::from_le_bytes(bytes[4..8].try_into().unwrap()), 3);
1402 builder.set_document_units(true);
1403 bytes.clear();
1404 write_chunk_maps(&mut bytes, &[(0, &builder)], &[]).unwrap();
1405 assert_eq!(u32::from_le_bytes(bytes[4..8].try_into().unwrap()), 4);
1406 let file = read_chunk_maps(OwnedBytes::new(bytes.clone())).unwrap();
1407 let map = &file.chunk_maps[&0];
1408 assert!(map.is_document_map());
1409 assert_eq!(map.resolve(0), (2, 0));
1410 for doc in 0..3 {
1411 let slot = map.document_slot(doc).unwrap();
1412 assert_eq!(map.resolve(slot), (doc, 0));
1413 }
1414 assert_eq!(map.document_slot(3), None);
1415 assert_eq!(map.document_slot(u32::MAX), None);
1416 assert_eq!(map.bm25_length(0), 10);
1417 assert_eq!(map.bm25_length(1), 20);
1418 assert_eq!(map.slots_for_document(0).collect::<Vec<_>>(), vec![(0, 1)]);
1419 bytes[4..8].copy_from_slice(&3u32.to_le_bytes());
1420 assert!(read_chunk_maps(OwnedBytes::new(bytes)).is_err());
1421 }
1422
1423 fn build(entries: &[(u32, u16, u32)]) -> ChunkMapBuilder {
1424 let mut builder = ChunkMapBuilder::default();
1425 for &(doc, ord, len) in entries {
1426 builder.push(doc, ord, len).unwrap();
1427 }
1428 builder
1429 }
1430
1431 #[test]
1432 fn batched_length_reads_preserve_missing_values_chunk_floors_and_column_bytes() {
1433 let docs = DocLengths::from_lengths(&[0, 1, u16::MAX, 120]);
1434 let original = docs.length_bytes().to_vec();
1435 let ids = [u32::MAX, 1, 2, 3, 0, 2];
1436 let mut out = [999; 8];
1437 docs.gather_lengths(&ids, &mut out);
1438 assert_eq!(
1439 out,
1440 [
1441 0,
1442 1,
1443 u32::from(u16::MAX),
1444 120,
1445 0,
1446 u32::from(u16::MAX),
1447 999,
1448 999
1449 ]
1450 );
1451 for (&id, &value) in ids.iter().zip(&out) {
1452 assert_eq!(value, docs.length(id));
1453 }
1454 docs.gather_lengths(&[], &mut []);
1455 assert_eq!(docs.length_bytes(), original);
1456
1457 let mut builder = ChunkMapBuilder::default();
1458 for id in 0..100 {
1459 builder
1460 .push(
1461 id,
1462 0,
1463 if id == 99 {
1464 300
1465 } else if id == 0 {
1466 0
1467 } else {
1468 10
1469 },
1470 )
1471 .unwrap();
1472 }
1473 let mut bytes = Vec::new();
1474 write_chunk_maps(&mut bytes, &[(0, &builder)], &[]).unwrap();
1475 let file = read_chunk_maps(OwnedBytes::new(bytes)).unwrap();
1476 let map = &file.chunk_maps[&0];
1477 let original = map.length_bytes().to_vec();
1478 let ids = [0, 99, 50, 1];
1479 let mut out = [999; 6];
1480 map.gather_bm25_lengths(&ids, &mut out);
1481 assert_eq!(out, [10, 300, 10, 10, 999, 999]);
1482 for (&id, &value) in ids.iter().zip(&out) {
1483 assert_eq!(value, map.bm25_length(id));
1484 }
1485 assert_eq!(map.length_bytes(), original);
1486 }
1487
1488 #[test]
1489 fn invalid_virtual_ids_score_as_one_token_and_are_reported_once() {
1490 let mut out = [0; 2];
1491 gather_lengths::<true>(&[7, 0], &[u32::MAX, 0], &mut out, 0);
1492 assert_eq!(out, [INVALID_CHUNK_LENGTH, 7]);
1493 assert!(invalid_chunk_id_reported());
1494 let mut floored = [0];
1495 gather_lengths::<true>(&[0, 0], &[u32::MAX], &mut floored, 10);
1496 assert_eq!(floored, [10], "the substitute still honours the BM25 floor");
1497
1498 let map = build(&[(0, 0, 10)]);
1499 let mut bytes = Vec::new();
1500 write_chunk_maps(&mut bytes, &[(0, &map)], &[]).unwrap();
1501 let map = &read_chunk_maps(OwnedBytes::new(bytes)).unwrap().chunk_maps[&0];
1502 assert_eq!(map.length(0), 10);
1503 assert_eq!(map.length(1), INVALID_CHUNK_LENGTH);
1504 assert_eq!(
1505 map.bm25_length(1),
1506 INVALID_CHUNK_LENGTH.max(map.length_floor())
1507 );
1508 let mut plain = [9];
1510 gather_lengths::<false>(&[0, 0], &[5], &mut plain, 0);
1511 assert_eq!(plain, [0]);
1512 }
1513
1514 #[test]
1515 fn round_trips_two_fields() {
1516 let a = build(&[(0, 0, 10), (0, 1, 20), (3, 0, 70_000)]);
1517 let b = build(&[(1, 0, 5)]);
1518 let mut out = Vec::new();
1519 write_chunk_maps(&mut out, &[(2, &a), (7, &b)], &[]).unwrap();
1520 let maps = read_chunk_maps(OwnedBytes::new(out)).unwrap().chunk_maps;
1521 let a = &maps[&2];
1522 assert_eq!(a.num_chunks(), 3);
1523 assert_eq!(a.resolve(0), (0, 0));
1524 assert_eq!(a.resolve(1), (0, 1));
1525 assert_eq!(a.resolve(2), (3, 0));
1526 assert_eq!(a.length(1), 20);
1527 assert_eq!(a.length(2), MAX_CHUNK_LENGTH, "lengths saturate at u16");
1528 assert_eq!(a.total_tokens(), 70_030);
1529 assert_eq!(maps[&7].resolve(0), (1, 0));
1530 assert_eq!(maps[&7].avg_len(), 5.0);
1531 }
1532
1533 #[test]
1534 fn merged_maps_offset_doc_ids_and_keep_ordinals() {
1535 let first = build(&[(0, 0, 10), (1, 0, 11), (1, 1, 12)]);
1536 let second = build(&[(0, 0, 20), (0, 1, 21)]);
1537 let mut raw_first = Vec::new();
1538 write_chunk_maps(&mut raw_first, &[(4, &first)], &[]).unwrap();
1539 let mut raw_second = Vec::new();
1540 write_chunk_maps(&mut raw_second, &[(4, &second)], &[]).unwrap();
1541 let first = read_chunk_maps(OwnedBytes::new(raw_first))
1542 .unwrap()
1543 .chunk_maps;
1544 let second = read_chunk_maps(OwnedBytes::new(raw_second))
1545 .unwrap()
1546 .chunk_maps;
1547
1548 let mut merged = Vec::new();
1549 write_merged_chunk_maps(
1550 &mut merged,
1551 &[(
1552 4,
1553 vec![
1554 ChunkMapSource {
1555 map: &first[&4],
1556 doc_offset: 0,
1557 },
1558 ChunkMapSource {
1559 map: &second[&4],
1560 doc_offset: 2,
1561 },
1562 ],
1563 )],
1564 &[],
1565 )
1566 .unwrap();
1567 let merged = read_chunk_maps(OwnedBytes::new(merged)).unwrap().chunk_maps;
1568 let map = &merged[&4];
1569 assert_eq!(map.num_chunks(), 5);
1570 assert_eq!(map.total_tokens(), 74);
1571 assert_eq!(
1572 (0..5).map(|v| map.resolve(v)).collect::<Vec<_>>(),
1573 vec![(0, 0), (1, 0), (1, 1), (2, 0), (2, 1)]
1574 );
1575 assert_eq!(
1576 (0..5).map(|v| map.length(v)).collect::<Vec<_>>(),
1577 vec![10, 11, 12, 20, 21]
1578 );
1579 }
1580
1581 #[test]
1582 fn rejects_foreign_or_truncated_files() {
1583 assert!(read_chunk_maps(OwnedBytes::new(vec![0u8; 4])).is_err());
1584 let mut bad_magic = Vec::new();
1585 bad_magic.write_u32::<LittleEndian>(0xDEAD_BEEF).unwrap();
1586 bad_magic.write_u32::<LittleEndian>(VERSION).unwrap();
1587 bad_magic.write_u32::<LittleEndian>(0).unwrap();
1588 assert!(read_chunk_maps(OwnedBytes::new(bad_magic)).is_err());
1589
1590 let a = build(&[(0, 0, 10)]);
1591 let mut out = Vec::new();
1592 write_chunk_maps(&mut out, &[(1, &a)], &[]).unwrap();
1593 out.truncate(out.len() - 1);
1594 assert!(read_chunk_maps(OwnedBytes::new(out)).is_err());
1595 }
1596
1597 #[test]
1598 fn doc_length_columns_round_trip_and_merge_with_zero_fill() {
1599 let a = build(&[(0, 0, 10)]);
1600 let column = [7u16, 0, 300];
1601 let mut out = Vec::new();
1602 write_chunk_maps(
1603 &mut out,
1604 &[(1, &a)],
1605 &[DocLengthsColumn {
1606 field_id: 5,
1607 lengths: &column,
1608 total_tokens: 307,
1609 }],
1610 )
1611 .unwrap();
1612 let file = read_chunk_maps(OwnedBytes::new(out)).unwrap();
1613 assert_eq!(file.chunk_maps[&1].num_chunks(), 1);
1614 let norms = &file.doc_lengths[&5];
1615 assert_eq!(norms.num_docs(), 3);
1616 assert_eq!(
1617 (0..4).map(|d| norms.length(d)).collect::<Vec<_>>(),
1618 vec![7, 0, 300, 0]
1619 );
1620 assert_eq!(norms.total_tokens(), 307);
1621 assert!(
1622 (norms.avg_len() - 153.5).abs() < 1e-3,
1623 "{}",
1624 norms.avg_len()
1625 );
1626
1627 let mut merged = Vec::new();
1629 write_merged_chunk_maps(
1630 &mut merged,
1631 &[],
1632 &[(
1633 5,
1634 vec![
1635 DocLengthsSource {
1636 lengths: None,
1637 num_docs: 2,
1638 },
1639 DocLengthsSource {
1640 lengths: Some(norms),
1641 num_docs: 3,
1642 },
1643 ],
1644 )],
1645 )
1646 .unwrap();
1647 let merged = read_chunk_maps(OwnedBytes::new(merged)).unwrap();
1648 assert!(merged.chunk_maps.is_empty());
1649 let norms = &merged.doc_lengths[&5];
1650 assert_eq!(norms.num_docs(), 5);
1651 assert_eq!(
1652 (0..5).map(|d| norms.length(d)).collect::<Vec<_>>(),
1653 vec![0, 0, 7, 0, 300]
1654 );
1655 assert_eq!(norms.total_tokens(), 307);
1656 }
1657
1658 #[test]
1659 fn version_one_files_still_read() {
1660 let a = build(&[(0, 0, 10), (2, 0, 4)]);
1661 let mut out = Vec::new();
1662 out.write_u32::<LittleEndian>(MAGIC).unwrap();
1663 out.write_u32::<LittleEndian>(1).unwrap();
1664 out.write_u32::<LittleEndian>(1).unwrap();
1665 out.write_u32::<LittleEndian>(9).unwrap();
1666 out.write_u32::<LittleEndian>(2).unwrap();
1667 out.write_u64::<LittleEndian>(14).unwrap();
1668 out.write_u64::<LittleEndian>((HEADER_SIZE + TOC_ENTRY_SIZE_V1) as u64)
1669 .unwrap();
1670 for doc in &a.doc_ids {
1671 out.write_u32::<LittleEndian>(*doc).unwrap();
1672 }
1673 for ord in &a.ordinals {
1674 out.write_u16::<LittleEndian>(*ord).unwrap();
1675 }
1676 for len in &a.lengths {
1677 out.write_u16::<LittleEndian>(*len).unwrap();
1678 }
1679 let file = read_chunk_maps(OwnedBytes::new(out)).unwrap();
1680 assert!(file.doc_lengths.is_empty());
1681 let map = &file.chunk_maps[&9];
1682 assert_eq!(map.resolve(1), (2, 0));
1683 assert_eq!(map.length(1), 4);
1684 }
1685
1686 #[test]
1687 fn addressed_chunk_maps_copy_and_remap_slots_without_losing_missing_ordinals() {
1688 let source = build(&[(2, 7, 11), (0, 3, 12), (2, 1, 13)]);
1689 let mut bytes = Vec::new();
1690 write_chunk_maps(&mut bytes, &[(1, &source)], &[]).unwrap();
1691 let file = read_chunk_maps(OwnedBytes::new(bytes)).unwrap();
1692 let map = &file.chunk_maps[&1];
1693 assert_eq!(
1694 map.slot_for_unit(super::super::logical_address::LogicalUnit { doc: 2, ordinal: 7 }),
1695 Some(0)
1696 );
1697 assert_eq!(
1698 map.slot_for_unit(super::super::logical_address::LogicalUnit { doc: 2, ordinal: 0 }),
1699 None
1700 );
1701 let mut merged = Vec::new();
1702 write_merged_chunk_maps(
1703 &mut merged,
1704 &[(
1705 1,
1706 vec![
1707 ChunkMapSource { map, doc_offset: 0 },
1708 ChunkMapSource { map, doc_offset: 4 },
1709 ],
1710 )],
1711 &[],
1712 )
1713 .unwrap();
1714 let file = read_chunk_maps(OwnedBytes::new(merged)).unwrap();
1715 let result = &file.chunk_maps[&1];
1716 assert_eq!(
1717 result.ordinal_bytes(),
1718 [map.ordinal_bytes(), map.ordinal_bytes()].concat()
1719 );
1720 assert_eq!(
1721 result.length_bytes(),
1722 [map.length_bytes(), map.length_bytes()].concat()
1723 );
1724 assert_eq!(
1725 result.slot_for_unit(super::super::logical_address::LogicalUnit { doc: 6, ordinal: 7 }),
1726 Some(3)
1727 );
1728 assert_eq!(
1729 result.slot_for_unit(super::super::logical_address::LogicalUnit { doc: 4, ordinal: 3 }),
1730 Some(4)
1731 );
1732 assert_eq!(
1733 result.slot_for_unit(super::super::logical_address::LogicalUnit { doc: 6, ordinal: 1 }),
1734 Some(5)
1735 );
1736 assert_eq!(
1737 result.slot_for_unit(super::super::logical_address::LogicalUnit { doc: 5, ordinal: 0 }),
1738 None
1739 );
1740 }
1741
1742 #[test]
1743 fn chunk_map_versions_preserve_legacy_capability_and_reject_corrupt_addressing() {
1744 let source = build(&[(2, 0, 11), (0, 0, 12)]);
1745 let mut bytes = Vec::new();
1746 write_chunk_maps(&mut bytes, &[(1, &source)], &[]).unwrap();
1747 let mut invalid = bytes.clone();
1748 let end = invalid.len();
1749 invalid[end - 4..].copy_from_slice(&1u32.to_le_bytes());
1750 assert!(
1751 read_chunk_maps(OwnedBytes::new(invalid)).is_err(),
1752 "duplicate slots"
1753 );
1754 let mut legacy = bytes.clone();
1755 legacy.truncate(legacy.len() - 8);
1756 legacy[16..20].copy_from_slice(&KIND_CHUNK_MAP.to_le_bytes());
1757 assert!(
1758 read_chunk_maps(OwnedBytes::new(legacy.clone())).is_err(),
1759 "V3 ordered kind must be ordered"
1760 );
1761 legacy[4..8].copy_from_slice(&2u32.to_le_bytes());
1762 let old = read_chunk_maps(OwnedBytes::new(legacy)).unwrap();
1763 let old_map = &old.chunk_maps[&1];
1764 assert!(!old_map.has_logical_addressing());
1765 let mut merged = Vec::new();
1766 write_merged_chunk_maps(
1767 &mut merged,
1768 &[(
1769 1,
1770 vec![ChunkMapSource {
1771 map: old_map,
1772 doc_offset: 0,
1773 }],
1774 )],
1775 &[],
1776 )
1777 .unwrap();
1778 assert!(
1779 !read_chunk_maps(OwnedBytes::new(merged)).unwrap().chunk_maps[&1]
1780 .has_logical_addressing()
1781 );
1782 let current = read_chunk_maps(OwnedBytes::new(bytes)).unwrap();
1783 let mut output = Vec::new();
1784 assert!(
1785 write_merged_chunk_maps(
1786 &mut output,
1787 &[(
1788 1,
1789 vec![
1790 ChunkMapSource {
1791 map: old_map,
1792 doc_offset: 0
1793 },
1794 ChunkMapSource {
1795 map: ¤t.chunk_maps[&1],
1796 doc_offset: 3
1797 },
1798 ]
1799 )],
1800 &[]
1801 )
1802 .is_err()
1803 );
1804 assert!(
1805 output.is_empty(),
1806 "reject incompatible sources before writing"
1807 );
1808 }
1809}