diff --git a/src/scan/exec/segments.rs b/src/scan/exec/segments.rs index 6e77bde..22aa0ea 100644 --- a/src/scan/exec/segments.rs +++ b/src/scan/exec/segments.rs @@ -240,6 +240,30 @@ unsafe fn lookup_segments_by_minmax_index( } } +/// Convert a batch-qual constant (raw PG datum bits) into the i64 domain the +/// bloom BUILD side hashed (`compress.rs` hashes `TypedColumn` values, which +/// store timestamps/dates as Unix-epoch microseconds and floats as raw bit +/// patterns). Probing with the raw PG-epoch datum would never match the +/// build-side hashes — every segment would be falsely bloom-rejected and +/// `col = const` on a timestamp/date column would return zero rows. +fn bloom_probe_encode(datum: pg_sys::Datum, type_oid: pg_sys::Oid) -> i64 { + match type_oid { + pg_sys::TIMESTAMPOID | pg_sys::TIMESTAMPTZOID => { + // PG-epoch µs → Unix-epoch µs (matches TypedColumn::Int64 build). + (datum.value() as i64) + crate::compress::PG_EPOCH_OFFSET_USEC + } + pg_sys::DATEOID => { + // PG-epoch days (int32 datum) → Unix-epoch µs. + ((datum.value() as i32 as i64) + crate::compress::PG_EPOCH_OFFSET_DAYS) * 86_400_000_000 + } + // f32 bit pattern occupies the low 32 datum bits (build hashes + // `x.to_bits() as i64`). + pg_sys::FLOAT4OID => (datum.value() as u32) as i64, + // Integers and f64 bit patterns are identity. + _ => datum.value() as i64, + } +} + /// Resolve `{partition}_` (where the partition name is derived /// from `meta_oid` by stripping the `_meta` suffix) to a relation OID in /// the same namespace as `meta_oid`. Returns `InvalidOid` when the table @@ -1922,18 +1946,21 @@ pub(super) unsafe fn load_segments_heap( let hashes = if bq.op == BatchCompareOp::InList { if let Some(ref vals) = bq.in_list_i64 { vals.iter() - .map(|&v| crate::bloom::hash_datum_i64(v)) + .map(|&v| { + crate::bloom::hash_datum_i64(bloom_probe_encode( + pg_sys::Datum::from(v as usize), + bq.type_oid, + )) + }) .collect() } else { continue; } } else { - let val_i64 = match bq.type_oid { - pg_sys::FLOAT4OID => (bq.const_datum.value() as u32) as i64, - pg_sys::FLOAT8OID => bq.const_datum.value() as i64, - _ => bq.const_datum.value() as i64, - }; - vec![crate::bloom::hash_datum_i64(val_i64)] + vec![crate::bloom::hash_datum_i64(bloom_probe_encode( + bq.const_datum, + bq.type_oid, + ))] }; bloom_checks.push(BloomCheck { col_idx: ci, @@ -2055,20 +2082,29 @@ pub(super) unsafe fn load_segments_heap( // --- Segment survived time/segment_by pruning --- - // Extract per-column min/max (time column from meta — identity encoding for timestamp/date) + // Extract per-column min/max. The meta table stores the time + // column's `_min_`/`_max_` in its NATIVE type (PG-epoch datum), + // but every `ColMinMax` consumer (`decode_encoded_to_datum`, + // `decode_encoded_to_pg_i64`, `segment_all_rows_pass`, ...) + // expects the order-preserving colstats encoding (Unix-epoch + // microseconds for timestamps/dates) — so convert here with the + // same `encode_datum_to_i64` rule used to populate colstats. let mut col_minmax = HashMap::new(); for (col_name, min_att, max_att, type_oid) in &minmax_col_attnos { let min_null = nulls[*min_att]; let max_null = nulls[*max_att]; + let encode_meta = |d: pg_sys::Datum| -> i64 { + encode_datum_to_i64(d, *type_oid).unwrap_or_else(|| d.value() as i64) + }; let min_enc = if min_null { 0i64 } else { - values[*min_att].value() as i64 + encode_meta(values[*min_att]) }; let max_enc = if max_null { 0i64 } else { - values[*max_att].value() as i64 + encode_meta(values[*max_att]) }; col_minmax.insert( col_name.clone(), @@ -2289,22 +2325,25 @@ pub(super) unsafe fn load_segments_heap( None => continue, }; - // For float in-list, re-encode from raw datum bits to order-preserving i64 - let encoded_in_list = bq.in_list_i64.as_ref().map(|vals| { - vals.iter() + // `in_list_i64` holds raw PG datum bits — re-encode each + // value into the colstats domain with the same rule as + // `const_i64` above (timestamps/dates: PG epoch → Unix + // epoch; floats: order-preserving bits; ints: identity). + let encoded_in_list = match bq.in_list_i64.as_ref() { + None => None, + Some(vals) => match vals + .iter() .map(|&v| { - match bq.type_oid { - pg_sys::FLOAT4OID => { - encode_f32_to_i64(f32::from_bits(v as u32)) - } - pg_sys::FLOAT8OID => { - encode_f64_to_i64(f64::from_bits(v as u64)) - } - _ => v, // int/timestamp/date: identity - } + encode_datum_to_i64(pg_sys::Datum::from(v as usize), bq.type_oid) }) - .collect() - }); + .collect::>>() + { + Some(enc) => Some(enc), + // Unencodable element — skip minmax pruning for + // this qual rather than risk a wrong prune. + None => continue, + }, + }; cs_minmax_filters.push(MinMaxFilter { col_idx: ci, @@ -4152,6 +4191,32 @@ mod tests { } } + #[test] + fn bloom_probe_encode_matches_build_domain() { + // Integers: identity. + assert_eq!( + bloom_probe_encode(pg_sys::Datum::from(42i64 as usize), pg_sys::INT8OID), + 42 + ); + // Timestamps: PG-epoch usec -> Unix-epoch usec (the domain + // TypedColumn::Int64 stores and the bloom build hashes). + assert_eq!( + bloom_probe_encode(pg_sys::Datum::from(0i64 as usize), pg_sys::TIMESTAMPTZOID), + crate::compress::PG_EPOCH_OFFSET_USEC + ); + // Dates: PG-epoch days -> Unix-epoch usec. + assert_eq!( + bloom_probe_encode(pg_sys::Datum::from(1i32 as usize), pg_sys::DATEOID), + (crate::compress::PG_EPOCH_OFFSET_DAYS + 1) * 86_400_000_000 + ); + // f32: low 32 datum bits = bit pattern. + let f = 1.5f32; + assert_eq!( + bloom_probe_encode(pg_sys::Datum::from(f.to_bits() as usize), pg_sys::FLOAT4OID), + f.to_bits() as i64 + ); + } + #[test] fn segment_all_rows_pass_eq_ranges() { let cm = mk_cm(10, 10, pg_sys::INT4OID); diff --git a/tests/test_compression.py b/tests/test_compression.py index b2d280b..6d06779 100644 --- a/tests/test_compression.py +++ b/tests/test_compression.py @@ -4684,3 +4684,187 @@ def test_same_named_partition_in_other_schema_not_hijacked(db): # 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" +# Epoch-encoding regression tests (Unix epoch vs PG epoch, offset 946,684,800s) +# --------------------------------------------------------------------------- + +class TestEpochEncoding: + """Regression tests for Unix-epoch vs PostgreSQL-epoch (2000-01-01, + offset 946,684,800 s) encoding confusions in segment metadata. + + Three distinct bugs, all fixed in src/scan/exec/segments.rs: + + 1. Bloom probes hashed the raw PG-epoch datum of `col = const` / + `col IN (...)` constants while the bloom BUILD side (compress.rs) + hashes the colstats encoding (Unix-epoch microseconds for + timestamps/dates). Every segment was falsely bloom-rejected, so + equality predicates on timestamp/date columns returned ZERO rows + on compressed partitions. + + 2. The colstats min/max in-list filter identity-encoded raw PG-epoch + datums for timestamp/date IN-lists on non-time columns, while the + colstats `_min`/`_max` it compares against are Unix-epoch + microseconds — every segment falsely pruned, zero rows. + + 3. `load_segments_heap` stored the time column's meta-table + `_min_`/`_max_` identity-encoded (PG-epoch datum) in `ColMinMax`, + whose consumers all decode the colstats encoding (Unix-epoch us) — + timestamps 30 years early on any path consuming that map. + + Year-2000-adjacent data makes the two encodings maximally divergent, + which is the shape that exposed these bugs. + """ + + EPOCH_NOW = "2000-01-15 12:00:00+00" + EPOCH_BASE = "2000-01-15 00:00:00+00" + + def _setup_and_compress(self, db, n_rows=1000): + """Timestamp-sort-key table with year-2000 data, one compressed + partition. Rows are 1 second apart starting at EPOCH_BASE; row i has + val = i and d alternating between 2000-01-15 and 2000-01-16.""" + db.execute(f"SET pg_deltax.mock_now = '{self.EPOCH_NOW}'") + db.execute(""" + CREATE TABLE epoch_t ( + ts TIMESTAMPTZ NOT NULL, + d DATE, + device_id TEXT NOT NULL, + val BIGINT + ) + """) + db.execute( + "SELECT deltax.deltax_create_table('epoch_t', 'ts', '1 day'::interval)" + ) + db.execute( + "SELECT deltax.deltax_enable_compression('epoch_t', " + "order_by => ARRAY['ts'])" + ) + db.execute(f""" + INSERT INTO epoch_t + SELECT '{self.EPOCH_BASE}'::timestamptz + (i || ' seconds')::interval, + CASE WHEN i % 2 = 0 THEN DATE '2000-01-15' ELSE DATE '2000-01-16' END, + 'dev-' || (i % 5)::text, + i + FROM generate_series(0, {n_rows - 1}) i + """) + db.commit() + + part_name = db.execute( + "SELECT partition_name FROM deltax.deltax_partition_info('epoch_t') " + "WHERE range_start <= '2000-01-15'::timestamptz " + "AND range_end > '2000-01-15'::timestamptz" + ).fetchone()[0] + db.execute(f"SELECT deltax.deltax_compress_partition('{part_name}')") + db.commit() + return part_name + + def test_timestamp_equality_returns_matching_row(self, db): + """`ts = const` on the timestamp sort column must return the row. + Broken before the bloom-probe fix: the probe hashed the PG-epoch + datum, never matched the Unix-epoch build hashes, and every segment + was bloom-rejected (zero rows).""" + self._setup_and_compress(db) + rows = db.execute( + "SELECT ts, val FROM epoch_t WHERE ts = '2000-01-15 00:01:40+00'" + ).fetchall() + assert len(rows) == 1 + assert rows[0][1] == 100 + + def test_timestamp_in_list_returns_matching_rows(self, db): + """`ts IN (...)` exercises the in-list bloom probe path.""" + self._setup_and_compress(db) + rows = db.execute( + "SELECT val FROM epoch_t " + "WHERE ts IN ('2000-01-15 00:01:40+00', '2000-01-15 00:03:20+00') " + "ORDER BY val" + ).fetchall() + assert [r[0] for r in rows] == [100, 200] + + def test_timestamp_equality_absent_value_returns_nothing(self, db): + """A constant inside the segment's [min, max] but absent from the + data must return zero rows — guards against an over-correction that + would defeat bloom pruning entirely.""" + self._setup_and_compress(db) + count = db.execute( + "SELECT count(*) FROM epoch_t WHERE ts = '2000-01-15 00:10:00.5+00'" + ).fetchone()[0] + assert count == 0 + + def test_date_equality_on_non_sort_column(self, db): + """`d = const` on a DATE payload column probes the per-segment bloom + with the DATE encoding (PG-epoch days -> Unix-epoch microseconds).""" + self._setup_and_compress(db) + count = db.execute( + "SELECT count(*) FROM epoch_t WHERE d = DATE '2000-01-15'" + ).fetchone()[0] + assert count == 500 + count = db.execute( + "SELECT count(*) FROM epoch_t WHERE d = DATE '2000-01-17'" + ).fetchone()[0] + assert count == 0 + + def test_date_in_list_on_non_sort_column(self, db): + """`d IN (...)` on a DATE payload column: both the bloom probe and + the colstats min/max in-list filter must compare in the colstats + domain (Unix-epoch microseconds). The in-list arm of the colstats + filter identity-encoded the raw PG-epoch datum, so every segment was + falsely min/max-pruned (zero rows).""" + self._setup_and_compress(db) + count = db.execute( + "SELECT count(*) FROM epoch_t " + "WHERE d IN (DATE '2000-01-15', DATE '2000-01-17')" + ).fetchone()[0] + assert count == 500 + count = db.execute( + "SELECT count(*) FROM epoch_t " + "WHERE d IN (DATE '2000-01-17', DATE '2000-01-18')" + ).fetchone()[0] + assert count == 0 + + def test_min_max_of_timestamp_sort_column(self, db): + """MIN/MAX of the timestamp column served from segment metadata must + not come back 30 years early (meta-table min/max encoding fix). + Several aggregate shapes, since they route to different paths + (DeltaXMinMax, DeltaXAgg metadata fast path).""" + self._setup_and_compress(db) + lo, hi = db.execute("SELECT min(ts), max(ts) FROM epoch_t").fetchone() + assert str(lo) == "2000-01-15 00:00:00+00:00" + assert str(hi) == "2000-01-15 00:16:39+00:00" + + lo, hi, cnt = db.execute( + "SELECT min(ts), max(ts), count(*) FROM epoch_t" + ).fetchone() + assert str(lo) == "2000-01-15 00:00:00+00:00" + assert str(hi) == "2000-01-15 00:16:39+00:00" + assert cnt == 1000 + + lo, avg = db.execute( + "SELECT min(ts), avg(val) FROM epoch_t" + ).fetchone() + assert str(lo) == "2000-01-15 00:00:00+00:00" + assert float(avg) == 499.5 + + def test_range_predicates_across_year_2000(self, db): + """Range predicates with year-2000-adjacent constants: segment + min/max metadata must neither falsely skip (zero rows) nor falsely + all-pass (every row) when classified against the encoded constant.""" + self._setup_and_compress(db) + sum_, cnt = db.execute( + "SELECT sum(val), count(*) FROM epoch_t " + "WHERE ts <= '2000-01-15 00:01:39+00'" + ).fetchone() + assert (sum_, cnt) == (4950, 100) + + sum_, cnt = db.execute( + "SELECT sum(val), count(*) FROM epoch_t " + "WHERE ts >= '2000-01-15 00:01:40+00'" + ).fetchone() + assert (sum_, cnt) == (494550, 900) + + # Fully outside the data range on either side of the data + cnt = db.execute( + "SELECT count(*) FROM epoch_t WHERE ts < '2000-01-15 00:00:00+00'" + ).fetchone()[0] + assert cnt == 0 + cnt = db.execute( + "SELECT count(*) FROM epoch_t WHERE ts > '2000-01-15 00:16:39+00'" + ).fetchone()[0] + assert cnt == 0