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

hermes_core/index/
mod.rs
