laurus 0.10.0

//! Merge engine for combining segments efficiently.
//!
//! This module provides the core functionality for merging multiple segments
//! into a single optimized segment with proper handling of deletions and updates.

use std::sync::Arc;

use ahash::{AHashMap, AHashSet};

use crate::error::{LaurusError, Result};
use crate::lexical::core::analyzed::{AnalyzedDocument, AnalyzedTerm};
use crate::lexical::index::inverted::reader::{InvertedIndexReader, SegmentReader};
use crate::lexical::index::inverted::segment::SegmentInfo;
use crate::lexical::index::inverted::segment::manager::{
    ManagedSegmentInfo, MergeCandidate, MergeStrategy,
};
use crate::lexical::index::inverted::writer::{InvertedIndexWriter, InvertedIndexWriterConfig};
use crate::lexical::index::structures::aabb::AABB;
use crate::lexical::index::structures::visitor::{CellRelation, IntersectVisitor};
use crate::lexical::reader::LexicalIndexReader;
use crate::storage::Storage;

/// Configuration for merge operations.
#[derive(Debug, Clone)]
pub struct MergeConfig {
    /// Maximum memory usage during merge (in bytes).
    pub max_memory_mb: u64,

    /// Number of documents to process in each batch.
    pub batch_size: usize,

    /// Enable compression during merge.
    pub enable_compression: bool,

    /// Remove deleted documents during merge.
    pub remove_deleted_docs: bool,

    /// Sort documents by ID during merge for better locality.
    pub sort_by_doc_id: bool,

    /// Verify integrity after merge.
    pub verify_after_merge: bool,
}

impl Default for MergeConfig {
    fn default() -> Self {
        MergeConfig {
            max_memory_mb: 256,
            batch_size: 10000,
            enable_compression: true,
            remove_deleted_docs: true,
            sort_by_doc_id: true,
            verify_after_merge: true,
        }
    }
}

/// Statistics about a merge operation.
#[derive(Debug, Clone, Default)]
pub struct MergeStats {
    /// Number of segments merged.
    pub segments_merged: usize,

    /// Number of documents processed.
    pub docs_processed: u64,

    /// Number of deleted documents removed.
    pub deleted_docs_removed: u64,

    /// Size before merge (in bytes).
    pub size_before: u64,

    /// Size after merge (in bytes).
    pub size_after: u64,

    /// Time taken for merge (in milliseconds).
    pub merge_time_ms: u64,

    /// Compression ratio achieved.
    pub compression_ratio: f64,

    /// Terms merged.
    pub terms_merged: u64,

    /// Postings merged.
    pub postings_merged: u64,

    /// Shard ID for the merged segment.
    pub shard_id: u16,
}

impl MergeStats {
    /// Calculate space savings percentage.
    pub fn space_savings(&self) -> f64 {
        if self.size_before == 0 {
            0.0
        } else {
            ((self.size_before - self.size_after) as f64 / self.size_before as f64) * 100.0
        }
    }
}

/// Result of a merge operation.
#[derive(Debug)]
pub struct MergeResult {
    /// Information about the new merged segment.
    pub new_segment: ManagedSegmentInfo,

    /// Statistics about the merge operation.
    pub stats: MergeStats,

    /// File paths of the new segment.
    pub file_paths: Vec<String>,
}

/// Core merge engine for segment operations (schema-less mode).
#[derive(Debug)]
pub struct MergeEngine {
    /// Configuration for merge operations.
    config: MergeConfig,

    /// Storage backend.
    storage: Arc<dyn Storage>,
}

impl MergeEngine {
    /// Create a new merge engine (schema-less mode).
    pub fn new(config: MergeConfig, storage: Arc<dyn Storage>) -> Self {
        MergeEngine { config, storage }
    }

