Skip to main content

summa_core/segment/
chunk_map.rs

1//! Virtual-id maps of chunked text fields (`seg_<id>.chunks`).
2//!
3//! A text field declared `chunked` indexes every value as its own scoring
4//! unit: term postings and positions are keyed by a dense, segment-local
5//! **virtual id** instead of the document id. This file maps each virtual id
6//! back to `(doc_id, ordinal)` and records the chunk's token count for BM25
7//! length normalisation. See `docs/chunked-text-fields.md`.
8//!
9//! ```text
10//! [magic "CHNK"][version u32, 1..=5][num_sections u32]
11//! TOC × num_sections: [field_id u32][kind u32][count u32][total_tokens u64][data_offset u64]
12//! kind 0 (chunk map):   doc_ids u32 × n | ordinals u16 × n | lengths u16 × n
13//! kind 2 (addressed map): kind 0 columns | logical-order physical slots u32 × n
14//! kind 1 (doc lengths): lengths u16 × num_docs        (norms of a plain text field)
15//! kind 3 (document map, V4): addressed map with document scoring semantics
16//! kind 4 (byte norms, V5): byte4 norm codes × num_docs, exact total_tokens in TOC
17//! ```
18//!
19//! Version 1 files have 24-byte entries without `kind` and hold chunk maps
20//! only; they are still read.
21//!
22//! Virtual ids are assigned in indexing order, and documents are indexed in
23//! doc-id order, so `doc_ids` starts out non-decreasing. A reorder pass on a
24//! field with the `reorder` attribute permutes the virtual ids (BP over the
25//! field's postings, `segment/text_reorder.rs`). Readers verify doc-id order
26//! at open before enabling ordered query paths. Merges concatenate sections and add the document offset to
27//! `doc_ids`; ordinals and lengths are copied verbatim.
28//!
29//! A doc-length section stores the token count of the field in every
30//! document of the segment (0 when the document has no value), so BM25 can
31//! normalise plain fields by their real length instead of `tf`.
32
33use 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; // "CHNK"
42const 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
54/// Token count stored per chunk; longer chunks saturate.
55pub const MAX_CHUNK_LENGTH: u32 = u16::MAX as u32;
56
57/// In-memory map of one chunked field while a segment is being built.
58#[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    /// Preserve the scoring-unit policy when rebuilding mapped text.
80    #[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    /// Number of chunks so far (the next virtual id).
101    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    /// Register the next chunk. Returns its virtual id.
110    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    /// Heap bytes held by this builder (memory-budget accounting).
125    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    /// Token count of virtual id `vid` (saturated at `MAX_CHUNK_LENGTH`).
149    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
156/// Per-document token counts of one plain text field, ready to be written.
157pub struct DocLengthsColumn<'a> {
158    pub field_id: u32,
159    /// One entry per document of the segment (0 = no value).
160    pub lengths: &'a [u16],
161    /// Sum of the unsaturated token counts.
162    pub total_tokens: u64,
163}
164
165/// Write every chunked field's map and every plain field's length column as
166/// one `.chunks` file.
167///
168/// `fields` must be sorted by field id and contain only non-empty builders;
169/// `norms` likewise sorted, one column per field.
170pub 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
178/// Norm inputs share one CHNK section writer. Existing columns keep their
179/// encoding and payload bytes; only newly built columns choose quantization.
180enum 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/// Preserve per-field normalization policy during row compaction of mixed
265/// generations. A representative length is never requantized into another value.
266#[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
357/// Set once a posting referenced a virtual chunk id past its field's chunk
358/// map, so the corruption is logged a single time per process instead of
359/// once per scored posting.
360static INVALID_CHUNK_ID_REPORTED: std::sync::atomic::AtomicBool =
361    std::sync::atomic::AtomicBool::new(false);
362
363/// Scoring length substituted for a virtual id the chunk map does not hold.
364/// One token is the loosest valid normalisation: search keeps running and the
365/// (logged) corruption is visible instead of a bounds-check panic.
366const 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/// Whether an out-of-range virtual chunk id has been reported (tests).
382#[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
462/// Read a bounded batch from the existing little-endian length column.
463/// Resolve the backing byte view once; two-byte arrays need no alignment.
464/// The bounds check is the only work on the common path; a missing id takes
465/// the cold reporting path (`STRICT`) or reads as absent (0).
466fn 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/// Read-only per-document lengths of one plain text field, backed by the
479/// mapped `.chunks` file.
480#[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    /// In-memory lengths column (tests).
490    #[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    /// Average length over documents that have the field (1.0 when none).
513    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    /// Scoring length of the field in `doc_id` (0 when absent or out of range).
528    /// U16 columns are saturated; byte columns return their rounded-down
529    /// representative. Token totals retain the original unsaturated counts.
530    #[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/// Everything a `.chunks` file holds.
573#[derive(Debug, Default)]
574pub struct ChunkMapFile {
575    pub chunk_maps: FxHashMap<u32, ChunkMap>,
576    pub doc_lengths: FxHashMap<u32, DocLengths>,
577}
578
579/// Read-only chunk map of one field, backed by the mapped `.chunks` file.
580#[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    /// Nominal chunk length: the 90th-percentile chunk length of the field in
588    /// this segment. BM25 floors every chunk length at this value so a short
589    /// tail chunk is not rewarded for being short (`docs/chunked-bm25.md`).
590    length_floor: u32,
591    logically_ordered: bool,
592    logical_slots: Option<OwnedBytes>,
593    /// Derived once at open; never persisted or assumed from schema flags.
594    doc_ids_monotonic: bool,
595    document_units: bool,
596}
597
598/// 90th-percentile of a little-endian `u16` length column (0 when empty).
599fn 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    // Smallest length such that at least 90 % of the chunks are ≤ it.
609    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    /// Represent an older unpermuted plain field without rebuilding postings.
626    /// The caller admits six bytes per document before creating these columns;
627    /// existing length bytes remain borrowed from the immutable segment.
628    #[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    /// Document maps are validated dense permutations: logical index equals
681    /// document ID, so their existing slot column is already the inverse map.
682    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    /// First virtual id owned by a document at or after `target`.
711    /// Only valid for a verified doc-ordered map; num_chunks means exhausted.
712    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    /// Number of chunks (virtual ids) in this segment.
727    #[inline]
728    pub fn num_chunks(&self) -> u32 {
729        self.num_chunks
730    }
731
732    /// Sum of all chunk token counts.
733    pub fn total_tokens(&self) -> u64 {
734        self.total_tokens
735    }
736
737    /// Average chunk length in tokens (1.0 when empty).
738    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    /// Nominal chunk length (90th percentile), the BM25 length floor.
747    #[inline]
748    pub fn length_floor(&self) -> u32 {
749        self.length_floor
750    }
751
752    /// BM25 length of virtual id `vid`: its token count floored at the
753    /// nominal chunk length.
754    #[inline]
755    pub fn bm25_length(&self, vid: u32) -> u32 {
756        self.length(vid).max(self.length_floor)
757    }
758
759    /// Document owning virtual id `vid`.
760    #[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    /// Ordinal (value index within the document) of virtual id `vid`.
768    #[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    /// Token count of virtual id `vid` (saturated at `MAX_CHUNK_LENGTH`).
776    /// An id past the map is corrupt: it is logged once and scored as one
777    /// token instead of panicking.
778    #[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    /// `(doc_id, ordinal)` of virtual id `vid`.
792    #[inline]
793    pub fn resolve(&self, vid: u32) -> (DocId, u16) {
794        (self.doc_id(vid), self.ordinal(vid))
795    }
796
797    /// Raw little-endian document-id column (merge copy).
798    pub(crate) fn doc_id_bytes(&self) -> &[u8] {
799        self.doc_ids.as_slice()
800    }
801
802    /// Raw little-endian ordinal column (merge copy).
803    pub(crate) fn ordinal_bytes(&self) -> &[u8] {
804        self.ordinals.as_slice()
805    }
806
807    /// Raw little-endian length column (merge copy).
808    pub(crate) fn length_bytes(&self) -> &[u8] {
809        self.lengths.as_slice()
810    }
811}
812
813/// Parse a `.chunks` file into per-field chunk maps and length columns.
814pub 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
1006/// One source section of a merged chunk map.
1007pub struct ChunkMapSource<'a> {
1008    pub map: &'a ChunkMap,
1009    /// Added to every document id of the source.
1010    pub doc_offset: u32,
1011}
1012
1013/// One source of a merged length column: the source segment's column when it
1014/// has one, and its document count (zeros are written for a missing column).
1015pub 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
1027/// Write the merged `.chunks` file: per field, the sources' sections are
1028/// concatenated in order (virtual ids of a later source are offset by the
1029/// chunk counts of the earlier ones, matching the posting merge; length
1030/// columns follow the document order of the merge).
1031///
1032/// `fields` and `norms` must be sorted by field id; a field with zero total
1033/// chunks is skipped.
1034pub 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
1042/// A validated permutation of concatenated source physical units.
1043pub(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    // Never silently discard addressing on a prepared source. Pure legacy
1060    // maps can still copy-merge in their original version until explicit reorder.
1061    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        // V4 document maps reuse addressed columns but do not apply the
1400        // chunked scoring policy. Older readers must reject this section.
1401        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        // Non-strict document lengths keep reporting absence as 0.
1509        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        // Merge: a source without the column contributes zeros for its docs.
1628        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: &current.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}