mod load;
use super::{IVFRecordKey as Key, IVFRecordLayout, IVFRecordValue as Value};
use crate::mvcc::vector::resolve::{bytes, replace};
use crate::mvcc::{
commit::RecordWriteKind, CommittedRecordSnapshot, PreparedRecordWrite, PrivateRecordChanges,
VersionError, VersionResult,
};
use crate::read_control::StorageReadControl;
pub(in crate::mvcc) fn merge(
key: &[u8],
operations: &crate::mvcc::vector::VectorOperations<'_>,
changes: &PrivateRecordChanges,
current: &dyn CommittedRecordSnapshot,
layout: &dyn IVFRecordLayout,
kind: RecordWriteKind,
control: &StorageReadControl,
) -> VersionResult<()> {
control.cancellation().check()?;
let row = current.get(key, control)?;
let template = bytes(row.as_ref()).ok_or(VersionError::InvalidEncoding(
"missing current IVF definition",
))?;
let header = layout.header(key, template, control)?;
let mut prepared = load::index(key, header, current, layout, control)?;
operations.visit(control, |_, operation| {
prepared.apply_evaluated(operation.borrowed().ivf())?;
Ok(())
})?;
let snapshot = prepared.header();
let next = header
.revision
.map(|revision| {
revision
.checked_add(operations.len() as u64)
.ok_or(VersionError::InvalidEncoding("IVF revision exhausted"))
})
.transpose()?;
for address in [Key::Centroids, Key::Assignments] {
let prefix = layout.key(key, address, control)?;
current.visit_keys(&prefix, None, usize::MAX, control, &mut |key, record| {
if record.live {
changes.apply_owned(
&[
PreparedRecordWrite::copy_bytes(key, record.revision, None, control)?
.with_kind(kind),
],
control,
)?;
}
Ok(true)
})?;
}
let header_value = layout.encode(
key,
template,
Value::Header {
snapshot,
revision: next,
},
control,
)?;
replace(changes, current, key, &header_value, kind, control)?;
for (centroid, vector) in snapshot.centroids.iter().enumerate() {
let address = layout.key(key, Key::Centroid(centroid), control)?;
let value = layout.encode(&address, template, Value::Centroid(vector), control)?;
replace(changes, current, &address, &value, kind, control)?;
}
for entry in prepared.assignments() {
let (document, ordinal, centroid) = entry?;
let address = layout.key(key, Key::Assignment(document, ordinal), control)?;
let value = layout.encode(&address, template, Value::Assignment(centroid), control)?;
replace(changes, current, &address, &value, kind, control)?;
}
Ok(())
}