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};
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        let metadata = IndexMetadata::new((*schema).clone());
246
247        let segment_manager = Arc::new(crate::merge::SegmentManager::new(
248            Arc::clone(&directory),
249            Arc::clone(&schema),
250            metadata,
251            config.merge_policy.clone_box(),
252            config.term_cache_blocks,
253            config.max_concurrent_merges,
254            Arc::clone(&config.background_merge_permits),
255            config.merge_bp_time_budget,
256            config.bp_memory_budget_bytes,
257            Arc::clone(&config.background_reorder_permits),
258            config.background_reorder_pool.clone(),
259        ));
260
261        // Save initial metadata
262        segment_manager.update_metadata(|_| {}).await?;
263
264        Ok(Self {
265            directory,
266            schema,
267            config,
268            search_resources,
269            segment_manager,
270            cached_reader: tokio::sync::OnceCell::new(),
271        })
272    }
273
274    /// Open an existing index from a directory
275    pub async fn open(directory: D, config: IndexConfig) -> Result<Self> {
276        let search_resources = searcher::SearcherResources::new(
277            config.term_cache_blocks,
278            config.store_cache_blocks,
279            config.num_threads,
280        )?;
281        let directory = Arc::new(directory);
282
283        // Load metadata (includes schema)
284        let metadata = IndexMetadata::load(directory.as_ref()).await?;
285        let schema = Arc::new(metadata.schema.clone());
286        // Directory-layer metrics (cold writes, lazy reads) carry the index label
287        directory.set_index_label(schema.index_label());
288
289        let segment_manager = Arc::new(crate::merge::SegmentManager::new(
290            Arc::clone(&directory),
291            Arc::clone(&schema),
292            metadata,
293            config.merge_policy.clone_box(),
294            config.term_cache_blocks,
295            config.max_concurrent_merges,
296            Arc::clone(&config.background_merge_permits),
297            config.merge_bp_time_budget,
298            config.bp_memory_budget_bytes,
299            Arc::clone(&config.background_reorder_permits),
300            config.background_reorder_pool.clone(),
301        ));
302
303        // Load trained structures into SegmentManager's ArcSwap
304        segment_manager.try_load_and_publish_trained().await?;
305
306        Ok(Self {
307            directory,
308            schema,
309            config,
310            search_resources,
311            segment_manager,
312            cached_reader: tokio::sync::OnceCell::new(),
313        })
314    }
315
316    /// Get the schema
317    pub fn schema(&self) -> &Schema {
318        &self.schema
319    }
320
321    /// Get the schema as an Arc reference (avoids clone when Arc is needed)
322    pub fn schema_arc(&self) -> &Arc<Schema> {
323        &self.schema
324    }
325
326    /// Get a reference to the underlying directory
327    pub fn directory(&self) -> &D {
328        &self.directory
329    }
330
331    /// Get the segment manager
332    pub fn segment_manager(&self) -> &Arc<crate::merge::SegmentManager<D>> {
333        &self.segment_manager
334    }
335
336    /// Get an IndexReader for searching (with reload policy)
337    ///
338    /// The reader is cached and reused across calls. The reader's internal
339    /// searcher will reload segments based on its reload interval (configurable via IndexConfig).
340    pub async fn reader(&self) -> Result<&IndexReader<D>> {
341        self.cached_reader
342            .get_or_try_init(|| async {
343                IndexReader::from_segment_manager_with_resources(
344                    Arc::clone(&self.schema),
345                    Arc::clone(&self.segment_manager),
346                    self.config.reload_interval_ms,
347                    self.search_resources.clone(),
348                )
349                .await
350            })
351            .await
352    }
353
354    /// Get the config
355    pub fn config(&self) -> &IndexConfig {
356        &self.config
357    }
358
359    /// Get segment readers for query execution (convenience method)
360    pub async fn segment_readers(&self) -> Result<Vec<Arc<crate::segment::SegmentReader>>> {
361        let reader = self.reader().await?;
362        let searcher = reader.searcher().await?;
363        Ok(searcher.segment_readers().to_vec())
364    }
365
366    /// Total number of documents across all segments
367    pub async fn num_docs(&self) -> Result<u32> {
368        let reader = self.reader().await?;
369        let searcher = reader.searcher().await?;
370        Ok(searcher.num_docs())
371    }
372
373    /// Get default fields for search
374    pub fn default_fields(&self) -> Vec<crate::Field> {
375        if !self.schema.default_fields().is_empty() {
376            self.schema.default_fields().to_vec()
377        } else {
378            self.schema
379                .fields()
380                .filter(|(_, entry)| {
381                    entry.indexed && entry.field_type == crate::dsl::FieldType::Text
382                })
383                .map(|(field, _)| field)
384                .collect()
385        }
386    }
387
388    /// Get tokenizer registry
389    pub fn tokenizers(&self) -> Arc<crate::tokenizer::TokenizerRegistry> {
390        Arc::new(crate::tokenizer::TokenizerRegistry::default())
391    }
392
393    /// Create a query parser for this index
394    pub fn query_parser(&self) -> crate::dsl::QueryLanguageParser {
395        let default_fields = self.default_fields();
396        let tokenizers = self.tokenizers();
397
398        let query_routers = self.schema.query_routers();
399        if !query_routers.is_empty()
400            && let Ok(router) = crate::dsl::QueryFieldRouter::from_rules(query_routers)
401        {
402            return crate::dsl::QueryLanguageParser::with_router(
403                Arc::clone(&self.schema),
404                default_fields,
405                tokenizers,
406                router,
407            );
408        }
409
410        crate::dsl::QueryLanguageParser::new(Arc::clone(&self.schema), default_fields, tokenizers)
411    }
412
413    /// Parse and search using a query string
414    pub async fn query(
415        &self,
416        query_str: &str,
417        limit: usize,
418    ) -> Result<crate::query::SearchResponse> {
419        self.query_offset(query_str, limit, 0).await
420    }
421
422    /// Query with offset for pagination
423    pub async fn query_offset(
424        &self,
425        query_str: &str,
426        limit: usize,
427        offset: usize,
428    ) -> Result<crate::query::SearchResponse> {
429        let parser = self.query_parser();
430        let query = parser
431            .parse(query_str)
432            .map_err(crate::error::Error::Query)?;
433        self.search_offset(query.as_ref(), limit, offset).await
434    }
435
436    /// Search and return results
437    pub async fn search(
438        &self,
439        query: &dyn crate::query::Query,
440        limit: usize,
441    ) -> Result<crate::query::SearchResponse> {
442        self.search_offset(query, limit, 0).await
443    }
444
445    /// Search with offset for pagination
446    pub async fn search_offset(
447        &self,
448        query: &dyn crate::query::Query,
449        limit: usize,
450        offset: usize,
451    ) -> Result<crate::query::SearchResponse> {
452        let reader = self.reader().await?;
453        let searcher = reader.searcher().await?;
454
455        #[cfg(feature = "sync")]
456        let (results, total_seen) = {
457            // Sync search: rayon handles segment parallelism internally.
458            // On multi-threaded tokio, use block_in_place to yield the worker;
459            // on single-threaded (tests), call directly.
460            let runtime_flavor = tokio::runtime::Handle::current().runtime_flavor();
461            if runtime_flavor == tokio::runtime::RuntimeFlavor::MultiThread {
462                tokio::task::block_in_place(|| {
463                    searcher.search_with_offset_and_count_sync(query, limit, offset)
464                })?
465            } else {
466                searcher.search_with_offset_and_count_sync(query, limit, offset)?
467            }
468        };
469
470        #[cfg(not(feature = "sync"))]
471        let (results, total_seen) = {
472            searcher
473                .search_with_offset_and_count(query, limit, offset)
474                .await?
475        };
476
477        let total_hits = total_seen;
478        let hits: Vec<crate::query::SearchHit> = results
479            .into_iter()
480            .map(|result| crate::query::SearchHit {
481                address: crate::query::DocAddress::new(result.segment_id, result.doc_id),
482                score: result.score,
483                matched_fields: result.extract_ordinals(),
484            })
485            .collect();
486
487        Ok(crate::query::SearchResponse { hits, total_hits })
488    }
489
490    /// Get a document by its unique address
491    pub async fn get_document(
492        &self,
493        address: &crate::query::DocAddress,
494    ) -> Result<Option<crate::dsl::Document>> {
495        let reader = self.reader().await?;
496        let searcher = reader.searcher().await?;
497        searcher.get_document(address).await
498    }
499
500    /// Get posting lists for a term across all segments
501    pub async fn get_postings(
502        &self,
503        field: crate::Field,
504        term: &[u8],
505    ) -> Result<
506        Vec<(
507            Arc<crate::segment::SegmentReader>,
508            crate::structures::BlockPostingList,
509        )>,
510    > {
511        let segments = self.segment_readers().await?;
512        let mut results = Vec::new();
513
514        for segment in segments {
515            if let Some(postings) = segment.get_postings(field, term).await? {
516                results.push((segment, postings));
517            }
518        }
519
520        Ok(results)
521    }
522}
523
524/// Native-only methods for Index
525#[cfg(feature = "native")]
526impl<D: crate::directories::DirectoryWriter + 'static> Index<D> {
527    /// Get an IndexWriter for adding documents
528    pub fn writer(&self) -> writer::IndexWriter<D> {
529        writer::IndexWriter::from_index(self)
530    }
531}
532
533#[cfg(test)]
534mod tests;
535
536// (tests moved to index/tests/ module)
hermes_core/index/mod.rs

hermes_core/index/
mod.rs
