persy/index/

serialization.rs

use crate::{
    id::{PersyId, RecRef},
    index::{
        bytevec::ByteVec,
        string_wrapper::StringWrapper,
        tree::nodes::{Leaf, LeafEntry, Node, TreeNode, TreeNodeRef, Value},
    },
    util::io::{
        ArcSliceRead, InfallibleRead, InfallibleReadFormat, InfallibleReadVarInt, InfallibleWrite,
        InfallibleWriteFormat, InfallibleWriteVarInt,
    },
};

pub trait IndexSerialization: Sized {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite);
    fn deserialize(value: &mut dyn InfallibleRead) -> Self;
}

pub fn reuse_deserialize<K: IndexSerialization, V: IndexSerialization>(
    mut value: ArcSliceRead,
    nodes: Option<Node<K>>,
) -> TreeNode<K, V> {
    let version = value.read_u8();
    match version {
        0u8 => deserialize_v0(value, nodes),
        _ => panic!("not compatible disk version"),
    }
}

pub fn deserialize<K: IndexSerialization, V: IndexSerialization>(value: ArcSliceRead) -> TreeNode<K, V> {
    reuse_deserialize(value, None)
}
fn destruct_reuse<K>(reuse: Option<Node<K>>) -> (Option<Vec<K>>, Option<Vec<RecRef>>) {
    if let Some(Node {
        mut keys, mut pointers, ..
    }) = reuse
    {
        keys.clear();
        pointers.clear();
        (Some(keys), Some(pointers))
    } else {
        (None, None)
    }
}

pub fn deserialize_v0<K: IndexSerialization, V: IndexSerialization>(
    mut reader: ArcSliceRead,
    reuse: Option<Node<K>>,
) -> TreeNode<K, V> {
    let t = reader.read_varint_u8();
    match t {
        1 => {
            let prev = if reader.read_varint_u8() == 1 {
                Some(K::deserialize(&mut reader))
            } else {
                None
            };
            let size = reader.read_varint_u32();
            let mut entries = Vec::with_capacity(size as usize);
            for _ in 0..size {
                let key = K::deserialize(&mut reader);
                let value_type = reader.read_varint_u8();
                if value_type == 1 {
                    let val_size = reader.read_varint_u32();
                    let mut value = Vec::with_capacity(val_size as usize);
                    for _ in 0..val_size {
                        value.push(V::deserialize(&mut reader));
                    }
                    entries.push(LeafEntry {
                        key,
                        value: Value::Cluster(value),
                    });
                } else {
                    let value = V::deserialize(&mut reader);
                    entries.push(LeafEntry {
                        key,
                        value: Value::Single(value),
                    });
                }
            }
            let next = if reader.read_varint_u8() == 1 {
                Some(K::deserialize(&mut reader))
            } else {
                None
            };
            TreeNode::Leaf(Leaf { entries, prev, next })
        }
        2 => {
            let (reuse_keys, reuse_pointers) = destruct_reuse(reuse);
            let prev = if reader.read_varint_u8() == 1 {
                Some(K::deserialize(&mut reader))
            } else {
                None
            };
            let size = reader.read_varint_u32();
            let mut keys = reuse_keys.unwrap_or_else(|| Vec::with_capacity(size as usize));
            for _ in 0..size {
                let key = K::deserialize(&mut reader);
                keys.push(key);
            }
            let size = reader.read_varint_u32();
            let mut pointers = reuse_pointers.unwrap_or_else(|| Vec::with_capacity(size as usize));
            for _ in 0..size {
                let page = reader.read_varint_u64();
                let pos = reader.read_varint_u32();
                pointers.push(RecRef::new(page, pos));
            }
            let next = if reader.read_varint_u8() == 1 {
                Some(K::deserialize(&mut reader))
            } else {
                None
            };
            TreeNode::Node(Node {
                keys,
                pointers,
                prev,
                next,
            })
        }
        _ => panic!("error on index node deserialization"),
    }
}

