Update JSTable format to include timestamp and implement timestamp-based conflict resolution during compaction

Author: michaelmior

specs/storage.md

@@ -7,12 +7,14 @@ Each JSTable is stored in a single file.
 This file must contain a JSON Schema as well as all the associated documents.
 
 The format is as follows:
-1.  The schema, serialized as JSON, followed by a newline.
+1.  A JSON object containing metadata about the table. It must include:
+    *   `timestamp`: The time the table was created, as a Unix timestamp in milliseconds.
+    *   `schema`: The JSON Schema for the documents in the table.
 2.  A sequence of records, one per line. Each record is a JSON array `[id, document]`.
 
 Example:
 ```
-{"type":"object","properties":{"a":{"type":["integer"]}}}
+{"timestamp": 1686776400000, "schema": {"type":"object","properties":{"a":{"type":["integer"]}}}}
 ["01H4J3J4J3J4J3J4J3J4J3J4J3", {"a": 1}]
 ["01H4J3J4J3J4J3J4J3J4J3J4J4", {"a": 2}]
 ["01H4J3J4J3J4J3J4J3J4J3J4J5", null]

src/db.rs

@@ -194,8 +194,10 @@ mod tests {
         db.delete(&id1);
 
         let db2 = DB::new(dir.path().to_str().unwrap());
-        assert_eq!(db2.memtable.len(), 1);
+        // Memtable should contain 2 entries: id2 (doc) and id1 (null/tombstone)
+        assert_eq!(db2.memtable.len(), 2);
         assert_eq!(*db2.memtable.documents.get(&id2).unwrap(), doc2);
+        assert!(db2.memtable.documents.get(&id1).unwrap().is_null());
     }
 
     #[test]

src/jstable.rs

@@ -1,23 +1,35 @@
 use crate::schema::Schema;
+use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use std::collections::BTreeMap;
 use std::io::{self, BufReader, BufRead, Write};
 use std::fs::File;
 
 pub struct JSTable {
+    pub timestamp: u64,
     pub schema: Schema,
     pub documents: BTreeMap<String, Value>,
 }
 
+#[derive(Serialize, Deserialize)]
+struct JSTableHeader {
+    timestamp: u64,
+    schema: Schema,
+}
+
 impl JSTable {
-    pub fn new(schema: Schema, documents: BTreeMap<String, Value>) -> Self {
-        JSTable { schema, documents }
+    pub fn new(timestamp: u64, schema: Schema, documents: BTreeMap<String, Value>) -> Self {
+        JSTable { timestamp, schema, documents }
     }
 
     pub fn write(&self, path: &str) -> io::Result<()> {
         let mut file = File::create(path)?;
-        let schema_json = serde_json::to_string(&self.schema)?;
-        writeln!(file, "{}", schema_json)?;
+        let header = JSTableHeader {
+            timestamp: self.timestamp,
+            schema: self.schema.clone(),
+        };
+        let header_json = serde_json::to_string(&header)?;
+        writeln!(file, "{}", header_json)?;
         for (id, doc) in &self.documents {
             let record: (String, &Value) = (id.clone(), doc);
             let record_json = serde_json::to_string(&record)?;
@@ -32,8 +44,8 @@ pub fn read_jstable(path: &str) -> io::Result<JSTable> {
     let reader = BufReader::new(file);
     let mut lines = reader.lines();
 
-    let schema_line = lines.next().ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Missing schema line"))??;
-    let schema: Schema = serde_json::from_str(&schema_line)?;
+    let header_line = lines.next().ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Missing header line"))??;
+    let header: JSTableHeader = serde_json::from_str(&header_line)?;
 
     let mut documents = BTreeMap::new();
     for line_result in lines {
@@ -45,14 +57,25 @@ pub fn read_jstable(path: &str) -> io::Result<JSTable> {
         documents.insert(record.0, record.1);
     }
 
-    Ok(JSTable { schema, documents })
+    Ok(JSTable { 
+        timestamp: header.timestamp,
+        schema: header.schema, 
+        documents 
+    })
 }
 
 pub fn merge_jstables(tables: &[JSTable]) -> JSTable {
+    let mut sorted_tables: Vec<&JSTable> = tables.iter().collect();
+    sorted_tables.sort_by_key(|t| t.timestamp);
+
     let mut merged_schema = Schema::new(crate::schema::SchemaType::Object);
     let mut merged_documents = BTreeMap::new();
+    let mut max_timestamp = 0;
 
-    for table in tables {
+    for table in sorted_tables {
+        if table.timestamp > max_timestamp {
+            max_timestamp = table.timestamp;
+        }
         merged_schema.merge(table.schema.clone());
         for (id, doc) in &table.documents {
             merged_documents.insert(id.clone(), doc.clone());
@@ -61,7 +84,7 @@ pub fn merge_jstables(tables: &[JSTable]) -> JSTable {
 
     merged_documents.retain(|_, v| !v.is_null());
 
-    JSTable::new(merged_schema, merged_documents)
+    JSTable::new(max_timestamp, merged_schema, merged_documents)
 }
 
 #[cfg(test)]
@@ -75,19 +98,25 @@ mod tests {
     #[test]
     fn test_read_jstable() -> Result<(), Box<dyn std::error::Error>> {
         let mut file = NamedTempFile::new().unwrap();
-        let schema_json = serde_json::to_string(&Schema {
+        let schema = Schema {
             types: vec![SchemaType::Object],
             properties: Some(BTreeMap::from([
                 ("a".to_string(), Schema::new(SchemaType::Integer)),
             ])),
             items: None,
-        }).unwrap();
-        writeln!(file, "{}", schema_json).unwrap();
+        };
+        let header = JSTableHeader {
+            timestamp: 12345,
+            schema: schema,
+        };
+        let header_json = serde_json::to_string(&header).unwrap();
+        writeln!(file, "{}", header_json).unwrap();
         writeln!(file, "{}", serde_json::to_string(&json!(["id1", {"a": 1}])).unwrap()).unwrap();
         writeln!(file, "{}", serde_json::to_string(&json!(["id2", {"a": 2}])).unwrap()).unwrap();
 
         let jstable = read_jstable(file.path().to_str().unwrap()).unwrap();
 
+        assert_eq!(jstable.timestamp, 12345);
         assert_eq!(jstable.schema.types, vec![SchemaType::Object]);
         assert_eq!(jstable.documents.len(), 2);
         assert_eq!(*jstable.documents.get("id1").unwrap(), json!({"a": 1}));
@@ -107,16 +136,53 @@ mod tests {
         let mut documents = BTreeMap::new();
         documents.insert("id1".to_string(), json!({"a": 1}));
         documents.insert("id2".to_string(), json!({"a": 2}));
-        let jstable = JSTable::new(schema, documents);
+        let jstable = JSTable::new(67890, schema, documents);
 
         let file = NamedTempFile::new().unwrap();
         jstable.write(file.path().to_str().unwrap()).unwrap();
 
         let content = std::fs::read_to_string(file.path()).unwrap();
-        assert!(content.contains("{\"type\":\"object\",\"properties\":{\"a\":{\"type\":[\"integer\"]}}}"));
+        let lines: Vec<&str> = content.lines().collect();
+        let header: JSTableHeader = serde_json::from_str(lines[0])?;
+        
+        assert_eq!(header.timestamp, 67890);
+        assert_eq!(header.schema.types, vec![SchemaType::Object]);
+
         assert!(content.contains("[\"id1\",{\"a\":1}]"));
         assert!(content.contains("[\"id2\",{\"a\":2}]"));
 
         Ok(())
     }
+
+    #[test]
+    fn test_merge_jstables_conflict_resolution() {
+        let schema = Schema::new(SchemaType::Object);
+        
+        let mut docs1 = BTreeMap::new();
+        docs1.insert("id1".to_string(), json!({"v": 1}));
+        let t1 = JSTable::new(100, schema.clone(), docs1);
+
+        let mut docs2 = BTreeMap::new();
+        docs2.insert("id1".to_string(), json!({"v": 2}));
+        let t2 = JSTable::new(200, schema.clone(), docs2);
+
+        // Case 1: t1 (older) then t2 (newer) in the slice
+        // Note: Creating array [t1, t2] moves them.
+        let merged = merge_jstables(&[t1, t2]);
+        assert_eq!(*merged.documents.get("id1").unwrap(), json!({"v": 2}));
+        assert_eq!(merged.timestamp, 200);
+
+        // Case 2: Reverse order
+        let mut docs1 = BTreeMap::new();
+        docs1.insert("id1".to_string(), json!({"v": 1}));
+        let t1b = JSTable::new(100, schema.clone(), docs1);
+
+        let mut docs2 = BTreeMap::new();
+        docs2.insert("id1".to_string(), json!({"v": 2}));
+        let t2b = JSTable::new(200, schema.clone(), docs2);
+
+        let merged_reverse = merge_jstables(&[t2b, t1b]);
+        assert_eq!(*merged_reverse.documents.get("id1").unwrap(), json!({"v": 2}));
+        assert_eq!(merged_reverse.timestamp, 200);
+    }
 }

src/storage.rs

@@ -25,7 +25,11 @@ impl MemTable {
     }
 
     pub fn flush(&self, path: &str) -> std::io::Result<()> {
-        let jstable = JSTable::new(self.schema.clone(), self.documents.clone());
+        let timestamp = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+        let jstable = JSTable::new(timestamp, self.schema.clone(), self.documents.clone());
         jstable.write(path)
     }