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 = "sync")]
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/// Index configuration
60#[derive(Debug, Clone)]
61pub struct IndexConfig {
62    /// Number of threads shared by CPU-intensive search work.
63    ///
64    /// Indexes in the same process that request the same width reuse one Rayon
65    /// pool. A value of zero is invalid and is rejected by `Index::create` and
66    /// `Index::open`.
67    pub num_threads: usize,
68    /// Number of parallel segment builders (documents distributed round-robin)
69    pub num_indexing_threads: usize,
70    /// Number of threads for parallel block compression within each segment
71    pub num_compression_threads: usize,
72    /// Block cache size for term dictionary per segment
73    pub term_cache_blocks: usize,
74    /// Block cache size for document store per segment
75    pub store_cache_blocks: usize,
76    /// Max memory (bytes) across all builders before auto-commit (global limit)
77    pub max_indexing_memory_bytes: usize,
78    /// Merge policy for background segment merging
79    pub merge_policy: Box<dyn crate::merge::MergePolicy>,
80    /// Index optimization mode (adaptive, size-optimized, performance-optimized)
81    pub optimization: crate::structures::IndexOptimization,
82    /// Reload interval in milliseconds for IndexReader (how often to check for new segments)
83    pub reload_interval_ms: u64,
84    /// Maximum number of concurrent background merges per index (default: 4)
85    pub max_concurrent_merges: usize,
86    /// Application-wide background merge gate shared by clones of this
87    /// config. The per-index limit alone multiplied large merge working sets
88    /// by the number of active indexes.
89    #[cfg(feature = "native")]
90    pub background_merge_permits: Arc<tokio::sync::Semaphore>,
91    /// Wall-clock budget for merge-time BP reorder per field (only applies
92    /// when the index has `reorder_on_merge`). A truncated pass still writes
93    /// a valid, better-ordered segment; it is marked `bp_converged = false`
94    /// and the background optimizer deepens it later (warm-started).
95    /// `None` = unbudgeted (BP runs to full depth inside the merge, which can
96    /// hold a merge slot for 10-30+ minutes on 10M+ doc outputs).
97    pub merge_bp_time_budget: Option<std::time::Duration>,
98    /// Memory budget (bytes) for the BP forward index during reorder passes
99    /// (merge-time and background). When a large segment's forward index
100    /// would exceed this, the highest-df dims are dropped from BP's input
101    /// (logged loudly) — clustering quality degrades gracefully. Production
102    /// evidence: 18M-doc merges exceeded the former 2 GB default and dropped
103    /// ~10% of eligible dims; hosts with less headroom may lower this.
104    pub bp_memory_budget_bytes: usize,
105    /// Hard limit on simultaneous whole-segment BP rewrites. This is shared
106    /// by all indexes opened from clones of this config and applies to
107    /// optimizer, merge-time, and manual reorder passes. It is deliberately
108    /// separate from the Rayon pool width: one pass can already use every
109    /// background CPU thread and consume the full BP memory budget.
110    #[cfg(feature = "native")]
111    pub background_reorder_permits: Arc<tokio::sync::Semaphore>,
112    /// Optional process/application-owned Rayon pool for BP work. Supplying
113    /// one lets every index and the optimizer share the same worker threads;
114    /// `None` lazily uses one process-wide cores/2 fallback pool.
115    #[cfg(feature = "native")]
116    pub background_reorder_pool: Option<Arc<rayon::ThreadPool>>,
117}
118
119/// Search pools are shared process-wide by width. This avoids multiplying OS
120/// threads by the number of open indexes while still allowing applications to
121/// deliberately isolate indexes that need different CPU budgets.
122#[cfg(feature = "sync")]
123static SEARCH_CPU_POOLS: OnceLock<parking_lot::Mutex<HashMap<usize, Weak<rayon::ThreadPool>>>> =
124    OnceLock::new();
125
126#[cfg(feature = "sync")]
127fn shared_search_pool(num_threads: usize) -> Result<Arc<rayon::ThreadPool>> {
128    if num_threads == 0 {
129        return Err(crate::Error::Internal(
130            "IndexConfig.num_threads must be greater than zero".into(),
131        ));
132    }
133
134    let mut pools = SEARCH_CPU_POOLS
135        .get_or_init(|| parking_lot::Mutex::new(HashMap::new()))
136        .lock();
137    if let Some(pool) = pools.get(&num_threads).and_then(Weak::upgrade) {
138        return Ok(pool);
139    }
140
141    // Build while holding the registry lock. Index construction is cold-path
142    // work, and serialization here prevents two concurrent opens from creating
143    // duplicate pools for the same width.
144    let pool = Arc::new(
145        rayon::ThreadPoolBuilder::new()
146            .num_threads(num_threads)
147            .thread_name(move |idx| format!("hermes-search-{}-{}", num_threads, idx))
148            .build()
149            .map_err(|error| {
150                crate::Error::Internal(format!(
151                    "failed to create {num_threads}-thread search pool: {error}"
152                ))
153            })?,
154    );
155    pools.retain(|_, pool| pool.strong_count() > 0);
156    pools.insert(num_threads, Arc::downgrade(&pool));
157    log::info!("[search] process-wide CPU pool: {} thread(s)", num_threads);
158    Ok(pool)
159}
160
161impl Default for IndexConfig {
162    fn default() -> Self {
163        #[cfg(feature = "native")]
164        let compression_threads = crate::default_compression_threads();
165        #[cfg(not(feature = "native"))]
166        let compression_threads = 1;
167
168        #[cfg(feature = "native")]
169        let search_threads = crate::default_search_threads();
170        #[cfg(not(feature = "native"))]
171        let search_threads = 1;
172
173        Self {
174            num_threads: search_threads,
175            num_indexing_threads: 1, // Increase to 2+ for production to avoid stalls during segment build
176            num_compression_threads: compression_threads,
177            term_cache_blocks: 256,
178            store_cache_blocks: 32,
179            max_indexing_memory_bytes: 256 * 1024 * 1024, // 256 MB default
180            // large_scale: wide fan-in + budget/scored selection. Safe for
181            // small indexes too (tier floors only shape *when* segments
182            // merge); merge-time BP is wall-clock budgeted, so giant merges
183            // cannot hold slots indefinitely.
184            merge_policy: Box::new(crate::merge::TieredMergePolicy::large_scale()),
185            optimization: crate::structures::IndexOptimization::default(),
186            reload_interval_ms: 1000, // 1 second default
187            max_concurrent_merges: 4,
188            #[cfg(feature = "native")]
189            background_merge_permits: Arc::new(tokio::sync::Semaphore::new(4)),
190            merge_bp_time_budget: Some(std::time::Duration::from_secs(600)),
191            // 24 GB — mirrors segment::reorder::DEFAULT_MEMORY_BUDGET (that
192            // module is native-only; IndexConfig also compiles for wasm).
193            // A cap, not an allocation: usage is proportional to the segment
194            // being reordered (~4 B/posting + ~32 B/doc). Sized from prod
195            // evidence: a 58M-doc/5B-posting pass estimated 20.1 GB, which
196            // 8/16 GB budgets trimmed by dropping highest-df dims.
197            // 24 GB overflows 32-bit usize (wasm32) — reorder never runs
198            // there, so any large value works; use usize::MAX.
199            #[cfg(target_pointer_width = "64")]
200            bp_memory_budget_bytes: 24 * 1024 * 1024 * 1024,
201            #[cfg(not(target_pointer_width = "64"))]
202            bp_memory_budget_bytes: usize::MAX,
203            #[cfg(feature = "native")]
204            background_reorder_permits: Arc::new(tokio::sync::Semaphore::new(2)),
205            #[cfg(feature = "native")]
206            background_reorder_pool: None,
207        }
208    }
209}
210
211/// Multi-segment async Index
212///
213/// The central concept for search. Owns segment lifecycle and provides:
214/// - `Index::create()` / `Index::open()` - create or open an index
215/// - `index.writer()` - get an IndexWriter for adding documents
216/// - `index.reader()` - get an IndexReader for searching with reload policy
217///
218/// All segment management is delegated to SegmentManager.
219#[cfg(feature = "native")]
220pub struct Index<D: crate::directories::DirectoryWriter + 'static> {
221    directory: Arc<D>,
222    schema: Arc<Schema>,
223    config: IndexConfig,
224    /// Cache and CPU policy used by every searcher reload.
225    search_resources: searcher::SearcherResources,
226    /// Segment manager - owns segments, tracker, metadata, and trained structures
227    segment_manager: Arc<crate::merge::SegmentManager<D>>,
228    /// Cached reader (created lazily, reused across calls)
229    cached_reader: tokio::sync::OnceCell<IndexReader<D>>,
230}
231
232#[cfg(feature = "native")]
233impl<D: crate::directories::DirectoryWriter + 'static> Index<D> {
234    /// Create a new index in the directory
235    pub async fn create(directory: D, schema: Schema, config: IndexConfig) -> Result<Self> {
236        let search_resources = searcher::SearcherResources::new(
237            config.term_cache_blocks,
238            config.store_cache_blocks,
239            config.num_threads,
240        )?;
241        let directory = Arc::new(directory);
242        let schema = Arc::new(schema);
243        // Directory-layer metrics (cold writes, lazy reads) carry the index label
244        directory.set_index_label(schema.index_label());
245
246        // Refuse to clobber an existing index: persisting a fresh empty
247        // metadata.json would orphan every committed segment, and the next
248        // writer open's orphan sweep would permanently delete them.
249        if directory
250            .exists(std::path::Path::new(INDEX_META_FILENAME))
251            .await?
252        {
253            return Err(crate::Error::Internal(format!(
254                "refusing to create index: {} already exists in this directory; \
255                 use Index::open to open the existing index, or delete the \
256                 directory first if you really want to start over",
257                INDEX_META_FILENAME
258            )));
259        }
260
261        let metadata = IndexMetadata::new((*schema).clone());
262
263        let segment_manager = Arc::new(crate::merge::SegmentManager::new(
264            Arc::clone(&directory),
265            Arc::clone(&schema),
266            metadata,
267            config.merge_policy.clone_box(),
268            config.term_cache_blocks,
269            config.max_concurrent_merges,
270            Arc::clone(&config.background_merge_permits),
271            config.merge_bp_time_budget,
272            config.bp_memory_budget_bytes,
273            Arc::clone(&config.background_reorder_permits),
274            config.background_reorder_pool.clone(),
275        ));
276
277        // Save initial metadata
278        segment_manager.update_metadata(|_| {}).await?;
279
280        Ok(Self {
281            directory,
282            schema,
283            config,
284            search_resources,
285            segment_manager,
286            cached_reader: tokio::sync::OnceCell::new(),
287        })
288    }
289
290    /// Open an existing index from a directory
291    pub async fn open(directory: D, config: IndexConfig) -> Result<Self> {
292        let search_resources = searcher::SearcherResources::new(
293            config.term_cache_blocks,
294            config.store_cache_blocks,
295            config.num_threads,
296        )?;
297        let directory = Arc::new(directory);
298
299        // Load metadata (includes schema)
300        let metadata = IndexMetadata::load(directory.as_ref()).await?;
301        let schema = Arc::new(metadata.schema.clone());
302        // Directory-layer metrics (cold writes, lazy reads) carry the index label
303        directory.set_index_label(schema.index_label());
304
305        let segment_manager = Arc::new(crate::merge::SegmentManager::new(
306            Arc::clone(&directory),
307            Arc::clone(&schema),
308            metadata,
309            config.merge_policy.clone_box(),
310            config.term_cache_blocks,
311            config.max_concurrent_merges,
312            Arc::clone(&config.background_merge_permits),
313            config.merge_bp_time_budget,
314            config.bp_memory_budget_bytes,
315            Arc::clone(&config.background_reorder_permits),
316            config.background_reorder_pool.clone(),
317        ));
318
319        // Load trained structures into SegmentManager's ArcSwap
320        segment_manager.try_load_and_publish_trained().await?;
321
322        Ok(Self {
323            directory,
324            schema,
325            config,
326            search_resources,
327            segment_manager,
328            cached_reader: tokio::sync::OnceCell::new(),
329        })
330    }
331
332    /// Get the schema
333    pub fn schema(&self) -> &Schema {
334        &self.schema
335    }
336
337    /// Get the schema as an Arc reference (avoids clone when Arc is needed)
338    pub fn schema_arc(&self) -> &Arc<Schema> {
339        &self.schema
340    }
341
342    /// Get a reference to the underlying directory
343    pub fn directory(&self) -> &D {
344        &self.directory
345    }
346
347    /// Get the segment manager
348    pub fn segment_manager(&self) -> &Arc<crate::merge::SegmentManager<D>> {
349        &self.segment_manager
350    }
351
352    /// Get an IndexReader for searching (with reload policy)
353    ///
354    /// The reader is cached and reused across calls. The reader's internal
355    /// searcher will reload segments based on its reload interval (configurable via IndexConfig).
356    pub async fn reader(&self) -> Result<&IndexReader<D>> {
357        self.cached_reader
358            .get_or_try_init(|| async {
359                IndexReader::from_segment_manager_with_resources(
360                    Arc::clone(&self.schema),
361                    Arc::clone(&self.segment_manager),
362                    self.config.reload_interval_ms,
363                    self.search_resources.clone(),
364                )
365                .await
366            })
367            .await
368    }
369
370    /// Get the config
371    pub fn config(&self) -> &IndexConfig {
372        &self.config
373    }
374
375    /// Get segment readers for query execution (convenience method)
376    pub async fn segment_readers(&self) -> Result<Vec<Arc<crate::segment::SegmentReader>>> {
377        let reader = self.reader().await?;
378        let searcher = reader.searcher().await?;
379        Ok(searcher.segment_readers().to_vec())
380    }
381
382    /// Total number of documents across all segments
383    pub async fn num_docs(&self) -> Result<u32> {
384        let reader = self.reader().await?;
385        let searcher = reader.searcher().await?;
386        Ok(searcher.num_docs())
387    }
388
389    /// Get default fields for search
390    pub fn default_fields(&self) -> Vec<crate::Field> {
391        if !self.schema.default_fields().is_empty() {
392            self.schema.default_fields().to_vec()
393        } else {
394            self.schema
395                .fields()
396                .filter(|(_, entry)| {
397                    entry.indexed && entry.field_type == crate::dsl::FieldType::Text
398                })
399                .map(|(field, _)| field)
400                .collect()
401        }
402    }
403
404    /// Get tokenizer registry
405    pub fn tokenizers(&self) -> Arc<crate::tokenizer::TokenizerRegistry> {
406        Arc::new(crate::tokenizer::TokenizerRegistry::default())
407    }
408
409    /// Create a query parser for this index
410    pub fn query_parser(&self) -> crate::dsl::QueryLanguageParser {
411        let default_fields = self.default_fields();
412        let tokenizers = self.tokenizers();
413
414        let query_routers = self.schema.query_routers();
415        if !query_routers.is_empty()
416            && let Ok(router) = crate::dsl::QueryFieldRouter::from_rules(query_routers)
417        {
418            return crate::dsl::QueryLanguageParser::with_router(
419                Arc::clone(&self.schema),
420                default_fields,
421                tokenizers,
422                router,
423            );
424        }
425
426        crate::dsl::QueryLanguageParser::new(Arc::clone(&self.schema), default_fields, tokenizers)
427    }
428
429    /// Parse and search using a query string
430    pub async fn query(
431        &self,
432        query_str: &str,
433        limit: usize,
434    ) -> Result<crate::query::SearchResponse> {
435        self.query_offset(query_str, limit, 0).await
436    }
437
438    /// Query with offset for pagination
439    pub async fn query_offset(
440        &self,
441        query_str: &str,
442        limit: usize,
443        offset: usize,
444    ) -> Result<crate::query::SearchResponse> {
445        let parser = self.query_parser();
446        let query = parser
447            .parse(query_str)
448            .map_err(crate::error::Error::Query)?;
449        self.search_offset(query.as_ref(), limit, offset).await
450    }
451
452    /// Search and return results
453    pub async fn search(
454        &self,
455        query: &dyn crate::query::Query,
456        limit: usize,
457    ) -> Result<crate::query::SearchResponse> {
458        self.search_offset(query, limit, 0).await
459    }
460
461    /// Search with offset for pagination
462    pub async fn search_offset(
463        &self,
464        query: &dyn crate::query::Query,
465        limit: usize,
466        offset: usize,
467    ) -> Result<crate::query::SearchResponse> {
468        let reader = self.reader().await?;
469        let searcher = reader.searcher().await?;
470
471        #[cfg(feature = "sync")]
472        let (results, total_seen) = {
473            // Sync search: rayon handles segment parallelism internally.
474            // On multi-threaded tokio, use block_in_place to yield the worker;
475            // on single-threaded (tests), call directly.
476            let runtime_flavor = tokio::runtime::Handle::current().runtime_flavor();
477            if runtime_flavor == tokio::runtime::RuntimeFlavor::MultiThread {
478                tokio::task::block_in_place(|| {
479                    searcher.search_with_offset_and_count_sync(query, limit, offset)
480                })?
481            } else {
482                searcher.search_with_offset_and_count_sync(query, limit, offset)?
483            }
484        };
485
486        #[cfg(not(feature = "sync"))]
487        let (results, total_seen) = {
488            searcher
489                .search_with_offset_and_count(query, limit, offset)
490                .await?
491        };
492
493        let total_hits = total_seen;
494        let hits: Vec<crate::query::SearchHit> = results
495            .into_iter()
496            .map(|result| crate::query::SearchHit {
497                address: crate::query::DocAddress::new(result.segment_id, result.doc_id),
498                score: result.score,
499                matched_fields: result.extract_ordinals(),
500            })
501            .collect();
502
503        Ok(crate::query::SearchResponse { hits, total_hits })
504    }
505
506    /// Get a document by its unique address
507    pub async fn get_document(
508        &self,
509        address: &crate::query::DocAddress,
510    ) -> Result<Option<crate::dsl::Document>> {
511        let reader = self.reader().await?;
512        let searcher = reader.searcher().await?;
513        searcher.get_document(address).await
514    }
515
516    /// Get posting lists for a term across all segments
517    pub async fn get_postings(
518        &self,
519        field: crate::Field,
520        term: &[u8],
521    ) -> Result<
522        Vec<(
523            Arc<crate::segment::SegmentReader>,
524            crate::structures::BlockPostingList,
525        )>,
526    > {
527        let segments = self.segment_readers().await?;
528        let mut results = Vec::new();
529
530        for segment in segments {
531            if let Some(postings) = segment.get_postings(field, term).await? {
532                results.push((segment, postings));
533            }
534        }
535
536        Ok(results)
537    }
538}
539
540/// Native-only methods for Index
541#[cfg(feature = "native")]
542impl<D: crate::directories::DirectoryWriter + 'static> Index<D> {
543    /// Get an IndexWriter for adding documents
544    pub fn writer(&self) -> writer::IndexWriter<D> {
545        writer::IndexWriter::from_index(self)
546    }
547}
548
549#[cfg(test)]
550mod tests;
551
552// (tests moved to index/tests/ module)
hermes_core/index/mod.rs

hermes_core/index/
mod.rs
