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

hermes_core/index/
mod.rs
