fix: validate snapshot hint and CRC invariants

kernel/src/crc/file_size_histogram.rs

@@ -115,41 +115,47 @@ impl FileSizeHistogram {
         file_counts: Vec<i64>,
         total_bytes: Vec<i64>,
     ) -> DeltaResult<Self> {
+        let histogram = Self {
+            sorted_bin_boundaries,
+            file_counts,
+            total_bytes,
+        };
+        histogram.check_shape()?;
+        Ok(histogram)
+    }
+
+    fn check_shape(&self) -> DeltaResult<()> {
         require!(
-            sorted_bin_boundaries.len() >= 2,
+            self.sorted_bin_boundaries.len() >= 2,
             Error::internal_error(format!(
                 "sorted_bin_boundaries must have at least 2 elements, got {}",
-                sorted_bin_boundaries.len()
+                self.sorted_bin_boundaries.len()
             ))
         );
         require!(
-            sorted_bin_boundaries[0] == 0,
+            self.sorted_bin_boundaries[0] == 0,
             Error::internal_error(format!(
                 "First boundary must be 0, got {}",
-                sorted_bin_boundaries[0]
+                self.sorted_bin_boundaries[0]
             ))
         );
         require!(
-            sorted_bin_boundaries.len() == file_counts.len()
-                && sorted_bin_boundaries.len() == total_bytes.len(),
+            self.sorted_bin_boundaries.len() == self.file_counts.len()
+                && self.sorted_bin_boundaries.len() == self.total_bytes.len(),
             Error::internal_error(format!(
                 "All arrays must have the same length: boundaries={}, file_counts={}, total_bytes={}",
-                sorted_bin_boundaries.len(),
-                file_counts.len(),
-                total_bytes.len()
+                self.sorted_bin_boundaries.len(),
+                self.file_counts.len(),
+                self.total_bytes.len()
             ))
         );
         require!(
-            sorted_bin_boundaries.windows(2).all(|w| w[0] < w[1]),
+            self.sorted_bin_boundaries.windows(2).all(|w| w[0] < w[1]),
             Error::internal_error(
                 "sorted_bin_boundaries must be sorted in strictly ascending order"
             )
         );
-        Ok(Self {
-            sorted_bin_boundaries,
-            file_counts,
-            total_bytes,
-        })
+        Ok(())
     }
 
     /// Creates an empty histogram with the given bin boundaries and zero counts/bytes.
@@ -273,6 +279,64 @@ impl FileSizeHistogram {
         }
         Ok(self)
     }
