///|
let segment_magic : Bytes = b"MOONSEG\x00"
///|
let segment_format_version : UInt = 5U
///|
let previous_segment_format_version : UInt = 4U
///|
let v3_segment_format_version : UInt = 3U
///|
let v2_segment_format_version : UInt = 2U
///|
let legacy_segment_format_version : UInt = 1U
///|
fn encode_schema(buffer : @buffer.Buffer, schema : Schema?) -> Unit {
match schema {
None => write_bool(buffer, false)
Some(schema) => {
write_bool(buffer, true)
buffer.write_int_be(schema.field_count())
for field_index in 0.. name
None => abort("schema field disappeared during encoding")
}
let field_type = match schema.field_type(field_id) {
Some(field_type) => field_type
None => abort("schema field type disappeared during encoding")
}
write_string(buffer, name)
buffer.write_byte(
match field_type {
@schema.Text => b'\x00'
@schema.Keyword => b'\x01'
@schema.I64 => b'\x02'
@schema.U64 => b'\x03'
@schema.F64 => b'\x04'
@schema.Bool => b'\x05'
@schema.Date => b'\x06'
@schema.Bytes => b'\x07'
},
)
write_bool(buffer, schema.is_indexed(field_id))
write_bool(buffer, schema.is_stored(field_id))
write_bool(buffer, schema.is_fast(field_id))
write_bool(buffer, schema.is_multi_valued(field_id))
write_string(
buffer,
match schema.tokenizer_name(field_id) {
Some(name) => name
None => ""
},
)
write_string(
buffer,
match schema.analyzer_version(field_id) {
Some(version) => version
None => ""
},
)
}
}
}
}
///|
fn encode_field_value(buffer : @buffer.Buffer, value : FieldValue) -> Unit {
match value {
@core.Text(value) => {
buffer.write_byte(b'\x00')
write_string(buffer, value)
}
@core.Keyword(value) => {
buffer.write_byte(b'\x01')
write_string(buffer, value)
}
@core.I64(value) => {
buffer.write_byte(b'\x02')
write_int64(buffer, value)
}
@core.U64(value) => {
buffer.write_byte(b'\x03')
write_uint64(buffer, value)
}
@core.F64(value) => {
buffer.write_byte(b'\x04')
write_double(buffer, value)
}
@core.Bool(value) => {
buffer.write_byte(b'\x05')
write_bool(buffer, value)
}
@core.Date(value) => {
buffer.write_byte(b'\x06')
write_int64(buffer, value)
}
@core.Bytes(value) => {
buffer.write_byte(b'\x07')
buffer.write_int_be(value.length())
buffer.write_bytes(value[:])
}
}
}
///|
fn common_text_prefix(left : String, right : String) -> Int {
let left_chars = left.to_array()
let right_chars = right.to_array()
let limit = if left_chars.length() < right_chars.length() {
left_chars.length()
} else {
right_chars.length()
}
let mut length = 0
while length < limit && left_chars[length] == right_chars[length] {
length += 1
}
length
}
///|
fn text_suffix(text : String, prefix_length : Int) -> String {
let chars = text.to_array()
let suffix : Array[Char] = []
for index in prefix_length.. String? {
let previous_chars = previous.to_array()
if prefix_length < 0 || prefix_length > previous_chars.length() {
return None
}
let chars : Array[Char] = []
for index in 0.. Bytes {
let buffer = @buffer.Buffer()
encode_schema(buffer, segment.schema)
buffer.write_int_be(segment.document_count)
buffer.write_int_be(segment.indexed_terms.length())
let mut previous_term_field = FieldId::new(-1)
let mut previous_term_text = ""
for term_index in 0.. Bytes {
let buffer = @buffer.Buffer(
size_hint=segment_header_length + payload.length(),
)
buffer.write_bytes(segment_magic[:])
buffer.write_uint_be(segment_format_version)
buffer.write_int_be(payload.length())
buffer.write_uint_be(segment_checksum(payload))
buffer.write_bytes(payload[:])
buffer.to_bytes()
}
///|
pub fn encode_segment(segment : Segment) -> Bytes {
encode_container(encode_payload(segment))
}
///|
fn decode_schema(
decoder : SegmentDecoder,
version : UInt,
) -> Schema? raise PersistenceError {
if !decoder.read_bool("schema marker") {
return None
}
let field_count = decoder.read_count("schema field count")
let builder = SchemaBuilder::new()
for _ in 0.. @schema.Text
b'\x01' => @schema.Keyword
b'\x02' => @schema.I64
b'\x03' => @schema.U64
b'\x04' => @schema.F64
b'\x05' => @schema.Bool
b'\x06' => @schema.Date
b'\x07' => @schema.Bytes
_ => raise PersistenceError::InvalidFormat("invalid field type")
}
let indexed = decoder.read_bool("indexed option")
let stored = decoder.read_bool("stored option")
let fast = decoder.read_bool("fast option")
let multi_valued = decoder.read_bool("multi-valued option")
let tokenizer_name = decoder.read_string("tokenizer name")
let analyzer_version = if version == segment_format_version {
decoder.read_string("analyzer version")
} else {
""
}
let field_id = match field_type {
@schema.Text => {
guard valid_tokenizer_name(tokenizer_name) else {
raise PersistenceError::InvalidFormat("invalid tokenizer name")
}
builder.add_text_field(
name,
TextOptions::new(indexed, stored)
.with_fast(fast)
.with_multi_valued(multi_valued)
.with_tokenizer(tokenizer_name)
.with_analyzer_version(analyzer_version),
)
}
_ => {
guard tokenizer_name == "" else {
raise PersistenceError::InvalidFormat(
"non-text field cannot have a tokenizer",
)
}
guard analyzer_version == "" else {
raise PersistenceError::InvalidFormat(
"non-text field cannot have an analyzer version",
)
}
let options = FieldOptions::new(indexed, stored, fast).with_multi_valued(
multi_valued,
)
match field_type {
@schema.Keyword => builder.add_keyword_field(name, options)
@schema.I64 => builder.add_i64_field(name, options)
@schema.U64 => builder.add_u64_field(name, options)
@schema.F64 => builder.add_f64_field(name, options)
@schema.Bool => builder.add_bool_field(name, options)
@schema.Date => builder.add_date_field(name, options)
@schema.Bytes => builder.add_bytes_field(name, options)
@schema.Text => abort("unreachable text field type")
}
}
}
ignore(field_id)
} else {
let indexed = decoder.read_bool("indexed option")
let stored = decoder.read_bool("stored option")
let tokenizer_name = if version == legacy_segment_format_version {
"whitespace"
} else {
decoder.read_string("tokenizer name")
}
guard valid_tokenizer_name(tokenizer_name) else {
raise PersistenceError::InvalidFormat("invalid tokenizer name")
}
ignore(
builder.add_text_field(
name,
TextOptions::new(indexed, stored).with_tokenizer(tokenizer_name),
),
)
}
}
Some(builder.build())
}
///|
fn valid_field_id(schema : Schema?, field_id : FieldId) -> Bool {
if field_id.value < 0 {
return false
}
match schema {
Some(schema) => field_id.value < schema.field_count()
None => true
}
}
///|
fn decode_block_postings(
decoder : SegmentDecoder,
document_count : Int,
) -> ReadOnlyArray[Posting] raise PersistenceError {
let block_count = decoder.read_varint("posting block count")
guard block_count <= decoder.remaining() else {
raise PersistenceError::InvalidFormat("invalid posting block count")
}
let postings : Array[Posting] = []
let mut previous_doc_id = -1
let mut previous_block_max = -1
for _ in 0.. 0 &&
count <= posting_block_size &&
block_max_doc > previous_block_max &&
block_max_doc < document_count else {
raise PersistenceError::InvalidFormat("invalid posting block metadata")
}
for _ in 0..= 0 && doc_id < document_count && doc_id > previous_doc_id else {
raise PersistenceError::InvalidFormat(
"invalid posting doc id: \{doc_id}",
)
}
guard term_freq > 0 && position_count == term_freq else {
raise PersistenceError::InvalidFormat(
"term frequency and positions disagree",
)
}
let positions : Array[Int] = []
let position_lengths : Array[Int] = []
let start_offsets : Array[Int] = []
let end_offsets : Array[Int] = []
let value_indexes : Array[Int] = []
let mut previous_position = -1
for position_index in 0..= 0 && position >= previous_position else {
raise PersistenceError::InvalidFormat("invalid term position")
}
positions.push(position)
let position_length = decoder.read_varint("position length")
let start_offset = decoder.read_varint("start offset")
let end_offset = decoder.read_varint("end offset")
let value_index = decoder.read_varint("value index")
guard position_length > 0 && end_offset >= start_offset else {
raise PersistenceError::InvalidFormat("invalid posting occurrence")
}
position_lengths.push(position_length)
start_offsets.push(start_offset)
end_offsets.push(end_offset)
value_indexes.push(value_index)
previous_position = position
}
postings.push({
doc_id: DocId::new(doc_id),
term_freq,
positions: ReadOnlyArray::from_array(positions),
position_lengths: ReadOnlyArray::from_array(position_lengths),
start_offsets: ReadOnlyArray::from_array(start_offsets),
end_offsets: ReadOnlyArray::from_array(end_offsets),
value_indexes: ReadOnlyArray::from_array(value_indexes),
})
previous_doc_id = doc_id
}
guard previous_doc_id == block_max_doc else {
raise PersistenceError::InvalidFormat("posting block max doc mismatch")
}
previous_block_max = block_max_doc
}
ReadOnlyArray::from_array(postings)
}
///|
fn decode_postings(
decoder : SegmentDecoder,
document_count : Int,
version : UInt,
) -> ReadOnlyArray[Posting] raise PersistenceError {
if version == segment_format_version {
return decode_block_postings(decoder, document_count)
}
let uses_varint = version == previous_segment_format_version ||
version == v3_segment_format_version
let posting_count = if uses_varint {
decoder.read_varint("posting count")
} else {
decoder.read_count("posting count")
}
guard posting_count <= decoder.remaining() else {
raise PersistenceError::InvalidFormat("invalid posting count")
}
let postings : Array[Posting] = []
let mut previous_doc_id = -1
for posting_index in 0.. 0 {
previous_doc_id + encoded_doc_id
} else {
encoded_doc_id
}
let term_freq = if uses_varint {
decoder.read_varint("term frequency")
} else {
decoder.read_count("term frequency")
}
let position_count = if uses_varint {
decoder.read_varint("position count")
} else {
decoder.read_count("position count")
}
guard position_count <= decoder.remaining() else {
raise PersistenceError::InvalidFormat("invalid position count")
}
guard doc_id >= 0 && doc_id < document_count && doc_id > previous_doc_id else {
raise PersistenceError::InvalidFormat("invalid posting doc id: \{doc_id}")
}
guard term_freq > 0 && position_count == term_freq else {
raise PersistenceError::InvalidFormat(
"term frequency and positions disagree",
)
}
let positions : Array[Int] = []
let mut previous_position = -1
for position_index in 0.. 0 {
previous_position + encoded_position
} else {
encoded_position
}
guard position >= 0 && position >= previous_position else {
raise PersistenceError::InvalidFormat("invalid term position")
}
positions.push(position)
previous_position = position
}
postings.push({
doc_id: DocId::new(doc_id),
term_freq,
positions: ReadOnlyArray::from_array(positions),
position_lengths: ReadOnlyArray::from_array(
Array::make(position_count, 1),
),
start_offsets: ReadOnlyArray::from_array(Array::make(position_count, 0)),
end_offsets: ReadOnlyArray::from_array(Array::make(position_count, 0)),
value_indexes: ReadOnlyArray::from_array(Array::make(position_count, 0)),
})
previous_doc_id = doc_id
}
ReadOnlyArray::from_array(postings)
}
///|
fn decode_stored_document(
decoder : SegmentDecoder,
schema : Schema?,
version : UInt,
) -> StoredDocument raise PersistenceError {
let value_count = decoder.read_count("stored value count")
let field_ids : Array[FieldId] = []
let texts : Array[String] = []
for _ in 0.. {
let field_type = match schema.field_type(field_id) {
Some(field_type) => field_type
None => raise PersistenceError::InvalidFormat("unknown typed field")
}
guard field_value_matches_type(field_type, value) else {
raise PersistenceError::InvalidFormat("typed field value mismatch")
}
}
None => ()
}
value_field_ids.push(field_id)
values.push(value)
}
}
StoredDocument::from_all_fields(field_ids, texts, value_field_ids, values)
}
///|
fn decode_field_value(
decoder : SegmentDecoder,
) -> FieldValue raise PersistenceError {
match decoder.read_byte("field value type") {
b'\x00' => @core.Text(decoder.read_string("text value"))
b'\x01' => @core.Keyword(decoder.read_string("keyword value"))
b'\x02' => @core.I64(decoder.read_int64("i64 value"))
b'\x03' => @core.U64(decoder.read_uint64("u64 value"))
b'\x04' => @core.F64(decoder.read_double("f64 value"))
b'\x05' => @core.Bool(decoder.read_bool("bool value"))
b'\x06' => @core.Date(decoder.read_int64("date value"))
b'\x07' => @core.Bytes(decoder.read_bytes("bytes value"))
_ => raise PersistenceError::InvalidFormat("invalid field value type")
}
}
///|
fn decode_document_values(
decoder : SegmentDecoder,
schema : Schema?,
document_count : Int,
) -> ReadOnlyArray[ReadOnlyArray[FieldValueEntry]] raise PersistenceError {
let count = decoder.read_count("typed document count")
guard count == document_count else {
raise PersistenceError::InvalidFormat("typed document count mismatch")
}
let documents : Array[ReadOnlyArray[FieldValueEntry]] = []
for _ in 0.. {
let field_type = match schema.field_type(field_id) {
Some(field_type) => field_type
None => raise PersistenceError::InvalidFormat("unknown typed field")
}
guard field_value_matches_type(field_type, value) else {
raise PersistenceError::InvalidFormat("typed field value mismatch")
}
guard schema.is_multi_valued(field_id) ||
entries.search_by(entry => entry.field_id == field_id) is None else {
raise PersistenceError::InvalidFormat(
"multiple values for single-valued field",
)
}
}
None => ()
}
entries.push({ field_id, value })
}
documents.push(ReadOnlyArray::from_array(entries))
}
ReadOnlyArray::from_array(documents)
}
///|
fn decode_field_lengths(
decoder : SegmentDecoder,
schema : Schema?,
) -> ReadOnlyArray[FieldLength] raise PersistenceError {
let field_count = decoder.read_count("document field length count")
let lengths : Array[FieldLength] = []
for _ in 0.. entry.field_id == field_id) is None else {
raise PersistenceError::InvalidFormat("invalid field length entry")
}
lengths.push({ field_id, length })
}
ReadOnlyArray::from_array(lengths)
}
///|
fn decode_field_stats(
decoder : SegmentDecoder,
schema : Schema?,
document_count : Int,
) -> ReadOnlyArray[FieldStats] raise PersistenceError {
let field_count = decoder.read_count("field stats count")
let stats_entries : Array[FieldStats] = []
for _ in 0.. entry.field_id == field_id) is None else {
raise PersistenceError::InvalidFormat("invalid field statistics")
}
stats_entries.push({
field_id,
document_count: field_document_count,
total_length,
})
}
ReadOnlyArray::from_array(stats_entries)
}
///|
fn decode_payload(
payload : Bytes,
version : UInt,
) -> Segment raise PersistenceError {
let decoder = SegmentDecoder::new(payload)
let schema = decode_schema(decoder, version)
let document_count = decoder.read_count("document count")
let term_count = decoder.read_count("term count")
let terms : Array[Term] = []
let posting_lists : Array[ReadOnlyArray[Posting]] = []
let mut previous_term_field = FieldId::new(-1)
let mut previous_term_text = ""
for _ in 0.. text
None => raise PersistenceError::InvalidFormat("invalid term prefix")
}
} else {
decoder.read_string("term text")
}
let term = Term::new(field_id, term_text)
guard terms.search_by(existing => existing == term) is None else {
raise PersistenceError::InvalidFormat("duplicate term")
}
terms.push(term)
posting_lists.push(decode_postings(decoder, document_count, version))
previous_term_field = field_id
previous_term_text = term_text
}
let stored_field_blocks = if version == segment_format_version {
let block_count = decoder.read_count("stored block count")
let blocks : Array[StoredFieldBlock] = []
for _ in 0.. Segment raise PersistenceError {
guard bytes.length() >= segment_header_length else {
raise PersistenceError::InvalidFormat("truncated segment header")
}
for index in 0..