use std::collections::HashSet;
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use crate::mmap_index::{write_file_atomic, MmapPostingData};
pub const META_FILE: &str = "meta.json";
pub const META_VERSION: u32 = 1;
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SegmentMeta {
pub id: String,
pub num_vectors: u32,
#[serde(default)]
pub deleted: Vec<u64>,
}
impl SegmentMeta {
pub fn live_vectors(&self) -> usize {
(self.num_vectors as usize).saturating_sub(self.deleted.len())
}
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct IndexMeta {
#[serde(default)]
pub version: u32,
#[serde(default)]
pub segments: Vec<SegmentMeta>,
}
impl Default for IndexMeta {
fn default() -> Self {
Self { version: META_VERSION, segments: Vec::new() }
}
}
impl IndexMeta {
pub fn read(base: &Path) -> Result<Self, String> {
let path = base.join(META_FILE);
if !path.exists() {
return Ok(Self::default());
}
let data = std::fs::read(&path)
.map_err(|e| format!("cannot read {}: {e}", path.display()))?;
let meta: Self = serde_json::from_slice(&data)
.map_err(|e| format!("cannot parse {}: {e}", path.display()))?;
if meta.version > META_VERSION {
return Err(format!(
"{} was written by a newer version ({}, this build writes {META_VERSION})",
path.display(), meta.version,
));
}
Ok(meta)
}
pub fn write(&self, base: &Path) -> Result<(), String> {
let data = serde_json::to_vec_pretty(self)
.map_err(|e| format!("cannot serialize {META_FILE}: {e}"))?;
write_file_atomic(&base.join(META_FILE), &data)
}
pub fn live_vectors(&self) -> usize {
self.segments.iter().map(SegmentMeta::live_vectors).sum()
}
pub fn files(&self) -> Vec<String> {
let mut files: Vec<String> = Vec::with_capacity(self.segments.len() * 2 + 1);
for s in &self.segments {
files.push(segment_file(&s.id));
files.push(ids_file(&s.id));
}
files.push(META_FILE.to_string());
files
}
}
pub fn segment_file(id: &str) -> String {
format!("seg_{id}.mmap")
}
pub fn ids_file(id: &str) -> String {
format!("seg_{id}.ids")
}
pub fn encode_ids(ids: &[u64]) -> Vec<u8> {
let mut out = Vec::with_capacity(ids.len() * 8);
for id in ids {
out.extend_from_slice(&id.to_le_bytes());
}
out
}
fn decode_ids(data: &[u8]) -> Result<Vec<u64>, String> {
if data.len() % 8 != 0 {
return Err(format!("id list is {} bytes, not a multiple of 8", data.len()));
}
Ok(data.chunks_exact(8).map(|c| u64::from_le_bytes(c.try_into().unwrap())).collect())
}
pub fn new_segment_id(counter: u64) -> String {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
format!("{:08x}{:04x}{:04x}", nanos & 0xffff_ffff, std::process::id() & 0xffff, counter & 0xffff)
}
pub struct Segment {
pub meta: SegmentMeta,
pub data: MmapPostingData,
deleted: HashSet<u64>,
ids: std::cell::OnceCell<Vec<u64>>,
base: PathBuf,
}
impl Segment {
pub fn open(base: &Path, meta: SegmentMeta) -> Result<Self, String> {
let data = MmapPostingData::open(&base.join(segment_file(&meta.id)))?;
let deleted = meta.deleted.iter().copied().collect();
Ok(Self { meta, data, deleted, ids: std::cell::OnceCell::new(), base: base.to_path_buf() })
}
pub fn is_live(&self, id: u64) -> bool {
!self.deleted.contains(&id)
}
pub fn ids(&self) -> Result<&[u64], String> {
if let Some(ids) = self.ids.get() {
return Ok(ids);
}
let path = self.base.join(ids_file(&self.meta.id));
let data = std::fs::read(&path)
.map_err(|e| format!("cannot read {}: {e}", path.display()))?;
let ids = decode_ids(&data)?;
Ok(self.ids.get_or_init(|| ids))
}
pub fn holds(&self, id: u64) -> Result<bool, String> {
Ok(self.ids()?.binary_search(&id).is_ok())
}
pub fn tombstone(&mut self, id: u64) -> bool {
if !self.deleted.insert(id) {
return false;
}
if let Err(pos) = self.meta.deleted.binary_search(&id) {
self.meta.deleted.insert(pos, id);
}
true
}
pub fn path(&self) -> PathBuf {
self.base.join(segment_file(&self.meta.id))
}
pub fn ids_path(&self) -> PathBuf {
self.base.join(ids_file(&self.meta.id))
}
}
pub fn merge_segments(
base: &Path,
inputs: &[&Segment],
new_id: &str,
) -> Result<SegmentMeta, String> {
use crate::wand::Postings;
let mut cursors: Vec<std::iter::Peekable<_>> =
inputs.iter().map(|s| s.data.tokens().peekable()).collect();
let mut tokens: Vec<u32> = Vec::new();
loop {
let next = cursors.iter_mut()
.filter_map(|c| c.peek().map(|(t, _)| *t))
.min();
let Some(token) = next else { break };
for c in cursors.iter_mut() {
if c.peek().is_some_and(|(t, _)| *t == token) {
c.next();
}
}
tokens.push(token);
}
let mut postings: Vec<Postings> = Vec::with_capacity(tokens.len());
let mut ids: std::collections::BTreeSet<u64> = std::collections::BTreeSet::new();
let mut pairs: Vec<(u64, f32)> = Vec::new();
for &token in &tokens {
pairs.clear();
for seg in inputs {
for e in seg.data.entries_of_token(token) {
if seg.is_live(e.record_id) {
pairs.push((e.record_id, e.weight));
ids.insert(e.record_id);
}
}
}
pairs.sort_by_key(|(id, _)| *id);
postings.push(Postings::from_sorted_pairs(&pairs));
}
let ids: Vec<u64> = ids.into_iter().collect();
crate::mmap_index::write_mmap_file(
&base.join(segment_file(new_id)),
&postings,
&tokens,
ids.len() as u32,
)?;
write_file_atomic(&base.join(ids_file(new_id)), &encode_ids(&ids))?;
Ok(SegmentMeta { id: new_id.to_string(), num_vectors: ids.len() as u32, deleted: Vec::new() })
}