persist, storage, repr: address review feedback

Finish the "fail open on undecodable part stats" change. `LazyPartStats`
loses its panicking `decode`, so no read path can reintroduce the panic:
the fast-path peek filter keeps the part, `EXPLAIN FILTER PUSHDOWN`
reports it as selected, and inspect-state serializes the undecodable
marker its `Debug` impl already renders. All three ran on stats straight
off durable state, so version skew or corruption panicked clusterd and
environmentd respectively.

Extend that to the 'ok' column shape checks in `RelationPartStats`, which
panicked on a wrong-shaped column on the very call stack the peek filter
runs. A mismatch now counts in `mismatched_count` and reports the column
range as unknown.

Use the whole-second leap time as the interesting `TIME` datum.
`from_hms_micro_opt(23, 59, 59, 1_999_999)` is a fractional leap second
encoding to 86_400_999_999 microseconds, outside PostgreSQL's
[0, 86_400_000_000] `time` domain, where a `postgres-types` client wraps
it to 00:00:00.999999. Since `interesting_datums` also feeds the DATUMS
load generator, that value reached a user-visible table. It is
unreachable from SQL besides, which falsified the premise for keeping the
leap representation in `TIME` at all. `TIME '23:59:60'` exercises the
same `frac >= 1e9` path and encodes to exactly PostgreSQL's bound.

Redact the `DataflowError` in the new error-row audit arm: those events
go to Sentry, and `DecodeError` carries the raw source record bytes.

Also drop the duplicated `value_between` doc summary line, document that
the fuzz runner's interrupted classification treats a kernel OOM SIGKILL
as interrupted too, relying on libFuzzer's rss limit to catch memory
blowups as artifact-producing OOMs first, and replace the
clippy-disallowed `prop_oneof!` and `Iterator::zip` in the new proptest
strategies with `Union` and `zip_eq`.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: def-

src/adapter/src/coord/sequencer.rs

@@ -1149,10 +1149,12 @@ pub(crate) async fn explain_pushdown_future_inner<
                 let bytes = u64::cast_from(*bytes);
                 total_bytes += bytes;
                 total_parts += 1u64;