    /// Merge segments according to the merge candidate.
    pub fn merge_segments(
        &self,
        candidate: &MergeCandidate,
        segments: &[ManagedSegmentInfo],
        next_generation: u64,
    ) -> Result<MergeResult> {
        let start_millis = crate::util::time::now_millis();

        // Filter segments to merge
        let segments_to_merge: Vec<_> = segments
            .iter()
            .filter(|seg| candidate.segments.contains(&seg.segment_info.segment_id))
            .collect();

        if segments_to_merge.is_empty() {
            return Err(LaurusError::index("No segments found to merge"));
        }

        // Create new segment ID
        let new_segment_id = format!("merged_{next_generation}");

        // Initialize merge statistics
        let mut stats = MergeStats {
            segments_merged: segments_to_merge.len(),
            size_before: segments_to_merge.iter().map(|s| s.size_bytes).sum(),
            ..Default::default()
        };

        // Perform merge based on strategy
        let merge_result = match candidate.strategy {
            MergeStrategy::SizeBased => self.merge_by_size(&segments_to_merge, &new_segment_id)?,
            MergeStrategy::DeletionBased => {
                self.merge_by_deletion(&segments_to_merge, &new_segment_id)?
            }
            MergeStrategy::TimeBased => self.merge_by_time(&segments_to_merge, &new_segment_id)?,
            MergeStrategy::Balanced => self.merge_balanced(&segments_to_merge, &new_segment_id)?,
        };

        // Calculate final statistics
        let end_millis = crate::util::time::now_millis();
        stats.merge_time_ms = end_millis.saturating_sub(start_millis);

        stats.size_after = merge_result.new_segment.size_bytes;
        stats.compression_ratio = if stats.size_before > 0 {
            stats.size_after as f64 / stats.size_before as f64
        } else {
            1.0
        };

        // Update merge result stats
        let mut final_result = merge_result;
        final_result.stats = stats;

        // Verify merge if configured
        if self.config.verify_after_merge {
            self.verify_merged_segment(&final_result.new_segment)?;
        }

        Ok(final_result)
    }

    /// Merge segments prioritizing size efficiency.
    fn merge_by_size(
        &self,
        segments: &[&ManagedSegmentInfo],
        new_segment_id: &str,
    ) -> Result<MergeResult> {
        // Sort segments by size (smallest first for better merging efficiency)
        let mut sorted_segments = segments.to_vec();
        sorted_segments.sort_by_key(|s| s.size_bytes);

        self.perform_merge(&sorted_segments, new_segment_id)
    }

    /// Merge segments prioritizing deletion removal.
    fn merge_by_deletion(
        &self,
        segments: &[&ManagedSegmentInfo],
        new_segment_id: &str,
    ) -> Result<MergeResult> {
        // Sort by deletion ratio (highest first for better compaction)
        let mut sorted_segments = segments.to_vec();
        sorted_segments.sort_by(|a, b| b.deletion_ratio().total_cmp(&a.deletion_ratio()));

        self.perform_merge(&sorted_segments, new_segment_id)
    }

    /// Merge segments prioritizing age.
    fn merge_by_time(
        &self,
        segments: &[&ManagedSegmentInfo],
        new_segment_id: &str,
    ) -> Result<MergeResult> {
        // Sort by creation time (oldest first)
        let mut sorted_segments = segments.to_vec();
        sorted_segments.sort_by_key(|s| s.created_at);

        self.perform_merge(&sorted_segments, new_segment_id)
    }

    /// Balanced merge considering multiple factors.
    fn merge_balanced(
        &self,
        segments: &[&ManagedSegmentInfo],
        new_segment_id: &str,
    ) -> Result<MergeResult> {
        // Calculate composite score for each segment
        let mut scored_segments: Vec<_> = segments
            .iter()
            .map(|seg| {
                let size_score = 1.0 / (seg.size_bytes as f64 + 1.0); // Prefer smaller
                let deletion_score = seg.deletion_ratio() * 2.0; // Prefer high deletion
                let age_score = 1.0 / (seg.created_at as f64 + 1.0); // Prefer older

                let composite_score = size_score + deletion_score + age_score;
                (*seg, composite_score)
            })
            .collect();

        // Sort by composite score (highest first)
        scored_segments.sort_by(|a, b| b.1.total_cmp(&a.1));

        let sorted_segments: Vec<_> = scored_segments.into_iter().map(|(seg, _)| seg).collect();

        self.perform_merge(&sorted_segments, new_segment_id)
    }

