expr: don't read unordered stats bounds as an empty range (PER-53)

`ResultSpec::value_between` collapsed `min > max` to `nothing()`, i.e. "no
value can be here". That only holds for bounds a caller derived coherently.
Persist part statistics are not such a caller: arrow orders floats totally,
putting `-NaN` below `-Infinity`, while `OrderedFloat`, the `Datum` order
`value_between` compares in, ranks every NaN above every finite value. A part
holding `-NaN` therefore reports `lower = -NaN` against a finite `upper`, the
range read back as empty, and filter pushdown discarded the part, losing every
other row in it. `'-NaN'` is ordinary user input, so no corrupt storage is
needed to reach this.

Widening unordered bounds is not sufficient on its own. A part holding NaNs
of both signs records the total-order bounds `(-NaN, +NaN)`, which decode to
`NaN == NaN`: a seemingly valid, non-inverted range claiming the part holds
nothing but NaN, so a filter matching a finite row still discarded the part.
Only the stats decode still sees the NaN signs, so `col_values` additionally
refuses to produce float bounds from a negative-NaN lower, unless the upper
shows the whole column is NaN.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Q5GiWioZM6CWrwWgfBADDR

Author: def-

src/expr/src/interpret.rs

@@ -257,6 +257,14 @@ impl<'a> ResultSpec<'a> {
     }
 
     /// A spec that matches values between the given (non-null) min and max.
+    /// A spec for the values between `min` and `max` inclusive.
+    ///
+    /// Unordered bounds widen to [`ResultSpec::value_all`] instead of collapsing
+    /// to [`ResultSpec::nothing`]: they mean the bounds are unusable, not that
+    /// the column holds nothing. Persist float stats produce them, because arrow
+    /// orders floats totally, putting `-NaN` below `-Infinity`, while the
+    /// [`Datum`] order compared here ranks every NaN above every finite value.
+    /// Collapsing lost every other row in such a part (PER-53).
     pub fn value_between(min: Datum<'a>, max: Datum<'a>) -> ResultSpec<'a> {
         assert!(!min.is_null());
         assert!(!max.is_null());
@@ -266,7 +274,7 @@ impl<'a> ResultSpec<'a> {
                 ..ResultSpec::nothing()
             }
         } else {
-            ResultSpec::nothing()
+            ResultSpec::value_all()
         }
     }
 

src/repr/src/stats.rs

@@ -96,6 +96,28 @@ pub fn fixed_stats_from_column(
     .into()
 }
 
+/// Persist computes float column bounds in IEEE-754 total order, in which
+/// negative NaNs sort below -Infinity. `Datum` floats compare via
+/// `OrderedFloat`, which ranks every NaN (of either sign) above every other
+/// value. A lower bound that is a negative NaN therefore admits both NaNs
+/// (the largest values under `Datum` ordering) and ordinary values up to
+/// `upper`, a set no `Datum` interval can bound, so no bounds are returned
+/// and the column is treated as unconstrained. The one exception is an upper
+/// bound that is also a negative NaN, which under total order means every
+/// value in the column is a NaN.
+///
+/// NOTE: the widening `ResultSpec::value_between` does for inverted bounds is
+/// not sufficient here. A part holding both `-NaN` and `+NaN` decodes to the
+/// non-inverted bounds `(NaN, NaN)`, which would wrongly claim the part holds
+/// nothing but NaN. Only this layer still sees the NaN signs.
+fn float_bounds<F: num_traits::Float>(lower: F, upper: F) -> Option<(F, F)> {
+    if lower.is_nan() && lower.is_sign_negative() && !(upper.is_nan() && upper.is_sign_negative()) {
+        None
+    } else {
+        Some((lower, upper))
+    }
+}
+
 /// Returns a `(lower, upper)` bound from the provided [`ColumnStatKinds`], if applicable.
 pub fn col_values<'a>(
     typ: &SqlScalarType,
