extern crate memmap;
#[cfg(test)]
extern crate tempfile;
mod mmap_util;
use mmap_util::*;
mod iter;
use iter::*;
mod filter;
use filter::*;
mod ints;
pub use ints::*;
#[allow(unused_imports)]
use std::mem::{size_of, size_of_val};
use std::fs::{File};
use std::io::{self, Write};
use std::str::FromStr;
use memmap::{Mmap, MmapMut};
pub const MAGIC_NUMBER: u64 = 0x4242_0868_7537_8673;
pub const FILE_FORMAT_VERSION: u16 = 0;
#[derive(Debug)]
pub struct Storage {
header: Header,
file: File,
mmap: MmapMut,
}
impl Storage {
pub fn create(file: File, schema: &str) -> io::Result<Storage> {
let header = Header::new(schema).unwrap();
file.set_len(10*1024*1024)?;
let mmap = unsafe { MmapMut::map_mut(&file)? };
let mut s = Storage { header, file, mmap };
s.persist()?;
Ok(s)
}
fn calc_required_space(&self) -> u64 {
let header = self.header.serialized_size();
header + self.header.n_tuples * self.header.schema.data_size
}
fn calc_offset_location(&self, offset: u64) -> u64 {
assert!(offset <= self.header.n_tuples, "Storage::calc_offset_location({}) too big - n_tuples: {}", offset, self.header.n_tuples);
let header = self.header.serialized_size();
header + offset * self.header.schema.data_size
}
pub fn persist(&mut self) -> io::Result<()> {
let space = self.calc_required_space();
self.file.set_len(space)?;
self.header.persist(&mut self.mmap);
self.mmap.flush()?;
Ok( () )
}
pub fn open(file: File) -> io::Result<Storage> {
file.set_len(10*1024*1024)?;
let mut reader = MmapReader::new(&file)?;
let header = Header::open(&mut reader).unwrap();
let mmap = unsafe { MmapMut::map_mut(&file)? };
Ok(Storage { header, file, mmap })
}
pub fn insert(&mut self, tuple: &mut Tuple) -> Result<(), SchemaError> {
self.header.schema.matches(&tuple)?;
if self.iter().any(|t| t.data[0] == tuple.data[0]) {
return Err(SchemaError::DuplicateKey)
}
let mut current_offset = self.calc_required_space();
let new_size = current_offset + self.header.schema.data_size;
assert!(self.file.set_len(new_size).is_ok(), "Storage::insert file.set_len({}) failed", new_size);
assert!((new_size as usize) <= self.mmap.len(), "Storage::insert: out of mapped space");
let start = self.iter().enumerate()
.fold(0, |b, (i, t)| {
if t.data[0] < tuple.data[0] {
i as u64
} else {
b
}
});
if start < self.header.n_tuples {
let mut i = self.header.n_tuples;
while i > start {
let offset = self.calc_offset_location(i);
let mut last = self.get(i - 1).expect("Storage::insert moving out of range");
let mut writer = MmapWriter::at_offset(
&mut self.mmap,
offset as usize);
last.persist(&mut writer);
i -= 1;
}
current_offset = self.calc_offset_location(start);
}
let mut writer = MmapWriter::at_offset(&mut self.mmap, current_offset as usize);
tuple.persist(&mut writer);
self.header.n_tuples += 1;
Ok( () )
}
pub fn append(&mut self, tuple: &mut Tuple) -> Result<(), SchemaError> {
self.header.schema.matches(&tuple)?;
let current_offset = self.calc_required_space();
let new_size = current_offset + self.header.schema.data_size;
assert!(self.file.set_len(new_size).is_ok(), "Storage::append file.set_len({}) failed", new_size);
assert!((new_size as usize) <= self.mmap.len(), "Storage::append: out of mapped space");
let mut writer = MmapWriter::at_offset(&mut self.mmap, current_offset as usize);
tuple.persist(&mut writer);
self.header.n_tuples += 1;
Ok( () )
}
pub fn update(&mut self, tuple: &mut Tuple) -> Result<usize, SchemaError> {
self.header.schema.matches(tuple)?;
let mut count = 0;
for i in 0..self.header.n_tuples {
let t = self.get(i).expect("Tuple must exist in Range");
if tuple.data[0] == t.data[0] {
let current_offset = self.header.serialized_size()
+ i as u64 * self.header.schema.data_size;
let mut writer = MmapWriter::at_offset(&mut self.mmap, current_offset as usize);
tuple.persist(&mut writer);
count += 1;
}
}
Ok(count)
}
pub fn remove_all(&mut self, filter: &Filter) -> usize {
let mut deleted = 0;
for i in (0..self.header.n_tuples).map(|i| i).rev() {
let tuple = self.get(i).expect("in Range access current");
if filter.matches(&tuple) {
let mut last = self.get(self.header.n_tuples - 1).expect("in Range access last");
let current_offset = self.calc_offset_location(i);
let mut writer = MmapWriter::at_offset(&mut self.mmap, current_offset as usize);
last.persist(&mut writer);
deleted += 1;
self.header.n_tuples -= 1;
}
}
deleted
}
pub fn remove(&mut self, filter: &Filter) -> usize {
let mut deleted = 0;
let mut i = 0;
while i < self.header.n_tuples {
let tuple = self.get(i).expect("in Range access current");
if filter.matches(&tuple) {
for j in i+1..self.header.n_tuples {
let mut current = self.get(j).expect("in Range access moving");
let offset = self.calc_offset_location(j-1);
let mut writer = MmapWriter::at_offset(&mut self.mmap, offset as usize);
current.persist(&mut writer);
}
deleted += 1;
self.header.n_tuples -= 1;
} else {
i += 1;
}
}
deleted
}
pub fn iter(&self) -> StorageIterator {
StorageIterator::new(self)
}
pub fn get(&self, offset: u64) -> Option<Tuple> {
if offset < self.header.n_tuples {
let offset = self.calc_offset_location(offset);
let mut reader = MmapReader::at_offset(&self.file, offset as usize).unwrap();
Some(Tuple::from(&mut reader, &self.header.schema))
} else {
None
}
}
pub fn create_filter(&self, filter: &str) -> Result<Filter, SchemaParseError> {
Filter::create(&self.header.schema, filter)
}
pub fn schema_to_string(&self) -> String {
format!("({})", self.header.schema.schema.join(", "))
}
}
impl Drop for Storage {
fn drop(&mut self) {
assert!(self.persist().is_ok());
assert!(self.mmap.flush().is_ok());
assert!(self.file.flush().is_ok());
}
}
impl std::cmp::PartialEq for Storage {
fn eq(&self, other: &Self) -> bool {
self.header == other.header && self.iter().zip(other.iter()).all(|(a, b)| a == b)
}
}
#[derive(Debug, PartialEq)]
pub struct Header {
magic_number: u64,
format_version: u16,
header_size: u16,
n_tuples: u64,
schema: Schema,
}
impl Header {
fn new(schema: &str) -> Result<Header, SchemaParseError> {
Ok(Header {
magic_number: MAGIC_NUMBER,
format_version: FILE_FORMAT_VERSION,
header_size: Self::base_size(),
n_tuples: 0,
schema: schema.parse()?,
})
}
fn open(reader: &mut MmapReader) -> Result<Header, SchemaParseError> {
use SchemaParseError::*;
if &MAGIC_NUMBER != reader.get() {
return Err(InvalidMagicNumber)
}
if &FILE_FORMAT_VERSION != reader.get() {
return Err(InvalidFileFormatVersion)
}
let header_size = reader.get();
if *header_size < Self::base_size() {
return Err(HeaderSizeTooSmall(*header_size))
}
let n_tuples = reader.get();
let slice = reader.get_str_slice((header_size - Self::base_size()) as usize);
let schema = slice.parse().unwrap();
Ok(Header {
magic_number: MAGIC_NUMBER,
format_version: FILE_FORMAT_VERSION,
header_size: *header_size,
n_tuples: *n_tuples,
schema,
})
}
fn base_size() -> u16 {
let size = size_of::<u64>()
+ size_of::<u16>()
+ size_of::<u16>()
+ size_of::<u64>()
;
assert!(size <= u16::max_value() as usize, "base_size outgrew data_type {}", size);
size as u16
}
fn serialized_size(&self) -> u64 {
Self::base_size() as u64 + self.schema.serialized_size()
}
fn persist(&mut self, mmap: &mut MmapMut) {
self.header_size = self.serialized_size() as u16;
let mut writer = MmapWriter::new(mmap);
writer.append(&self.magic_number);
writer.append(&self.format_version);
writer.append(&(self.serialized_size() as u16));
writer.append(&self.n_tuples);
self.schema.persist(&mut writer);
}
}
#[derive(Debug, PartialEq)]
pub struct Schema {
schema: Vec<String>,
data_size: u64,
}
impl FromStr for Schema {
type Err = SchemaParseError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let schema = s.trim_matches(|p| p == '(' || p == ')' )
.split(",")
.map(|s| s.trim())
.map(|s| {
match s {
"u64" | "i64" | "u32" | "i32" | "u16" | "i16" | "u8" | "i8" => Ok(s.into()),
t => Err(SchemaParseError::InvalidType(t.into()))
}
})
.collect::<Result<Vec<String>, Self::Err>>()?;
let data_size = schema.iter()
.map(|s: &String| {
match &**s {
"u64" | "i64" => 8,
"u32" | "i32" => 4,
"u16" | "i16" => 2,
"u8" | "i8" => 1,
_ => panic!("unhandled type {:?}", s),
}
})
.sum();
Ok(Schema { schema, data_size })
}
}
impl Schema {
fn serialize_schema(&self) -> String {
let mut s = "(".to_string();
s += &self.schema.join(",");
s += ")";
s
}
fn serialized_size(&self) -> u64 {
self.serialize_schema().len() as u64
}
fn persist(&mut self, writer: &mut MmapWriter) {
let schema = self.serialize_schema();
writer.append_slice(&schema.as_bytes());
}
fn matches(&self, tuple: &Tuple) -> Result<(), SchemaError> {
use SchemaError::*;
if self.schema.len() != tuple.data.len() {
return Err(MismatchedLength)
}
for (s, t) in self.schema.iter().zip(tuple.data.iter()) {
let t_s = t.str_repr();
if s != t_s {
return Err(InvalidType(t_s.into()))
}
}
Ok( () )
}
}
#[derive(Debug, PartialEq)]
pub enum SchemaParseError {
InvalidType(String),
InvalidMagicNumber,
InvalidFileFormatVersion,
HeaderSizeTooSmall(u16),
FilterInvalidOperand(String),
}
#[derive(Debug, PartialEq)]
pub enum SchemaError {
MismatchedLength,
InvalidType(String),
DuplicateKey,
}
#[derive(Debug, PartialEq)]
pub struct Tuple {
data: Vec<Ints>,
}
impl Tuple {
pub fn new(data: Vec<Ints>) -> Tuple {
Tuple { data }
}
pub fn from(reader: &mut MmapReader, schema: &Schema) -> Tuple {
let mut data = vec![];
for t in &schema.schema {
let x = match &t[..] {
"u64" => U64(*reader.get()),
"i64" => I64(*reader.get()),
"u32" => U32(*reader.get()),
"i32" => I32(*reader.get()),
"u16" => U16(*reader.get()),
"i16" => I16(*reader.get()),
"u8" => U8(*reader.get()),
"i8" => I8(*reader.get()),
_ => unimplemented!("Tuple::from({:?}) missing type {:?}", schema, t),
};
data.push(x);
}
Tuple { data }
}
pub fn from_str(storage: &Storage, input: &str) -> Tuple {
let ref schema = storage.header.schema;
Self::from_schema_and_str(schema, input)
}
fn from_schema_and_str(schema: &Schema, input: &str) -> Tuple {
let input = input.trim_matches(|p| p == '(' || p == ')' )
.split(",")
.map(|s| s.trim())
.collect::<Vec<_>>();
assert_eq!(schema.schema.len(), input.len(), "Tuple::from_str() input len does not match");
let mut data = vec![];
for (t, i) in schema.schema.iter().zip(input.iter()) {
let x = match &t[..] {
"u64" => U64(i.parse().unwrap()),
"i64" => I64(i.parse().unwrap()),
"u32" => U32(i.parse().unwrap()),
"i32" => I32(i.parse().unwrap()),
"u16" => U16(i.parse().unwrap()),
"i16" => I16(i.parse().unwrap()),
"u8" => U8 (i.parse().unwrap()),
"i8" => I8 (i.parse().unwrap()),
_ => unimplemented!("Tuple::from({:?}) missing type {:?}", schema, t),
};
data.push(x);
}
Tuple { data }
}
fn persist(&mut self, writer: &mut MmapWriter) {
for val in &self.data {
match val {
U64(value) => writer.append(value),
I64(value) => writer.append(value),
U32(value) => writer.append(value),
I32(value) => writer.append(value),
U16(value) => writer.append(value),
I16(value) => writer.append(value),
U8(value) => writer.append(value),
I8(value) => writer.append(value),
}
}
}
pub fn to_string(&self) -> String {
let mut parts = vec![];
for val in &self.data {
parts.push(match val {
U64(value) => format!("{}", value),
I64(value) => format!("{}", value),
U32(value) => format!("{}", value),
I32(value) => format!("{}", value),
U16(value) => format!("{}", value),
I16(value) => format!("{}", value),
U8(value) => format!("{}", value),
I8(value) => format!("{}", value),
});
}
format!("({})", parts.join(", "))
}
}
#[cfg(test)]
mod test {
use super::*;
use tempfile::{TempDir};
fn random_file() -> io::Result<(TempDir, File)> {
let dir = tempfile::tempdir()?;
println!("{:?}", dir);
let path = dir.path().join("demo.ts");
let file = std::fs::OpenOptions::new()
.read(true).write(true).create(true)
.open(&path)?;
Ok((dir, file))
}
#[test]
fn serialize_empty() {
let expected = [
0x73, 0x86, 0x37, 0x75, 0x68, 0x08, 0x42, 0x42,
0x00, 0x00, 0x1d, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x28, 0x75, 0x36, 0x34,
0x2c, 0x69, 0x36, 0x34, 0x29,
];
let (dir, file) = random_file().unwrap();
{
let mut s = Storage::create(file, "(u64, i64)").unwrap();
assert!(s.persist().is_ok());
println!("{:?}", s);
}
let r = std::fs::read(dir.path().join("demo.ts")).unwrap();
assert_eq!(expected, *r);
}
#[test]
fn deserialize_empty() {
let blob = [
0x73, 0x86, 0x37, 0x75, 0x68, 0x08, 0x42, 0x42,
0x00, 0x00, 0x1d, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x28, 0x75, 0x36, 0x34,
0x2c, 0x69, 0x36, 0x34, 0x29,
];
let (dir, file) = random_file().unwrap();
std::fs::write(dir.path().join("demo.ts"), &blob).is_ok();
let r = Storage::open(file).unwrap();
println!("{:?}", r);
let (_dir, file) = random_file().unwrap();
let expected = Storage::create(file, "(u64, i64)").unwrap();
assert_eq!(expected, r);
}
#[test]
fn serialize_one() {
let expected = vec![
0x73, 0x86, 0x37, 0x75, 0x68, 0x08, 0x42, 0x42,
0x00, 0x00, 0x1d, 0x00, 0x01, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x28, 0x75, 0x36, 0x34,
0x2c, 0x69, 0x36, 0x34, 0x29,
0x0c, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0xc0, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
];
let (dir, file) = random_file().unwrap();
{
let mut s = Storage::create(file, "(u64, i64)").unwrap();
assert!(s.append(&mut Tuple{ data: vec![U64(12), I64(-64)]}).is_ok());
assert!(s.persist().is_ok());
println!("{:?}", s);
}
let r = std::fs::read(dir.path().join("demo.ts")).unwrap();
assert_eq!(expected, r, "\n expected: `{:x?}`\n read: `{:x?}`", expected, r);
}
#[test]
fn deserialize_one() {
let blob = vec![
0x73, 0x86, 0x37, 0x75, 0x68, 0x08, 0x42, 0x42,
0x00, 0x00,
0x1d, 0x00,
0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x28, 0x75, 0x36, 0x34, 0x2c, 0x69, 0x36, 0x34,
0x29,
0x0c, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0xc0, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
];
let (dir, file) = random_file().unwrap();
std::fs::write(dir.path().join("demo.ts"), &blob).is_ok();
let r = Storage::open(file).unwrap();
println!("{:?}", r);
let (_dir, file) = random_file().unwrap();
let mut expected = Storage::create(file, "(u64, i64)").unwrap();
assert!(expected.append(&mut Tuple{ data: vec![U64(12), I64(-64)]}).is_ok());
assert!(expected.persist().is_ok());
assert_eq!(expected, r);
let mut it = expected.iter();
let expected = Tuple { data: vec![ U64(12), I64(-64) ]};
assert_eq!(Some(expected), it.next());
assert_eq!(None, it.next());
}
#[test]
fn update_one() {
let blob = vec![
0x73, 0x86, 0x37, 0x75, 0x68, 0x08, 0x42, 0x42,
0x00, 0x00,
0x1d, 0x00,
0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x28, 0x75, 0x36, 0x34, 0x2c, 0x69, 0x36, 0x34,
0x29,
0x0c, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0xc0, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
];
let expected = vec![
0x73, 0x86, 0x37, 0x75, 0x68, 0x08, 0x42, 0x42,
0x00, 0x00,
0x1d, 0x00,
0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x28, 0x75, 0x36, 0x34, 0x2c, 0x69, 0x36, 0x34,
0x29,
0x0c, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x40, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
];
let (dir, file) = random_file().unwrap();
std::fs::write(dir.path().join("demo.ts"), &blob).is_ok();
let mut s = Storage::open(file).unwrap();
println!("{:?}", s);
let mut t = Tuple { data: vec![ U64(12), I64(64) ]};
assert_eq!(Ok(1), s.update(&mut t));
s.persist().unwrap();
let r = std::fs::read(dir.path().join("demo.ts")).unwrap();
assert_eq!(expected, r, "\n expected: `{:x?}`\n read: `{:x?}`", expected, r);
let mut it = s.iter();
assert_eq!(Some(t), it.next());
assert_eq!(None, it.next());
}
#[test]
fn insert_duplicate() {
let schema = "(u8, i8)";
let (_dir, file) = random_file().unwrap();
let mut s = Storage::create(file, schema).unwrap();
let mut t = Tuple { data: vec![ U8(12), I8(64) ]};
assert_eq!(Ok(()), s.insert(&mut t));
assert_eq!(1, s.header.n_tuples);
assert_eq!(Err(SchemaError::DuplicateKey), s.insert(&mut t));
assert_eq!(1, s.header.n_tuples);
}
#[test]
fn insert_sorted() {
let schema = "(u8, i8)";
let (_dir, file) = random_file().unwrap();
let mut s = Storage::create(file, schema).unwrap();
let mut t0 = Tuple { data: vec![ U8(12), I8(64) ]};
assert_eq!(Ok(()), s.insert(&mut t0));
assert_eq!(1, s.header.n_tuples);
let mut t1 = Tuple { data: vec![ U8(8), I8(64) ]};
assert_eq!(Ok(()), s.insert(&mut t1));
assert_eq!(2, s.header.n_tuples);
let mut t2 = Tuple { data: vec![ U8(6), I8(64) ]};
assert_eq!(Ok(()), s.insert(&mut t2));
assert_eq!(3, s.header.n_tuples);
let mut iter = s.iter();
assert_eq!(Some(t2), iter.next(), "t2");
assert_eq!(Some(t1), iter.next(), "t1");
assert_eq!(Some(t0), iter.next(), "t0");
}
#[test]
fn remove_all() {
let schema = "(u8, i8)";
let (_dir, file) = random_file().unwrap();
let mut s = Storage::create(file, schema).unwrap();
let mut t0 = Tuple { data: vec![ U8(12), I8(64) ]};
assert_eq!(Ok(()), s.insert(&mut t0));
let mut t1 = Tuple { data: vec![ U8(8), I8(64) ]};
assert_eq!(Ok(()), s.insert(&mut t1));
let filter = Filter::Equal(U8(12));
assert_eq!(1, s.remove_all(&filter));
assert_eq!(1, s.header.n_tuples);
let mut iter = s.iter();
assert_eq!(Some(t1), iter.next(), "t1");
assert_eq!(None, iter.next(), "should be empty");
}
#[test]
fn remove_mid() {
let schema = "(u8, i8)";
let (_dir, file) = random_file().unwrap();
let mut s = Storage::create(file, schema).unwrap();
let mut t0 = Tuple { data: vec![ U8(12), I8(64) ]};
assert_eq!(Ok(()), s.insert(&mut t0));
let mut t1 = Tuple { data: vec![ U8(8), I8(64) ]};
assert_eq!(Ok(()), s.insert(&mut t1));
let mut t2 = Tuple { data: vec![ U8(6), I8(64) ]};
assert_eq!(Ok(()), s.insert(&mut t2));
let filter = Filter::Equal(U8(8));
assert_eq!(1, s.remove(&filter));
assert_eq!(2, s.header.n_tuples);
let mut iter = s.iter();
assert_eq!(Some(t2), iter.next(), "t2");
assert_eq!(Some(t0), iter.next(), "t0");
assert_eq!(None, iter.next(), "should be empty");
}
#[test]
fn remove_first() {
let schema = "(u8, i8)";
let (_dir, file) = random_file().unwrap();
let mut s = Storage::create(file, schema).unwrap();
let mut t0 = Tuple { data: vec![ U8(12), I8(0) ]};
assert_eq!(Ok(()), s.insert(&mut t0));
let mut t1 = Tuple { data: vec![ U8(8), I8(1) ]};
assert_eq!(Ok(()), s.insert(&mut t1));
let mut t2 = Tuple { data: vec![ U8(6), I8(2) ]};
assert_eq!(Ok(()), s.insert(&mut t2));
let filter = Filter::Less(U8(8));
assert_eq!(1, s.remove(&filter));
assert_eq!(2, s.header.n_tuples);
let mut iter = s.iter();
assert_eq!(Some(t1), iter.next(), "t1");
assert_eq!(Some(t0), iter.next(), "t0");
assert_eq!(None, iter.next(), "should be empty");
}
#[test]
fn remove_sorted_all() {
let schema = "(i8, i8)";
let (_dir, file) = random_file().unwrap();
let mut s = Storage::create(file, schema).unwrap();
let mut t0 = Tuple { data: vec![ I8(2), I8(0) ]};
assert_eq!(Ok(()), s.insert(&mut t0));
let mut t1 = Tuple { data: vec![ I8(0), I8(1) ]};
assert_eq!(Ok(()), s.insert(&mut t1));
let mut t2 = Tuple { data: vec![ I8(-1), I8(2) ]};
assert_eq!(Ok(()), s.insert(&mut t2));
let filter = Filter::Greater(I8(-10));
assert_eq!(3, s.remove(&filter));
assert_eq!(0, s.header.n_tuples);
let mut iter = s.iter();
assert_eq!(None, iter.next(), "should be empty");
}
#[test]
fn schema_to_string() {
let schema = "(i8, i8)";
let (_dir, file) = random_file().unwrap();
let s = Storage::create(file, schema).unwrap();
assert_eq!(schema, s.schema_to_string());
}
#[test]
fn tuple_to_string() {
let expected = "(-666, 747)";
let t = Tuple { data: vec![ I32(-666), U16(747) ] };
assert_eq!(expected, t.to_string())
}
}