use crate::index::{ColumnLearnedRange, ColumnLearnedRangeSnapshot};
use crate::rowid::RowId;
use crate::schema::{IndexKind, Schema};
use crate::{MongrelError, Result};
use bincode::Options;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::{HashMap, HashSet};
use std::io::Read;
use std::path::{Path, PathBuf};
pub const IDX_MAGIC: [u8; 8] = *b"MONGRIDX";
pub const IDX_VERSION: u16 = 3;
pub const IDX_DIR: &str = "_idx";
pub const IDX_FILENAME: &str = "global.idx";
const MAX_GLOBAL_INDEX_BYTES: u64 = 64 * 1024 * 1024;
const MAX_INDEX_RECORDS: usize = 65_536;
const MAX_INDEX_PAYLOAD_BYTES: usize = MAX_GLOBAL_INDEX_BYTES as usize;
const K_HOT: u8 = 1;
const K_BITMAP: u8 = 3;
const K_FM: u8 = 4;
const K_ANN: u8 = 5;
const K_SPARSE: u8 = 9;
const K_LEARNED: u8 = 10;
const K_MINHASH: u8 = 11;
pub struct IndexSnapshot<'a> {
pub hot: &'a crate::index::HotIndex,
pub bitmap: &'a HashMap<u16, crate::index::BitmapIndex>,
pub ann: &'a HashMap<u16, crate::index::AnnIndex>,
pub fm: &'a HashMap<u16, crate::index::FmIndex>,
pub sparse: &'a HashMap<u16, crate::index::SparseIndex>,
pub minhash: &'a HashMap<u16, crate::index::MinHashIndex>,
pub learned_range: &'a HashMap<u16, ColumnLearnedRange>,
}
pub struct LoadedIndexes {
pub epoch_built: u64,
pub hot: crate::index::HotIndex,
pub bitmap: HashMap<u16, crate::index::BitmapIndex>,
pub ann: HashMap<u16, crate::index::AnnIndex>,
pub fm: HashMap<u16, crate::index::FmIndex>,
pub sparse: HashMap<u16, crate::index::SparseIndex>,
pub minhash: HashMap<u16, crate::index::MinHashIndex>,
pub learned_range: HashMap<u16, ColumnLearnedRange>,
}
#[derive(Serialize, Deserialize)]
struct GlobalIdxBody {
format_version: u16,
table_id: u64,
epoch_built: u64,
#[serde(deserialize_with = "deserialize_records")]
records: Vec<Record>,
}
#[derive(Serialize, Deserialize)]
struct Record {
kind: u8,
column_id: u16,
#[serde(deserialize_with = "deserialize_payload")]
payload: Vec<u8>,
}
fn deserialize_records<'de, D>(deserializer: D) -> std::result::Result<Vec<Record>, D::Error>
where
D: serde::Deserializer<'de>,
{
struct Visitor;
impl<'de> serde::de::Visitor<'de> for Visitor {
type Value = Vec<Record>;
fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("a bounded global-index record list")
}
fn visit_seq<A>(self, mut sequence: A) -> std::result::Result<Self::Value, A::Error>
where
A: serde::de::SeqAccess<'de>,
{
let capacity = sequence.size_hint().unwrap_or(0);
if capacity > MAX_INDEX_RECORDS {
return Err(serde::de::Error::custom("too many global-index records"));
}
let mut records = Vec::with_capacity(capacity);
while let Some(record) = sequence.next_element()? {
if records.len() == MAX_INDEX_RECORDS {
return Err(serde::de::Error::custom("too many global-index records"));
}
records.push(record);
}
Ok(records)
}
}
deserializer.deserialize_seq(Visitor)
}
fn deserialize_payload<'de, D>(deserializer: D) -> std::result::Result<Vec<u8>, D::Error>
where
D: serde::Deserializer<'de>,
{
struct Visitor;
impl<'de> serde::de::Visitor<'de> for Visitor {
type Value = Vec<u8>;
fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("a bounded global-index payload")
}
fn visit_seq<A>(self, mut sequence: A) -> std::result::Result<Self::Value, A::Error>
where
A: serde::de::SeqAccess<'de>,
{
let capacity = sequence.size_hint().unwrap_or(0);
if capacity > MAX_INDEX_PAYLOAD_BYTES {
return Err(serde::de::Error::custom(
"global-index payload is too large",
));
}
let mut payload = Vec::with_capacity(capacity);
while let Some(byte) = sequence.next_element()? {
if payload.len() == MAX_INDEX_PAYLOAD_BYTES {
return Err(serde::de::Error::custom(
"global-index payload is too large",
));
}
payload.push(byte);
}
Ok(payload)
}
}
deserializer.deserialize_seq(Visitor)
}
pub fn path(dir: &Path) -> PathBuf {
dir.join(IDX_DIR).join(IDX_FILENAME)
}
fn encode_file(plain: Vec<u8>, dek: Option<&[u8; 32]>) -> Result<Vec<u8>> {
#[cfg(feature = "encryption")]
{
if let Some(k) = dek {
return crate::encryption::encrypt_blob(k, &plain);
}
}
#[cfg(not(feature = "encryption"))]
{
let _ = dek;
}
Ok(plain)
}
fn decode_file(raw: Vec<u8>, dek: Option<&[u8; 32]>) -> Option<Vec<u8>> {
#[cfg(feature = "encryption")]
{
if let Some(k) = dek {
return crate::encryption::decrypt_blob(k, &raw).ok();
}
}
#[cfg(not(feature = "encryption"))]
{
let _ = dek;
}
Some(raw)
}
pub fn write_atomic(
dir: &Path,
table_id: u64,
epoch_built: u64,
snap: IndexSnapshot<'_>,
dek: Option<&[u8; 32]>,
) -> Result<()> {
let table_root = crate::durable_file::DurableRoot::open(dir)?;
let idx_root = table_root.create_directory_all_pinned(IDX_DIR)?;
write_atomic_root(&idx_root, table_id, epoch_built, snap, dek)
}
pub(crate) fn write_atomic_root(
idx_root: &crate::durable_file::DurableRoot,
table_id: u64,
epoch_built: u64,
snap: IndexSnapshot<'_>,
dek: Option<&[u8; 32]>,
) -> Result<()> {
let mut records = Vec::new();
let hot_entries = snap.hot.entries();
if !hot_entries.is_empty() {
records.push(Record {
kind: K_HOT,
column_id: 0,
payload: bincode::serialize(&hot_entries)?,
});
}
for (&cid, bm) in snap.bitmap {
let entries = bm.entries();
if !entries.is_empty() {
records.push(Record {
kind: K_BITMAP,
column_id: cid,
payload: bincode::serialize(&entries)?,
});
}
}
for (&cid, ann) in snap.ann {
if !ann.is_empty() {
records.push(Record {
kind: K_ANN,
column_id: cid,
payload: ann.freeze(),
});
}
}
for (&cid, fm) in snap.fm {
if fm.doc_count() > 0 {
let docs = fm.docs();
records.push(Record {
kind: K_FM,
column_id: cid,
payload: bincode::serialize(&docs)?,
});
}
}
for (&cid, sp) in snap.sparse {
if !sp.is_empty() {
let entries = sp.entries();
records.push(Record {
kind: K_SPARSE,
column_id: cid,
payload: bincode::serialize(&entries)?,
});
}
}
for (&cid, mh) in snap.minhash {
if !mh.is_empty() {
records.push(Record {
kind: K_MINHASH,
column_id: cid,
payload: bincode::serialize(&mh.snapshot())?,
});
}
}
for (&cid, lr) in snap.learned_range {
let s = lr.snapshot();
records.push(Record {
kind: K_LEARNED,
column_id: cid,
payload: bincode::serialize(&s)?,
});
}
let body = GlobalIdxBody {
format_version: IDX_VERSION,
table_id,
epoch_built,
records,
};
let body_bytes = bincode::serialize(&body)?;
let mut out = Vec::with_capacity(8 + body_bytes.len() + 8 + 32);
out.extend_from_slice(&IDX_MAGIC);
out.extend_from_slice(&body_bytes);
out.extend_from_slice(&IDX_MAGIC);
let hash: [u8; 32] = Sha256::digest(&out).into();
out.extend_from_slice(&hash);
let out = encode_file(out, dek)?;
idx_root.write_atomic(IDX_FILENAME, &out)?;
Ok(())
}
pub fn read(
dir: &Path,
expected_table_id: u64,
schema: &Schema,
dek: Option<&[u8; 32]>,
) -> Result<Option<LoadedIndexes>> {
let path = path(dir);
let file = match crate::durable_file::open_regular_nofollow(&path) {
Ok(file) => file,
Err(MongrelError::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => {
return Ok(None)
}
Err(error) => return Err(error),
};
read_file(file, expected_table_id, schema, dek)
}
pub(crate) fn read_durable_for(
root: &crate::durable_file::DurableRoot,
relative_dir: impl AsRef<Path>,
expected_table_id: u64,
schema: &Schema,
dek: Option<&[u8; 32]>,
) -> Result<Option<LoadedIndexes>> {
let relative = relative_dir.as_ref().join(IDX_DIR).join(IDX_FILENAME);
let file = match root.open_regular(relative) {
Ok(file) => file,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(error.into()),
};
read_file(file, expected_table_id, schema, dek)
}
pub(crate) fn read_root(
idx_root: &crate::durable_file::DurableRoot,
expected_table_id: u64,
schema: &Schema,
dek: Option<&[u8; 32]>,
) -> Result<Option<LoadedIndexes>> {
let file = match idx_root.open_regular(IDX_FILENAME) {
Ok(file) => file,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(error.into()),
};
read_file(file, expected_table_id, schema, dek)
}
fn read_file(
file: std::fs::File,
expected_table_id: u64,
schema: &Schema,
dek: Option<&[u8; 32]>,
) -> Result<Option<LoadedIndexes>> {
let length = file.metadata()?.len();
if length > MAX_GLOBAL_INDEX_BYTES {
return Ok(None);
}
let mut raw = Vec::with_capacity(length as usize);
file.take(MAX_GLOBAL_INDEX_BYTES + 1)
.read_to_end(&mut raw)?;
if raw.len() as u64 != length {
return Ok(None);
}
let bytes = match decode_file(raw, dek) {
Some(b) => b,
None => return Ok(None),
};
if bytes.len() < 8 + 8 + 32 {
return Ok(None); }
let header_magic = &bytes[..8];
if header_magic != IDX_MAGIC {
return Ok(None);
}
let footer_start = bytes.len() - 32 - 8;
let footer_magic = &bytes[footer_start..footer_start + 8];
if footer_magic != IDX_MAGIC {
return Ok(None);
}
let stored_hash = &bytes[bytes.len() - 32..];
let recomputed: [u8; 32] = Sha256::digest(&bytes[..bytes.len() - 32]).into();
if stored_hash != recomputed {
return Ok(None);
}
let body: GlobalIdxBody = match decode_bounded(&bytes[8..footer_start]) {
Some(body) => body,
None => return Ok(None),
};
if body.format_version != IDX_VERSION {
return Ok(None);
}
if body.table_id != expected_table_id {
return Ok(None);
}
let mut hot = crate::index::HotIndex::new();
let mut bitmap = HashMap::new();
let mut ann = HashMap::new();
let mut fm = HashMap::new();
let mut sparse = HashMap::new();
let mut minhash = HashMap::new();
let mut learned_range = HashMap::new();
let mut identities = HashSet::new();
for rec in body.records {
if !identities.insert((rec.kind, rec.column_id)) || !record_matches_schema(&rec, schema) {
return Ok(None);
}
match rec.kind {
K_HOT => {
let Some(entries) = decode_bounded::<Vec<(Vec<u8>, RowId)>>(&rec.payload) else {
return Ok(None);
};
hot = crate::index::HotIndex::from_entries(entries);
}
K_BITMAP => {
let Some(entries) = decode_bounded::<Vec<(Vec<u8>, Vec<u8>)>>(&rec.payload) else {
return Ok(None);
};
let Ok(index) = crate::index::BitmapIndex::from_entries(entries) else {
return Ok(None);
};
bitmap.insert(rec.column_id, index);
}
K_FM => {
let Some(docs) = decode_bounded::<Vec<(Vec<u8>, RowId)>>(&rec.payload) else {
return Ok(None);
};
fm.insert(rec.column_id, crate::index::FmIndex::from_docs(docs));
}
K_ANN => {
let Ok(idx) =
crate::index::AnnIndex::thaw_bounded(&rec.payload, MAX_GLOBAL_INDEX_BYTES)
else {
return Ok(None);
};
ann.insert(rec.column_id, idx);
}
K_SPARSE => {
let Some(entries) = decode_bounded::<Vec<(u32, Vec<(RowId, f32)>)>>(&rec.payload)
else {
return Ok(None);
};
sparse.insert(
rec.column_id,
crate::index::SparseIndex::from_entries(entries),
);
}
K_MINHASH => {
let Some(snapshot) =
decode_bounded::<crate::index::minhash::MinHashSnapshot>(&rec.payload)
else {
return Ok(None);
};
minhash.insert(
rec.column_id,
crate::index::MinHashIndex::from_snapshot(snapshot),
);
}
K_LEARNED => {
let Some(snap) = decode_bounded::<ColumnLearnedRangeSnapshot>(&rec.payload) else {
return Ok(None);
};
learned_range.insert(rec.column_id, ColumnLearnedRange::from_snapshot(snap));
}
_ => return Ok(None),
}
}
Ok(Some(LoadedIndexes {
epoch_built: body.epoch_built,
hot,
bitmap,
ann,
fm,
sparse,
minhash,
learned_range,
}))
}
fn decode_bounded<T>(bytes: &[u8]) -> Option<T>
where
T: serde::de::DeserializeOwned,
{
bincode::DefaultOptions::new()
.with_fixint_encoding()
.reject_trailing_bytes()
.with_limit(MAX_GLOBAL_INDEX_BYTES)
.deserialize(bytes)
.ok()
}
fn record_matches_schema(record: &Record, schema: &Schema) -> bool {
let expected_kind = match record.kind {
K_HOT => return record.column_id == 0,
K_BITMAP => IndexKind::Bitmap,
K_FM => IndexKind::FmIndex,
K_ANN => IndexKind::Ann,
K_SPARSE => IndexKind::Sparse,
K_LEARNED => IndexKind::LearnedRange,
K_MINHASH => IndexKind::MinHash,
_ => return false,
};
schema
.indexes
.iter()
.any(|index| index.column_id == record.column_id && index.kind == expected_kind)
}
pub fn remove(dir: &Path) {
let _ = std::fs::remove_file(path(dir));
}
#[derive(Debug, Clone)]
pub struct IndexRecordSize {
pub kind: &'static str,
pub column_id: u16,
pub payload_bytes: u64,
}
pub fn plaintext_record_sizes(dir: &Path) -> Result<Vec<IndexRecordSize>> {
let bytes = std::fs::read(path(dir))?;
if bytes.len() < 48 || bytes[..8] != IDX_MAGIC {
return Err(MongrelError::MagicMismatch {
what: "global index",
expected: IDX_MAGIC,
got: bytes
.get(..8)
.and_then(|value| value.try_into().ok())
.unwrap_or([0; 8]),
});
}
let footer_start = bytes.len() - 40;
let body: GlobalIdxBody = bincode::deserialize(&bytes[8..footer_start])?;
Ok(body
.records
.into_iter()
.map(|record| IndexRecordSize {
kind: match record.kind {
K_HOT => "hot_primary",
K_BITMAP => "bitmap",
K_FM => "fm",
K_ANN => "ann",
K_SPARSE => "sparse",
K_LEARNED => "learned_range",
K_MINHASH => "minhash",
_ => "unknown",
},
column_id: record.column_id,
payload_bytes: record.payload.len() as u64,
})
.collect())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::index::{AnnIndex, BitmapIndex, FmIndex, HotIndex, SparseIndex};
use crate::rowid::RowId;
use crate::schema::{IndexDef, IndexOptions};
use tempfile::tempdir;
fn indexed_schema() -> Schema {
Schema {
indexes: [
(7, IndexKind::Bitmap),
(9, IndexKind::FmIndex),
(11, IndexKind::Ann),
(13, IndexKind::Sparse),
(15, IndexKind::MinHash),
]
.into_iter()
.map(|(column_id, kind)| IndexDef {
name: format!("idx_{column_id}"),
column_id,
kind,
predicate: None,
options: IndexOptions::default(),
})
.collect(),
..Schema::default()
}
}
fn write_body(dir: &Path, body: &GlobalIdxBody) {
std::fs::create_dir_all(dir.join(IDX_DIR)).unwrap();
let mut bytes = IDX_MAGIC.to_vec();
bytes.extend(bincode::serialize(body).unwrap());
bytes.extend(IDX_MAGIC);
let hash: [u8; 32] = Sha256::digest(&bytes).into();
bytes.extend(hash);
std::fs::write(path(dir), bytes).unwrap();
}
#[test]
fn roundtrip_all_index_kinds() {
let dir = tempdir().unwrap();
let mut hot = HotIndex::new();
hot.insert(b"alice".to_vec(), RowId(1));
hot.insert(b"bob".to_vec(), RowId(2));
let mut bitmap = HashMap::new();
let mut bm = BitmapIndex::new();
bm.insert(b"red".to_vec(), RowId(1));
bm.insert(b"red".to_vec(), RowId(3));
bm.insert(b"blue".to_vec(), RowId(5));
bitmap.insert(7u16, bm);
let mut fm_map = HashMap::new();
let mut fm = FmIndex::new();
fm.insert(b"the quick brown fox".to_vec(), RowId(1));
fm.insert(b"fox in socks".to_vec(), RowId(2));
fm_map.insert(9u16, fm);
let mut ann_map = HashMap::new();
let mut ann = AnnIndex::new(8);
ann.insert(&[1.0, -1.0, 1.0, -1.0, 1.0, -1.0, 1.0, -1.0], RowId(0))
.unwrap();
ann.insert(&[-1.0, 1.0, -1.0, 1.0, -1.0, 1.0, -1.0, 1.0], RowId(1))
.unwrap();
ann_map.insert(11u16, ann);
let mut sparse_map = HashMap::new();
let mut sp = SparseIndex::new();
sp.insert(&[(1, 2.0), (2, 1.0)], RowId(0));
sp.insert(&[(1, 1.0)], RowId(1));
sparse_map.insert(13u16, sp);
let mut minhash_map = HashMap::new();
let mut mh = crate::index::MinHashIndex::new();
let toks = |ts: &[&str]| -> Vec<u64> {
ts.iter()
.map(|t| crate::index::minhash_token_hash(t))
.collect()
};
mh.insert(&toks(&["a", "b", "c", "d"]), RowId(0));
mh.insert(&toks(&["x", "y", "z", "w"]), RowId(1));
minhash_map.insert(15u16, mh);
let lr_map = HashMap::<u16, ColumnLearnedRange>::new();
let snap = IndexSnapshot {
hot: &hot,
bitmap: &bitmap,
ann: &ann_map,
fm: &fm_map,
sparse: &sparse_map,
minhash: &minhash_map,
learned_range: &lr_map,
};
write_atomic(dir.path(), 42, 7, snap, None).unwrap();
let loaded = read(dir.path(), 42, &indexed_schema(), None)
.unwrap()
.expect("checkpoint present");
assert_eq!(loaded.epoch_built, 7);
assert_eq!(loaded.hot.get(b"alice"), Some(RowId(1)));
assert_eq!(loaded.hot.get(b"bob"), Some(RowId(2)));
let red = loaded.bitmap[&7].get(b"red");
let red_ids: Vec<u32> = red.iter().collect();
assert_eq!(red_ids, vec![1, 3]);
assert_eq!(loaded.fm[&9].locate(b"fox").len(), 2);
let top = loaded.ann[&11]
.search(&[1.0, -1.0, 1.0, -1.0, 1.0, -1.0, 1.0, -1.0], 1)
.unwrap();
assert_eq!(top[0].0, RowId(0));
let sp_top = loaded.sparse[&13].search(&[(1, 1.0), (2, 1.0)], 2);
assert_eq!(sp_top[0].0, RowId(0));
let mh_top = loaded.minhash[&15].search(&toks(&["a", "b", "c", "d"]), 5);
assert_eq!(mh_top[0].0, RowId(0));
assert!(mh_top[0].1 > 0.95);
}
#[test]
fn read_returns_none_when_absent() {
let dir = tempdir().unwrap();
assert!(read(dir.path(), 1, &Schema::default(), None)
.unwrap()
.is_none());
}
#[test]
fn old_checkpoint_version_is_rejected_for_rebuild() {
let dir = tempdir().unwrap();
let body = GlobalIdxBody {
format_version: IDX_VERSION - 1,
table_id: 1,
epoch_built: 1,
records: vec![],
};
write_body(dir.path(), &body);
assert!(read(dir.path(), 1, &Schema::default(), None)
.unwrap()
.is_none());
}
#[test]
fn detects_corruption() {
let dir = tempdir().unwrap();
let hot = HotIndex::new();
let bitmap = HashMap::new();
let ann = HashMap::new();
let fm = HashMap::new();
let sparse = HashMap::new();
let minhash = HashMap::new();
let lr = HashMap::new();
write_atomic(
dir.path(),
1,
1,
IndexSnapshot {
hot: &hot,
bitmap: &bitmap,
ann: &ann,
fm: &fm,
sparse: &sparse,
minhash: &minhash,
learned_range: &lr,
},
None,
)
.unwrap();
let p = path(dir.path());
let mut bytes = std::fs::read(&p).unwrap();
bytes[12] ^= 0xFF;
std::fs::write(&p, bytes).unwrap();
assert!(read(dir.path(), 1, &Schema::default(), None)
.unwrap()
.is_none());
}
#[test]
fn checkpoint_is_bound_to_table_and_schema() {
let dir = tempdir().unwrap();
let hot = HotIndex::new();
let mut bitmap = HashMap::new();
let mut index = BitmapIndex::new();
index.insert(b"value".to_vec(), RowId(1));
bitmap.insert(7, index);
write_atomic(
dir.path(),
42,
1,
IndexSnapshot {
hot: &hot,
bitmap: &bitmap,
ann: &HashMap::new(),
fm: &HashMap::new(),
sparse: &HashMap::new(),
minhash: &HashMap::new(),
learned_range: &HashMap::new(),
},
None,
)
.unwrap();
assert!(read(dir.path(), 41, &indexed_schema(), None)
.unwrap()
.is_none());
assert!(read(dir.path(), 42, &Schema::default(), None)
.unwrap()
.is_none());
}
#[test]
fn duplicate_and_unknown_records_trigger_rebuild() {
let dir = tempdir().unwrap();
let bitmap_payload = bincode::serialize(&Vec::<(Vec<u8>, Vec<u8>)>::new()).unwrap();
let mut body = GlobalIdxBody {
format_version: IDX_VERSION,
table_id: 1,
epoch_built: 1,
records: vec![
Record {
kind: K_BITMAP,
column_id: 7,
payload: bitmap_payload.clone(),
},
Record {
kind: K_BITMAP,
column_id: 7,
payload: bitmap_payload,
},
],
};
write_body(dir.path(), &body);
assert!(read(dir.path(), 1, &indexed_schema(), None)
.unwrap()
.is_none());
body.records = vec![Record {
kind: 255,
column_id: 7,
payload: Vec::new(),
}];
write_body(dir.path(), &body);
assert!(read(dir.path(), 1, &indexed_schema(), None)
.unwrap()
.is_none());
}
}