Persist pushdown hardening

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

@@ -1977,6 +1977,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 +2667,28 @@ 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")
+    }
+
+    /// `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();
+    }
+
+    #[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/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/repr/src/relation.rs

@@ -1369,6 +1369,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.
@@ -2028,6 +2041,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,9 @@ 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.
+                Datum::Time(NaiveTime::from_hms_micro_opt(23, 59, 59, 1_999_999).unwrap()),
             ])
         });
         static TIMESTAMP: LazyLock<Row> = LazyLock::new(|| {
@@ -4225,6 +4237,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 +4286,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(&[