fix: epoch-encoding bugs — timestamp equality/IN returns ZERO rows on compressed partitions

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}_<suffix>` (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::<Option<Vec<i64>>>()
+                        {
+                            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);

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