-                let selected = match stats {
+                let selected = match stats.as_ref().and_then(|x| x.try_decode().ok()) {
+                    // Also the arm for stats that do not decode, which a
+                    // newer writer's stats kind can produce. Both report the
+                    // part as selected, matching what a read of it would do.
                     None => true,
                     Some(stats) => {
-                        let stats = stats.decode();
                         let stats = RelationPartStats::new(
                             name.as_str(),
                             &snapshot_stats.metrics.pushdown.part_stats,

src/expr/src/interpret.rs

@@ -256,7 +256,6 @@ 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

src/persist-client/src/internal/encoding.rs

@@ -1756,17 +1756,11 @@ impl LazyPartStats {
     /// This does not cache the returned value, it decodes each time it's
     /// called.
     ///
-    /// Panics if the encoded bytes are malformed. Only call this where the value
-    /// is known to have come from `Self::encode` rather than straight off blob.
-    pub fn decode(&self) -> PartStats {
-        self.try_decode().expect("valid stats")
-    }
-
-    /// Like [Self::decode], but surfaces a malformed encoding as an error.
-    ///
-    /// The bytes are stored undecoded (see the [RustType] impl), so a corrupted
-    /// or crafted blob reaches here intact. Anything running on state that has
-    /// not been validated yet must use this.
+    /// The bytes are stored undecoded (see the [RustType] impl) and are never
+    /// validated on the way in, so a corrupted, crafted, or newer-version blob
+    /// reaches here intact. There is deliberately no infallible variant: every
+    /// caller reads stats straight off durable state, where a decode failure
+    /// must fail open (keep the part, report it selected) rather than panic.
     pub fn try_decode(&self) -> Result<PartStats, TryFromProtoError> {
         let key = self
             .key
@@ -2678,15 +2672,12 @@ mod tests {
         LazyPartStats::from_proto(Bytes::from(bytes)).expect("stats bytes are stored undecoded")
     }
 
-    /// `decode` panics on stats from a newer version, which is why the read
-    /// paths (the shard_source filter and the `stats()` accessors in fetch)
-    /// must use `try_decode` and fail open to fetching the part.
-    #[mz_ore::test]
-    #[should_panic(expected = "valid stats")]
-    fn part_stats_decode_panics_on_unknown_variant() {
-        let _ = version_skewed_part_stats().decode();
-    }
-
+    /// Stats from a newer version are an error rather than a value this
+    /// version misreads, which is what lets every read path fail open on
+    /// them: the `shard_source` filter and the fast-path peek filter keep the
+    /// part, the `stats()` accessors in fetch report `None`, `EXPLAIN FILTER
+    /// PUSHDOWN` reports the part as selected, and inspect-state serializes
+    /// the stats as absent.
     #[mz_ore::test]
     fn part_stats_try_decode_fails_open_on_unknown_variant() {
         assert_err!(version_skewed_part_stats().try_decode());

src/persist-client/src/internal/state.rs

@@ -2739,8 +2739,19 @@ fn serialize_part_stats<S: Serializer>(
     val: &Option<LazyPartStats>,
     s: S,
 ) -> Result<S::Ok, S::Error> {
-    let val = val.as_ref().map(|x| x.decode().key);
-    val.serialize(s)
+    // These bytes come from blob and are never validated on the way in, so a
+    // malformed or newer-version encoding reaches here intact. Report it as
+    // absent rather than panicking, and keep the field's shape stable for
+    // consumers of the inspect-state output by logging the failure instead of
+    // serializing a differently typed value in its place.
+    let stats = val.as_ref().and_then(|x| match x.try_decode() {
+        Ok(stats) => Some(stats.key),
+        Err(err) => {
+            tracing::warn!("undecodable part stats, reporting as absent: {err}");
+            None
+        }
+    });
+    stats.serialize(s)
 }
 
 fn serialize_diffs_sum<S: Serializer>(val: &Option<[u8; 8]>, s: S) -> Result<S::Ok, S::Error> {

src/repr/src/scalar.rs

@@ -4113,8 +4113,12 @@ impl SqlScalarType {
                 Datum::Time(NaiveTime::from_hms_micro_opt(0, 0, 0, 0).unwrap()),
                 Datum::Time(NaiveTime::from_hms_micro_opt(23, 59, 59, 999_999).unwrap()),
                 // Leap second: chrono represents it as a fractional part of
-                // one second or more.
-                Datum::Time(NaiveTime::from_hms_micro_opt(23, 59, 59, 1_999_999).unwrap()),
+                // one second or more. `TIME '23:59:60'` is the largest value
+                // parsing admits, since fractional leap seconds are rejected,
+                // and it encodes to exactly PostgreSQL's 24:00:00 bound. A
+                // fractional leap second here would leave the type's
+                // PostgreSQL wire domain.
+                Datum::Time(NaiveTime::from_hms_micro_opt(23, 59, 59, 1_000_000).unwrap()),
             ])
         });
         static TIMESTAMP: LazyLock<Row> = LazyLock::new(|| {

src/storage-operators/src/persist_source.rs

@@ -711,10 +711,15 @@ impl PendingWork {
                         sentry::with_scope(
                             |scope| scope.set_tag("alert_id", "persist_pushdown_audit_violation"),
                             || {
+                                // `err` is redacted for the same reason the
+                                // `Ok`-row arm redacts its MFP output: these
+                                // events go to Sentry, and a `DecodeError`
+                                // carries the raw source record bytes while
+                                // several `EvalError`s embed user input.
                                 error!(
                                     ?stats,
                                     name,
-                                    ?err,
+                                    err = ?redact(&err),
                                     "persist filter pushdown correctness violation!"
                                 );
                                 if self.panic_on_audit_failure {
@@ -1534,6 +1539,7 @@ mod tests {
     /// column stat range that fails to contain a real value). See
     /// database-issues#9656 / PER-50.
     mod filter_pushdown_audit {
+        use itertools::Itertools;
         use mz_expr::func::variadic::{And, Or};
         use mz_expr::func::{
             AddFloat32, AddTimestampInterval, CastNumericToFloat32, CastNumericToMzTimestamp, Eq,
@@ -1549,6 +1555,7 @@ mod tests {
         use mz_repr::{Diff, ReprScalarType, SqlScalarType};
         use proptest::prelude::*;
         use proptest::sample::{Index, select};
+        use proptest::strategy::Union;
 
         use super::*;
 
@@ -1945,14 +1952,14 @@ mod tests {
             let ok_row = prop::collection::vec(any::<Index>(), WIDE_ARITY).prop_map(move |picks| {
                 let datums = picks
                     .iter()
-                    .zip(&pools)
+                    .zip_eq(&pools)
                     .map(|(pick, pool)| pool[pick.index(pool.len())]);
                 SourceData(Ok(Row::pack(datums)))
             });
             let err_row = Just(SourceData(Err(DataflowError::from(
                 EvalError::DivisionByZero,
             ))));
-            let row = prop_oneof![9 => ok_row, 1 => err_row];
+            let row = Union::new_weighted(vec![(9, ok_row.boxed()), (1, err_row.boxed())]);
             prop::collection::vec(row, 2..8)
         }
 
@@ -2154,37 +2161,47 @@ mod tests {
         }
 
         fn arb_wide_predicate() -> impl Strategy<Value = MirScalarExpr> {
-            let leaf = prop_oneof![
-                arb_cmp_col_lit(),
-                arb_is_null_pred(),
-                arb_jsonb_pred(),
-                arb_case_jsonb_pred(),
-                arb_iso_parse_pred(),
-                arb_ts_interval_pred(),
-                arb_float_mul_pred(),
-                arb_temporal_pred(),
+            let leaf = Union::new(vec![
+                arb_cmp_col_lit().boxed(),
+                arb_is_null_pred().boxed(),
+                arb_jsonb_pred().boxed(),
+                arb_case_jsonb_pred().boxed(),
+                arb_iso_parse_pred().boxed(),
+                arb_ts_interval_pred().boxed(),
+                arb_float_mul_pred().boxed(),
+                arb_temporal_pred().boxed(),
                 // The numeric shapes from the narrow test, aimed at the
                 // fallible-interior mechanisms; both reference column 0.
                 (
                     select(vec![0i32, 2, -5, 24699]),
                     f32_consts(),
                     f32_consts(),
                     f32_consts(),
-                    comparison_funcs()
+                    comparison_funcs(),
                 )
-                    .prop_map(|(s, a, b, c, cmp)| float_arith_predicate(s, a, b, c, cmp)),
+                    .prop_map(|(s, a, b, c, cmp)| float_arith_predicate(s, a, b, c, cmp))
+                    .boxed(),
                 (select(vec![0u64, 1, 2, 100, u64::MAX]), comparison_funcs())
-                    .prop_map(|(ts, cmp)| cast_mz_timestamp_predicate(ts, cmp)),
-            ]
+                    .prop_map(|(ts, cmp)| cast_mz_timestamp_predicate(ts, cmp))
+                    .boxed(),
+            ])
             .boxed();
-            prop_oneof![
-                3 => leaf.clone(),
-                1 => (leaf.clone(), leaf.clone(), any::<bool>()).prop_map(|(a, b, is_and)| {
-                    let func = if is_and { And.into() } else { Or.into() };
-                    MirScalarExpr::CallVariadic { func, exprs: vec![a, b] }
-                }),
-                1 => leaf.prop_map(not),
-            ]
+            Union::new_weighted(vec![
+                (3, leaf.clone()),
+                (
+                    1,
+                    (leaf.clone(), leaf.clone(), any::<bool>())
+                        .prop_map(|(a, b, is_and)| {
+                            let func = if is_and { And.into() } else { Or.into() };
+                            MirScalarExpr::CallVariadic {
+                                func,
+                                exprs: vec![a, b],
+                            }
+                        })
+                        .boxed(),
+                ),
+                (1, leaf.prop_map(not).boxed()),
+            ])
         }
 
         /// The zero-column count(*) path: when the read desc projects away

src/storage-operators/src/stats.rs

@@ -48,7 +48,11 @@ impl StatsCursor {
         let should_fetch = |name: &'static str, errors: bool| {
             move |stats: Option<&LazyPartStats>| {
                 let Some(stats) = stats else { return true };
-                let stats = stats.decode();
+                // Stats written by a newer version may not decode. The sound
+                // fallback is to fetch the part.
+                let Ok(stats) = stats.try_decode() else {
+                    return true;
+                };
                 let metrics = &metrics.pushdown.part_stats;
                 let relation_stats = RelationPartStats::new(name, metrics, desc, &stats);
                 if errors {

src/storage-types/src/stats.rs

@@ -124,9 +124,20 @@ impl RelationPartStats<'_> {
         let typ = &self.desc.get_type(idx);
 
         let ok_stats = self.stats.key.col("ok")?;
-        let ok_stats = ok_stats
-            .try_as_optional_struct()
-            .expect("ok column should be nullable struct");
+        // These stats come straight off durable state, so a corrupt or
+        // version-skewed encoding can carry any shape here. Report the column
+        // range as unknown rather than panicking the process reading it.
+        let ok_stats = match ok_stats.try_as_optional_struct() {
+            Ok(ok_stats) => ok_stats,
+            Err(err) => {
+                self.metrics.mismatched_count.inc();
+                tracing::error!(
+                    "expected nullable struct stats for the 'ok' column of {}: {err}",
+                    self.name
+                );
+                return None;
+            }
+        };
         let col_stats = ok_stats.some.cols.get(name.as_str())?;
 
         if let SqlColumnType {
@@ -200,8 +211,16 @@ impl RelationPartStats<'_> {
         let typ = self.desc.get_type(idx);
 
         let ok_stats = self.stats.key.cols.get("ok")?;
+        // See the note in `col_json`: durable stats can be any shape, so a
+        // wrong-shaped 'ok' column makes the range unknown, not a panic.
         let ColumnStatKinds::Struct(ok_stats) = &ok_stats.values else {
-            panic!("'ok' column stats should be a struct")
+            self.metrics.mismatched_count.inc();
+            tracing::error!(
+                "expected struct stats for the 'ok' column of {}, found {:?}",
+                self.name,
+                ok_stats.values
+            );
+            return None;
         };
         let col_stats = ok_stats.cols.get(name.as_str())?;
 
@@ -516,6 +535,101 @@ mod tests {
         assert_eq!(stats.err_count(), None);
     }
 
+    /// Wrong-shaped ok-column stats must read as "column range unknown",
+    /// which fails open to keeping the part. Both column paths run here: a
+    /// plain column goes through `col_values`, a JSON one adds `col_json`,
+    /// and neither may panic the process reading durable state.
+    #[mz_ore::test]
+    #[cfg_attr(miri, ignore)] // too slow
+    fn malformed_ok_stats_fail_open() {
+        use mz_expr::{BinaryFunc, MirScalarExpr, func};
+        use mz_persist_types::stats::{ColumnNullStats, ColumnarStats, PrimitiveStats};
+        use mz_repr::ReprScalarType;
+
+        let schema = RelationDesc::builder()
+            .with_column("col", SqlScalarType::Int32.nullable(false))
+            .with_column("json", SqlScalarType::Jsonb.nullable(true))
+            .finish();
+        let mut builder = PartBuilder::new(&schema, &UnitSchema);
+        builder.push(
+            &SourceData(Ok(Row::pack_slice(&[Datum::Int32(1), Datum::JsonNull]))),
+            &(),
+            1u64,
+            1i64,
+        );
+        let part = builder.finish();
+        let key_col = part.key.as_struct();
+        let decoder = <RelationDesc as Schema<SourceData>>::decoder(&schema, key_col.clone())
+            .expect("success");
+
+        let json_idx = schema
+            .iter_all()
+            .map(|(idx, _name, _typ)| idx)
+            .nth(1)
+            .expect("two columns");
+        // A filter no Ok row in this part satisfies. With well-shaped stats
+        // the part is skipped, so keeping it proves the fallback ran.
+        let mfp = MapFilterProject::new(2).filter(std::iter::once(MirScalarExpr::CallBinary {
+            func: BinaryFunc::Eq(func::Eq),
+            expr1: Box::new(MirScalarExpr::column(0)),
+            expr2: Box::new(MirScalarExpr::literal_ok(
+                Datum::Int32(999),
+                ReprScalarType::Int32,
+            )),
+        }));
+
+        let metrics = PartStatsMetrics::new(&MetricsRegistry::new());
+        let arena = RowArena::new();
+        let well_shaped = PartStats {
+            key: decoder.stats(),
+        };
+        let well_shaped = RelationPartStats::new("test", &metrics, &schema, &well_shaped);
+        assert!(!well_shaped.may_match_mfp(ResultSpec::anything(), &mfp));
+
+        // `col_values` matched the ok column's kind infallibly. A wrong kind
+        // makes every column's range unknown.
+        let mut not_a_struct = decoder.stats();
+        not_a_struct.cols.insert(
+            "ok".to_string(),
+            ColumnarStats {
+                nulls: Some(ColumnNullStats { count: 0 }),
+                values: PrimitiveStats {
+                    lower: 0i32,
+                    upper: 0i32,
+                }
+                .into(),
+            },
+        );
+        let not_a_struct = PartStats { key: not_a_struct };
+        let not_a_struct = RelationPartStats::new("test", &metrics, &schema, &not_a_struct);
+        for (idx, _name, _typ) in schema.iter_all() {
+            assert_eq!(not_a_struct.col_stats(idx, &arena), ResultSpec::anything());
+        }
+        assert!(not_a_struct.may_match_mfp(ResultSpec::anything(), &mfp));
+
+        // `col_json` additionally required the ok column to be nullable. A
+        // struct that is not widens the JSON range alone, so assert on that
+        // rather than on the part-level decision.
+        let mut not_nullable = decoder.stats();
+        match not_nullable.cols.get_mut("ok") {
+            Some(ok_stats) => ok_stats.nulls = None,
+            None => panic!("ok stats missing"),
+        }
+        let not_nullable = PartStats { key: not_nullable };
+        let not_nullable = RelationPartStats::new("test", &metrics, &schema, &not_nullable);
+        let other_json = Datum::String("a");
+        assert!(
+            !well_shaped
+                .col_stats(json_idx, &arena)
+                .may_contain(other_json)
+        );
+        assert!(
+            not_nullable
+                .col_stats(json_idx, &arena)
+                .may_contain(other_json)
+        );
+    }
+
     #[mz_ore::test]
     #[ignore] // TODO(parkmycar): Re-enable this test with a smaller sample size.
     fn statistics_stability() {

test/cargo-fuzz/mzcompose.py

@@ -437,7 +437,10 @@ def _reap(self, job: Job) -> None:
         ):
             # Killed by a signal (Ctrl-C, step timeout, an external kill)
             # without a crash artifact: an interrupted run, not a crash.
-            # Crashes always leave an artifact, so this cannot mask one.
+            # libFuzzer-detected crashes always leave an artifact, so this
+            # cannot mask one. A kernel OOM SIGKILL is also reported as
+            # interrupted; the rss limit passed to libFuzzer catches memory
+            # blowups as artifact-producing OOMs well before the kernel does.
             self.succeeded.append(job)
             say(
                 f"- {job.name} interrupted by signal {-job.returncode} [{secs}s]  "