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/free_join/mod.rs

@@ -732,6 +732,12 @@ impl Database {
     pub(crate) fn plan_query(&mut self, query: Query) -> Plan {
         plan::plan_query(query)
     }
+
+    pub fn stabilize(&mut self) {
+        for (_, t) in self.tables.iter_mut() {
+            t.table.stabilize();
+        }
+    }
 }
 
 impl Drop for Database {

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,33 +35,93 @@ impl<'de> Deserialize<'de> for RowBuffer {
     where
         D: Deserializer<'de>,
     {
-        #[derive(Deserialize)]
-        struct Partial {
-            n_columns: usize,
-            total_rows: usize,
-            data: Vec<Cell<Value>>,
-        }
+        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")
+            }
 
-        let helper = Partial::deserialize(deserializer)?;
+            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),
+                })
+            }
+        }
 
-        Ok(RowBuffer {
-            n_columns: helper.n_columns,
-            total_rows: helper.total_rows,
-            data: Pooled::new(helper.data),
-        })
+        deserializer.deserialize_bytes(RowBufferVisitor)
     }
 }
 
+/// Serialize an u32
+/// The highest bit of each byte represents whether this is the last byte (0 = no; 1 = yes).
+/// The lower seven bits encodes seven bits from the number.
+/// This encoding uses fewer than 4 bytes if the number is small as shown in the following function.
+/// ```
+/// 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
+///     }
+/// }
+/// ```
+/// In practice, small number usually out-proportions large numbers, so this encoding saves space.
+
+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 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)
     }
 }