+
+    /// Validates an absolute histogram against its complete file statistics.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the histogram shape is invalid, a bin has impossible aggregate
+    /// statistics for its bounds, a total overflows, or the totals differ from `num_files` and
+    /// `table_size_bytes`.
+    pub(crate) fn validate_complete(
+        &self,
+        num_files: i64,
+        table_size_bytes: i64,
+    ) -> DeltaResult<()> {
+        self.check_shape()?;
+        for i in 0..self.sorted_bin_boundaries.len() {
+            let count = i128::from(self.file_counts[i]);
+            let bytes = i128::from(self.total_bytes[i]);
+            let lower = i128::from(self.sorted_bin_boundaries[i]);
+            let upper = self
+                .sorted_bin_boundaries
+                .get(i + 1)
+                .copied()
+                .map(i128::from);
+            let aggregates_are_possible = (count == 0 && bytes == 0)
+                || (count > 0
+                    && bytes >= lower * count
+                    && upper.is_none_or(|upper| bytes < upper * count));
+            require!(
+                aggregates_are_possible,
+                Error::internal_error(format!(
+                    "Histogram bin {i} has count {count} and total bytes {bytes}, which are \
+                     inconsistent with its bounds"
+                ))
+            );
+        }
+
+        let file_count = self.file_counts.iter().try_fold(0_i64, |sum, count| {
+            sum.checked_add(*count)
+                .ok_or_else(|| Error::internal_error("Histogram file count overflow"))
+        })?;
+        let total_bytes = self.total_bytes.iter().try_fold(0_i64, |sum, bytes| {
+            sum.checked_add(*bytes)
+                .ok_or_else(|| Error::internal_error("Histogram total bytes overflow"))
+        })?;
+        require!(
+            file_count == num_files,
+            Error::internal_error(format!(
+                "Histogram file count {file_count} does not match numFiles {num_files}"
+            ))
+        );
+        require!(
+            total_bytes == table_size_bytes,
+            Error::internal_error(format!(
+                "Histogram total bytes {total_bytes} does not match tableSizeBytes {table_size_bytes}"
+            ))
+        );
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -311,6 +375,73 @@ mod tests {
         assert_eq!(hist.total_bytes, vec![200, 900]);
     }
 
+    #[rstest]
+    #[case::empty_bin_has_bytes(vec![0, 10], vec![0, 0], vec![1, 0], 0, 1)]
+    #[case::below_lower_bound(vec![0, 10], vec![0, 1], vec![0, 9], 1, 9)]
+    #[case::at_exclusive_upper_bound(vec![0, 10], vec![1, 0], vec![10, 0], 1, 10)]
+    fn validate_complete_rejects_impossible_bin_aggregates(
+        #[case] boundaries: Vec<i64>,
+        #[case] file_counts: Vec<i64>,
+        #[case] total_bytes: Vec<i64>,
+        #[case] num_files: i64,
+        #[case] table_size_bytes: i64,
+    ) {
+        let histogram = FileSizeHistogram::try_new(boundaries, file_counts, total_bytes).unwrap();
+        assert_result_error_with_message(
+            histogram.validate_complete(num_files, table_size_bytes),
+            "inconsistent with its bounds",
+        );
+    }
+
+    #[rstest]
+    #[case::empty(vec![0, 10], vec![0, 0], vec![0, 0], 0, 0)]
+    #[case::exclusive_upper_bound(vec![0, 10], vec![2, 0], vec![18, 0], 2, 18)]
+    #[case::last_bin_unbounded(vec![0, 10], vec![0, 1], vec![0, i64::MAX], 1, i64::MAX)]
+    fn validate_complete_accepts_achievable_bin_aggregates(
+        #[case] boundaries: Vec<i64>,
+        #[case] file_counts: Vec<i64>,
+        #[case] total_bytes: Vec<i64>,
+        #[case] num_files: i64,
+        #[case] table_size_bytes: i64,
+    ) {
+        let histogram = FileSizeHistogram::try_new(boundaries, file_counts, total_bytes).unwrap();
+        histogram
+            .validate_complete(num_files, table_size_bytes)
+            .unwrap();
+    }
+
+    #[rstest]
+    #[case::file_count(
+        vec![0, 10],
+        vec![i64::MAX, 1],
+        vec![0, 10],
+        i64::MAX,
+        10,
+        "file count overflow"
+    )]
+    #[case::total_bytes(
+        vec![0, i64::MAX],
+        vec![1, 1],
+        vec![i64::MAX - 1, i64::MAX],
+        2,
+        i64::MAX,
+        "total bytes overflow"
+    )]
+    fn validate_complete_rejects_overflowing_totals(
+        #[case] boundaries: Vec<i64>,
+        #[case] file_counts: Vec<i64>,
+        #[case] total_bytes: Vec<i64>,
+        #[case] num_files: i64,
+        #[case] table_size_bytes: i64,
+        #[case] expected: &str,
+    ) {
+        let histogram = FileSizeHistogram::try_new(boundaries, file_counts, total_bytes).unwrap();
+        assert_result_error_with_message(
+            histogram.validate_complete(num_files, table_size_bytes),
+            expected,
+        );
+    }
+
     #[rstest]
     #[case::empty_boundaries(vec![], vec![], vec![], "at least 2 elements")]
     #[case::single_boundary(vec![0], vec![0], vec![0], "at least 2 elements")]

kernel/src/crc/mod.rs

@@ -204,6 +204,11 @@ impl Crc {
                 )));
             }
         }
+        if let Some(histogram) = raw.file_size_histogram.as_ref() {
+            histogram
+                .validate_complete(raw.num_files, raw.table_size_bytes)
+                .map_err(|error| Error::generic(error.to_string()))?;
+        }
         // A CRC file on disk is by definition complete; we never deserialize a degraded state.
         let file_stats_state = FileStatsState::Complete(FileStats {
             num_files: raw.num_files,
@@ -252,6 +257,11 @@ impl TryFrom<&Crc> for CrcRaw {
                 crc.file_stats_state
             )));
         };
