///|
/// Result of memory-budgeted indexing. `flushed_segments` counts immutable
/// segments emitted before the final flush; `peak_estimated_bytes` is a
/// conservative input-side estimate rather than allocator RSS.
pub struct BudgetedWriterStats {
memory_budget_bytes : Int
flushed_segments : Int
indexed_documents : Int
peak_estimated_bytes : Int
}
///|
pub fn BudgetedWriterStats::memory_budget_bytes(
self : BudgetedWriterStats,
) -> Int {
self.memory_budget_bytes
}
///|
pub fn BudgetedWriterStats::flushed_segments(self : BudgetedWriterStats) -> Int {
self.flushed_segments
}
///|
pub fn BudgetedWriterStats::indexed_documents(
self : BudgetedWriterStats,
) -> Int {
self.indexed_documents
}
///|
pub fn BudgetedWriterStats::peak_estimated_bytes(
self : BudgetedWriterStats,
) -> Int {
self.peak_estimated_bytes
}
///|
fn estimated_value_bytes(value : FieldValue) -> Int {
match value {
@core.Text(text) | @core.Keyword(text) => @utf8.encode(text).length() + 16
@core.Bytes(bytes) => bytes.length() + 16
@core.I64(_) | @core.U64(_) | @core.F64(_) | @core.Date(_) => 16
@core.Bool(_) => 8
}
}
///|
/// Estimates the mutable indexing footprint contributed by a document. The
/// multiplier accounts for term text, postings, positions, offsets and stored
/// fields and intentionally errs high so automatic flush happens early.
pub fn estimate_document_memory(document : Document) -> Int {
let mut bytes = 64
for index in 0.. bytes += @utf8.encode(text).length() * 3 + 48
None => ()
}
}
for index in 0.. bytes += estimated_value_bytes(value)
None => ()
}
}
bytes
}
///|
/// Schema-aware Segment writer that automatically seals segments at a memory
/// budget. It is deterministic on every MoonBit backend and is the unit that
/// native executors can shard across independent indexing lanes.
pub struct BudgetedSegmentWriter {
schema : Schema
tokenizers : TokenizerManager
memory_budget_bytes : Int
mut current : SegmentWriter[WhitespaceAnalyzer]
mut current_estimated_bytes : Int
mut current_documents : Int
segments : Array[Segment]
mut indexed_documents : Int
mut peak_estimated_bytes : Int
}
///|
pub fn BudgetedSegmentWriter::new(
schema : Schema,
tokenizers : TokenizerManager,
memory_budget_bytes : Int,
) -> BudgetedSegmentWriter raise AnalysisError {
guard memory_budget_bytes > 0 else {
abort("index memory budget must be positive")
}
let snapshot = tokenizers.snapshot()
{
schema,
tokenizers: snapshot,
memory_budget_bytes,
current: SegmentWriter::with_schema_and_tokenizers(schema, snapshot),
current_estimated_bytes: 0,
current_documents: 0,
segments: [],
indexed_documents: 0,
peak_estimated_bytes: 0,
}
}
///|
fn BudgetedSegmentWriter::seal_current(
self : BudgetedSegmentWriter,
) -> Unit raise AnalysisError {
if self.current_documents == 0 {
return
}
self.segments.push(self.current.finish())
self.current = SegmentWriter::with_schema_and_tokenizers(
self.schema,
self.tokenizers,
)
self.current_estimated_bytes = 0
self.current_documents = 0
}
///|
/// Adds a document and flushes before it when the next document would exceed
/// the budget. A single oversized document is accepted into its own segment.
pub fn BudgetedSegmentWriter::add_document(
self : BudgetedSegmentWriter,
document : Document,
) -> DocId raise AnalysisError {
let estimated = estimate_document_memory(document)
if self.current_documents > 0 &&
self.current_estimated_bytes + estimated > self.memory_budget_bytes {
self.seal_current()
}
let doc_id = self.current.add_document(document)
self.current_documents += 1
self.indexed_documents += 1
self.current_estimated_bytes += estimated
if self.current_estimated_bytes > self.peak_estimated_bytes {
self.peak_estimated_bytes = self.current_estimated_bytes
}
if self.current_estimated_bytes >= self.memory_budget_bytes {
self.seal_current()
}
doc_id
}
///|
pub fn BudgetedSegmentWriter::flush(
self : BudgetedSegmentWriter,
) -> Unit raise AnalysisError {
self.seal_current()
}
///|
pub fn BudgetedSegmentWriter::finish(
self : BudgetedSegmentWriter,
) -> ReadOnlyArray[Segment] raise AnalysisError {
self.seal_current()
ReadOnlyArray::from_array(self.segments.copy())
}
///|
pub fn BudgetedSegmentWriter::stats(
self : BudgetedSegmentWriter,
) -> BudgetedWriterStats {
{
memory_budget_bytes: self.memory_budget_bytes,
flushed_segments: self.segments.length(),
indexed_documents: self.indexed_documents,
peak_estimated_bytes: self.peak_estimated_bytes,
}
}
///|
/// Portable capability report for segment-sharded indexing. MoonSearch keeps
/// the sharding contract backend-neutral; the current built-in executor is
/// cooperative and therefore reports effective parallelism 1.
pub struct IndexingExecutionPlan {
requested_parallelism : Int
effective_parallelism : Int
degraded : Bool
}
///|
pub fn IndexingExecutionPlan::portable(
requested_parallelism : Int,
) -> IndexingExecutionPlan {
guard requested_parallelism > 0 else {
abort("requested indexing parallelism must be positive")
}
{
requested_parallelism,
effective_parallelism: 1,
degraded: requested_parallelism > 1,
}
}
///|
pub fn IndexingExecutionPlan::requested_parallelism(
self : IndexingExecutionPlan,
) -> Int {
self.requested_parallelism
}
///|
pub fn IndexingExecutionPlan::effective_parallelism(
self : IndexingExecutionPlan,
) -> Int {
self.effective_parallelism
}
///|
pub fn IndexingExecutionPlan::is_degraded(self : IndexingExecutionPlan) -> Bool {
self.degraded
}