Remove the serialization size bottleneck

Cargo.toml

@@ -52,10 +52,11 @@ getrandom = "0.3"
 once_cell = "1.21"
 num-bigint = { version = "0.4", features = ["serde"] }
 num-rational = {version = "0.4", features = ["serde"]}
-csv = "1.3"
+csv = "1.4"
 typetag = "0.2"
 serde = { version = "1.0", features = ["derive", "rc"] }
 serde_json = "1.0"
+flexbuffers = "25.12.19"
 
 ######################
 # build dependencies
@@ -162,6 +163,7 @@ serde_json_diff = "0.2.0"
 anyhow.workspace = true
 walkdir = "2.5.0"
 egglog-reports = { workspace = true }
+flexbuffers.workspace = true
 
 [build-dependencies]
 chrono = { workspace = true, features = ["now"], optional = true }

core-relations/src/hash_index/mod.rs

@@ -915,6 +915,7 @@ static THREAD_POOL: Lazy<rayon::ThreadPool> = Lazy::new(|| {
 /// to the beginning of an unused vector.
 #[derive(Default, Clone, Serialize, Deserialize)]
 pub(super) struct FreeList {
+    #[serde(skip)]
     data: HashMap<usize, Vec<BufferIndex>>,
 }
 impl FreeList {

core-relations/src/row_buffer/mod.rs

@@ -6,7 +6,7 @@ use std::{cell::Cell, mem, ops::Deref};
 use crate::numeric_id::NumericId;
 use egglog_concurrency::ParallelVecWriter;
 use rayon::iter::ParallelIterator;
-use serde::{ser::SerializeStruct, Deserialize, Deserializer, Serialize};
+use serde::{Deserialize, Deserializer, Serialize};
 use smallvec::SmallVec;
 
 use crate::{
@@ -35,6 +35,7 @@ impl<'de> Deserialize<'de> for RowBuffer {
     where
         D: Deserializer<'de>,
     {
+        /*
         #[derive(Deserialize)]
         struct Partial {
             n_columns: usize,
@@ -49,19 +50,105 @@ impl<'de> Deserialize<'de> for RowBuffer {
             total_rows: helper.total_rows,
             data: Pooled::new(helper.data),
         })
+        */
+
+        struct RowBufferVisitor;
+
+        impl<'de> serde::de::Visitor<'de> for RowBufferVisitor {
+            type Value = RowBuffer;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+                formatter.write_str("Expecting a byte array")
+            }
+
+            fn visit_bytes<E>(self, bytes: &[u8]) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                let mut it = bytes.iter();
+                let n_columns = deserialize_compressed(&mut it);
+                let total_rows = deserialize_compressed(&mut it);
+                let mut data = <Vec<Cell<Value>>>::new();
+                for _i in 0..n_columns * total_rows {
+                    data.push(Cell::new(Value::new(deserialize_compressed(&mut it))));
+                }
+                Ok(RowBuffer {
+                    n_columns: n_columns.try_into().unwrap(),
+                    total_rows: total_rows.try_into().unwrap(),
+                    data: Pooled::new(data),
+                })
+            }
+        }
+
+        deserializer.deserialize_bytes(RowBufferVisitor)
     }
 }
 
+#[allow(dead_code)]
+fn get_n_compressed_bytes(x: u32) -> usize {
+    if x < (1u32 << 7) {
+        1
+    } else if x < (1u32 << 14) {
+        2
+    } else if x < (1u32 << 21) {
+        3
+    } else if x < (1u32 << 28) {
+        4
+    } else {
+        5
+    }
+}
+
+fn compressed_serialize(buf: &mut Vec<u8>, x: u32) {
+    let mut rem = x;
+    while rem >= (1u32 << 7) {
+        buf.push((rem & ((1u32 << 7) - 1)).try_into().unwrap());
+        rem = rem >> 7;
+    }
+    buf.push((rem | (1u32 << 7)).try_into().unwrap());
+}
+
+fn deserialize_compressed<'a, T: Iterator<Item = &'a u8>>(it: &mut T) -> u32 {
+    let mut ret = 0u32;
+    let mut delta = 0u32;
+    let mut val: u32 = <u8>::into(*it.next().unwrap());
+    while val < (1u32 << 7) {
+        ret = ret | (val << delta);
+        delta += 7;
+        val = <u8>::into(*it.next().unwrap());
+    }
+    let last = (val ^ (1u32 << 7)) << delta;
+    ret | last
+}
+
 impl Serialize for RowBuffer {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: serde::Serializer,
     {
+        /*
         let mut state = serializer.serialize_struct("RowBuffer", 3)?;
         state.serialize_field("n_columns", &self.n_columns)?;
         state.serialize_field("total_rows", &self.total_rows)?;
         state.serialize_field("data", &*self.data)?;
         state.end()
+        */
+        //let len = mem::size_of::<usize>() * 2 + self.n_columns * self.total_rows * mem::size_of::<u32>();
+        /*
+        let mut len = get_n_compressed_bytes(self.n_columns.try_into().unwrap()) + get_n_compressed_bytes(self.total_rows.try_into().unwrap());
+        for r in self.data.iter() {
+            len = len + get_n_compressed_bytes(r.get().rep);
+        }
+        let mut buf = vec![0u8; len];
+        //TODO: put data in
+        */
+        let mut buf = Vec::new();
+        compressed_serialize(&mut buf, self.n_columns.try_into().unwrap());
+        compressed_serialize(&mut buf, self.total_rows.try_into().unwrap());
+        for r in self.data.iter() {
+            compressed_serialize(&mut buf, r.get().rep);
+        }
+        serializer.serialize_bytes(&buf)
     }
 }