hermes_core/index/
mod.rs

1//! Index - multi-segment async search index
2//!
3//! The `Index` is the central concept that provides:
4//! - `Index::create()` / `Index::open()` - create or open an index
5//! - `index.writer()` - get an IndexWriter for adding documents
6//! - `index.reader()` - get an IndexReader for searching (with reload policy)
7//!
8//! The Index owns the SegmentManager which handles segment lifecycle and tracking.
9
10#[cfg(feature = "native")]
11use crate::dsl::Schema;
12#[cfg(feature = "native")]
13use crate::error::Result;
14#[cfg(feature = "sync")]
15use std::collections::HashMap;
16#[cfg(feature = "native")]
17use std::sync::Arc;
18#[cfg(feature = "native")]
19use std::sync::{OnceLock, Weak};
20
21mod searcher;
22pub use searcher::Searcher;
23
24#[cfg(feature = "native")]
25mod primary_key;
26#[cfg(feature = "native")]
27mod reader;
28#[cfg(feature = "native")]
29mod vector_builder;
30#[cfg(all(feature = "wasm", not(feature = "native")))]
31mod wasm_writer;
32#[cfg(feature = "native")]
33mod writer;
34#[cfg(feature = "native")]
35pub use primary_key::PrimaryKeyIndex;
36#[cfg(feature = "native")]
37pub use reader::IndexReader;
38#[cfg(all(feature = "wasm", not(feature = "native")))]
39pub use wasm_writer::IndexWriter as WasmIndexWriter;
40#[cfg(feature = "native")]
41pub use writer::{IndexWriter, PreparedCommit, WRITER_LOCK_FILENAME};
42
43mod metadata;
44pub use metadata::{
45    FieldVectorMeta, INDEX_META_FILENAME, IndexMetadata, SegmentMetaInfo, VectorIndexState,
46};
47
48#[cfg(feature = "native")]
49mod helpers;
50#[cfg(feature = "native")]
51pub use helpers::{
52    IndexingStats, SchemaConfig, SchemaFieldConfig, create_index_at_path, create_index_from_sdl,
53    index_documents_from_reader, index_json_document, parse_schema,
54};
55
56/// Default file name for the slice cache
57pub const SLICE_CACHE_FILENAME: &str = "index.slicecache";
58
59/// A BP pass can consume every background CPU worker and the complete
60/// per-pass memory allowance. More than two simultaneous passes only
61/// oversubscribe the same pool and multiply memory-bandwidth pressure.
62#[cfg(feature = "native")]
63pub const MAX_CONCURRENT_REORDER_PASSES: usize = 2;
64
65#[cfg(feature = "native")]
66#[derive(Clone, Copy, Debug, Eq, PartialEq)]
67pub(crate) enum ReorderPriority {
68    /// Periodic optimizer and explicit standalone reorder work. This class
69    /// must retain capacity while automatic merges continuously arrive,
70    /// otherwise fresh segments stay unordered for minutes behind giant BP
71    /// passes and query pruning recovers slowly after ingestion.
72    Optimizer,
73    /// BP performed while producing an automatic merge output.
74    AutomaticMerge,
75    Foreground,
76}
77
78/// Application-wide gate shared by optimizer, merge-time, and manual BP.
79///
80/// Besides enforcing the hard two-pass ceiling, the gate lets an explicit
81/// force merge reserve all but one slot. Background passes already running
82/// finish normally; new ones wait until the force merge releases its guard.
83#[cfg(feature = "native")]
84#[derive(Debug)]
85pub struct ReorderConcurrencyGate {
86    permits: Arc<tokio::sync::Semaphore>,
87    /// Automatic merges may consume all but one whole-pass slot. The reserved
88    /// slot lets short optimizer passes continuously retire fresh segments.
89    /// With a one-pass configuration both classes share the only slot.
90    automatic_merge_permits: Arc<tokio::sync::Semaphore>,
91    limit: usize,
92    foreground_lock: Arc<tokio::sync::Mutex<()>>,
93    foreground_active: std::sync::atomic::AtomicBool,
94    foreground_finished: tokio::sync::Notify,
95}
96
97/// Process-wide cap on simultaneously active BMP segment scorers.
98///
99/// Each scorer performs random mmap reads. Letting every segment of every
100/// concurrent query run at once multiplies page faults without increasing
101/// useful NVMe throughput, so this gate is independent from the CPU pool.
102#[cfg(feature = "native")]
103#[derive(Debug)]
104pub(crate) struct BmpIoGate {
105    limit: usize,
106    active: parking_lot::Mutex<usize>,
107    available: parking_lot::Condvar,
108    async_available: tokio::sync::Notify,
109}
110
111#[cfg(feature = "native")]
112impl BmpIoGate {
113    fn new(limit: usize) -> Self {
114        Self {
115            limit,
116            active: parking_lot::Mutex::new(0),
117            available: parking_lot::Condvar::new(),
118            async_available: tokio::sync::Notify::new(),
119        }
120    }
121
122    fn acquire(&self) -> BmpIoPermit<'_> {
123        let mut active = self.active.lock();
124        while *active >= self.limit {
125            self.available.wait(&mut active);
126        }
127        *active += 1;
128        BmpIoPermit { gate: self }
129    }
130
131    async fn acquire_async(&self) -> BmpIoPermit<'_> {
132        loop {
133            // Register before checking the counter, so a release between the
134            // check and await cannot be lost.
135            let notified = self.async_available.notified();
136            {
137                let mut active = self.active.lock();
138                if *active < self.limit {
139                    *active += 1;
140                    return BmpIoPermit { gate: self };
141                }
142            }
143            notified.await;
144        }
145    }
146}
147
148#[cfg(feature = "native")]
149struct BmpIoPermit<'a> {
150    gate: &'a BmpIoGate,
151}
152
153#[cfg(feature = "native")]
154impl Drop for BmpIoPermit<'_> {
155    fn drop(&mut self) {
156        let mut active = self.gate.active.lock();
157        *active -= 1;
158        self.gate.available.notify_one();
159        self.gate.async_available.notify_one();
160    }
161}
162
163#[cfg(feature = "native")]
164impl ReorderConcurrencyGate {
165    pub fn new(requested_limit: usize) -> Self {
166        let limit = requested_limit.clamp(1, MAX_CONCURRENT_REORDER_PASSES);
167        let automatic_merge_limit = limit.saturating_sub(1).max(1);
168        Self {
169            permits: Arc::new(tokio::sync::Semaphore::new(limit)),
170            automatic_merge_permits: Arc::new(tokio::sync::Semaphore::new(automatic_merge_limit)),
171            limit,
172            foreground_lock: Arc::new(tokio::sync::Mutex::new(())),
173            foreground_active: std::sync::atomic::AtomicBool::new(false),
174            foreground_finished: tokio::sync::Notify::new(),
175        }
176    }
177
178    pub fn limit(&self) -> usize {
179        self.limit
180    }
181
182    pub(crate) async fn acquire(
183        self: &Arc<Self>,
184        priority: ReorderPriority,
185    ) -> std::result::Result<ReorderPermit, tokio::sync::AcquireError> {
186        match priority {
187            ReorderPriority::Optimizer => self.acquire_background(None).await,
188            ReorderPriority::AutomaticMerge => {
189                let merge_permit = Arc::clone(&self.automatic_merge_permits)
190                    .acquire_owned()
191                    .await?;
192                self.acquire_background(Some(merge_permit)).await
193            }
194            ReorderPriority::Foreground => self.acquire_foreground().await,
195        }
196    }
197
198    /// Acquire capacity for periodic optimizer or automatic merge work.
199    async fn acquire_background(
200        self: &Arc<Self>,
201        automatic_merge: Option<tokio::sync::OwnedSemaphorePermit>,
202    ) -> std::result::Result<ReorderPermit, tokio::sync::AcquireError> {
203        loop {
204            if self
205                .foreground_active
206                .load(std::sync::atomic::Ordering::Acquire)
207            {
208                let notified = self.foreground_finished.notified();
209                if self
210                    .foreground_active
211                    .load(std::sync::atomic::Ordering::Acquire)
212                {
213                    notified.await;
214                    continue;
215                }
216            }
217
218            let permit = Arc::clone(&self.permits).acquire_owned().await?;
219            if !self
220                .foreground_active
221                .load(std::sync::atomic::Ordering::Acquire)
222            {
223                return Ok(ReorderPermit {
224                    _permit: permit,
225                    _automatic_merge: automatic_merge,
226                });
227            }
228            // A foreground operation started between the check and permit
229            // acquisition. Yield the slot instead of extending its queue.
230            drop(permit);
231        }
232    }
233
234    /// Acquire the one BP slot left available to a foreground force merge.
235    async fn acquire_foreground(
236        self: &Arc<Self>,
237    ) -> std::result::Result<ReorderPermit, tokio::sync::AcquireError> {
238        let permit = Arc::clone(&self.permits).acquire_owned().await?;
239        Ok(ReorderPermit {
240            _permit: permit,
241            _automatic_merge: None,
242        })
243    }
244
245    /// Prioritize one explicit force merge across all indexes using this gate.
246    ///
247    /// Foreground operations are serialized to avoid two force merges each
248    /// reserving one slot and then waiting for the other. The guard is
249    /// cancellation-safe and releases reservations on drop.
250    pub(crate) async fn begin_foreground(
251        self: &Arc<Self>,
252    ) -> std::result::Result<ForegroundReorderGuard, tokio::sync::AcquireError> {
253        let exclusive = Arc::clone(&self.foreground_lock).lock_owned().await;
254        self.foreground_active
255            .store(true, std::sync::atomic::Ordering::Release);
256
257        // Construct the guard before awaiting capacity. If this future is
258        // cancelled while existing background work drains, Drop clears the
259        // active flag and releases the foreground mutex.
260        let mut guard = ForegroundReorderGuard {
261            gate: Arc::clone(self),
262            reserved: None,
263            _exclusive: exclusive,
264        };
265        if self.limit > 1 {
266            guard.reserved = Some(
267                Arc::clone(&self.permits)
268                    .acquire_many_owned((self.limit - 1) as u32)
269                    .await?,
270            );
271        }
272        Ok(guard)
273    }
274}
275
276#[cfg(feature = "native")]
277pub(crate) struct ReorderPermit {
278    _permit: tokio::sync::OwnedSemaphorePermit,
279    _automatic_merge: Option<tokio::sync::OwnedSemaphorePermit>,
280}
281
282#[cfg(feature = "native")]
283pub(crate) struct ForegroundReorderGuard {
284    gate: Arc<ReorderConcurrencyGate>,
285    reserved: Option<tokio::sync::OwnedSemaphorePermit>,
286    _exclusive: tokio::sync::OwnedMutexGuard<()>,
287}
288
289#[cfg(feature = "native")]
290impl Drop for ForegroundReorderGuard {
291    fn drop(&mut self) {
292        // Make capacity visible before waking background waiters.
293        drop(self.reserved.take());
294        self.gate
295            .foreground_active
296            .store(false, std::sync::atomic::Ordering::Release);
297        self.gate.foreground_finished.notify_waiters();
298    }
299}
300
301/// Index configuration
302#[derive(Debug, Clone)]
303pub struct IndexConfig {
304    /// Number of threads shared by CPU-intensive search work.
305    ///
306    /// Indexes in the same process that request the same width reuse one Rayon
307    /// pool. A value of zero is invalid and is rejected by `Index::create` and
308    /// `Index::open`.
309    pub num_threads: usize,
310    /// Maximum BMP segment scorers issuing random mmap reads concurrently
311    /// across the process. CPU parallelism remains controlled by
312    /// `num_threads`; this separate cap protects the page cache and storage
313    /// queue from segment/query fan-out.
314    pub bmp_io_concurrency: usize,
315    /// Number of parallel segment builders (documents distributed round-robin)
316    pub num_indexing_threads: usize,
317    /// Number of threads for parallel block compression within each segment
318    pub num_compression_threads: usize,
319    /// Block cache size for term dictionary per segment
320    pub term_cache_blocks: usize,
321    /// Process-wide byte budget for decompressed document-store blocks.
322    ///
323    /// Indexes opened with the same budget share one read-concurrent,
324    /// byte-bounded cache. This is a byte limit rather than a block count
325    /// because a stored document can legitimately make one decompressed block
326    /// tens of MiB.
327    pub store_cache_budget_bytes: usize,
328    /// Max memory (bytes) across all builders before auto-commit (global limit)
329    pub max_indexing_memory_bytes: usize,
330    /// Maximum vectors retained for one field's global ANN training sample.
331    /// The byte budget below is applied at the same time; the smaller bound
332    /// wins. Fields are sampled and trained serially.
333    pub vector_training_max_samples: usize,
334    /// Maximum raw vector bytes retained for one field's ANN training sample.
335    pub vector_training_memory_bytes: usize,
336    /// Merge policy for background segment merging
337    pub merge_policy: Box<dyn crate::merge::MergePolicy>,
338    /// Index optimization mode (adaptive, size-optimized, performance-optimized)
339    pub optimization: crate::structures::IndexOptimization,
340    /// Reload interval in milliseconds for IndexReader (how often to check for new segments)
341    pub reload_interval_ms: u64,
342    /// Maximum number of concurrent background merges per index (default: 4)
343    pub max_concurrent_merges: usize,
344    /// Application-wide background merge gate shared by clones of this
345    /// config. The per-index limit alone multiplied large merge working sets
346    /// by the number of active indexes.
347    #[cfg(feature = "native")]
348    pub background_merge_permits: Arc<tokio::sync::Semaphore>,
349    /// Wall-clock budget for merge-time BP reorder per field (only applies
350    /// when the index has `reorder_on_merge`). A truncated pass still writes
351    /// a valid, better-ordered segment; it is marked `bp_converged = false`
352    /// and the background optimizer deepens it later (warm-started).
353    /// `None` = unbudgeted (BP runs to full depth inside the merge, which can
354    /// hold a merge slot for 10-30+ minutes on 10M+ doc outputs).
355    pub merge_bp_time_budget: Option<std::time::Duration>,
356    /// Memory budget (bytes) for the BP forward index during reorder passes
357    /// (merge-time and background). When a large segment's forward index
358    /// would exceed this, the highest-df dims are dropped from BP's input
359    /// (logged loudly) — clustering quality degrades gracefully. Production
360    /// evidence: 18M-doc merges exceeded the former 2 GB default and dropped
361    /// ~10% of eligible dims; hosts with less headroom may lower this.
362    pub bp_memory_budget_bytes: usize,
363    /// Hard limit on simultaneous whole-segment BP rewrites. This is shared
364    /// by all indexes opened from clones of this config and applies to
365    /// optimizer, merge-time, and manual reorder passes. It is deliberately
366    /// separate from the Rayon pool width: one pass can already use every
367    /// background CPU thread and consume the full BP memory budget.
368    #[cfg(feature = "native")]
369    pub background_reorder_permits: Arc<ReorderConcurrencyGate>,
370    /// Optional process/application-owned Rayon pool for BP work. Supplying
371    /// one lets every index and the optimizer share the same worker threads;
372    /// `None` lazily uses one process-wide cores/2 fallback pool.
373    #[cfg(feature = "native")]
374    pub background_reorder_pool: Option<Arc<rayon::ThreadPool>>,
375}
376
377/// Search pools are shared process-wide by width. This avoids multiplying OS
378/// threads by the number of open indexes while still allowing applications to
379/// deliberately isolate indexes that need different CPU budgets.
380#[cfg(feature = "sync")]
381static SEARCH_CPU_POOLS: OnceLock<parking_lot::Mutex<HashMap<usize, Weak<rayon::ThreadPool>>>> =
382    OnceLock::new();
383
384/// Store caches are shared process-wide by configured byte budget, just like
385/// search CPU pools are shared by width. `IndexRegistry` clones one config for
386/// every index, but standalone callers with the same policy also converge on
387/// the same bounded cache.
388#[cfg(feature = "native")]
389static STORE_CACHE_POOLS: OnceLock<
390    parking_lot::Mutex<std::collections::HashMap<usize, Weak<crate::segment::SharedStoreCache>>>,
391> = OnceLock::new();
392
393#[cfg(feature = "native")]
394static BMP_IO_GATES: OnceLock<
395    parking_lot::Mutex<std::collections::HashMap<usize, Weak<BmpIoGate>>>,
396> = OnceLock::new();
397
398#[cfg(feature = "native")]
399pub(crate) fn shared_bmp_io_gate(limit: usize) -> Arc<BmpIoGate> {
400    let mut gates = BMP_IO_GATES
401        .get_or_init(|| parking_lot::Mutex::new(std::collections::HashMap::new()))
402        .lock();
403    if let Some(gate) = gates.get(&limit).and_then(Weak::upgrade) {
404        return gate;
405    }
406    let gate = Arc::new(BmpIoGate::new(limit));
407    gates.retain(|_, gate| gate.strong_count() > 0);
408    gates.insert(limit, Arc::downgrade(&gate));
409    log::info!("[bmp] process-wide random-I/O concurrency={limit}");
410    gate
411}
412
413#[cfg(feature = "native")]
414pub(crate) fn shared_store_cache(budget_bytes: usize) -> Arc<crate::segment::SharedStoreCache> {
415    let mut caches = STORE_CACHE_POOLS
416        .get_or_init(|| parking_lot::Mutex::new(std::collections::HashMap::new()))
417        .lock();
418    if let Some(cache) = caches.get(&budget_bytes).and_then(Weak::upgrade) {
419        return cache;
420    }
421    let cache = Arc::new(crate::segment::SharedStoreCache::new(budget_bytes));
422    caches.retain(|_, cache| cache.strong_count() > 0);
423    caches.insert(budget_bytes, Arc::downgrade(&cache));
424    log::info!(
425        "[store_cache] process-wide budget={}",
426        crate::format_bytes(budget_bytes as u64)
427    );
428    cache
429}
430
431#[cfg(feature = "sync")]
432fn shared_search_pool(num_threads: usize) -> Result<Arc<rayon::ThreadPool>> {
433    if num_threads == 0 {
434        return Err(crate::Error::Internal(
435            "IndexConfig.num_threads must be greater than zero".into(),
436        ));
437    }
438
439    let mut pools = SEARCH_CPU_POOLS
440        .get_or_init(|| parking_lot::Mutex::new(HashMap::new()))
441        .lock();
442    if let Some(pool) = pools.get(&num_threads).and_then(Weak::upgrade) {
443        return Ok(pool);
444    }
445
446    // Build while holding the registry lock. Index construction is cold-path
447    // work, and serialization here prevents two concurrent opens from creating
448    // duplicate pools for the same width.
449    let pool = Arc::new(
450        rayon::ThreadPoolBuilder::new()
451            .num_threads(num_threads)
452            .thread_name(move |idx| format!("hermes-search-{}-{}", num_threads, idx))
453            .build()
454            .map_err(|error| {
455                crate::Error::Internal(format!(
456                    "failed to create {num_threads}-thread search pool: {error}"
457                ))
458            })?,
459    );
460    pools.retain(|_, pool| pool.strong_count() > 0);
461    pools.insert(num_threads, Arc::downgrade(&pool));
462    log::info!("[search] process-wide CPU pool: {} thread(s)", num_threads);
463    Ok(pool)
464}
465
466impl Default for IndexConfig {
467    fn default() -> Self {
468        #[cfg(feature = "native")]
469        let compression_threads = crate::default_compression_threads();
470        #[cfg(not(feature = "native"))]
471        let compression_threads = 1;
472
473        #[cfg(feature = "native")]
474        let search_threads = crate::default_search_threads();
475        #[cfg(not(feature = "native"))]
476        let search_threads = 1;
477
478        Self {
479            num_threads: search_threads,
480            bmp_io_concurrency: 4,
481            num_indexing_threads: 1, // Increase to 2+ for production to avoid stalls during segment build
482            num_compression_threads: compression_threads,
483            term_cache_blocks: 256,
484            // Stored bodies can be much larger than the writer's nominal
485            // 16-KiB block target. Keep this process-wide and byte bounded so
486            // segment fan-out cannot multiply it into tens of GiB.
487            #[cfg(target_pointer_width = "64")]
488            store_cache_budget_bytes: 2 * 1024 * 1024 * 1024,
489            #[cfg(not(target_pointer_width = "64"))]
490            store_cache_budget_bytes: 32 * 1024 * 1024,
491            max_indexing_memory_bytes: 256 * 1024 * 1024, // 256 MB default
492            vector_training_max_samples: 10_000_000,
493            #[cfg(target_pointer_width = "64")]
494            vector_training_memory_bytes: 4 * 1024 * 1024 * 1024,
495            #[cfg(not(target_pointer_width = "64"))]
496            vector_training_memory_bytes: usize::MAX,
497            // large_scale: wide fan-in + budget/scored selection. Safe for
498            // small indexes too (tier floors only shape *when* segments
499            // merge); merge-time BP is wall-clock budgeted, so giant merges
500            // cannot hold slots indefinitely.
501            merge_policy: Box::new(crate::merge::TieredMergePolicy::large_scale()),
502            optimization: crate::structures::IndexOptimization::default(),
503            reload_interval_ms: 1000, // 1 second default
504            max_concurrent_merges: 4,
505            #[cfg(feature = "native")]
506            background_merge_permits: Arc::new(tokio::sync::Semaphore::new(4)),
507            merge_bp_time_budget: Some(std::time::Duration::from_secs(600)),
508            // 24 GB — mirrors segment::reorder::DEFAULT_MEMORY_BUDGET (that
509            // module is native-only; IndexConfig also compiles for wasm).
510            // A cap, not an allocation: usage is proportional to the segment
511            // being reordered (~4 B/posting + ~32 B/doc). Sized from prod
512            // evidence: a 58M-doc/5B-posting pass estimated 20.1 GB, which
513            // 8/16 GB budgets trimmed by dropping highest-df dims.
514            // 24 GB overflows 32-bit usize (wasm32) — reorder never runs
515            // there, so any large value works; use usize::MAX.
516            #[cfg(target_pointer_width = "64")]
517            bp_memory_budget_bytes: 24 * 1024 * 1024 * 1024,
518            #[cfg(not(target_pointer_width = "64"))]
519            bp_memory_budget_bytes: usize::MAX,
520            #[cfg(feature = "native")]
521            background_reorder_permits: Arc::new(ReorderConcurrencyGate::new(2)),
522            #[cfg(feature = "native")]
523            background_reorder_pool: None,
524        }
525    }
526}
527
528/// Multi-segment async Index
529///
530/// The central concept for search. Owns segment lifecycle and provides:
531/// - `Index::create()` / `Index::open()` - create or open an index
532/// - `index.writer()` - get an IndexWriter for adding documents
533/// - `index.reader()` - get an IndexReader for searching with reload policy
534///
535/// All segment management is delegated to SegmentManager.
536#[cfg(feature = "native")]
537pub struct Index<D: crate::directories::DirectoryWriter + 'static> {
538    directory: Arc<D>,
539    schema: Arc<Schema>,
540    config: IndexConfig,
541    /// Cache and CPU policy used by every searcher reload.
542    search_resources: searcher::SearcherResources,
543    /// Segment manager - owns segments, tracker, metadata, and trained structures
544    segment_manager: Arc<crate::merge::SegmentManager<D>>,
545    /// Cached reader (created lazily, reused across calls)
546    cached_reader: tokio::sync::OnceCell<IndexReader<D>>,
547}
548
549#[cfg(feature = "native")]
550impl<D: crate::directories::DirectoryWriter + 'static> Index<D> {
551    /// Create a new index in the directory
552    pub async fn create(directory: D, schema: Schema, config: IndexConfig) -> Result<Self> {
553        let search_resources = searcher::SearcherResources::new(
554            config.term_cache_blocks,
555            config.store_cache_budget_bytes,
556            config.num_threads,
557            config.bmp_io_concurrency,
558        )?;
559        let directory = Arc::new(directory);
560        let schema = Arc::new(schema);
561        // Directory-layer metrics (cold writes, lazy reads) carry the index label
562        directory.set_index_label(schema.index_label());
563
564        // Refuse to clobber an existing index: persisting a fresh empty
565        // metadata.json would orphan every committed segment, and the next
566        // writer open's orphan sweep would permanently delete them.
567        if directory
568            .exists(std::path::Path::new(INDEX_META_FILENAME))
569            .await?
570        {
571            return Err(crate::Error::Internal(format!(
572                "refusing to create index: {} already exists in this directory; \
573                 use Index::open to open the existing index, or delete the \
574                 directory first if you really want to start over",
575                INDEX_META_FILENAME
576            )));
577        }
578
579        let metadata = IndexMetadata::new((*schema).clone());
580
581        let segment_manager = Arc::new(crate::merge::SegmentManager::new(
582            Arc::clone(&directory),
583            Arc::clone(&schema),
584            metadata,
585            config.merge_policy.clone_box(),
586            config.term_cache_blocks,
587            config.max_concurrent_merges,
588            Arc::clone(&config.background_merge_permits),
589            config.merge_bp_time_budget,
590            config.bp_memory_budget_bytes,
591            Arc::clone(&config.background_reorder_permits),
592            config.background_reorder_pool.clone(),
593        ));
594
595        // Save initial metadata
596        segment_manager.update_metadata(|_| {}).await?;
597
598        Ok(Self {
599            directory,
600            schema,
601            config,
602            search_resources,
603            segment_manager,
604            cached_reader: tokio::sync::OnceCell::new(),
605        })
606    }
607
608    /// Open an existing index from a directory
609    pub async fn open(directory: D, config: IndexConfig) -> Result<Self> {
610        let search_resources = searcher::SearcherResources::new(
611            config.term_cache_blocks,
612            config.store_cache_budget_bytes,
613            config.num_threads,
614            config.bmp_io_concurrency,
615        )?;
616        let directory = Arc::new(directory);
617
618        // Load metadata (includes schema)
619        let metadata = IndexMetadata::load(directory.as_ref()).await?;
620        let schema = Arc::new(metadata.schema.clone());
621        // Directory-layer metrics (cold writes, lazy reads) carry the index label
622        directory.set_index_label(schema.index_label());
623
624        let segment_manager = Arc::new(crate::merge::SegmentManager::new(
625            Arc::clone(&directory),
626            Arc::clone(&schema),
627            metadata,
628            config.merge_policy.clone_box(),
629            config.term_cache_blocks,
630            config.max_concurrent_merges,
631            Arc::clone(&config.background_merge_permits),
632            config.merge_bp_time_budget,
633            config.bp_memory_budget_bytes,
634            Arc::clone(&config.background_reorder_permits),
635            config.background_reorder_pool.clone(),
636        ));
637
638        // Load trained structures into SegmentManager's ArcSwap
639        segment_manager.try_load_and_publish_trained().await?;
640
641        Ok(Self {
642            directory,
643            schema,
644            config,
645            search_resources,
646            segment_manager,
647            cached_reader: tokio::sync::OnceCell::new(),
648        })
649    }
650
651    /// Get the schema
652    pub fn schema(&self) -> &Schema {
653        &self.schema
654    }
655
656    /// Get the schema as an Arc reference (avoids clone when Arc is needed)
657    pub fn schema_arc(&self) -> &Arc<Schema> {
658        &self.schema
659    }
660
661    /// Get a reference to the underlying directory
662    pub fn directory(&self) -> &D {
663        &self.directory
664    }
665
666    /// Get the segment manager
667    pub fn segment_manager(&self) -> &Arc<crate::merge::SegmentManager<D>> {
668        &self.segment_manager
669    }
670
671    /// Get an IndexReader for searching (with reload policy)
672    ///
673    /// The reader is cached and reused across calls. The reader's internal
674    /// searcher will reload segments based on its reload interval (configurable via IndexConfig).
675    pub async fn reader(&self) -> Result<&IndexReader<D>> {
676        self.cached_reader
677            .get_or_try_init(|| async {
678                IndexReader::from_segment_manager_with_resources(
679                    Arc::clone(&self.schema),
680                    Arc::clone(&self.segment_manager),
681                    self.config.reload_interval_ms,
682                    self.search_resources.clone(),
683                )
684                .await
685            })
686            .await
687    }
688
689    /// Get the config
690    pub fn config(&self) -> &IndexConfig {
691        &self.config
692    }
693
694    /// Get segment readers for query execution (convenience method)
695    pub async fn segment_readers(&self) -> Result<Vec<Arc<crate::segment::SegmentReader>>> {
696        let reader = self.reader().await?;
697        let searcher = reader.searcher().await?;
698        Ok(searcher.segment_readers().to_vec())
699    }
700
701    /// Total number of documents across all segments
702    pub async fn num_docs(&self) -> Result<u32> {
703        let reader = self.reader().await?;
704        let searcher = reader.searcher().await?;
705        Ok(searcher.num_docs())
706    }
707
708    /// Get default fields for search
709    pub fn default_fields(&self) -> Vec<crate::Field> {
710        if !self.schema.default_fields().is_empty() {
711            self.schema.default_fields().to_vec()
712        } else {
713            self.schema
714                .fields()
715                .filter(|(_, entry)| {
716                    entry.indexed && entry.field_type == crate::dsl::FieldType::Text
717                })
718                .map(|(field, _)| field)
719                .collect()
720        }
721    }
722
723    /// Get tokenizer registry
724    pub fn tokenizers(&self) -> Arc<crate::tokenizer::TokenizerRegistry> {
725        Arc::new(crate::tokenizer::TokenizerRegistry::default())
726    }
727
728    /// Create a query parser for this index
729    pub fn query_parser(&self) -> crate::dsl::QueryLanguageParser {
730        let default_fields = self.default_fields();
731        let tokenizers = self.tokenizers();
732
733        let query_routers = self.schema.query_routers();
734        if !query_routers.is_empty()
735            && let Ok(router) = crate::dsl::QueryFieldRouter::from_rules(query_routers)
736        {
737            return crate::dsl::QueryLanguageParser::with_router(
738                Arc::clone(&self.schema),
739                default_fields,
740                tokenizers,
741                router,
742            );
743        }
744
745        crate::dsl::QueryLanguageParser::new(Arc::clone(&self.schema), default_fields, tokenizers)
746    }
747
748    /// Parse and search using a query string
749    pub async fn query(
750        &self,
751        query_str: &str,
752        limit: usize,
753    ) -> Result<crate::query::SearchResponse> {
754        self.query_offset(query_str, limit, 0).await
755    }
756
757    /// Query with offset for pagination
758    pub async fn query_offset(
759        &self,
760        query_str: &str,
761        limit: usize,
762        offset: usize,
763    ) -> Result<crate::query::SearchResponse> {
764        let parser = self.query_parser();
765        let query = parser
766            .parse(query_str)
767            .map_err(crate::error::Error::Query)?;
768        self.search_offset(query.as_ref(), limit, offset).await
769    }
770
771    /// Search and return results
772    pub async fn search(
773        &self,
774        query: &dyn crate::query::Query,
775        limit: usize,
776    ) -> Result<crate::query::SearchResponse> {
777        self.search_offset(query, limit, 0).await
778    }
779
780    /// Search with offset for pagination
781    pub async fn search_offset(
782        &self,
783        query: &dyn crate::query::Query,
784        limit: usize,
785        offset: usize,
786    ) -> Result<crate::query::SearchResponse> {
787        let reader = self.reader().await?;
788        let searcher = reader.searcher().await?;
789
790        #[cfg(feature = "sync")]
791        let (results, total_seen) = {
792            // Sync search: rayon handles segment parallelism internally.
793            // On multi-threaded tokio, use block_in_place to yield the worker;
794            // on single-threaded (tests), call directly.
795            let runtime_flavor = tokio::runtime::Handle::current().runtime_flavor();
796            if runtime_flavor == tokio::runtime::RuntimeFlavor::MultiThread {
797                tokio::task::block_in_place(|| {
798                    searcher.search_with_offset_and_count_sync(query, limit, offset)
799                })?
800            } else {
801                searcher.search_with_offset_and_count_sync(query, limit, offset)?
802            }
803        };
804
805        #[cfg(not(feature = "sync"))]
806        let (results, total_seen) = {
807            searcher
808                .search_with_offset_and_count(query, limit, offset)
809                .await?
810        };
811
812        let total_hits = total_seen;
813        let hits: Vec<crate::query::SearchHit> = results
814            .into_iter()
815            .map(|result| crate::query::SearchHit {
816                address: crate::query::DocAddress::new(result.segment_id, result.doc_id),
817                score: result.score,
818                matched_fields: result.extract_ordinals(),
819            })
820            .collect();
821
822        Ok(crate::query::SearchResponse { hits, total_hits })
823    }
824
825    /// Get a document by its unique address
826    pub async fn get_document(
827        &self,
828        address: &crate::query::DocAddress,
829    ) -> Result<Option<crate::dsl::Document>> {
830        let reader = self.reader().await?;
831        let searcher = reader.searcher().await?;
832        searcher.get_document(address).await
833    }
834
835    /// Get posting lists for a term across all segments
836    pub async fn get_postings(
837        &self,
838        field: crate::Field,
839        term: &[u8],
840    ) -> Result<
841        Vec<(
842            Arc<crate::segment::SegmentReader>,
843            crate::structures::BlockPostingList,
844        )>,
845    > {
846        let segments = self.segment_readers().await?;
847        let mut results = Vec::new();
848
849        for segment in segments {
850            if let Some(postings) = segment.get_postings(field, term).await? {
851                results.push((segment, postings));
852            }
853        }
854
855        Ok(results)
856    }
857}
858
859/// Native-only methods for Index
860#[cfg(feature = "native")]
861impl<D: crate::directories::DirectoryWriter + 'static> Index<D> {
862    /// Get an IndexWriter for adding documents
863    pub fn writer(&self) -> writer::IndexWriter<D> {
864        writer::IndexWriter::from_index(self)
865    }
866}
867
868#[cfg(test)]
869mod tests;
870
871// (tests moved to index/tests/ module)
hermes_core/index/mod.rs

hermes_core/index/
mod.rs
