///|
/// One document occurrence in a posting list.
pub(all) struct Posting {
doc_id : DocId
term_freq : Int
positions : ReadOnlyArray[Int]
position_lengths : ReadOnlyArray[Int]
start_offsets : ReadOnlyArray[Int]
end_offsets : ReadOnlyArray[Int]
value_indexes : ReadOnlyArray[Int]
} derive(Eq, @debug.Debug)
///|
pub(all) struct PostingOccurrence {
position : Int
position_length : Int
start_offset : Int
end_offset : Int
value_index : Int
} derive(Eq, @debug.Debug)
///|
pub fn Posting::occurrences(self : Posting) -> ReadOnlyArray[PostingOccurrence] {
let occurrences : Array[PostingOccurrence] = []
for index in 0.. ReadOnlyArray[Term] {
self.terms
}
///|
fn compare_terms(left : Term, right : Term) -> Int {
let field_order = left.field_id.value.compare(right.field_id.value)
if field_order != 0 {
field_order
} else {
let left_bytes = @utf8.encode(left.text)
let right_bytes = @utf8.encode(right.text)
let common_length = if left_bytes.length() < right_bytes.length() {
left_bytes.length()
} else {
right_bytes.length()
}
for index in 0.. (ReadOnlyArray[Term], ReadOnlyArray[ReadOnlyArray[Posting]]) {
let order : Array[Int] = []
for index in 0.. compare_terms(terms[left], terms[right]))
let sorted_terms : Array[Term] = []
let sorted_postings : Array[ReadOnlyArray[Posting]] = []
for index in order {
sorted_terms.push(terms[index])
sorted_postings.push(posting_lists[index])
}
(
ReadOnlyArray::from_array(sorted_terms),
ReadOnlyArray::from_array(sorted_postings),
)
}
///|
/// Returns the ordinal of an exact term using a lower-bound binary search.
pub fn TermDictionary::seek(self : TermDictionary, term : Term) -> Int? {
let mut low = 0
let mut high = self.terms.length()
while low < high {
let middle = low + (high - low) / 2
if compare_terms(self.terms[middle], term) < 0 {
low = middle + 1
} else {
high = middle
}
}
if low < self.terms.length() && self.terms[low] == term {
Some(low)
} else {
None
}
}
///|
fn text_has_prefix(text : String, prefix : String) -> Bool {
let text_bytes = @utf8.encode(text)
let prefix_bytes = @utf8.encode(prefix)
if prefix_bytes.length() > text_bytes.length() {
return false
}
for index in 0.. ReadOnlyArray[Term] {
let probe = Term::new(field_id, prefix)
let mut low = 0
let mut high = self.terms.length()
while low < high {
let middle = low + (high - low) / 2
if compare_terms(self.terms[middle], probe) < 0 {
low = middle + 1
} else {
high = middle
}
}
let matches : Array[Term] = []
let mut index = low
while index < self.terms.length() {
let term = self.terms[index]
if term.field_id != field_id || !text_has_prefix(term.text, prefix) {
break
}
matches.push(term)
index += 1
}
ReadOnlyArray::from_array(matches)
}
///|
/// Forward-only view over an immutable posting list.
pub struct PostingCursor {
postings : ReadOnlyArray[Posting]
blocks : ReadOnlyArray[PostingSkipBlock]
mut cursor : Int
mut skipped_blocks : Int
}
///|
fn PostingCursor::new(postings : ReadOnlyArray[Posting]) -> PostingCursor {
let blocks : Array[PostingSkipBlock] = []
let mut start = 0
while start < postings.length() {
let end = if start + posting_block_size < postings.length() {
start + posting_block_size
} else {
postings.length()
}
blocks.push({ start, end, max_doc: postings[end - 1].doc_id })
start = end
}
{
postings,
blocks: ReadOnlyArray::from_array(blocks),
cursor: -1,
skipped_blocks: 0,
}
}
///|
pub fn PostingCursor::advance(self : PostingCursor) -> Bool {
let next = self.cursor + 1
if next >= 0 && next < self.postings.length() {
self.cursor = next
true
} else {
self.cursor = self.postings.length()
false
}
}
///|
/// Seeks to the first posting whose document ID is at least `target`.
pub fn PostingCursor::advance_to(self : PostingCursor, target : DocId) -> Bool {
if self.cursor >= 0 &&
self.cursor < self.postings.length() &&
self.postings[self.cursor].doc_id.value >= target.value {
return true
}
let search_start = if self.cursor < 0 { 0 } else { self.cursor + 1 }
let mut first_block = search_start / posting_block_size
let original_block = first_block
let mut block_high = self.blocks.length()
while first_block < block_high {
let middle = first_block + (block_high - first_block) / 2
if self.blocks[middle].max_doc.value < target.value {
first_block = middle + 1
} else {
block_high = middle
}
}
if first_block >= self.blocks.length() {
self.skipped_blocks += self.blocks.length() - original_block
self.cursor = self.postings.length()
return false
}
if first_block > original_block {
self.skipped_blocks += first_block - original_block
}
let block = self.blocks[first_block]
let mut low = if search_start > block.start {
search_start
} else {
block.start
}
let mut high = block.end
while low < high {
let middle = low + (high - low) / 2
if self.postings[middle].doc_id.value < target.value {
low = middle + 1
} else {
high = middle
}
}
if low < self.postings.length() {
self.cursor = low
true
} else {
self.cursor = self.postings.length()
false
}
}
///|
pub fn PostingCursor::skip_blocks(
self : PostingCursor,
) -> ReadOnlyArray[PostingSkipBlock] {
self.blocks
}
///|
/// Number of complete blocks bypassed by `advance_to` calls.
pub fn PostingCursor::skipped_block_count(self : PostingCursor) -> Int {
self.skipped_blocks
}
///|
pub fn PostingCursor::doc(self : PostingCursor) -> DocId {
match self.posting() {
Some(posting) => posting.doc_id
None => DocId::new(-1)
}
}
///|
pub fn PostingCursor::posting(self : PostingCursor) -> Posting? {
if self.cursor >= 0 && self.cursor < self.postings.length() {
Some(self.postings[self.cursor])
} else {
None
}
}
///|
pub fn PostingCursor::term_freq(self : PostingCursor) -> Int {
match self.posting() {
Some(posting) => posting.term_freq
None => 0
}
}
///|
pub fn PostingCursor::positions(self : PostingCursor) -> ReadOnlyArray[Int] {
match self.posting() {
Some(posting) => posting.positions
None => []
}
}
///|
struct FieldLength {
field_id : FieldId
mut length : Int
}
///|
struct FieldStats {
field_id : FieldId
mut document_count : Int
mut total_length : Int
}
///|
priv struct FieldPositionState {
field_id : FieldId
mut next_position : Int
}
///|
pub(all) struct FieldValueEntry {
field_id : FieldId
value : FieldValue
} derive(Eq, @debug.Debug)
///|
/// Column-oriented, docID-addressable values for one fast field.
pub struct FastFieldColumn {
field_id : FieldId
values : ReadOnlyArray[ReadOnlyArray[FieldValue]]
}
///|
pub fn FastFieldColumn::field_id(self : FastFieldColumn) -> FieldId {
self.field_id
}
///|
pub fn FastFieldColumn::values(
self : FastFieldColumn,
doc_id : DocId,
) -> ReadOnlyArray[FieldValue] {
if doc_id.value < 0 || doc_id.value >= self.values.length() {
[]
} else {
self.values[doc_id.value]
}
}
///|
/// Mutable builder for one in-memory segment.
pub struct SegmentWriter[A] {
tokenizer : A
tokenizer_manager : TokenizerManager?
schema : Schema?
terms : Array[Term]
term_indexes : @hashmap.HashMap[String, Int]
posting_doc_ids : Array[Array[DocId]]
posting_term_freqs : Array[Array[Int]]
posting_positions : Array[Array[Array[Int]]]
posting_position_lengths : Array[Array[Array[Int]]]
posting_start_offsets : Array[Array[Array[Int]]]
posting_end_offsets : Array[Array[Array[Int]]]
posting_value_indexes : Array[Array[Array[Int]]]
stored_documents : Array[StoredDocument]
document_values : Array[Array[FieldValueEntry]]
document_field_lengths : Array[Array[FieldLength]]
field_stats : Array[FieldStats]
mut next_doc_id : Int
}
///|
pub fn[A] SegmentWriter::new(tokenizer : A) -> SegmentWriter[A] {
{
tokenizer,
tokenizer_manager: None,
schema: None,
terms: [],
term_indexes: @hashmap.HashMap([]),
posting_doc_ids: [],
posting_term_freqs: [],
posting_positions: [],
posting_position_lengths: [],
posting_start_offsets: [],
posting_end_offsets: [],
posting_value_indexes: [],
stored_documents: [],
document_values: [],
document_field_lengths: [],
field_stats: [],
next_doc_id: 0,
}
}
///|
/// Creates a writer that applies the schema's indexed and stored text options.
pub fn[A] SegmentWriter::with_schema(
schema : Schema,
tokenizer : A,
) -> SegmentWriter[A] {
{
tokenizer,
tokenizer_manager: None,
schema: Some(schema),
terms: [],
term_indexes: @hashmap.HashMap([]),
posting_doc_ids: [],
posting_term_freqs: [],
posting_positions: [],
posting_position_lengths: [],
posting_start_offsets: [],
posting_end_offsets: [],
posting_value_indexes: [],
stored_documents: [],
document_values: [],
document_field_lengths: [],
field_stats: [],
next_doc_id: 0,
}
}
///|
fn Document::snapshot_with_schema(
self : Document,
schema : Schema,
) -> StoredDocument {
let field_ids : Array[FieldId] = []
let texts : Array[String] = []
let value_field_ids : Array[FieldId] = []
let values : Array[FieldValue] = []
for index in 0.. field_id
None => abort("document field identifier snapshot is inconsistent")
}
if schema.is_stored(field_id) {
field_ids.push(field_id)
match self.text_at(index) {
Some(text) => texts.push(text)
None => abort("document text snapshot is inconsistent")
}
}
}
for index in 0.. field_id
None => abort("document typed field identifier snapshot is inconsistent")
}
if schema.is_stored(field_id) {
value_field_ids.push(field_id)
match self.value_at(index) {
Some(value) => values.push(value)
None => abort("document typed value snapshot is inconsistent")
}
}
}
StoredDocument::from_all_fields(field_ids, texts, value_field_ids, values)
}
///|
/// Creates a schema-aware writer that resolves each indexed field's Tokenizer
/// pipeline through the supplied TokenizerManager before accepting documents.
pub fn SegmentWriter::with_schema_and_tokenizers(
schema : Schema,
tokenizer_manager : TokenizerManager,
) -> SegmentWriter[WhitespaceAnalyzer] raise AnalysisError {
tokenizer_manager.validate_schema(schema)
let tokenizer_snapshot = tokenizer_manager.snapshot()
{
tokenizer: WhitespaceAnalyzer::new(),
tokenizer_manager: Some(tokenizer_snapshot),
schema: Some(schema),
terms: [],
term_indexes: @hashmap.HashMap([]),
posting_doc_ids: [],
posting_term_freqs: [],
posting_positions: [],
posting_position_lengths: [],
posting_start_offsets: [],
posting_end_offsets: [],
posting_value_indexes: [],
stored_documents: [],
document_values: [],
document_field_lengths: [],
field_stats: [],
next_doc_id: 0,
}
}
///|
fn field_value_matches_type(field_type : FieldType, value : FieldValue) -> Bool {
match (field_type, value) {
(@schema.Keyword, @core.Keyword(_)) => true
(@schema.I64, @core.I64(_)) => true
(@schema.U64, @core.U64(_)) => true
(@schema.F64, @core.F64(_)) => true
(@schema.Bool, @core.Bool(_)) => true
(@schema.Date, @core.Date(_)) => true
(@schema.Bytes, @core.Bytes(_)) => true
_ => false
}
}
///|
fn[A] SegmentWriter::add_occurrence(
self : SegmentWriter[A],
term : Term,
doc_id : DocId,
position : Int,
position_length : Int,
start_offset : Int,
end_offset : Int,
value_index : Int,
) -> Unit {
let term_key = "\{term.field_id.value}:\{term.text}"
match self.term_indexes.get(term_key) {
Some(term_index) => {
let posting_index = self.posting_doc_ids[term_index].length() - 1
if posting_index >= 0 &&
self.posting_doc_ids[term_index][posting_index] == doc_id {
self.posting_term_freqs[term_index][posting_index] += 1
self.posting_positions[term_index][posting_index].push(position)
self.posting_position_lengths[term_index][posting_index].push(
position_length,
)
self.posting_start_offsets[term_index][posting_index].push(start_offset)
self.posting_end_offsets[term_index][posting_index].push(end_offset)
self.posting_value_indexes[term_index][posting_index].push(value_index)
} else {
self.posting_doc_ids[term_index].push(doc_id)
self.posting_term_freqs[term_index].push(1)
self.posting_positions[term_index].push([position])
self.posting_position_lengths[term_index].push([position_length])
self.posting_start_offsets[term_index].push([start_offset])
self.posting_end_offsets[term_index].push([end_offset])
self.posting_value_indexes[term_index].push([value_index])
}
}
None => {
self.term_indexes.set(term_key, self.terms.length())
self.terms.push(term)
self.posting_doc_ids.push([doc_id])
self.posting_term_freqs.push([1])
self.posting_positions.push([[position]])
self.posting_position_lengths.push([[position_length]])
self.posting_start_offsets.push([[start_offset]])
self.posting_end_offsets.push([[end_offset]])
self.posting_value_indexes.push([[value_index]])
}
}
}
///|
fn field_position_state(
states : Array[FieldPositionState],
field_id : FieldId,
) -> Int {
match states.search_by(state => state.field_id == field_id) {
Some(index) => index
None => {
states.push({ field_id, next_position: 0 })
states.length() - 1
}
}
}
///|
fn record_document_field_length(
lengths : Array[FieldLength],
field_id : FieldId,
token_count : Int,
) -> Unit {
match lengths.search_by(entry => entry.field_id == field_id) {
Some(index) => lengths[index].length += token_count
None => lengths.push({ field_id, length: token_count })
}
}
///|
fn[A] SegmentWriter::record_field_stats(
self : SegmentWriter[A],
lengths : Array[FieldLength],
) -> Unit {
for entry in lengths {
match
self.field_stats.search_by(stats => stats.field_id == entry.field_id) {
Some(index) => {
self.field_stats[index].document_count += 1
self.field_stats[index].total_length += entry.length
}
None =>
self.field_stats.push({
field_id: entry.field_id,
document_count: 1,
total_length: entry.length,
})
}
}
}
///|
/// Adds a document and returns its sequential segment-local identifier.
pub fn[A : @analysis.Tokenizer] SegmentWriter::add_document(
self : SegmentWriter[A],
document : Document,
) -> DocId {
let doc_id = DocId::new(self.next_doc_id)
let stored_document = match self.schema {
Some(schema) => document.snapshot_with_schema(schema)
None => document.snapshot()
}
let field_lengths : Array[FieldLength] = []
let field_positions : Array[FieldPositionState] = []
for field_index in 0.. {
guard schema.field_type(field_id) == Some(@schema.Text) else {
abort("text value does not match schema field type")
}
}
None => ()
}
let should_index = match self.schema {
Some(schema) => schema.is_indexed(field_id)
None => true
}
if !should_index {
continue
}
let token_stream = match (self.schema, self.tokenizer_manager) {
(Some(schema), Some(manager)) => {
let tokenizer_name = match schema.tokenizer_name(field_id) {
Some(name) => name
None => abort("document contains an unknown schema field")
}
manager.token_stream(tokenizer_name, document.texts[field_index]) catch {
error => abort(error.to_string())
}
}
_ => self.tokenizer.token_stream(document.texts[field_index])
}
let state_index = field_position_state(field_positions, field_id)
let position_base = field_positions[state_index].next_position
let mut token_count = 0
let mut maximum_position = -1
let mut previous_position = -1
while token_stream.advance() {
match token_stream.token() {
Some(token) => {
guard token.position >= 0 &&
token.position >= previous_position &&
token.position_length > 0 else {
abort(
"token positions must be non-decreasing and position_length must be positive",
)
}
token_count += 1
let token_final_position = token.position + token.position_length - 1
guard token_final_position >= token.position else {
abort("token end position overflowed")
}
if token_final_position > maximum_position {
maximum_position = token_final_position
}
self.add_occurrence(
Term::new(field_id, token.text),
doc_id,
position_base + token.position,
token.position_length,
token.start_offset,
token.end_offset,
value_index,
)
previous_position = token.position
}
None => ()
}
}
record_document_field_length(field_lengths, field_id, token_count)
if maximum_position >= 0 {
field_positions[state_index].next_position = position_base +
maximum_position +
2
}
}
let document_values : Array[FieldValueEntry] = []
for value_index in 0.. field_id
None => abort("document typed field identifier is inconsistent")
}
let value = match document.value_at(value_index) {
Some(value) => value
None => abort("document typed value is inconsistent")
}
match self.schema {
Some(schema) => {
let field_type = match schema.field_type(field_id) {
Some(field_type) => field_type
None => abort("document contains an unknown schema field")
}
guard field_value_matches_type(field_type, value) else {
abort("typed value does not match schema field type")
}
guard schema.is_multi_valued(field_id) ||
document_values.search_by(entry => entry.field_id == field_id) is None else {
abort("single-valued field received multiple values")
}
if schema.is_indexed(field_id) {
match value {
@core.Keyword(keyword) =>
self.add_occurrence(
Term::new(field_id, keyword),
doc_id,
0,
1,
0,
@utf8.encode(keyword).length(),
0,
)
_ => ()
}
}
}
None => ()
}
document_values.push({ field_id, value })
}
self.stored_documents.push(stored_document)
self.document_values.push(document_values)
self.record_field_stats(field_lengths)
self.document_field_lengths.push(field_lengths)
self.next_doc_id += 1
doc_id
}
///|
/// Immutable in-memory index for one batch of documents.
pub struct Segment {
schema : Schema?
indexed_terms : ReadOnlyArray[Term]
posting_lists : ReadOnlyArray[ReadOnlyArray[Posting]]
stored_field_blocks : ReadOnlyArray[StoredFieldBlock]
document_values : ReadOnlyArray[ReadOnlyArray[FieldValueEntry]]
fast_fields : ReadOnlyArray[FastFieldColumn]
document_field_lengths : ReadOnlyArray[ReadOnlyArray[FieldLength]]
field_stats : ReadOnlyArray[FieldStats]
document_count : Int
}
///|
fn freeze_field_values(
values : Array[FieldValueEntry],
) -> ReadOnlyArray[FieldValueEntry] {
let frozen : Array[FieldValueEntry] = []
for value in values {
frozen.push(value)
}
ReadOnlyArray::from_array(frozen)
}
///|
fn build_fast_fields(
schema : Schema?,
documents : ReadOnlyArray[ReadOnlyArray[FieldValueEntry]],
) -> ReadOnlyArray[FastFieldColumn] {
let columns : Array[FastFieldColumn] = []
match schema {
Some(schema) =>
for field_index in 0.. ()
}
ReadOnlyArray::from_array(columns)
}
///|
fn freeze_positions(positions : Array[Int]) -> ReadOnlyArray[Int] {
let frozen : Array[Int] = []
for position in positions {
frozen.push(position)
}
ReadOnlyArray::from_array(frozen)
}
///|
fn freeze_field_lengths(
lengths : Array[FieldLength],
) -> ReadOnlyArray[FieldLength] {
let frozen : Array[FieldLength] = []
for entry in lengths {
frozen.push(entry)
}
ReadOnlyArray::from_array(frozen)
}
///|
/// Finishes the current batch as an immutable in-memory segment.
pub fn[A] SegmentWriter::finish(self : SegmentWriter[A]) -> Segment {
let term_order : Array[Int] = []
for term_index in 0.. {
compare_terms(self.terms[left], self.terms[right])
})
let indexed_terms : Array[Term] = []
let posting_lists : Array[ReadOnlyArray[Posting]] = []
for term_index in term_order {
indexed_terms.push(self.terms[term_index])
let postings : Array[Posting] = []
for posting_index in 0.. Schema? {
self.schema
}
///|
pub fn Segment::doc_count(self : Segment) -> Int {
self.document_count
}
///|
/// Returns all indexed terms in deterministic dictionary order.
pub fn Segment::terms(self : Segment) -> ReadOnlyArray[Term] {
self.indexed_terms
}
///|
pub fn Segment::term_dictionary(self : Segment) -> TermDictionary {
{ terms: self.indexed_terms }
}
///|
pub fn Segment::term_ordinal(self : Segment, term : Term) -> Int? {
self.term_dictionary().seek(term)
}
///|
pub fn Segment::posting_cursor(self : Segment, term : Term) -> PostingCursor {
PostingCursor::new(self.postings_for(term))
}
///|
/// Returns the immutable stored document for a segment-local document ID.
pub fn Segment::doc(self : Segment, doc_id : DocId) -> StoredDocument? {
self.stored_fields_reader().doc(doc_id)
}
///|
pub fn Segment::stored_fields_reader(self : Segment) -> StoredFieldsReader {
StoredFieldsReader::new(self.stored_field_blocks, self.document_count)
}
///|
/// Returns typed indexed values for one document field.
pub fn Segment::field_values(
self : Segment,
doc_id : DocId,
field_id : FieldId,
) -> ReadOnlyArray[FieldValue] {
if doc_id.value < 0 || doc_id.value >= self.document_values.length() {
return []
}
let values : Array[FieldValue] = []
for entry in self.document_values[doc_id.value] {
if entry.field_id == field_id {
values.push(entry.value)
}
}
ReadOnlyArray::from_array(values)
}
///|
/// Returns docID-addressable fast-field values. Non-fast fields return empty.
pub fn Segment::fast_values(
self : Segment,
field_id : FieldId,
doc_id : DocId,
) -> ReadOnlyArray[FieldValue] {
match self.fast_fields.search_by(column => column.field_id == field_id) {
Some(index) => self.fast_fields[index].values(doc_id)
None => []
}
}
///|
pub fn Segment::fast_field(
self : Segment,
field_id : FieldId,
) -> FastFieldColumn? {
match self.fast_fields.search_by(column => column.field_id == field_id) {
Some(index) => Some(self.fast_fields[index])
None => None
}
}
///|
/// Returns the analyzed token count for one document field.
pub fn Segment::field_length(
self : Segment,
doc_id : DocId,
field_id : FieldId,
) -> Int {
if doc_id.value < 0 || doc_id.value >= self.document_field_lengths.length() {
return 0
}
match
self.document_field_lengths[doc_id.value].search_by(entry => {
entry.field_id == field_id
}) {
Some(index) => self.document_field_lengths[doc_id.value][index].length
None => 0
}
}
///|
pub fn Segment::has_field(
self : Segment,
doc_id : DocId,
field_id : FieldId,
) -> Bool {
if doc_id.value < 0 || doc_id.value >= self.document_field_lengths.length() {
false
} else {
self.document_field_lengths[doc_id.value].search_by(entry => {
entry.field_id == field_id
})
is Some(_) ||
self.document_values[doc_id.value].search_by(entry => {
entry.field_id == field_id
})
is Some(_)
}
}
///|
/// Number of documents that contain the field, including empty field values.
pub fn Segment::field_doc_count(self : Segment, field_id : FieldId) -> Int {
match self.field_stats.search_by(stats => stats.field_id == field_id) {
Some(index) => self.field_stats[index].document_count
None => 0
}
}
///|
/// Average analyzed token count among documents that contain the field.
pub fn Segment::average_field_length(
self : Segment,
field_id : FieldId,
) -> Double {
match self.field_stats.search_by(stats => stats.field_id == field_id) {
Some(index) => {
let stats = self.field_stats[index]
if stats.document_count == 0 {
0.0
} else {
stats.total_length.to_double() / stats.document_count.to_double()
}
}
None => 0.0
}
}
///|
/// Looks up a field-qualified term in this segment.
pub fn Segment::postings_for(
self : Segment,
term : Term,
) -> ReadOnlyArray[Posting] {
match self.term_ordinal(term) {
Some(index) => self.posting_lists[index]
None => []
}
}