fix: non-text column reads from compressed partitions (text[], inet, numeric)

Cargo.toml

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

src/compress.rs

@@ -1509,6 +1509,38 @@ pub(crate) unsafe fn jsonb_text_to_binary(text: &str) -> Vec<u8> {
     }
 }
 
+/// Convert a binary jsonb varlena payload (everything after the varlena
+/// header, as produced by `jsonb_text_to_binary` / `JsonbRaw`) back to its
+/// canonical JSON text by wrapping it in a fresh varlena header and calling
+/// PG's jsonb output function. Exact inverse of `jsonb_text_to_binary`.
+/// Must only be called from the main backend thread.
+pub(crate) unsafe fn jsonb_binary_to_text(payload: &[u8]) -> String {
+    unsafe {
+        let total_len = pgrx::pg_sys::VARHDRSZ + payload.len();
+        let varlena = pgrx::pg_sys::palloc(total_len) as *mut pgrx::pg_sys::varlena;
+        pgrx::set_varsize_4b(varlena, total_len as i32);
+        std::ptr::copy_nonoverlapping(
+            payload.as_ptr(),
+            (varlena as *mut u8).add(pgrx::pg_sys::VARHDRSZ),
+            payload.len(),
+        );
+        let mut typoutput: pgrx::pg_sys::Oid = pgrx::pg_sys::InvalidOid;
+        let mut typisvarlena: bool = false;
+        pgrx::pg_sys::getTypeOutputInfo(pgrx::pg_sys::JSONBOID, &mut typoutput, &mut typisvarlena);
+        let cstr = pgrx::pg_sys::OidOutputFunctionCall(
+            typoutput,
+            pgrx::pg_sys::Datum::from(varlena as usize),
+        );
+        let text = std::ffi::CStr::from_ptr(cstr)
+            .to_str()
+            .expect("jsonb_out produced invalid UTF-8")
+            .to_string();
+        pgrx::pg_sys::pfree(cstr.cast());
+        pgrx::pg_sys::pfree(varlena.cast());
+        text
+    }
+}
+
 /// Sort typed columns in-place by the given order_by column indices.
 /// Computes a permutation from the sort keys, then reorders all columns by that permutation.
 pub(crate) fn sort_typed_columns(
@@ -3487,6 +3519,51 @@ fn decompress_column_values(blob: &[u8], data_type: &str) -> Vec<Option<String>>
     let total_count = cc.row_count as usize;
     let dt = data_type.to_lowercase();
 
+    // jsonb blobs store PG's BINARY jsonb varlena payload (not UTF-8 text —
+    // see `jsonb_text_to_binary` / upstream #27), so they must be decoded via
+    // the byte-level codec entry points and converted back to canonical JSON
+    // text through jsonb's output function. Routing them through the
+    // UTF-8-validating text decoders below panics with "invalid UTF-8 in
+    // dictionary" / "invalid UTF-8 in LZ4 data".
+    if dt == "jsonb" {
+        let non_null_count = count_non_null(&cc.null_bitmap, total_count);
+        let strings: Vec<String> = match cc.type_tag {
+            CompressionType::Dictionary => {
+                compression::dictionary::decode_to_byte_slices(&cc.data, non_null_count)
+                    .iter()
+                    .map(|b| unsafe { jsonb_binary_to_text(b) })
+                    .collect()
+            }
+            CompressionType::DictionaryLz4 => {
+                let normalized = compression::dictionary::normalize_lz4(&cc.data);
+                compression::dictionary::decode_to_byte_slices(&normalized, non_null_count)
+                    .iter()
+                    .map(|b| unsafe { jsonb_binary_to_text(b) })
+                    .collect()
+            }
+            CompressionType::Lz4 => {
+                let (buf, ranges) = compression::lz4::decode_to_ranges(&cc.data, non_null_count);
+                ranges
+                    .iter()
+                    .map(|&(off, len)| unsafe { jsonb_binary_to_text(&buf[off..off + len]) })
+                    .collect()
+            }
+            CompressionType::Lz4Blocked => {
+                let (buf, ranges) =
+                    compression::lz4::decode_to_ranges_blocked(&cc.data, non_null_count, None);
+                ranges
+                    .iter()
+                    .map(|&(off, len)| unsafe { jsonb_binary_to_text(&buf[off..off + len]) })
+                    .collect()
+            }
+            other => pgrx::error!(
+                "pg_deltax: unexpected compression type {:?} for jsonb column",
+                other
+            ),
+        };
+        return compression::reinsert_nulls(&strings, &cc.null_bitmap, total_count);
+    }
+
     match cc.type_tag {
         CompressionType::Gorilla => {
             if dt.contains("timestamp") || dt == "date" {

src/compression/dictionary.rs

@@ -37,11 +37,22 @@ pub fn parse_header(data: &[u8]) -> DictHeader<'_> {
     offset += 2;
 
     let mut dict: Vec<&str> = Vec::with_capacity(dict_size);
-    for _ in 0..dict_size {
+    for entry_idx in 0..dict_size {
         let str_len = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap()) as usize;
         offset += 4;
-        let s = std::str::from_utf8(&data[offset..offset + str_len])
-            .expect("invalid UTF-8 in dictionary");
+        let s = std::str::from_utf8(&data[offset..offset + str_len]).unwrap_or_else(|e| {
+            // Reaching this means a non-text payload (e.g. binary jsonb) was
+            // routed through the text dictionary decoder — an encoding-dispatch
+            // bug in the caller, which should use `parse_header_bytes` /
+            // `decode_to_byte_slices` instead. The panic unwinds into a regular
+            // PG ERROR (pgrx catches it at the FFI boundary); it never takes
+            // down the backend.
+            panic!(
+                "invalid UTF-8 in dictionary entry {} of {} (len {}): {}; \
+                 non-text payloads must be decoded via the byte-level dictionary API",
+                entry_idx, dict_size, str_len, e
+            )
+        });
         offset += str_len;
         dict.push(s);
     }

src/scan/exec/datum_utils.rs

@@ -413,7 +413,7 @@ pub(super) unsafe fn decompress_blob_to_datums_truncated(
 ///
 /// Instead of allocating a PG varlena datum for every row and then filtering,
 /// this matches the LIKE pattern against raw `&str` slices (zero-copy) and only
-/// calls `str_to_text_datum()` for rows that match. Non-matching rows get a
+/// builds real datums (via `str_slices_to_text_datums_arena`) for rows that match. Non-matching rows get a
 /// dummy datum that will never be read (the returned selection vector marks them
 /// as filtered out).
 ///
@@ -1419,8 +1419,6 @@ pub(super) unsafe fn decompress_jsonb_blob_with_selection(
     }
 }
 
-/// Create a text/varchar/bpchar datum from a Rust string.
-/// Allocates in the current memory context.
 /// Compare two strings using PG's collation-aware comparison.
 /// Returns negative if a < b, 0 if equal, positive if a > b.
 #[inline]
@@ -1436,50 +1434,47 @@ pub(super) unsafe fn collation_strcmp(a: &str, b: &str) -> i32 {
     }
 }
 
-pub(super) unsafe fn str_to_text_datum(
-    s: &str,
-    type_oid: pg_sys::Oid,
-    typmod: i32,
-) -> pg_sys::Datum {
-    unsafe {
-        // bpchar needs the type input function for padding; jsonb stores
-        // as canonical text and needs the input function to produce a real
-        // jsonb binary Datum (otherwise jsonb operators segfault).
-        if type_oid == pg_sys::BPCHAROID || type_oid == pg_sys::JSONBOID {
-            let cstr = std::ffi::CString::new(s).unwrap();
-            let mut typinput: pg_sys::Oid = pg_sys::InvalidOid;
-            let mut typioparam: pg_sys::Oid = pg_sys::InvalidOid;
-            pg_sys::getTypeInputInfo(type_oid, &mut typinput, &mut typioparam);
-            pg_sys::OidInputFunctionCall(typinput, cstr.as_ptr() as *mut _, typioparam, typmod)
-        } else {
-            // text/varchar: direct varlena construction (avoids type input function lookup)
-            let text = pg_sys::cstring_to_text_with_len(s.as_ptr() as *const _, s.len() as i32);
-            pg_sys::Datum::from(text as usize)
-        }
-    }
-}
-
 /// Allocate text/varchar datums from string slices using a single contiguous allocation.
 ///
 /// Instead of N individual palloc calls (one per string), this allocates one
 /// large block and packs all varlena headers + string data sequentially.
 /// This dramatically improves cache locality during the per-row emit loop.
 ///
-/// For bpchar, falls back to per-string allocation (needs type input function for padding).
+/// For anything that isn't text/varchar, falls back to per-string
+/// reconstruction through the type input function (bpchar padding, and
+/// non-text types stored via their text rendering: text[], inet, numeric, ...).
 pub(super) unsafe fn str_slices_to_text_datums_arena(
     slices: &[&str],
     type_oid: pg_sys::Oid,
     typmod: i32,
 ) -> Vec<pg_sys::Datum> {
-    // bpchar needs the input function for padding. jsonb should normally go
-    // through `byte_slices_to_jsonb_datums_arena` (the bytes are binary, not
-    // UTF-8); this branch is only a safety net for any caller that still
-    // hands us text.
-    if type_oid == pg_sys::BPCHAROID || type_oid == pg_sys::JSONBOID {
+    // Only text/varchar attributes may take the raw-varlena arena fast path.
+    // Any other type_oid means the stored strings are TEXT RENDERINGS of a
+    // different type — bpchar (padding), jsonb (safety net; normally goes
+    // through `byte_slices_to_jsonb_datums_arena`), and the classify_column
+    // fallthrough types (text[], inet, numeric, uuid, ...). Those must be
+    // reconstructed via the type input function so the resulting Datum
+    // matches the attribute's real binary representation — handing a raw
+    // text varlena to a slot whose attribute type is e.g. text[] makes PG
+    // read text bytes as an ArrayType header: garbage dims/pointers, silent
+    // NULLs or a backend crash. One getTypeInputInfo lookup, then one
+    // input-function call per value.
+    if !matches!(type_oid, pg_sys::TEXTOID | pg_sys::VARCHAROID) {
         return unsafe {
+            let mut typinput: pg_sys::Oid = pg_sys::InvalidOid;
+            let mut typioparam: pg_sys::Oid = pg_sys::InvalidOid;
+            pg_sys::getTypeInputInfo(type_oid, &mut typinput, &mut typioparam);
             slices
                 .iter()
-                .map(|s| str_to_text_datum(s, type_oid, typmod))
+                .map(|s| {
+                    let cstr = std::ffi::CString::new(*s).unwrap();
+                    pg_sys::OidInputFunctionCall(
+                        typinput,
+                        cstr.as_ptr() as *mut _,
+                        typioparam,
+                        typmod,
+                    )
+                })
                 .collect()
         };
     }

tests/correctness/test_codecs_extended.py

@@ -275,7 +275,6 @@ def test_direct_backfill_null_segment_by_regression(db):
     )
 
 
-@pytest.mark.xfail(strict=True, reason="numeric fallback columns currently decode with invalid numeric text")
 def test_numeric_fallback_type_regression(db):
     db.execute("SET pg_deltax.mock_now = '2025-01-20 12:00:00+00'")
     for table_name in ("numeric_fallback_plain", "numeric_fallback"):
@@ -333,7 +332,6 @@ def test_numeric_fallback_type_regression(db):
     )
 
 
-@pytest.mark.xfail(strict=True, reason="time fallback columns currently decode to invalid time values")
 def test_time_fallback_type_regression(db):
     db.execute("SET pg_deltax.mock_now = '2025-01-20 12:00:00+00'")
     for table_name in ("time_fallback_plain", "time_fallback"):
@@ -385,7 +383,6 @@ def test_time_fallback_type_regression(db):
     )
 
 
-@pytest.mark.xfail(strict=True, reason="uuid fallback columns currently decode from raw text bytes")
 def test_uuid_fallback_type_regression(db):
     db.execute("SET pg_deltax.mock_now = '2025-01-20 12:00:00+00'")
     for table_name in ("uuid_fallback_plain", "uuid_fallback"):
@@ -437,7 +434,6 @@ def test_uuid_fallback_type_regression(db):
     )
 
 
-@pytest.mark.xfail(strict=True, reason="bytea fallback columns currently decode escaped text bytes")
 def test_bytea_fallback_type_regression(db):
     db.execute("SET pg_deltax.mock_now = '2025-01-20 12:00:00+00'")
     for table_name in ("bytea_fallback_plain", "bytea_fallback"):