fix(compress): cherry-pick upstream #27 — read jsonb SPI datums as JsonB, not text

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pg_deltax"
-version = "0.2.0-dotcms.3"
+version = "0.2.0-dotcms.4"
 edition = "2024"
 resolver = "3"
 license = "Apache-2.0"

src/compress.rs

@@ -1384,19 +1384,20 @@ fn append_row_to_columns(
                 }
             }
             ColumnKind::Jsonb => {
-                // Classic compression path (post-INSERT). jsonb comes through
-                // SPI as canonical JSON text (via jsonb_out); we re-parse via
-                // jsonb_in to store the binary varlena representation, so the
-                // scan path can skip jsonb_in per row. rtabench's hot path is
-                // direct-backfill, not this one — the extra roundtrip is fine.
-                let text_opt = row
+                // Classic compression path (post-INSERT). SPI returns native
+                // jsonb Datums, so we read the on-disk binary varlena payload
+                // directly via `JsonbRaw` — no jsonb_out/serde_json/jsonb_in
+                // round-trip (which would be lossy for high-precision numbers
+                // and leak a jsonb_in datum per row). The bytes are the same
+                // canonical jsonb container the COPY path produces via
+                // `jsonb_text_to_binary`, so the scan path is unaffected.
+                let v = row
                     .get_datum_by_ordinal(ordinal)
                     .unwrap()
-                    .value::<String>()
+                    .value::<JsonbRaw>()
                     .unwrap();
-                let bytes_opt = text_opt.map(|t| unsafe { jsonb_text_to_binary(&t) });
                 if let TypedColumn::Bytes(vec) = &mut typed_cols[i] {
-                    vec.push(bytes_opt);
+                    vec.push(v.map(|j| j.0));
                 }
             }
         }
@@ -1413,6 +1414,55 @@ thread_local! {
         const { std::cell::Cell::new(std::ptr::null_mut()) };
 }
 
+/// A `jsonb` Datum read as its raw on-disk binary varlena payload (everything
+/// after the varlena header), with no parse/serialize round-trip.
+///
+/// Reading a `jsonb` column via pgrx's `JsonB` would route the value through
+/// `jsonb_out` → `serde_json::Value` → `jsonb_in`, which is both expensive and
+/// lossy: pgrx's `serde_json` has no `arbitrary_precision`, so numbers outside
+/// i64/u64/f64 range (high-precision decimals, large integers) get rounded.
+/// We instead detoast and copy the binary container verbatim — the same bytes
+/// `jsonb_text_to_binary` produces, so the scan path reconstructs identically.
+pub(crate) struct JsonbRaw(pub Vec<u8>);
+
+impl FromDatum for JsonbRaw {
+    unsafe fn from_polymorphic_datum(
+        datum: pgrx::pg_sys::Datum,
+        is_null: bool,
+        _typoid: pgrx::pg_sys::Oid,
+    ) -> Option<Self> {
+        if is_null {
+            return None;
+        }
+        unsafe {
+            let varlena = datum.cast_mut_ptr::<pgrx::pg_sys::varlena>();
+            let detoasted = pgrx::pg_sys::pg_detoast_datum(varlena);
+            let total_len = pgrx::varsize_any_exhdr(detoasted);
+            let data_ptr = pgrx::vardata_any(detoasted).cast::<u8>();
+            let bytes = std::slice::from_raw_parts(data_ptr, total_len).to_vec();
+            // pg_detoast_datum allocates a copy in CurrentMemoryContext only
+            // when the datum was actually toasted; free it so a long compress
+            // loop doesn't accumulate detoasted copies.
+            if detoasted != varlena {
+                pgrx::pg_sys::pfree(detoasted.cast());
+            }
+            Some(JsonbRaw(bytes))
+        }
+    }
+}
+
+impl IntoDatum for JsonbRaw {
+    fn into_datum(self) -> Option<pgrx::pg_sys::Datum> {
+        // Read-only helper: only the FromDatum side is used (via SpiHeapTupleDataEntry::value).
+        // Required by the IntoDatum bound on `value::<T>()` and the binary-coercibility check.
+        unreachable!("JsonbRaw is read-only and must not be converted back into a Datum")
+    }
+
+    fn type_oid() -> pgrx::pg_sys::Oid {
+        pgrx::pg_sys::JSONBOID
+    }
+}
+
 /// Convert canonical JSON text to the binary jsonb varlena payload
 /// (everything after the varlena header) by calling PG's `jsonb_in`.
 /// Caller stores the returned bytes verbatim; to reconstruct a Datum, wrap

src/partition.rs

@@ -635,7 +635,14 @@ pub fn auto_drop_partitions(client: &mut SpiClient, ht: &catalog::DeltatableInfo
 
     for (schema, name, is_compressed) in &partitions {
         if *is_compressed {
-            for suffix in ["blobs", "blooms", "text_lengths", "colstats", "meta"] {
+            for suffix in [
+                "blobs",
+                "blooms",
+                "text_lengths",
+                "valbitmap",
+                "colstats",
+                "meta",
+            ] {
                 let fqn = format!("\"_deltax_compressed\".\"{}_{}\"", name, suffix);
                 client
                     .update(&format!("DROP TABLE IF EXISTS {}", fqn), None, &[])