//! Using the tuple_storage as a library or a stand alone binary
//!
//! # Installation
//!
//! ```bash
//! $ cargo install tuple_storage
//! ```
//!
//! Find an example usage in the [README](https://gitlab.com/dns2utf8/tuple_storage/blob/master/README.md).

//extern crate fs2;
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};

/// The Byte Tag to identify the `Storage`
pub const MAGIC_NUMBER: u64 = 0x4242_0868_7537_8673;
/// The file format version
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();
        // FIXME random length, can no grow
        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> {
        // FIXME random length, can no grow
        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 })
    }

    /// Insert does not allow duplicate key and
    /// inserts the Tuple in the set in a sorted way
    pub fn insert(&mut self, tuple: &mut Tuple) -> Result<(), SchemaError> {
        self.header.schema.matches(&tuple)?;

        // TODO optimieren
        if self.iter().any(|t| t.data[0] == tuple.data[0]) {
            return Err(SchemaError::DuplicateKey)
        }

        // add space to memmap
        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)| {
                            //println!("i: {}", i);
                            if t.data[0] < tuple.data[0] {
                                i as u64
                            } else {
                                b
                            }
                        });
        //println!("Storage::insert({:?}) => start {}", tuple, start);
        if start < self.header.n_tuples {
            // shift the rest
            let mut i = self.header.n_tuples;
            while i > start {
                //println!("{}", i);
                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( () )
    }


    /// Append allows duplicate key
    pub fn append(&mut self, tuple: &mut Tuple) -> Result<(), SchemaError> {
        self.header.schema.matches(&tuple)?;

        // add to memmap
        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( () )
    }
    
    /// Update all tuples matching the key
    /// 
    /// returns how many tuples were updated
    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)
    }
    
    /// Remove all tuples matching the key.
    /// Working from the end to the front of the Storage.
    pub fn remove_all(&mut self, filter: &Filter) -> usize {
        //println!("Storage::remove_all");
        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");
            //println!("filter.matches({:?}) == {}", tuple, filter.matches(&tuple));
            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
    }
    /// Delete all tuples matching the key sorted
    pub fn remove(&mut self, filter: &Filter) -> usize {
        //println!("Storage::remove({:?})", filter);
        let mut deleted = 0;

        let mut i = 0;
        while i < self.header.n_tuples {
            let tuple = self.get(i).expect("in Range access current");
            //println!("filter.matches({:?}) == {} | {}", tuple, filter.matches(&tuple), i);
            if filter.matches(&tuple) {
                for j in i+1..self.header.n_tuples {
                    let mut current = self.get(j).expect("in Range access moving");
                    //println!("    {} => {:?}", j, current);
                    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 iter_mut(&mut self) -> StorageIteratorMut {
        StorageIteratorMut::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,
    /// Size of header
    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);

        //println!("##########################\n  extracted schema: {:?}", slice);
        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>()
            // ignore schema
            ;

        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>,
    //index_size: u64,
    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 {
        // TODO
        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 {
        //println!("Tuple::from()");
        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) {
        //println!("Tuple::persist({:?})", self);
        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))
    }

    /*
    fn flush_stdout() {
        let stdout = io::stdout();
        let mut handle = stdout.lock();
        handle.flush().is_ok();
    }
    */

    #[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,
            // Data
            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,
            // format version
            0x00, 0x00,
            // serialized size
            0x1d, 0x00,
            // n_tuples
            0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
            // schema String
            0x28, 0x75, 0x36, 0x34, 0x2c, 0x69, 0x36, 0x34,
            0x29,
            // Data
            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,
            // format version
            0x00, 0x00,
            // serialized size
            0x1d, 0x00,
            // n_tuples
            0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
            // schema String
            0x28, 0x75, 0x36, 0x34, 0x2c, 0x69, 0x36, 0x34,
            0x29,
            // Data: (U64(12), I64(-64))
            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,
            // format version
            0x00, 0x00,
            // serialized size
            0x1d, 0x00,
            // n_tuples
            0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
            // schema String
            0x28, 0x75, 0x36, 0x34, 0x2c, 0x69, 0x36, 0x34,
            0x29,
            // Data: (U64(12), I64(64))
            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())
    }
}