    /// Core merge implementation.
    fn perform_merge(
        &self,
        segments: &[&ManagedSegmentInfo],
        new_segment_id: &str,
    ) -> Result<MergeResult> {
        let mut stats = MergeStats {
            segments_merged: segments.len(),
            shard_id: segments
                .first()
                .map(|s| s.segment_info.shard_id)
                .unwrap_or(0),
            ..Default::default()
        };

        // Reconstruct every live document's analyzed form from each source
        // segment's postings + stored fields (no re-tokenization; #753). The
        // postings are the source of truth for the inverted index, so
        // index-only (non-stored) fields are preserved, and original doc_ids
        // are kept (they encode the shard and are referenced by deletion
        // bitmaps / external-id maps).
        //
        // `docs` keyed by doc_id with `order` tracking first-seen order makes
        // doc_id collisions across segments (an update re-wrote a doc in a
        // newer segment) resolve to the last-processed version.
        let mut order: Vec<u64> = Vec::new();
        let mut docs: AHashMap<u64, AnalyzedDocument> = AHashMap::new();
        // Whether the source segments stored term positions; detected from the
        // first posting seen so the merged segment matches.
        let mut store_positions: Option<bool> = None;

        for segment in segments {
            let reader = SegmentReader::open(segment.segment_info.clone(), self.storage.clone())?;
            let deleted = self.load_deleted_docs(&segment.segment_info)?;
            let reconstructed = self.reconstruct_segment(
                &reader,
                &segment.segment_info.segment_id,
                &deleted,
                &mut store_positions,
            )?;
            stats.deleted_docs_removed += deleted.len() as u64;
            for (doc_id, analyzed) in reconstructed {
                if docs.insert(doc_id, analyzed).is_none() {
                    order.push(doc_id);
                }
            }
        }

        if self.config.sort_by_doc_id {
            order.sort_unstable();
        }

        let doc_count = order.len() as u64;
        let min_doc_id = order.iter().copied().min().unwrap_or(0);
        let max_doc_id = order.iter().copied().max().unwrap_or(0);

        // Replay the reconstructed analyzed documents through a writer so the
        // merged segment is written by the same complete, typed write path as a
        // normal flush, then flush to the merged segment's name. Buffers are
        // unbounded so the merge produces exactly one output segment.
        let writer_config = InvertedIndexWriterConfig {
            store_term_positions: store_positions.unwrap_or(true),
            shard_id: stats.shard_id,
            max_buffered_docs: usize::MAX,
            max_buffer_memory: usize::MAX,
            ..Default::default()
        };
        let mut writer = InvertedIndexWriter::new(self.storage.clone(), writer_config)?;
        for doc_id in &order {
            if let Some(analyzed) = docs.remove(doc_id) {
                writer.upsert_analyzed_document(*doc_id, analyzed)?;
            }
        }
        let file_paths = writer.flush_buffered_to_segment(new_segment_id)?;

        stats.docs_processed = doc_count;
        stats.postings_merged = doc_count;

        // Create new segment info
        let segment_info = SegmentInfo {
            segment_id: new_segment_id.to_string(),
            doc_count,
            min_doc_id,
            max_doc_id,
            generation: 0,        // Will be assigned by segment manager
            has_deletions: false, // New merged segment has no deleted docs until updated
            shard_id: stats.shard_id,
        };

        // Calculate segment size
        let size_bytes = file_paths
            .iter()
            .map(|path| {
                self.storage
                    .metadata(path)
                    .map(|meta| meta.size)
                    .unwrap_or(0)
            })
            .sum();

        // Create managed segment info
        let mut managed_info = ManagedSegmentInfo::new(segment_info);
        managed_info.size_bytes = size_bytes;
        managed_info.file_paths = file_paths.clone();

        Ok(MergeResult {
            new_segment: managed_info,
            stats,
            file_paths,
        })
    }

    /// Load the set of deleted doc_ids for a segment from its `.delmap`.
    fn load_deleted_docs(&self, segment_info: &SegmentInfo) -> Result<AHashSet<u64>> {
        let mut deleted = AHashSet::new();
        if !segment_info.has_deletions {
            return Ok(deleted);
        }
        let bitmap_file = format!("{}.delmap", segment_info.segment_id);
        if let Ok(input) = self.storage.open_input(&bitmap_file) {
            use crate::maintenance::deletion::DeletionBitmap;
            use crate::storage::structured::StructReader;

            if let Ok(mut reader) = StructReader::new(input)
                && let Ok(bitmap) = DeletionBitmap::read_from_storage(&mut reader)
            {
                for doc_id in bitmap.get_deleted_docs() {
                    deleted.insert(doc_id);
                }
            }
        }
        Ok(deleted)
    }

