From 82f4ff0a11d8656448bde5bee516a04829641388 Mon Sep 17 00:00:00 2001 From: agentxagi Date: Fri, 12 Jun 2026 15:16:48 -0300 Subject: [PATCH 1/4] fix(compress): read jsonb SPI datums as JsonB, not text (#27) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(compress): read jsonb SPI datums as JsonB, not text Post-INSERT compression via SPI returns native jsonb Datums. The classic path assumed jsonb_out text and failed with DatumError(IncompatibleTypes). Read JsonB and extract the binary varlena payload directly; keep text fallback for COPY-style callers. ValorBrain usage_transactions.reason_metadata compresses at 3.7–4.1x. * replace jsonb/serde roundtrip --------- Co-authored-by: Reviewer Agent Co-authored-by: Tudor Golubenco --- src/compress.rs | 68 ++++++++++++++++++++++++---- tests/test_compression.py | 94 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 153 insertions(+), 9 deletions(-) diff --git a/src/compress.rs b/src/compress.rs index 81c4b44..01892c9 100644 --- a/src/compress.rs +++ b/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::() + .value::() .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); + +impl FromDatum for JsonbRaw { + unsafe fn from_polymorphic_datum( + datum: pgrx::pg_sys::Datum, + is_null: bool, + _typoid: pgrx::pg_sys::Oid, + ) -> Option { + if is_null { + return None; + } + unsafe { + let varlena = datum.cast_mut_ptr::(); + 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::(); + 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 { + // Read-only helper: only the FromDatum side is used (via SpiHeapTupleDataEntry::value). + // Required by the IntoDatum bound on `value::()` 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 diff --git a/tests/test_compression.py b/tests/test_compression.py index 300f3c9..e6770d7 100644 --- a/tests/test_compression.py +++ b/tests/test_compression.py @@ -4500,3 +4500,97 @@ def test_no_eligible_partitions_returns_empty(self, db): assert results == [], ( f"expected zero eligible partitions, got {results}" ) + + +class TestJsonbCompressionFidelity: + """The post-INSERT compression path reads jsonb back via SPI. It must store + the jsonb container verbatim — NOT round-trip through serde_json, which + would silently round numbers outside i64/u64/f64 range. + + Comparisons use server-side `::text` (jsonb_out is full-precision) rather + than letting psycopg parse jsonb into Python floats, which would itself lose + precision and mask a regression. + """ + + def _setup(self, db, table="jfidelity"): + db.execute(f"SET pg_deltax.mock_now = '{MOCK_NOW}'") + db.execute(f""" + CREATE TABLE {table} ( + ts TIMESTAMPTZ NOT NULL, + device_id TEXT NOT NULL, + data JSONB + ) + """) + db.execute( + f"SELECT deltax.deltax_create_table('{table}', 'ts', '1 day'::interval)" + ) + db.commit() + + def test_high_precision_numbers_survive_compression(self, db): + table = "jfidelity" + self._setup(db, table) + + # Values chosen to break a serde_json (no arbitrary_precision) round-trip: + # - a 29-significant-digit decimal (f64 keeps ~17) + # - a 30-digit integer, well beyond u64::MAX (~1.8e19) + # - a plain nested object as a control + payloads = [ + '{"amount": 0.12345678901234567890123456789}', + '{"big": 123456789012345678901234567890}', + '{"nested": {"k": 1, "arr": [1, 2, 3]}, "s": "hello"}', + ] + for i, p in enumerate(payloads): + db.execute( + f"INSERT INTO {table} VALUES (" + f"'{BASE_TS}'::timestamptz + interval '{i} minutes', " + f"'dev-{i}', '{p}'::jsonb)" + ) + db.commit() + + # Canonical, full-precision text straight from PG, before compression. + before = db.execute( + f"SELECT data::text FROM {table} ORDER BY ts" + ).fetchall() + assert len(before) == len(payloads) + + db.execute( + f"SELECT deltax.deltax_enable_compression('{table}', " + f"segment_by => ARRAY['device_id'], order_by => ARRAY['ts'])" + ) + db.commit() + _compress_all_partitions(db, table) + + after = db.execute( + f"SELECT data::text FROM {table} ORDER BY ts" + ).fetchall() + + assert after == before, ( + "jsonb changed across compression — the SPI compress path is " + f"corrupting values.\nbefore={before}\nafter={after}" + ) + + def test_null_jsonb_survives_compression(self, db): + table = "jfidelity_null" + self._setup(db, table) + db.execute( + f"INSERT INTO {table} VALUES " + f"('{BASE_TS}'::timestamptz, 'dev-0', NULL), " + f"('{BASE_TS}'::timestamptz + interval '1 minute', 'dev-0', '{{\"a\": 1}}'::jsonb)" + ) + db.commit() + before = db.execute( + f"SELECT data::text FROM {table} ORDER BY ts" + ).fetchall() + + db.execute( + f"SELECT deltax.deltax_enable_compression('{table}', " + f"segment_by => ARRAY['device_id'], order_by => ARRAY['ts'])" + ) + db.commit() + _compress_all_partitions(db, table) + + after = db.execute( + f"SELECT data::text FROM {table} ORDER BY ts" + ).fetchall() + assert after == before + assert before[0][0] is None # NULL preserved as NULL, not '{}' From 8010fcd9dff32b04625e19de2605e434b4f90ae7 Mon Sep 17 00:00:00 2001 From: Will Ezell Date: Tue, 14 Jul 2026 17:45:46 -0400 Subject: [PATCH 2/4] chore: version 0.2.0-dotcms.4 (jsonb compression fix release) --- Cargo.lock | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6dddd21..fbe93b4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1214,7 +1214,7 @@ dependencies = [ [[package]] name = "pg_deltax" -version = "0.2.0-dotcms.2" +version = "0.2.0-dotcms.4" dependencies = [ "ahash", "cardinality-estimator", diff --git a/Cargo.toml b/Cargo.toml index 8282f4c..ec1c032 100644 --- a/Cargo.toml +++ b/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" From 6b465d8d4b6cb92c5c4df63ef78c81af753f95db Mon Sep 17 00:00:00 2001 From: Alexis Rico Date: Mon, 15 Jun 2026 15:45:23 +0200 Subject: [PATCH 3/4] Fix pre-existing valbitmap correctness bugs (#42) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: valbitmap correctness — fail-safe pruning, retention leak, cross-schema companion lookup Three pre-existing correctness bugs found during review of #31/#34/#35: 1. Text valbitmap wrong-empty-results on segment overflow. A segment with more than VALBITMAP_MAX_DISTINCT (32) distinct text values writes no valbitmap row and contributes nothing to the partition-level column_valmap. When the union of the remaining segments stayed <= 32, querying a value unique to the overflowed segment missed the valmap and took the 'prune every segment without reading the bitmap table' shortcut — returning zero rows instead of the overflowed segment's rows. Pruning now always goes through the per-segment bitmap rows: a valmap miss (empty wanted_bits) prunes exactly the segments that wrote a bitmap row; segments without one are never prunable by valmap logic. The now-redundant ValbitmapCheck::prune_all field is removed. 2. Retention drop leaked _valbitmap companions. auto_drop_partitions dropped blobs/blooms/text_lengths/colstats/meta but not valbitmap, leaving orphaned tables in _deltax_compressed forever. (decompress and the empty-partition cleanup already covered all six.) 3. check_compressed_partition matched companions by partition NAME only. Companion tables live in the single shared _deltax_compressed schema and embed only the table name, so a same-named partition in another schema (or any same-named plain table) was treated as compressed and served the other partition's data. The lookup now confirms via deltax.deltax_partition that the exact (schema, table) pair is the compressed one — at most one partition of a given name can be compressed, so is_compressed disambiguates. Regression tests: tests/test_valbitmap.py (one-segment-overflow shape), tests/test_worker.py (no orphaned companions after retention drop), tests/test_compression.py (two-schema same-name partition). * test: make overflow regression probe robust to minmax/dictionary segment skipping The plan-shape assertions (vb_skipped=1, segments=1) now ride on the present-value 'ov07' query — where only the valbitmap can skip the bitmap-covered segment and the overflowed segment must be scanned. The absent-value probe keeps only the row-count assertion: an absent constant can legitimately eliminate the overflowed segment through exact per-segment evidence (text minmax, dictionary value check) regardless of the valbitmap. * rustfmt: format worker.rs launcher fan-out (inherited unformatted from #38 merge) (cherry picked from commit 1649c98695610b5d1758dcab1ad44c6dc9846ffa) --- src/partition.rs | 9 +- src/scan/exec/segments.rs | 314 ++++++++++++++++++-------------------- src/scan/hook.rs | 37 ++++- tests/test_compression.py | 90 +++++++++++ tests/test_valbitmap.py | 115 ++++++++++++++ tests/test_worker.py | 107 +++++++++++++ 6 files changed, 505 insertions(+), 167 deletions(-) diff --git a/src/partition.rs b/src/partition.rs index 36957aa..75872db 100644 --- a/src/partition.rs +++ b/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, &[]) diff --git a/src/scan/exec/segments.rs b/src/scan/exec/segments.rs index fd7349e..6e77bde 100644 --- a/src/scan/exec/segments.rs +++ b/src/scan/exec/segments.rs @@ -1879,13 +1879,15 @@ pub(super) unsafe fn load_segments_heap( // Build valbitmap checks from batch quals (text Eq on low-card columns // whose partition-level value list is in `column_valmap`). Each check // carries the bit indices the segment must contain at least one of. - // `prune_all = true` means the queried constant doesn't appear in - // ANY segment of this partition — every segment can be skipped without - // even reading the bitmap table. + // An empty `wanted_bits` means the queried constant is absent from + // the partition-level valmap — every segment that wrote a bitmap row + // fails the check and gets pruned. Segments WITHOUT a bitmap row + // (they overflowed VALBITMAP_MAX_DISTINCT at compress time, so the + // valmap doesn't cover their values) must never be pruned: the + // constant may well live only in them. struct ValbitmapCheck { col_idx: u16, wanted_bits: Vec, - prune_all: bool, } let mut bloom_checks: Vec = Vec::new(); let mut valbitmap_checks: Vec = Vec::new(); @@ -1948,25 +1950,20 @@ pub(super) unsafe fn load_segments_heap( && let Some(ref needle) = bq.text_const && let Some(values) = valmap.get(col_name) { - let bit = values.iter().position(|v| v == needle); - match bit { - Some(idx) => { - valbitmap_checks.push(ValbitmapCheck { - col_idx: ci, - wanted_bits: vec![idx as u8], - prune_all: false, - }); - } - None => { - // Constant never appeared at compress time → no segment - // can match. Mark the column for "prune everything". - valbitmap_checks.push(ValbitmapCheck { - col_idx: ci, - wanted_bits: vec![], - prune_all: true, - }); - } - } + // A valmap miss leaves `wanted_bits` empty: no segment that + // contributed to the valmap can match. Segments that + // overflowed the per-segment distinct cap contributed + // nothing (and have no bitmap row), so presence is still + // re-checked per segment below. + let wanted_bits: Vec = values + .iter() + .position(|v| v == needle) + .map(|idx| vec![idx as u8]) + .unwrap_or_default(); + valbitmap_checks.push(ValbitmapCheck { + col_idx: ci, + wanted_bits, + }); } } let mut segments_bloom_skipped: u64 = 0; @@ -3041,172 +3038,159 @@ pub(super) unsafe fn load_segments_heap( // PK on `(_col_idx, _segment_id)`, fetch `_bits`, test the bit // recorded in `valmap` for the queried constant. Exact (no false // positives), so a clear bit guarantees the segment can be skipped. + // + // Presence is decided strictly per segment: only segments that wrote + // a bitmap row for the column are candidates for pruning. A segment + // with no row exceeded VALBITMAP_MAX_DISTINCT at compress time — its + // values are not covered by the partition valmap, so neither a clear + // bit nor a valmap miss (empty `wanted_bits`) says anything about it. // ---------------------------------------------------------------- if !valbitmap_checks.is_empty() && !segments.is_empty() { - // First handle "constant absent from partition entirely" — no - // need to even open the bitmap table for those. - if valbitmap_checks.iter().any(|c| c.prune_all) { - let pruned = segments.len(); - segments.clear(); - surviving_segment_ids.clear(); - segments_skipped += pruned as u64; - segments_valbitmap_skipped += pruned as u64; - } else { - let meta_name_ptr = pg_sys::get_rel_name(meta_oid); - let meta_name_str = std::ffi::CStr::from_ptr(meta_name_ptr) - .to_string_lossy() - .into_owned(); - let meta_ns_oid = pg_sys::get_rel_namespace(meta_oid); - let partition_name = meta_name_str - .strip_suffix("_meta") - .unwrap_or(&meta_name_str); - let valbitmap_name = format!("{}_valbitmap", partition_name); - let valbitmap_cname = std::ffi::CString::new(valbitmap_name).unwrap(); - let valbitmap_oid = - pg_sys::get_relname_relid(valbitmap_cname.as_ptr(), meta_ns_oid); - - if valbitmap_oid != pg_sys::InvalidOid { - let mut seg_id_to_idx: HashMap = HashMap::new(); - for (idx, &sid) in surviving_segment_ids.iter().enumerate() { - seg_id_to_idx.insert(sid, idx); + let meta_name_ptr = pg_sys::get_rel_name(meta_oid); + let meta_name_str = std::ffi::CStr::from_ptr(meta_name_ptr) + .to_string_lossy() + .into_owned(); + let meta_ns_oid = pg_sys::get_rel_namespace(meta_oid); + let partition_name = meta_name_str + .strip_suffix("_meta") + .unwrap_or(&meta_name_str); + let valbitmap_name = format!("{}_valbitmap", partition_name); + let valbitmap_cname = std::ffi::CString::new(valbitmap_name).unwrap(); + let valbitmap_oid = pg_sys::get_relname_relid(valbitmap_cname.as_ptr(), meta_ns_oid); + + if valbitmap_oid != pg_sys::InvalidOid { + let mut seg_id_to_idx: HashMap = HashMap::new(); + for (idx, &sid) in surviving_segment_ids.iter().enumerate() { + seg_id_to_idx.insert(sid, idx); + } + + let vb_rel = + pg_sys::table_open(valbitmap_oid, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + let vb_tupdesc = (*vb_rel).rd_att; + let vb_natts = (*vb_tupdesc).natts as usize; + + let mut seg_id_att: Option = None; + let mut bits_att: Option = None; + for i in 0..vb_natts { + let attr = &*tupdesc_get_attr(vb_tupdesc, i); + let name = + std::ffi::CStr::from_ptr(attr.attname.data.as_ptr()).to_string_lossy(); + if name == "_segment_id" { + seg_id_att = Some(i); + } else if name == "_bits" { + bits_att = Some(i); } + } + + let pk_index_oid = primary_key_index_oid(vb_rel); - let vb_rel = pg_sys::table_open( - valbitmap_oid, + if let (Some(sid_att), Some(bits_a), true) = + (seg_id_att, bits_att, pk_index_oid != pg_sys::InvalidOid) + { + let snapshot = pg_sys::GetActiveSnapshot(); + let idx_rel = pg_sys::index_open( + pk_index_oid, pg_sys::AccessShareLock as pg_sys::LOCKMODE, ); - let vb_tupdesc = (*vb_rel).rd_att; - let vb_natts = (*vb_tupdesc).natts as usize; - - let mut seg_id_att: Option = None; - let mut bits_att: Option = None; - for i in 0..vb_natts { - let attr = &*tupdesc_get_attr(vb_tupdesc, i); - let name = - std::ffi::CStr::from_ptr(attr.attname.data.as_ptr()).to_string_lossy(); - if name == "_segment_id" { - seg_id_att = Some(i); - } else if name == "_bits" { - bits_att = Some(i); - } - } + let mut vb_pruned_ids: std::collections::HashSet = + std::collections::HashSet::new(); - let pk_index_oid = primary_key_index_oid(vb_rel); + // Valmap-miss checks carry an empty `wanted_bits`, + // so every segment with a bitmap row fails `passes` + // below and is pruned; segments without a row + // survive. + for vc in &valbitmap_checks { + let mut skey = [pg_sys::ScanKeyData::default()]; + pg_sys::ScanKeyInit( + &mut skey[0], + 1, // attnum 1 = _col_idx + pg_sys::BTEqualStrategyNumber as u16, + pg_sys::F_INT2EQ.into(), + pg_sys::Datum::from(vc.col_idx as i16), + ); - if let (Some(sid_att), Some(bits_a), true) = - (seg_id_att, bits_att, pk_index_oid != pg_sys::InvalidOid) - { - let snapshot = pg_sys::GetActiveSnapshot(); - let idx_rel = pg_sys::index_open( - pk_index_oid, - pg_sys::AccessShareLock as pg_sys::LOCKMODE, + #[cfg(feature = "pg17")] + let scan = pg_sys::index_beginscan(vb_rel, idx_rel, snapshot, 1, 0); + #[cfg(feature = "pg18")] + let scan = pg_sys::index_beginscan( + vb_rel, + idx_rel, + snapshot, + std::ptr::null_mut(), + 1, + 0, ); - let mut vb_pruned_ids: std::collections::HashSet = - std::collections::HashSet::new(); + pg_sys::index_rescan(scan, skey.as_mut_ptr(), 1, std::ptr::null_mut(), 0); - for vc in &valbitmap_checks { - if vc.prune_all { - continue; - } - let mut skey = [pg_sys::ScanKeyData::default()]; - pg_sys::ScanKeyInit( - &mut skey[0], - 1, // attnum 1 = _col_idx - pg_sys::BTEqualStrategyNumber as u16, - pg_sys::F_INT2EQ.into(), - pg_sys::Datum::from(vc.col_idx as i16), - ); + let slot = pg_sys::table_slot_create(vb_rel, std::ptr::null_mut()); - #[cfg(feature = "pg17")] - let scan = pg_sys::index_beginscan(vb_rel, idx_rel, snapshot, 1, 0); - #[cfg(feature = "pg18")] - let scan = pg_sys::index_beginscan( - vb_rel, - idx_rel, - snapshot, - std::ptr::null_mut(), - 1, - 0, - ); - pg_sys::index_rescan( + loop { + if !pg_sys::index_getnext_slot( scan, - skey.as_mut_ptr(), - 1, - std::ptr::null_mut(), - 0, - ); - - let slot = pg_sys::table_slot_create(vb_rel, std::ptr::null_mut()); + pg_sys::ScanDirection::ForwardScanDirection, + slot, + ) { + break; + } + pg_sys::slot_getallattrs(slot); + let tts_values = (*slot).tts_values; + let tts_isnull = (*slot).tts_isnull; + if *tts_isnull.add(sid_att) || *tts_isnull.add(bits_a) { + continue; + } + let seg_id = (*tts_values.add(sid_att)).value() as i32; + if !seg_id_to_idx.contains_key(&seg_id) { + continue; + } - loop { - if !pg_sys::index_getnext_slot( - scan, - pg_sys::ScanDirection::ForwardScanDirection, - slot, - ) { - break; - } - pg_sys::slot_getallattrs(slot); - let tts_values = (*slot).tts_values; - let tts_isnull = (*slot).tts_isnull; - if *tts_isnull.add(sid_att) || *tts_isnull.add(bits_a) { - continue; - } - let seg_id = (*tts_values.add(sid_att)).value() as i32; - if !seg_id_to_idx.contains_key(&seg_id) { - continue; - } + let varlena_ptr = + (*tts_values.add(bits_a)).cast_mut_ptr::(); + let detoasted = pg_sys::pg_detoast_datum(varlena_ptr); + let data_ptr = pgrx::vardata_any(detoasted); + let data_len = pgrx::varsize_any_exhdr(detoasted); + #[allow(clippy::unnecessary_cast)] + let bits = std::slice::from_raw_parts(data_ptr as *const u8, data_len); - let varlena_ptr = - (*tts_values.add(bits_a)).cast_mut_ptr::(); - let detoasted = pg_sys::pg_detoast_datum(varlena_ptr); - let data_ptr = pgrx::vardata_any(detoasted); - let data_len = pgrx::varsize_any_exhdr(detoasted); - #[allow(clippy::unnecessary_cast)] - let bits = - std::slice::from_raw_parts(data_ptr as *const u8, data_len); - - // A segment passes if any wanted bit is set. - let passes = vc.wanted_bits.iter().any(|&bi| { - let byte = (bi / 8) as usize; - let mask = 1u8 << (bi % 8); - byte < bits.len() && (bits[byte] & mask) != 0 - }); - - if detoasted != varlena_ptr { - pg_sys::pfree(detoasted as *mut _); - } + // A segment passes if any wanted bit is set. + let passes = vc.wanted_bits.iter().any(|&bi| { + let byte = (bi / 8) as usize; + let mask = 1u8 << (bi % 8); + byte < bits.len() && (bits[byte] & mask) != 0 + }); - if !passes { - vb_pruned_ids.insert(seg_id); - } + if detoasted != varlena_ptr { + pg_sys::pfree(detoasted as *mut _); } - pg_sys::ExecDropSingleTupleTableSlot(slot); - pg_sys::index_endscan(scan); + if !passes { + vb_pruned_ids.insert(seg_id); + } } - pg_sys::index_close(idx_rel, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + pg_sys::ExecDropSingleTupleTableSlot(slot); + pg_sys::index_endscan(scan); + } - if !vb_pruned_ids.is_empty() { - let before = segments.len(); - let mut i = 0; - while i < segments.len() { - if vb_pruned_ids.contains(&surviving_segment_ids[i]) { - segments.swap_remove(i); - surviving_segment_ids.swap_remove(i); - } else { - i += 1; - } + pg_sys::index_close(idx_rel, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + + if !vb_pruned_ids.is_empty() { + let before = segments.len(); + let mut i = 0; + while i < segments.len() { + if vb_pruned_ids.contains(&surviving_segment_ids[i]) { + segments.swap_remove(i); + surviving_segment_ids.swap_remove(i); + } else { + i += 1; } - let pruned = before - segments.len(); - segments_skipped += pruned as u64; - segments_valbitmap_skipped += pruned as u64; } + let pruned = before - segments.len(); + segments_skipped += pruned as u64; + segments_valbitmap_skipped += pruned as u64; } - - pg_sys::table_close(vb_rel, pg_sys::AccessShareLock as pg_sys::LOCKMODE); } + + pg_sys::table_close(vb_rel, pg_sys::AccessShareLock as pg_sys::LOCKMODE); } } diff --git a/src/scan/hook.rs b/src/scan/hook.rs index bec5203..fe2b3a5 100644 --- a/src/scan/hook.rs +++ b/src/scan/hook.rs @@ -4551,7 +4551,42 @@ pub(crate) unsafe fn check_compressed_partition(rel_oid: pg_sys::Oid) -> pg_sys: // Check if _deltax_compressed._meta exists let meta_name = format!("{}_meta", rel_name); let companion_cname = std::ffi::CString::new(meta_name).unwrap(); - pg_sys::get_relname_relid(companion_cname.as_ptr(), compressed_ns_oid) + let meta_oid = pg_sys::get_relname_relid(companion_cname.as_ptr(), compressed_ns_oid); + if meta_oid == pg_sys::InvalidOid { + return pg_sys::InvalidOid; + } + + // Companion names embed only the table name and `_deltax_compressed` + // is a single shared namespace, so a same-named partition in ANOTHER + // schema finds this meta table too. Confirm via the catalog that THIS + // (schema-qualified) partition is the compressed one. At most one + // partition of a given name can be compressed — companion creation + // would collide otherwise — so `is_compressed` on the exact + // (schema, name) pair disambiguates. + let ns_name_ptr = pg_sys::get_namespace_name(rel_ns_oid); + if ns_name_ptr.is_null() { + return pg_sys::InvalidOid; + } + let rel_schema = std::ffi::CStr::from_ptr(ns_name_ptr) + .to_string_lossy() + .into_owned(); + let confirmed = Spi::connect(|client| { + client + .select( + "SELECT 1 FROM deltax.deltax_partition \ + WHERE schema_name = $1 AND table_name = $2 AND is_compressed", + Some(1), + &[rel_schema.into(), rel_name.into()], + ) + .ok() + .and_then(|table| table.first().get_one::().ok().flatten()) + .is_some() + }); + if confirmed { + meta_oid + } else { + pg_sys::InvalidOid + } } } diff --git a/tests/test_compression.py b/tests/test_compression.py index e6770d7..b2d280b 100644 --- a/tests/test_compression.py +++ b/tests/test_compression.py @@ -4594,3 +4594,93 @@ def test_null_jsonb_survives_compression(self, db): ).fetchall() assert after == before assert before[0][0] is None # NULL preserved as NULL, not '{}' + + +# --------------------------------------------------------------------------- +# Cross-schema companion disambiguation +# --------------------------------------------------------------------------- + +def test_same_named_partition_in_other_schema_not_hijacked(db): + """Two deltax tables with the same name in different schemas produce + identically-named partitions. Companion tables live in the single shared + `_deltax_compressed` namespace and embed only the partition NAME, so the + scan hook's companion lookup must verify via the catalog that the + (schema-qualified) partition it is planning is actually the compressed + one. Regression: compressing s1's partition made the hook treat s2's + same-named, UNCOMPRESSED partition as compressed and serve s1's data.""" + db.execute(f"SET pg_deltax.mock_now = '{MOCK_NOW}'") + for s in ("s1", "s2"): + db.execute(f"CREATE SCHEMA {s}") + db.execute(f""" + CREATE TABLE {s}.events ( + ts TIMESTAMPTZ NOT NULL, + val INT NOT NULL + ) + """) + db.execute( + f"SELECT deltax.deltax_create_table('{s}.events', 'ts', " + "'1 day'::interval)" + ) + db.commit() + + # Distinct payloads per schema so any cross-wiring is visible. + for s, base in (("s1", 1000), ("s2", 2000)): + values = ", ".join( + f"('{BASE_TS}'::timestamptz + interval '{i} minutes', {base + i})" + for i in range(50) + ) + db.execute(f"INSERT INTO {s}.events (ts, val) VALUES {values}") + db.commit() + + # Sanity: the partition names collide across the two schemas. + parts1 = { + r[0] + for r in db.execute( + "SELECT partition_name FROM deltax.deltax_partition_info('s1.events')" + ).fetchall() + } + parts2 = { + r[0] + for r in db.execute( + "SELECT partition_name FROM deltax.deltax_partition_info('s2.events')" + ).fetchall() + } + assert parts1 == parts2, f"expected colliding names, got {parts1} vs {parts2}" + + # Compress only s1's data-bearing partition. + db.execute( + "SELECT deltax.deltax_enable_compression('s1.events', " + "order_by => ARRAY['ts'])" + ) + db.commit() + target = None + for p in sorted(parts1): + if "default" in p: + continue + if db.execute(f'SELECT count(*) FROM s1."{p}"').fetchone()[0]: + target = p + break + assert target is not None + db.execute(f"SELECT deltax.deltax_compress_partition('s1.{target}')") + db.commit() + + # s1 reads its data through the companion. + vals1 = [ + r[0] + for r in db.execute("SELECT val FROM s1.events ORDER BY val").fetchall() + ] + assert vals1 == list(range(1000, 1050)) + + # s2 is NOT compressed — it must read its own heap, not s1's companion. + vals2 = [ + r[0] + for r in db.execute("SELECT val FROM s2.events ORDER BY val").fetchall() + ] + assert vals2 == list(range(2000, 2050)), ( + f"s2.events returned wrong rows — its partition was hijacked by " + f"s1's same-named compressed partition: {vals2[:5]}..." + ) + + # Scanning s2's partition directly must hit its heap too. + n = db.execute(f'SELECT count(*) FROM s2."{target}"').fetchone()[0] + assert n == 50, f"direct scan of s2 partition returned {n} rows" diff --git a/tests/test_valbitmap.py b/tests/test_valbitmap.py index b5a0011..7eba7c5 100644 --- a/tests/test_valbitmap.py +++ b/tests/test_valbitmap.py @@ -270,6 +270,121 @@ def test_direct_backfill_populates_valbitmap(self, db): f"expected vb_skipped > 0, got {match.group(1)}" ) + def test_overflowed_segment_survives_valmap_miss(self, db): + """Regression: a segment that exceeds VALBITMAP_MAX_DISTINCT (32) + writes no valbitmap row and contributes nothing to the partition + valmap. Querying a value that lives ONLY in that segment used to hit + the "constant absent from the valmap → prune every segment" shortcut + and return zero rows. The valmap-miss path may only skip segments + that wrote a bitmap row — never the overflowed one.""" + db.execute("DROP TABLE IF EXISTS evt_ov CASCADE") + db.execute(f"SET pg_deltax.mock_now = '{MOCK_NOW}'") + db.execute(""" + CREATE TABLE evt_ov ( + ts timestamptz NOT NULL, + order_id integer NOT NULL, + event_type text NOT NULL + ) + """) + db.execute( + "SELECT deltax.deltax_create_table('evt_ov', 'ts', '1 day'::interval)" + ) + db.commit() + + segment_size = 200 + rows = [] + # Segment 0 (order_id 0..199): 40 distinct values 'ov00'..'ov39' — + # exceeds the 32-distinct cap, so this segment gets NO bitmap row + # and its values are missing from the partition valmap. 'ov07' + # exists only here (order_id 7, 47, 87, 127, 167). + for i in range(segment_size): + ts = f"'{BASE_TS}'::timestamptz + interval '{i} seconds'" + rows.append(f"({ts}, {i}, 'ov{i % 40:02d}')") + # Segment 1 (order_id 200..399): 2 distinct values → bitmap row + # written; the partition valmap ends up ['aaa', 'zzz'] only. The + # values straddle the 'ov*' band so segment 1's [min,max] range + # covers the probe constants below — text minmax pruning stays out + # of the picture and only the valbitmap decides. + for i in range(segment_size): + order_id = segment_size + i + ts = f"'{BASE_TS}'::timestamptz + interval '{order_id} seconds'" + et = "aaa" if i % 2 == 0 else "zzz" + rows.append(f"({ts}, {order_id}, '{et}')") + db.execute( + "INSERT INTO evt_ov (ts, order_id, event_type) VALUES " + + ", ".join(rows) + ) + db.commit() + + db.execute( + "SELECT deltax.deltax_enable_compression('evt_ov', " + f"order_by => ARRAY['order_id'], segment_size => {segment_size})" + ) + db.commit() + parts = db.execute( + "SELECT partition_name FROM deltax.deltax_partition_info('evt_ov') " + "WHERE partition_name NOT LIKE '%default%'" + ).fetchall() + for (part_name,) in parts: + n = db.execute(f'SELECT count(*) FROM "{part_name}"').fetchone()[0] + if n: + db.execute( + f"SELECT deltax.deltax_compress_partition('{part_name}')" + ) + db.commit() + + # Sanity: the partition valmap must NOT contain the overflowed + # segment's values — that's exactly the shape that triggered the bug. + valmap = db.execute(""" + SELECT column_valmap::text + FROM deltax.deltax_partition + WHERE table_name LIKE 'evt_ov_%' AND is_compressed = true + """).fetchone()[0] + assert valmap is not None and '"aaa"' in valmap + assert '"ov07"' not in valmap, ( + f"expected overflowed segment's values absent from valmap, " + f"got: {valmap}" + ) + + # The rows in the overflowed segment must come back. Plan shape: + # the bitmap-covered segment is skipped via its bitmap row + # (vb_skipped=1; 'ov07' misses the ['aaa','zzz'] valmap and segment + # 1's [min,max] covers 'ov07', so only the valbitmap can skip it) + # while the overflowed segment — with NO bitmap row — must be + # decompressed and batch-filtered. Pre-fix, the valmap miss pruned + # BOTH segments without reading the bitmap table → zero rows. + decomp, vb_skipped = _explain_skip_counts( + db, + "SELECT * FROM evt_ov WHERE event_type = 'ov07' LIMIT 1000", + ) + assert vb_skipped == 1, ( + f"expected exactly the bitmap-covered segment skipped, " + f"got vb_skipped={vb_skipped}" + ) + assert decomp == 1, ( + f"expected the overflowed segment to be scanned, got " + f"segments={decomp}" + ) + got = [ + r[0] + for r in db.execute( + "SELECT order_id FROM evt_ov WHERE event_type = 'ov07' " + "ORDER BY order_id" + ).fetchall() + ] + assert got == [7, 47, 87, 127, 167], ( + f"overflowed segment was wrongly pruned, got rows: {got}" + ) + + # A value present nowhere still returns nothing. (No plan-shape + # assertion here: the overflowed segment may legitimately be + # skipped by exact per-segment evidence at decompress time, e.g. + # the dictionary codec's value check.) + n = db.execute( + "SELECT count(*) FROM evt_ov WHERE event_type = 'ov0a'" + ).fetchone()[0] + assert n == 0 + def test_catalog_column_valmap_populated(self, db): """After compression, deltax.deltax_partition.column_valmap should hold the sorted value list for the low-cardinality `event_type` diff --git a/tests/test_worker.py b/tests/test_worker.py index f9eb0ad..70b5804 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -244,3 +244,110 @@ def test_worker_retention_drops_old_partitions(postgres_db): finally: _cleanup(db, table) + + +def test_worker_retention_drops_compressed_companions(postgres_db): + """When retention drops a COMPRESSED partition, every companion table in + _deltax_compressed (meta, colstats, blobs, blooms, text_lengths, + valbitmap) must be dropped with it — no orphans left behind.""" + db = postgres_db + table = _unique_table() + + try: + db.execute("SET pg_deltax.mock_now = '2025-06-15 00:00:00+00'") + db.execute( + f'CREATE TABLE "{table}" ' + "(ts TIMESTAMPTZ NOT NULL, event_type TEXT NOT NULL, val FLOAT8)" + ) + db.commit() + db.execute( + f"SELECT deltax.deltax_create_table('{table}', 'ts', '1 day', 2)" + ) + db.commit() + + # Fill the June 14 partition. The low-cardinality text column makes + # compression create a `_valbitmap` companion; the text column also + # produces `_text_lengths`. + values = ", ".join( + f"('2025-06-14 12:00:00+00'::timestamptz + interval '{i} seconds', " + f"'type{i % 4}', {i})" + for i in range(100) + ) + db.execute( + f'INSERT INTO "{table}" (ts, event_type, val) VALUES {values}' + ) + db.commit() + + db.execute( + f"SELECT deltax.deltax_enable_compression('{table}', " + "order_by => ARRAY['ts'])" + ) + # The June 14 partition is the only one with rows — find its name + # rather than hardcoding it (partition names render range_start in + # the session timezone). + old_part = None + parts = db.execute( + f"SELECT partition_name FROM deltax.deltax_partition_info('{table}') " + "WHERE partition_name NOT LIKE '%default%'" + ).fetchall() + for (p,) in parts: + if db.execute(f'SELECT count(*) FROM "{p}"').fetchone()[0]: + old_part = p + break + assert old_part is not None, f"no non-empty partition among {parts}" + db.execute(f"SELECT deltax.deltax_compress_partition('{old_part}')") + db.commit() + + def list_companions(): + return [ + r[0] + for r in db.execute( + "SELECT c.relname FROM pg_class c " + "JOIN pg_namespace n ON n.oid = c.relnamespace " + "WHERE n.nspname = '_deltax_compressed' " + "AND c.relname LIKE %s AND c.relkind = 'r'", + (old_part + r"\_%",), + ).fetchall() + ] + + before = set(list_companions()) + assert f"{old_part}_meta" in before, f"companions missing: {before}" + assert f"{old_part}_valbitmap" in before, ( + f"expected a _valbitmap companion for the low-card text column, " + f"got: {before}" + ) + + db.execute(f"SELECT deltax.deltax_set_retention('{table}', '3 days')") + db.commit() + + # Jump time so the June 14 partition exceeds the retention window. + _alter_system( + "ALTER SYSTEM SET pg_deltax.mock_now = '2025-06-20 00:00:00+00'" + ) + + deadline = time.time() + 150 + dropped = False + while time.time() < deadline: + time.sleep(5) + still = { + row[0] + for row in db.execute( + f"SELECT partition_name FROM deltax.deltax_partition_info('{table}')" + ).fetchall() + } + db.commit() + if old_part not in still: + dropped = True + break + + assert dropped, f"worker did not drop partition {old_part}" + + leftovers = list_companions() + db.commit() + assert leftovers == [], ( + f"orphaned companion tables left in _deltax_compressed after " + f"retention drop: {leftovers}" + ) + + finally: + _cleanup(db, table) From 79ac0eafad63c72f7a8ed587eba58b1132bc7dc2 Mon Sep 17 00:00:00 2001 From: Will Ezell Date: Tue, 14 Jul 2026 17:59:32 -0400 Subject: [PATCH 4/4] style: rustfmt worker.rs launcher fan-out (pre-existing drift; upstream #43 equivalent) --- src/worker.rs | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/src/worker.rs b/src/worker.rs index c1d7cb2..947c92a 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -64,16 +64,14 @@ pub extern "C-unwind" fn deltax_launcher_main(_arg: pg_sys::Datum) { BackgroundWorker::attach_signal_handlers(SignalWakeFlags::SIGHUP | SignalWakeFlags::SIGTERM); let dbs = target_databases(); for (i, db) in dbs.iter().enumerate() { - let spawned = BackgroundWorkerBuilder::new(&format!( - "pg_deltax maintenance worker ({})", - db - )) - .set_function("deltax_worker_main") - .set_library("pg_deltax") - .set_argument((i as i32).into_datum()) - .enable_spi_access() - .set_restart_time(Some(Duration::from_secs(60))) - .load_dynamic(); + let spawned = + BackgroundWorkerBuilder::new(&format!("pg_deltax maintenance worker ({})", db)) + .set_function("deltax_worker_main") + .set_library("pg_deltax") + .set_argument((i as i32).into_datum()) + .enable_spi_access() + .set_restart_time(Some(Duration::from_secs(60))) + .load_dynamic(); match spawned { Ok(_) => log!("pg_deltax: launched maintenance worker for database {}", db), Err(e) => log!(