Persist pushdown hardening

src/adapter/src/coord/sequencer.rs

@@ -1163,10 +1163,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/scalar/func/impls/range.rs

@@ -84,6 +84,15 @@ impl fmt::Display for CastRangeToString {
     }
 }
 
+// The monotone claim survives this function mapping empty and
+// unbounded-lower ranges to NULL, which the interpreter's endpoint box
+// cannot represent, only because those inputs form a downward-closed
+// prefix of the range ordering (`None` inner sorts below `Some`, and a
+// `None` lower bound sorts below every finite one): a range whose
+// endpoints both yield values contains no NULL-yielding interior. Any
+// change to range ordering or to this function's NULL cases must revisit
+// the claim; see `try_parse_monotonic_iso8601_timestamp` for the
+// SpecialUnary alternative.
 #[sqlfunc(sqlname = "rangelower", is_monotone = true)]
 fn range_lower<T>(a: Range<T>) -> Option<T> {
     a.inner.map(|inner| inner.lower.bound).flatten()

src/persist-client/src/fetch.rs

@@ -656,8 +656,11 @@ where
     }
 
     /// Returns the pushdown stats for this part.
+    ///
+    /// Stats written by a newer version may not decode; those return `None`,
+    /// the same as a part that carries no stats.
     pub fn stats(&self) -> Option<PartStats> {
-        self.part.stats().map(|x| x.decode())
+        self.part.stats().and_then(|x| x.try_decode().ok())
     }
 
     /// Apply any relevant projection pushdown optimizations, assuming that the data in the part
@@ -688,6 +691,11 @@ where
             &[as_of] => as_of,
             _ => return,
         };
+        // NOTE: `diffs_sum` sums every row physically in the blob, while
+        // reads truncate rows outside the registered desc. Substituting it is
+        // sound only while no writer registers a batch with tighter bounds
+        // than the blob holds (none does today, and rewritten batches prove
+        // it), which nothing here can re-check without fetching the blob.
         let eligible = self.desc.upper().less_equal(as_of) && self.desc.since().less_equal(as_of);
         if !eligible {
             return;
@@ -870,9 +878,14 @@ impl<K: Codec, V: Codec, T: Timestamp + Lattice + Codec64, D> FetchedBlob<K, V,
     }
 
     /// Decodes and returns the pushdown stats for this part, if known.