    /// Load a segment reader for the given segment.
    fn load_segment_reader(
        &self,
        segment_info: &SegmentInfo,
    ) -> Result<Box<dyn LexicalIndexReader>> {
        // Create segment list with single segment
        let segments = vec![segment_info.clone()];

        // Use default config for reader
        let config = crate::lexical::index::inverted::reader::InvertedIndexReaderConfig::default();

        let reader = InvertedIndexReader::new(segments, self.storage.clone(), config)?;
        Ok(Box::new(reader) as Box<dyn LexicalIndexReader>)
    }

    /// Reconstruct every live document's [`AnalyzedDocument`] from one source
    /// segment, without re-tokenizing (Issue #753).
    ///
    /// `field_terms` are rebuilt from the segment's postings (the authoritative
    /// source for the inverted index, so index-only fields survive);
    /// `stored_fields` from the stored documents; `point_values` (BKD entries)
    /// are read back from the segment's BKD trees — the authoritative source
    /// for numeric/geo points, so index-only (`stored=false`) and multi-valued
    /// numeric fields are preserved (Issue #758); `field_lengths` are read back
    /// from the segment. Deleted docs are excluded.
    ///
    /// `store_positions` is set from the first posting seen (whether the source
    /// stored term positions) so the merged segment matches.
    fn reconstruct_segment(
        &self,
        reader: &SegmentReader,
        segment_id: &str,
        deleted: &AHashSet<u64>,
        store_positions: &mut Option<bool>,
    ) -> Result<Vec<(u64, AnalyzedDocument)>> {
        // Pass 1: bucket postings into per-doc analyzed terms.
        let mut field_terms: AHashMap<u64, AHashMap<String, Vec<AnalyzedTerm>>> = AHashMap::new();
        if let Some(dict) = reader.term_dictionary()? {
            for (term_key, _info) in dict.iter() {
                let Some((field, term)) = term_key.split_once(':') else {
                    continue;
                };
                if let Some(mut iter) = reader.postings(field, term)? {
                    while iter.next()? {
                        let doc_id = iter.doc_id();
                        if deleted.contains(&doc_id) {
                            continue;
                        }
                        let positions = iter.positions()?;
                        let freq = iter.term_freq();
                        if store_positions.is_none() {
                            *store_positions = Some(!positions.is_empty());
                        }
                        let terms = field_terms
                            .entry(doc_id)
                            .or_default()
                            .entry(field.to_string())
                            .or_default();
                        if positions.is_empty() {
                            // Frequency-only segment: one analyzed term carries
                            // the whole frequency.
                            terms.push(AnalyzedTerm {
                                term: term.to_string(),
                                position: 0,
                                frequency: freq as u32,
                                offset: (0, 0),
                            });
                        } else {
                            // One analyzed term per stored position so the
                            // rebuilt posting list reproduces the positions.
                            for pos in positions {
                                terms.push(AnalyzedTerm {
                                    term: term.to_string(),
                                    position: pos as u32,
                                    frequency: freq as u32,
                                    offset: (0, 0),
                                });
                            }
                        }
                    }
                }
            }
        }

        // Collect BKD points per (doc, field) straight from the segment's BKD
        // trees — the authoritative source for numeric/geo points. This keeps
        // index-only (`stored=false`) and multi-valued numeric fields, which a
        // stored-field derivation would miss (Issue #758).
        let mut points: AHashMap<u64, AHashMap<String, Vec<Vec<f64>>>> = AHashMap::new();
        let bkd_prefix = format!("{segment_id}.");
        let bkd_suffix = ".bkd";
        for file in self.storage.list_files()? {
            let Some(field) = file
                .strip_prefix(&bkd_prefix)
                .and_then(|rest| rest.strip_suffix(bkd_suffix))
            else {
                continue;
            };
            if let Some(tree) = reader.get_bkd_tree(field)? {
                let mut visitor = CollectPointsVisitor::default();
                tree.intersect(&mut visitor)?;
                for (doc_id, point) in visitor.entries {
                    if deleted.contains(&doc_id) {
                        continue;
                    }
                    points
                        .entry(doc_id)
                        .or_default()
                        .entry(field.to_string())
                        .or_default()
                        .push(point);
                }
            }
        }

        // Pass 2: assemble each live document.
        let mut out = Vec::new();
        for doc_id in reader.doc_ids()? {
            if deleted.contains(&doc_id) {
                continue;
            }
            let Some(stored) = reader.document(doc_id)? else {
                continue;
            };

            let mut analyzed = AnalyzedDocument::new();
            analyzed.field_terms = field_terms.remove(&doc_id).unwrap_or_default();
            analyzed.point_values = points.remove(&doc_id).unwrap_or_default();

            for (field_name, value) in &stored.fields {
                analyzed
                    .stored_fields
                    .insert(field_name.clone(), value.clone());
            }

            // Field lengths are read back from the segment so BM25 length
            // normalization is preserved exactly. Only indexed fields have a
            // recorded length.
            let indexed_fields: Vec<String> = analyzed.field_terms.keys().cloned().collect();
            for field_name in indexed_fields {
                if let Some(len) = reader.field_length(doc_id, &field_name)? {
                    analyzed.field_lengths.insert(field_name, len);
                }
            }

            out.push((doc_id, analyzed));
        }

        Ok(out)
    }