@@ -145,10 +167,18 @@ pub fn col_values<'a>(
             map_stats(stats, Datum::Int64)
         }
         (SqlScalarType::Float32, ColumnStatKinds::Primitive(F32(stats))) => {
-            map_stats(stats, |x| Datum::Float32(OrderedFloat(x)))
+            let (lower, upper) = float_bounds(stats.lower, stats.upper)?;
+            Some((
+                Datum::Float32(OrderedFloat(lower)),
+                Datum::Float32(OrderedFloat(upper)),
+            ))
         }
         (SqlScalarType::Float64, ColumnStatKinds::Primitive(F64(stats))) => {
-            map_stats(stats, |x| Datum::Float64(OrderedFloat(x)))
+            let (lower, upper) = float_bounds(stats.lower, stats.upper)?;
+            Some((
+                Datum::Float64(OrderedFloat(lower)),
+                Datum::Float64(OrderedFloat(upper)),
+            ))
         }
         (
             SqlScalarType::Numeric { .. },

src/storage-operators/src/persist_source.rs

@@ -1529,6 +1529,10 @@ mod tests {
             MirScalarExpr::literal_ok(Datum::from(x), ReprScalarType::Float32)
         }
 
+        fn f64_lit(x: f64) -> MirScalarExpr {
+            MirScalarExpr::literal_ok(Datum::from(x), ReprScalarType::Float64)
+        }
+
         fn numeric_datum(x: f64) -> Datum<'static> {
             Datum::from(Numeric::from(x))
         }
@@ -1713,5 +1717,130 @@ mod tests {
                 check(rows, predicate)?;
             });
         }
+
+        /// Assert that `filter_result` keeps a part it must keep, and that the
+        /// case is not vacuous: the MFP has to yield output on a real row for
+        /// "must keep" to mean anything.
+        fn assert_part_kept(desc: &RelationDesc, rows: &[Row], predicate: MirScalarExpr) {
+            let plan = MapFilterProject::new(1)
+                .filter(std::iter::once(predicate))
+                .into_plan()
+                .expect("into_plan");
+            assert!(
+                mfp_yields_output(&plan, rows),
+                "nothing to keep: the MFP yields no output on any of these rows.\n\
+                 rows={rows:?}\nplan={plan:?}",
+            );
+
+            let part_stats = build_part_stats(desc, rows);
+            let metrics = PartStatsMetrics::new(&MetricsRegistry::new());
+            let stats = RelationPartStats::new("test", &metrics, desc, &part_stats);
+            let decision = filter_result(desc, ResultSpec::anything(), stats, &plan);
+            assert!(
+                !matches!(decision, FilterResult::Discard),
+                "filter pushdown discarded a part whose MFP yields output on a real row.\n\
+                 rows={rows:?}\nplan={plan:?}",
+            );
+        }
+
+        /// A part holding `-NaN` must not hide the rest of its rows from a
+        /// filter on that column.
+        ///
+        /// `PrimitiveStats` takes a float column's bounds in arrow's *total*
+        /// order, where `-NaN` sorts below `-Infinity`, so such a part records
+        /// `lower = -NaN` against a finite `upper`. `OrderedFloat`, the `Datum`
+        /// order the interpreter compares in, ranks every NaN *above* every
+        /// finite value, so those bounds arrive unordered. Reading them as an
+        /// empty range discarded the part, which lost every other row in it
+        /// (PER-53). Both float widths take the same path.
+        #[mz_ore::test]
+        #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function
+        fn negative_nan_does_not_discard_matching_part() {
+            // Both rows have to land in the same part. Split across parts each
+            // one gets ordered bounds of its own and nothing is discarded.
+            let desc = RelationDesc::builder()
+                .with_column("c0", SqlScalarType::Float32.nullable(false))
+                .finish();
+            let rows = [
+                Row::pack_slice(&[Datum::from(-f32::NAN)]),
+                Row::pack_slice(&[Datum::from(0.0f32)]),
+            ];
+            assert_part_kept(
+                &desc,
+                &rows,
+                MirScalarExpr::CallBinary {
+                    func: BinaryFunc::Lt(Lt),
+                    expr1: Box::new(MirScalarExpr::column(0)),
+                    expr2: Box::new(f32_lit(1.0)),
+                },
+            );
+
+            let desc = RelationDesc::builder()
+                .with_column("c0", SqlScalarType::Float64.nullable(false))
+                .finish();
+            let rows = [
+                Row::pack_slice(&[Datum::from(-f64::NAN)]),
+                Row::pack_slice(&[Datum::from(0.0f64)]),
+            ];
+            assert_part_kept(
+                &desc,
+                &rows,
+                MirScalarExpr::CallBinary {
+                    func: BinaryFunc::Lt(Lt),
+                    expr1: Box::new(MirScalarExpr::column(0)),
+                    expr2: Box::new(f64_lit(1.0)),
+                },
+            );
+        }
+
+        /// A part holding NaNs of both signs must not hide its other rows.
+        ///
+        /// Arrow's total order puts `-NaN` below and `+NaN` above everything,
+        /// so such a part records the bounds `(-NaN, +NaN)`. Under
+        /// `OrderedFloat` those decode to `NaN == NaN`: not an inverted range
+        /// but a seemingly valid one claiming the part holds nothing but NaN,
+        /// so widening unordered bounds does not catch it. A filter that a
+        /// finite row matches but NaN does not then discarded the part. Only
+        /// the stats decode still sees the NaN signs, so the guard lives in
+        /// `mz_repr::stats::col_values`.
+        #[mz_ore::test]
+        #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function
+        fn mixed_sign_nans_do_not_discard_matching_part() {
+            let desc = RelationDesc::builder()
+                .with_column("c0", SqlScalarType::Float32.nullable(false))
+                .finish();
+            let rows = [
+                Row::pack_slice(&[Datum::from(-f32::NAN)]),
+                Row::pack_slice(&[Datum::from(f32::NAN)]),
+                Row::pack_slice(&[Datum::from(0.0f32)]),
+            ];
+            assert_part_kept(
+                &desc,
+                &rows,
+                MirScalarExpr::CallBinary {
+                    func: BinaryFunc::Eq(Eq),
+                    expr1: Box::new(MirScalarExpr::column(0)),
+                    expr2: Box::new(f32_lit(0.0)),
+                },
+            );
+
+            let desc = RelationDesc::builder()
+                .with_column("c0", SqlScalarType::Float64.nullable(false))
+                .finish();
+            let rows = [
+                Row::pack_slice(&[Datum::from(-f64::NAN)]),
+                Row::pack_slice(&[Datum::from(f64::NAN)]),
+                Row::pack_slice(&[Datum::from(0.0f64)]),
+            ];
+            assert_part_kept(
+                &desc,
+                &rows,
+                MirScalarExpr::CallBinary {
+                    func: BinaryFunc::Eq(Eq),
+                    expr1: Box::new(MirScalarExpr::column(0)),
+                    expr2: Box::new(f64_lit(0.0)),
+                },
+            );
+        }
     }
 }