+    ///
+    /// Stats written by a newer version may not decode; those return `None`,
+    /// the same as a part that carries no stats.
     pub fn stats(&self) -> Option<PartStats> {
         match &self.buf {
-            FetchedBlobBuf::Hollow { part, .. } => part.stats.as_ref().map(|x| x.decode()),
+            FetchedBlobBuf::Hollow { part, .. } => {
+                part.stats.as_ref().and_then(|x| x.try_decode().ok())
+            }
             FetchedBlobBuf::Inline { .. } => None,
         }
     }

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
@@ -1977,6 +1971,7 @@ impl<T: Timestamp + Codec64> RustType<ProtoU64Antichain> for Antichain<T> {
 #[cfg(test)]
 mod tests {
     use mz_ore::assert_none;
+    use mz_persist_types::stats::{ProtoDynStats, ProtoStructStats};
 
     use bytes::Bytes;
     use mz_build_info::DUMMY_BUILD_INFO;
@@ -2666,4 +2661,25 @@ mod tests {
         assert_err!(stats.try_decode());
         assert!(format!("{stats:?}").contains("undecodable"));
     }
+
+    /// The exact shape version skew produces: valid protobuf whose stats
+    /// oneof uses a variant this version does not know (a newer writer's new
+    /// stats kind reaching an older reader).
+    fn version_skewed_part_stats() -> LazyPartStats {
+        let mut proto = ProtoStructStats::default();
+        proto.cols.insert("c".into(), ProtoDynStats::default());
+        let bytes = prost::Message::encode_to_vec(&proto);
+        LazyPartStats::from_proto(Bytes::from(bytes)).expect("stats bytes are stored undecoded")
+    }
+
+    /// 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

@@ -2740,8 +2740,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/persist-client/src/operators/shard_source.rs

@@ -534,7 +534,19 @@ where
                         BatchPart::Hollow(x) => {
                             let should_fetch =
                                 x.stats.as_ref().map_or(FilterResult::Keep, |stats| {
-                                    filter_fn(&stats.decode(), current_frontier.borrow())
+                                    // Stats written by a newer version may
+                                    // not decode. The sound fallback is to
+                                    // fetch the part.
+                                    match stats.try_decode() {
+                                        Ok(stats) => filter_fn(&stats, current_frontier.borrow()),
+                                        Err(err) => {
+                                            tracing::warn!(
+                                                %err,
+                                                "could not decode part stats, fetching part"
+                                            );
+                                            FilterResult::Keep
+                                        }
+                                    }
                                 });
                             should_fetch
                         }

src/pgcopy/src/copy.rs

@@ -1358,6 +1358,16 @@ mod tests {
                         Datum::Timestamp(_) | Datum::TimestampTz(_) | Datum::Null => {
                             continue;
                         }
+                        // Text carries no sign for NaN, so `-NaN` decodes as
+                        // `NaN`, and `Row` equality compares the encoded bytes,
+                        // which differ. The positive NaN of the interesting set
+                        // covers the roundtrip itself.
+                        Datum::Float32(f) if f.is_nan() && f.is_sign_negative() => {
+                            continue;
+                        }
+                        Datum::Float64(f) if f.is_nan() && f.is_sign_negative() => {
+                            continue;
+                        }
                         Datum::String(s) => {
                             // TODO: The decoder cannot differentiate between empty string and null.
                             if s.trim() == copy_csv_params.null || s.trim().is_empty() {

src/repr/src/relation.rs

@@ -1399,6 +1399,19 @@ impl RelationDesc {
 
     /// Creates a new [`RelationDesc`] retaining only the columns specified in `demands`.
     pub fn apply_demand(&self, demands: &BTreeSet<usize>) -> RelationDesc {
+        // This filters `metadata` by raw ColumnIndex but `typ` by position,
+        // which only agree when the desc is dense. Every desc constructible
+        // today is (schema history is add-only), but a dropped column would
+        // desync the two and silently attach types, statistics, and filter
+        // specs to the wrong columns downstream.
+        debug_assert!(
+            self.metadata
+                .iter()
+                .enumerate()
+                .all(|(pos, (idx, meta))| idx.0 == pos && meta.typ_idx == pos),
+            "apply_demand requires a dense RelationDesc (ColumnIndex == typ_idx): {:?}",
+            self.metadata,
+        );
         let mut new_desc = self.clone();
 
         // Update ColumnMetadata.
@@ -2066,6 +2079,23 @@ mod tests {
     use super::*;
     use prost::Message;
 
+    /// `apply_demand`, and the stats and filter-spec plumbing downstream of
+    /// it, require dense descs. A desc with a dropped column must trip the
+    /// assertion rather than silently misattach columns.
+    #[mz_ore::test]
+    #[should_panic(expected = "dense RelationDesc")]
+    fn apply_demand_rejects_non_dense_desc() {
+        let desc = RelationDesc::builder()
+            .with_column("a", SqlScalarType::Int32.nullable(false))
+            .with_column("b", SqlScalarType::Int32.nullable(false))
+            .with_column("c", SqlScalarType::Int32.nullable(false))
+            .finish();
+        let mut versioned = VersionedRelationDesc::new(desc);
+        let version = versioned.drop_column("b");
+        let desc = versioned.at_version(RelationVersionSelector::Specific(version));
+        let _ = desc.apply_demand(&BTreeSet::from([0]));
+    }
+
     #[mz_ore::test]
     #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `pipe2` on OS `linux`
     fn smoktest_at_version() {

src/repr/src/row/encode.rs

@@ -1442,6 +1442,14 @@ impl RowColumnarEncoder {
 
                 // We name the Fields in Parquet with the column index, but for
                 // backwards compat use the column name for stats.
+                //
+                // NOTE: name-keyed stats are sound only while durable
+                // relations never carry duplicate column names (the planner
+                // enforces this) and a dropped column's name can never be
+                // reused by a later version (persist rejects schema
+                // migrations containing drops). Filter pushdown consults
+                // these stats by name; violating either invariant attaches
+                // one column's stats to another and yields wrong results.
                 let name = (col_idx.to_raw(), col_name.as_str().into());
 
                 (name, encoder)

src/repr/src/scalar.rs

@@ -4053,6 +4053,12 @@ impl SqlScalarType {
                 Datum::Float32(OrderedFloat(f32::MAX)),
                 Datum::Float32(OrderedFloat(f32::EPSILON)),
                 Datum::Float32(OrderedFloat(f32::NAN)),
+                // NOTE: -NaN and -0.0 have distinct bit patterns from NaN and
+                // 0.0 but compare equal under `OrderedFloat`. Orderings that
+                // look at the representation (e.g. arrow's total order, where
+                // -NaN < -Infinity) can disagree with `OrderedFloat` on them.
+                Datum::Float32(OrderedFloat(-f32::NAN)),
+                Datum::Float32(OrderedFloat(-0.0)),
                 Datum::Float32(OrderedFloat(f32::INFINITY)),
                 Datum::Float32(OrderedFloat(f32::NEG_INFINITY)),
             ])
@@ -4067,6 +4073,9 @@ impl SqlScalarType {
                 Datum::Float64(OrderedFloat(f64::MAX)),
                 Datum::Float64(OrderedFloat(f64::EPSILON)),
                 Datum::Float64(OrderedFloat(f64::NAN)),
+                // See the FLOAT32 note on -NaN and -0.0.
+                Datum::Float64(OrderedFloat(-f64::NAN)),
+                Datum::Float64(OrderedFloat(-0.0)),
                 Datum::Float64(OrderedFloat(f64::INFINITY)),
                 Datum::Float64(OrderedFloat(f64::NEG_INFINITY)),
             ])
@@ -4103,6 +4112,13 @@ impl SqlScalarType {
             Row::pack_slice(&[
                 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. `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(|| {
@@ -4225,6 +4241,13 @@ impl SqlScalarType {
                 Datum::String("."),
                 Datum::String("2015-09-18T23:56:04.123Z"),
                 Datum::String(&"x".repeat(100)),
+                // Persist stats truncate string bounds to 100 bytes: cover a
+                // string past that limit, one whose truncated upper bound
+                // cannot be incremented (every char is char::MAX), and one
+                // with a multibyte char straddling the truncation boundary.
+                Datum::String(&"x".repeat(101)),
+                Datum::String(&"\u{10FFFF}".repeat(101)),
+                Datum::String(&format!("{}\u{1F600}", "x".repeat(99))),
                 // Valid timezone.
                 Datum::String("JAPAN"),
                 Datum::String("1,2,3"),
@@ -4267,8 +4290,31 @@ impl SqlScalarType {
                 // JSON doesn't support NaN or Infinite numbers.
                 !(n.0.is_nan() || n.0.is_infinite())
             }));
-            // TODO: Add List, Map.
-            Row::pack_slice(&datums)
+            let mut row = Row::default();
+            let mut packer = row.packer();
+            for datum in datums {
+                packer.push(datum);
+            }
+            // Maps, including ones with disjoint key sets. Persist keeps
+            // per-key statistics for JSON maps, so a collection mixing maps
+            // where a key is present in one and absent in another exercises
+            // the absent-key handling in stats and their consumers.
+            packer.push_dict([("x", Datum::String("a"))]);
+            packer.push_dict([("y", Datum::String("b"))]);
+            packer.push_dict([("x", Datum::True), ("y", Datum::JsonNull)]);
+            packer.push_dict(std::iter::empty::<(&str, Datum)>());
+            // JSON map keys are not truncated in persist stats, unlike SQL
+            // string columns, so cover one past the string truncation limit.
+            let long_key = "k".repeat(101);
+            packer.push_dict([(long_key.as_str(), Datum::True)]);
+            packer.push_dict_with(|packer| {
+                packer.push(Datum::String("nested"));
+                packer.push_dict([("x", Datum::String("a"))]);
+            });
+            // Lists, including a heterogeneous one.
+            packer.push_list([Datum::True, Datum::JsonNull, Datum::String("a")]);
+            packer.push_list(std::iter::empty::<Datum>());
+            row
         });
         static UUID: LazyLock<Row> = LazyLock::new(|| {
             Row::pack_slice(&[