    /// Verify the integrity of a merged segment.
    fn verify_merged_segment(&self, segment: &ManagedSegmentInfo) -> Result<()> {
        // Load the segment and perform basic checks
        let reader = self.load_segment_reader(&segment.segment_info)?;

        // Check document count matches
        if reader.doc_count() != segment.segment_info.doc_count {
            return Err(LaurusError::index("Document count mismatch after merge"));
        }

        // TODO: Add more verification checks
        // - Term dictionary integrity
        // - Posting list consistency
        // - Document field validation

        Ok(())
    }

    /// Get merge configuration.
    pub fn get_config(&self) -> &MergeConfig {
        &self.config
    }
}

/// BKD visitor that enumerates **every** `(doc_id, point)` entry in a tree.
///
/// Used by the merge to read back all stored points (Issue #758). `compare`
/// always returns [`CellRelation::Crosses`] so the traversal descends to every
/// leaf and yields each point through `visit` (a `CellRelation::Inside` verdict
/// would report doc ids via `visit_inside` *without* the point coordinates,
/// which the merge needs). `visit_inside` is therefore never called.
#[derive(Default)]
struct CollectPointsVisitor {
    entries: Vec<(u64, Vec<f64>)>,
}

impl IntersectVisitor for CollectPointsVisitor {
    fn compare(&self, _cell: &AABB) -> CellRelation {
        // Force a full descent so every point is reported via `visit`.
        CellRelation::Crosses
    }

    fn visit_inside(&mut self, _doc_id: u64) {
        // Unreachable: `compare` never returns `Inside`. Points (not just doc
        // ids) are required, so all entries must arrive through `visit`.
    }

