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

hermes_core/index/
mod.rs