pub fn serialize<K: IndexSerialization, V: IndexSerialization>(node: &TreeNode<K, V>) -> Vec<u8> {
    let mut dest = Vec::new();
    dest.push(0u8);
    serialize_v0(node, &mut dest);
    dest
}
pub fn serialize_v0<K: IndexSerialization, V: IndexSerialization>(
    node: &TreeNode<K, V>,
    dest: &mut dyn InfallibleWrite,
) {
    let write_page_and_pos = |dest: &mut dyn InfallibleWrite, page, pos| {
        dest.write_varint_u64(page);
        dest.write_varint_u32(pos);
    };

    let write_opt_leafptr = |dest: &mut dyn InfallibleWrite, x: Option<&TreeNodeRef>| {
        if let Some(y) = x {
            write_page_and_pos(dest, y.page, y.pos);
        } else {
            write_page_and_pos(dest, 0, 0);
        }
    };

    match node {
        TreeNode::Leaf(leaf) => {
            dest.write_varint_u8(1);
            if let Some(pk) = &leaf.prev {
                dest.write_varint_u8(1);
                pk.serialize(dest);
            } else {
                dest.write_varint_u8(0);
            }
            dest.write_varint_u32(leaf.entries.len() as u32);
            for entry in &leaf.entries {
                entry.key.serialize(dest);
                match &entry.value {
                    Value::Cluster(cluster) => {
                        dest.write_varint_u8(1);
                        dest.write_varint_u32(cluster.len() as u32);
                        for val in cluster {
                            val.serialize(dest);
                        }
                    }
                    Value::Single(val) => {
                        dest.write_varint_u8(2);
                        val.serialize(dest);
                    }
                }
            }
            if let Some(pk) = &leaf.next {
                dest.write_varint_u8(1);
                pk.serialize(dest);
            } else {
                dest.write_varint_u8(0);
            }
        }
        TreeNode::Node(node) => {
            dest.write_varint_u8(2);
            if let Some(pk) = &node.prev {
                dest.write_varint_u8(1);
                pk.serialize(dest);
            } else {
                dest.write_varint_u8(0);
            }
            dest.write_varint_u32(node.keys.len() as u32);
            for k in &node.keys {
                k.serialize(dest);
            }
            dest.write_varint_u32(node.pointers.len() as u32);
            for p in &node.pointers {
                write_opt_leafptr(dest, Some(p));
            }
            if let Some(pk) = &node.next {
                dest.write_varint_u8(1);
                pk.serialize(dest);
            } else {
                dest.write_varint_u8(0);
            }
        }
    }
}

impl IndexSerialization for u8 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_u8(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_varint_u8()
    }
}

impl IndexSerialization for u16 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_u16(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_varint_u16()
    }
}

impl IndexSerialization for u32 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_u32(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_varint_u32()
    }
}

impl IndexSerialization for u64 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_u64(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_varint_u64()
    }
}

impl IndexSerialization for u128 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_u128(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_varint_u128()
    }
}

impl IndexSerialization for i8 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_i8(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_varint_i8()
    }
}

impl IndexSerialization for i16 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_i16(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_varint_i16()
    }
}

impl IndexSerialization for i32 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_i32(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_varint_i32()
    }
}

impl IndexSerialization for i64 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_i64(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_varint_i64()
    }
}

impl IndexSerialization for i128 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_i128(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_varint_i128()
    }
}

impl IndexSerialization for f32 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_f32(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_f32()
    }
}

impl IndexSerialization for f64 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_f64(*self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        value.read_f64()
    }
}

impl IndexSerialization for PersyId {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_u64(self.0.page);
        buffer.write_varint_u32(self.0.pos);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        let page = value.read_varint_u64();
        let pos = value.read_varint_u32();
        PersyId(RecRef::new(page, pos))
    }
}

impl IndexSerialization for StringWrapper {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_u32(self.len() as u32);
        buffer.write_all(self.slice());
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        let vec_size = value.read_varint_u32();
        value.read_slice(vec_size as usize).to_string_wrapper()
    }
}

impl IndexSerialization for ByteVec {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) {
        buffer.write_varint_u32(self.len() as u32);
        buffer.write_all(self);
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> Self {
        let vec_size = value.read_varint_u32();
        value.read_slice(vec_size as usize).to_byte_vec()
    }
}