test/sqllogictest/explain/pushdown.slt

@@ -123,6 +123,68 @@ WHERE ((CASE WHEN flag THEN j1 ELSE j2 END) ->> 'y') IS NULL
 ----
 3
 
+# Regression test for PER-53: A negative NaN must not poison the stats of the
+# float column it lands in.
+
+simple conn=mz_system,user=mz_system
+ALTER SYSTEM SET persist_stats_audit_percent = 0
+----
+COMPLETE 0
+
+statement ok
+CREATE TABLE floats (value double precision)
+
+# One INSERT, so both rows land in the same part and share its stats.
+statement ok
+INSERT INTO floats VALUES ('-NaN'), (1.0)
+
+# The part holds a matching row, so it must be selected.
+query TIIII
+EXPLAIN FILTER PUSHDOWN FOR SELECT * FROM floats WHERE value = 1.0::double precision
+----
+materialize.public.floats  1344  1344  1  1
+
+query I
+SELECT count(*) FROM floats WHERE value = 1.0::double precision
+----
+1
+
+# The same part, read through a dataflow rather than a one-shot peek.
+statement ok
+CREATE MATERIALIZED VIEW ones AS
+SELECT value FROM floats WHERE value = 1.0::double precision
+
+query I
+SELECT count(*) FROM ones
+----
+1
+
+# Variant with both NaN signs in one part: the bounds decode to the
+# non-inverted range (NaN, NaN), so widening inverted bounds alone does not
+# catch it. The spec then claimed the part holds nothing but NaN and a filter
+# on a finite value pruned the part, hiding the 0.0 row.
+
+statement ok
+CREATE TABLE mixed_nan (value double precision)
+
+statement ok
+INSERT INTO mixed_nan VALUES ('-NaN'), ('NaN'), (0.0)
+
+query TIIII
+EXPLAIN FILTER PUSHDOWN FOR SELECT * FROM mixed_nan WHERE value = 0.0::double precision
+----
+materialize.public.mixed_nan  1355  1355  1  1
+
+query I
+SELECT count(*) FROM mixed_nan WHERE value = 0.0::double precision
+----
+1
+
+simple conn=mz_system,user=mz_system
+ALTER SYSTEM RESET persist_stats_audit_percent
+----
+COMPLETE 0
+
 # Leap-second timestamps roll over at parse time (Postgres compat). The
 # chrono leap representation sorted below the next second while the epoch
 # family (extract epoch, casts to mz_timestamp, subtraction, date_bin)