Protocol and Index_File

appendable-rs/appendable/Cargo.toml

@@ -6,3 +6,9 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+protocol = { path="../protocol" }
+xxhash-rust = { version = "0.8.8", features = ["xxh3"] }
+
+[dev-dependencies]
+tempfile = "3.9.0"
+

appendable-rs/appendable/mock_data.jsonl

@@ -0,0 +1,2 @@
+{"name": "matteo", "id": 2, "alpha": ["a", "b", "c"]}
+{"name": "kevin", "id": 1, "alpha": ["x", "y", "z"]}

appendable-rs/appendable/src/handler/jsonl_handler.rs

@@ -0,0 +1,71 @@
+use crate::index_file::IndexFile;
+use crate::io::DataHandler;
+use std::io::{BufRead, BufReader, Cursor, Seek, SeekFrom};
+use xxhash_rust::xxh3::Xxh3;
+
+pub struct JSONLHandler {
+    // todo! change to borrowed type like &[u8] -- spent too long battling lifetimes
+    reader: BufReader<Cursor<Vec<u8>>>,
+    xxh3: Xxh3,
+}
+impl JSONLHandler {
+    pub fn new(data: Vec<u8>) -> Self {
+        JSONLHandler {
+            reader: BufReader::new(Cursor::new(data)),
+            xxh3: Xxh3::new(),
+        }
+    }
+}
+impl Seek for JSONLHandler {
+    fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
+        self.reader.seek(pos)
+    }
+}
+impl DataHandler for JSONLHandler {
+    fn synchronize(&mut self, index_file: &mut IndexFile) -> Result<(), String> {
+        let mut line = String::new();
+        let mut start_offset: u64 = 0;
+
+        while self
+            .reader
+            .read_line(&mut line)
+            .map_err(|e| e.to_string())?
+            > 0
+        {
+            let existing_count = index_file.end_byte_offsets.len();
+            // compute byte_offset for current line
+            let line_length = line.as_bytes().len() as u64;
+            let current_offset = start_offset + line_length + 1;
+            index_file.end_byte_offsets.push(current_offset);
+
+            // compute checksum
+            self.xxh3.update(line.as_bytes());
+            let checksum = self.xxh3.digest(); // produce the final hash value
+            index_file.checksums.push(checksum);
+
+            // Process the JSON line and update indexes
+            handle_json_object(
+                line.into_bytes(),
+                index_file,
+                &mut vec![],
+                existing_count as u64,
+                start_offset,
+            )?;
+
+            start_offset = current_offset;
+            line.clear();
+        }
+
+        Ok(())
+    }
+}
+
+fn handle_json_object(
+    json_line: Vec<u8>,
+    index_file: &mut IndexFile,
+    path: &mut Vec<String>,
+    data_index: u64,
+    data_offset: u64,
+) -> Result<usize, String> {
+    Ok(1)
+}

appendable-rs/appendable/src/index_file.rs

@@ -0,0 +1,127 @@
+use crate::io::DataHandler;
+use protocol::field_type::FieldFlags;
+use protocol::{FieldType, IndexRecord, Version};
+use std::collections::HashMap;
+use std::fmt;
+use std::fmt::Formatter;
+
+const CURRENT_VERSION: Version = 1;
+
+pub(crate) struct Index {
+    pub(crate) field_name: String,
+    pub(crate) field_type: FieldFlags,
+    pub(crate) index_records: HashMap<IndexKey, Vec<IndexRecord>>,
+}
+
+/// `IndexFile` is a representation of the entire index file.
+pub struct IndexFile {
+    version: Version,
+    pub(crate) indexes: Vec<Index>,
+    pub(crate) end_byte_offsets: Vec<u64>,
+    pub(crate) checksums: Vec<u64>,
+    tail: u32,
+}
+
+impl IndexFile {
+    pub fn new(mut data_handler: Box<dyn DataHandler>) -> Result<Self, String> {
+        let mut file = IndexFile {
+            version: CURRENT_VERSION,
+            indexes: Vec::new(),
+            end_byte_offsets: Vec::new(),
+            checksums: Vec::new(),
+            tail: 0,
+        };
+
+        data_handler.synchronize(&mut file)?;
+
+        Ok(file)
+    }
+
+    pub(crate) fn find_index(&mut self, name: &str, value: &IndexKey) -> usize {
+        if let Some((position, _)) = self
+            .indexes
+            .iter()
+            .enumerate()
+            .find(|(_, index)| index.field_name == name)
+        {
+            if !self.indexes[position]
+                .field_type
+                .contains(value.field_type())
+            {
+                self.indexes[position].field_type.set(value.field_type());
+            }
+
+            position
+        } else {
+            let mut new_index = Index {
+                field_name: name.to_string(),
+                field_type: FieldFlags::new(),
+                index_records: HashMap::new(),
+            };
+
+            new_index.field_type.set(value.field_type());
+            self.indexes.push(new_index);
+            self.indexes.len() - 1
+        }
+    }
+}
+
+/// `IndexKey` addresses the dynamic typing of keys in `IndexRecord` by stating all possible variants
+#[derive(Eq, PartialEq, Debug, Clone)]
+pub enum IndexKey {
+    String(String),
+    Number(String),
+    Boolean(bool),
+    Array(Vec<IndexKey>),
+    Object(HashMap<String, IndexKey>),
+}
+
+impl IndexKey {
+    fn field_type(&self) -> FieldType {
+        match self {
+            IndexKey::String(_) => FieldType::String,
+            IndexKey::Number(_) => FieldType::Number,
+            IndexKey::Boolean(_) => FieldType::Boolean,
+            IndexKey::Array(_) => FieldType::Array,
+            IndexKey::Object(_) => FieldType::Object,
+        }
+    }
+}
+
+impl fmt::Display for Index {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "Field\nname: {}\n\ttype: {:?}\n\tindex_records: {:?}",
+            self.field_name, self.field_type, self.index_records
+        )
+    }
+}
+
+impl fmt::Display for IndexKey {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        match self {
+            IndexKey::String(s) => write!(f, "{}", s),
+            IndexKey::Number(n) => write!(f, "{}", n),
+            IndexKey::Boolean(b) => write!(f, "{}", b),
+            IndexKey::Array(v) => {
+                let elements = v
+                    .iter()
+                    .map(|element| format!("{}", element))
+                    .collect::<Vec<String>>()
+                    .join(", ");
+
+                write!(f, "[{}]", elements)
+            }
+            IndexKey::Object(o) => {
+                let entries = o
+                    .iter()
+                    .map(|(key, value)| format!("{}: {}", key, value))
+                    .collect::<Vec<String>>()
+                    .join(", ");
+
+                write!(f, "{{{}}}", entries)
+            }
+        }
+    }
+}

appendable-rs/appendable/src/io.rs

@@ -0,0 +1,6 @@
+use crate::index_file::{Index, IndexFile};
+use std::io::Seek;
+
+pub trait DataHandler: Seek {
+    fn synchronize(&mut self, index_file: &mut IndexFile) -> Result<(), String>;
+}