+        if let Some(histogram) = stats.file_size_histogram.as_ref() {
+            histogram
+                .validate_complete(stats.num_files, stats.table_size_bytes)
+                .map_err(|error| Error::ChecksumWriteUnsupported(error.to_string()))?;
+        }
         Ok(CrcRaw {
             table_size_bytes: stats.table_size_bytes,
             num_files: stats.num_files,
@@ -340,6 +350,7 @@ mod tests {
     use std::collections::HashMap;
 
     use rstest::rstest;
+    use test_utils::assert_result_error_with_message;
 
     use super::{Crc, CrcRaw, DomainMetadataState, FileStats, FileStatsState, SetTransactionState};
     use crate::actions::{DomainMetadata, Protocol, SetTransaction};
@@ -779,6 +790,29 @@ mod tests {
         );
     }
 
+    #[rstest]
+    #[case::below_lower_bound(vec![0, 10], vec![0, 1], vec![0, 9], 1, 9)]
+    #[case::at_exclusive_upper_bound(vec![0, 10], vec![1, 0], vec![10, 0], 1, 10)]
+    fn de_impossible_file_size_histogram_bin_is_rejected(
+        #[case] boundaries: Vec<i64>,
+        #[case] file_counts: Vec<i64>,
+        #[case] total_bytes: Vec<i64>,
+        #[case] num_files: i64,
+        #[case] table_size_bytes: i64,
+    ) {
+        let mut crc: serde_json::Value =
+            serde_json::from_str(&crc_json_with_counts(table_size_bytes, num_files, 1, 1)).unwrap();
+        crc["fileSizeHistogram"] = serde_json::json!({
+            "sortedBinBoundaries": boundaries,
+            "fileCounts": file_counts,
+            "totalBytes": total_bytes,
+        });
+        assert_result_error_with_message(
+            Crc::try_from_json_bytes(crc.to_string().as_bytes(), 0),
+            "inconsistent with its bounds",
+        );
+    }
+
     // ===== protocol validation on the CRC deserialization path =====
 
     /// Minimal CRC JSON whose `protocol` is the supplied fragment. Proves CRC deserialization
@@ -853,11 +887,16 @@ mod tests {
     /// Minimal CRC JSON with a file size histogram field spliced in under the given field name
     /// (`fileSizeHistogram` per the Delta spec, or `histogramOpt` for legacy Delta-Spark
     /// compatibility).
-    fn crc_json_with_histogram(field_name: &str, histogram_json: &str) -> String {
+    fn crc_json_with_histogram(
+        field_name: &str,
+        histogram_json: &str,
+        table_size_bytes: i64,
+        num_files: i64,
+    ) -> String {
         format!(
             r#"{{
-                "tableSizeBytes": 0,
-                "numFiles": 0,
+                "tableSizeBytes": {table_size_bytes},
+                "numFiles": {num_files},
                 "numMetadata": 1,
                 "numProtocol": 1,
                 "metadata": {{
@@ -881,7 +920,9 @@ mod tests {
     fn de_valid_file_size_histogram_succeeds(#[case] field_name: &str) {
         let json = crc_json_with_histogram(
             field_name,
-            r#"{"sortedBinBoundaries": [0, 100, 200], "fileCounts": [1, 2, 3], "totalBytes": [10, 200, 300]}"#,
+            r#"{"sortedBinBoundaries": [0, 100, 200], "fileCounts": [1, 2, 3], "totalBytes": [10, 250, 900]}"#,
+            1160,
+            6,
         );
         let crc = Crc::try_from_json_bytes(json.as_bytes(), 0).unwrap();
         assert!(crc.file_stats().unwrap().file_size_histogram().is_some());
@@ -891,7 +932,7 @@ mod tests {
     #[case::spec_name("fileSizeHistogram")]
     #[case::legacy_name("histogramOpt")]
     fn de_null_file_size_histogram_deserializes_to_none(#[case] field_name: &str) {
-        let json = crc_json_with_histogram(field_name, "null");
+        let json = crc_json_with_histogram(field_name, "null", 0, 0);
         let crc = Crc::try_from_json_bytes(json.as_bytes(), 0).unwrap();
         assert!(crc.file_stats().unwrap().file_size_histogram().is_none());
     }
@@ -915,7 +956,7 @@ mod tests {
         #[case] histogram_json: &str,
         #[values("fileSizeHistogram", "histogramOpt")] field_name: &str,
     ) {
-        let json = crc_json_with_histogram(field_name, histogram_json);
+        let json = crc_json_with_histogram(field_name, histogram_json, 0, 0);
         assert!(Crc::try_from_json_bytes(json.as_bytes(), 0).is_err());
     }
 
@@ -927,6 +968,8 @@ mod tests {
         let legacy_json = crc_json_with_histogram(
             "histogramOpt",
             r#"{"sortedBinBoundaries": [0, 100], "fileCounts": [1, 0], "totalBytes": [50, 0]}"#,
+            50,
+            1,
         );
         let crc = Crc::try_from_json_bytes(legacy_json.as_bytes(), 0).unwrap();
 

kernel/src/crc/state.rs

@@ -45,9 +45,8 @@ impl FileStatsState {
         }
     }
 
-    /// Returns `true` if file stats are known-correct absolute totals. Also gates whether
-    /// the CRC is safe to write to disk: only `Complete` CRCs have well-defined on-disk
-    /// representations.
+    /// Returns `true` if file stats are known-correct absolute totals.
+    #[cfg(any(test, feature = "test-utils"))]
     pub fn is_complete(&self) -> bool {
         matches!(self, Self::Complete(_))
     }
@@ -60,8 +59,8 @@ impl FileStatsState {
     }
 }
 
-// TODO(#2568): make `Default` test-only. `Crc::default()` produces a Complete-zero CRC
-//              that passes `is_complete()` and could be silently written.
+// TODO(#2568): make `Default` test-only. `Crc::default()` produces a Complete-zero CRC that could
+//              be silently written.
 impl Default for FileStatsState {
     fn default() -> Self {
         Self::Complete(FileStats::default())