    fn visit(&mut self, doc_id: u64, point: &[f64]) {
        self.entries.push((doc_id, point.to_vec()));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::lexical::index::inverted::segment::SegmentInfo;
    use crate::lexical::index::inverted::segment::manager::ManagedSegmentInfo;

    use crate::storage::memory::MemoryStorage;
    use crate::storage::memory::MemoryStorageConfig;

    #[allow(dead_code)]
    fn create_test_segment(id: &str, doc_count: u64) -> ManagedSegmentInfo {
        let segment_info = SegmentInfo {
            segment_id: id.to_string(),
            doc_count,
            min_doc_id: 0,
            max_doc_id: doc_count.saturating_sub(1),
            generation: 1,
            has_deletions: false,
            shard_id: 0, // Added shard_id for test segments
        };

        ManagedSegmentInfo::new(segment_info)
    }

    #[test]
    fn test_merge_engine_creation() {
        let config = MergeConfig::default();
        let storage = Arc::new(MemoryStorage::new(MemoryStorageConfig::default()));

        let engine = MergeEngine::new(config, storage);
        assert_eq!(engine.config.batch_size, 10000);
        assert!(engine.config.remove_deleted_docs);
    }

    #[test]
    fn test_merge_config_default() {
        let config = MergeConfig::default();

        assert_eq!(config.max_memory_mb, 256);
        assert_eq!(config.batch_size, 10000);
        assert!(config.enable_compression);
        assert!(config.remove_deleted_docs);
        assert!(config.sort_by_doc_id);
        assert!(config.verify_after_merge);
    }

    #[test]
    fn test_merge_stats_space_savings() {
        let stats = MergeStats {
            size_before: 1000,
            size_after: 800,
            ..Default::default()
        };

        assert_eq!(stats.space_savings(), 20.0);

        let stats_zero = MergeStats {
            size_before: 0,
            size_after: 0,
            ..Default::default()
        };
        assert_eq!(stats_zero.space_savings(), 0.0);
    }

    use crate::data::{DataValue, Document};
    use crate::lexical::index::inverted::reader::{InvertedIndexReader, SegmentReader};
    use crate::lexical::index::inverted::writer::{InvertedIndexWriter, InvertedIndexWriterConfig};
    use crate::lexical::reader::LexicalIndexReader;

    fn text_int_doc(title: &str, num: i64) -> Document {
        Document::builder()
            .add_field("title", DataValue::Text(title.to_string()))
            .add_field("num", DataValue::Int64(num))
            .build()
    }

    fn load_segment_info(storage: &Arc<dyn Storage>, seg_id: &str) -> SegmentInfo {
        let mut input = storage.open_input(&format!("{seg_id}.meta")).unwrap();
        let mut buf = Vec::new();
        std::io::Read::read_to_end(&mut input, &mut buf).unwrap();
        serde_json::from_slice(&buf).unwrap()
    }

    /// End-to-end correctness of the rewritten merge (Issue #753, closes #556):
    /// merging two segments must produce one segment that preserves the
    /// documents, their *typed* stored fields (int stays int — the #556 bug
    /// stringified everything), and their searchable postings.
    #[test]
    fn merge_preserves_docs_typed_fields_and_postings() {
        let storage: Arc<dyn Storage> =
            Arc::new(MemoryStorage::new(MemoryStorageConfig::default()));

        // Two segments via two commits.
        let mut writer =
            InvertedIndexWriter::new(storage.clone(), InvertedIndexWriterConfig::default())
                .unwrap();
        let d0 = writer
            .add_document(text_int_doc("alpha bravo", 10))
            .unwrap();
        let d1 = writer
            .add_document(text_int_doc("bravo charlie", 20))
            .unwrap();
        writer.commit().unwrap(); // segment_000000
        let d2 = writer
            .add_document(text_int_doc("charlie delta", 30))
            .unwrap();
        writer.commit().unwrap(); // segment_000001
        drop(writer);

        let si0 = load_segment_info(&storage, "segment_000000");
        let si1 = load_segment_info(&storage, "segment_000001");
        let candidate = MergeCandidate {
            segments: vec![si0.segment_id.clone(), si1.segment_id.clone()],
            priority: 1.0,
            estimated_size: 0,
            strategy: MergeStrategy::SizeBased,
        };
        let engine = MergeEngine::new(MergeConfig::default(), storage.clone());
        let result = engine
            .merge_segments(
                &candidate,
                &[ManagedSegmentInfo::new(si0), ManagedSegmentInfo::new(si1)],
                1,
            )
            .unwrap();

        // All three docs survive the merge (verify_after_merge also checks this).
        assert_eq!(result.new_segment.segment_info.doc_count, 3);

        // Typed stored fields round-trip: `num` stays an Int64 (the #556 bug
        // wrote it as a stringified value).
        let merged =
            SegmentReader::open(result.new_segment.segment_info.clone(), storage.clone()).unwrap();
        for (doc_id, expected) in [(d0, 10i64), (d1, 20), (d2, 30)] {
            let doc = merged
                .document(doc_id)
                .unwrap()
                .unwrap_or_else(|| panic!("doc {doc_id} missing after merge"));
            match doc.fields.get("num") {
                Some(DataValue::Int64(n)) => assert_eq!(*n, expected, "doc {doc_id} num"),
                other => panic!("doc {doc_id} `num` not Int64 after merge: {other:?}"),
            }
        }

        // Postings are reconstructed (not empty): "bravo" appears in d0 and d1.
        let reader = InvertedIndexReader::new(
            vec![result.new_segment.segment_info.clone()],
            storage.clone(),
            Default::default(),
        )
        .unwrap();
        let mut got = Vec::new();
        if let Some(mut it) = reader.postings("title", "bravo").unwrap() {
            while it.next().unwrap() {
                got.push(it.doc_id());
            }
        }
        got.sort_unstable();
        let mut want = vec![d0, d1];
        want.sort_unstable();
        assert_eq!(got, want, "`title:bravo` postings after merge");
    }
}