feat: add StatsVerifier and commit-time stats validation

- Add StatsVerifier with verify() and verify_detailed() methods
- Add StatsValidationVisitor using RowVisitor pattern
- Integrate validation into commit() flow
- Add add_files_validated() and validate_add_files_stats() helpers
- Unit tests for verifier functionality

Author: DrakeLin

kernel/src/engine/default/stats.rs

@@ -1118,3 +1118,182 @@ mod tests {
         assert_eq!(value_null_count.value(0), 1);
     }
 }
+
+/// Verifies that collected statistics match the expected schema.
+/// Used for debugging and testing stats collection.
+pub(crate) struct StatsVerifier;
+
+impl StatsVerifier {
+    /// Verify stats and return a structured result.
+    #[allow(unused)]
+    pub(crate) fn verify(
+        stats: &StructArray,
+        expected_columns: &[String],
+    ) -> DeltaResult<StatsVerificationResult> {
+        use crate::arrow::array::Array;
+
+        let fields = stats.fields();
+        let field_names: Vec<&str> = fields.iter().map(|f| f.name().as_str()).collect();
+
+        // Check numRecords
+        let num_records = if field_names.contains(&"numRecords") {
+            let num_records_idx = fields
+                .iter()
+                .position(|f| f.name() == "numRecords")
+                .unwrap();
+            let num_records_array = stats.column(num_records_idx);
+            if let Some(int_array) = num_records_array.as_any().downcast_ref::<Int64Array>() {
+                int_array.value(0)
+            } else {
+                0
+            }
+        } else {
+            0
+        };
+
+        // Check tightBounds
+        let tight_bounds = if field_names.contains(&"tightBounds") {
+            let idx = fields
+                .iter()
+                .position(|f| f.name() == "tightBounds")
+                .unwrap();
+            let array = stats.column(idx);
+            if let Some(bool_array) = array.as_any().downcast_ref::<BooleanArray>() {
+                bool_array.value(0)
+            } else {
+                false
+            }
+        } else {
+            false
+        };
+
+        // Check nullCount columns
+        let mut present_null_count = Vec::new();
+        let mut missing_null_count = Vec::new();
+
+        if let Some(idx) = fields.iter().position(|f| f.name() == "nullCount") {
+            let null_count_array = stats.column(idx);
+            if let Some(null_struct) = null_count_array.as_any().downcast_ref::<StructArray>() {
+                let null_fields: Vec<&str> = null_struct
+                    .fields()
+                    .iter()
+                    .map(|f| f.name().as_str())
+                    .collect();
+                for col in expected_columns {
+                    if null_fields.contains(&col.as_str()) {
+                        present_null_count.push(col.clone());
+                    } else {
+                        missing_null_count.push(col.clone());
+                    }
+                }
+            }
+        }
+
+        // Get min/max columns
+        let mut min_max_columns = Vec::new();
+        if let Some(idx) = fields.iter().position(|f| f.name() == "minValues") {
+            let min_array = stats.column(idx);
+            if let Some(min_struct) = min_array.as_any().downcast_ref::<StructArray>() {
+                for field in min_struct.fields() {
+                    min_max_columns.push(field.name().clone());
+                }
+            }
+        }
+
+        Ok(StatsVerificationResult {
+            num_records,
+            tight_bounds,
+            present_null_count_columns: present_null_count,
+            missing_null_count_columns: missing_null_count,
+            min_max_columns,
+        })
+    }
+
+    /// Verify stats and return a detailed human-readable string.
+    #[allow(unused)]
+    pub(crate) fn verify_detailed(
+        stats: &StructArray,
+        expected_columns: &[String],
+    ) -> DeltaResult<String> {
+        let result = Self::verify(stats, expected_columns)?;
+        Ok(format!(
+            "Stats: numRecords={}, tightBounds={}, nullCount=[{}], minMax=[{}]",
+            result.num_records,
+            result.tight_bounds,
+            result.present_null_count_columns.join(", "),
+            result.min_max_columns.join(", ")
+        ))
+    }
+}
+
+/// Result of stats verification.
+#[allow(unused)]
+pub(crate) struct StatsVerificationResult {
+    pub num_records: i64,
+    pub tight_bounds: bool,
+    pub present_null_count_columns: Vec<String>,
+    pub missing_null_count_columns: Vec<String>,
+    pub min_max_columns: Vec<String>,
+}
+
+impl StatsVerificationResult {
+    /// Returns true if all expected columns have nullCount stats.
+    pub fn has_all_null_counts(&self) -> bool {
+        self.missing_null_count_columns.is_empty()
+    }
+}
+
+#[cfg(test)]
+mod verifier_tests {
+    use super::*;
+    use crate::arrow::datatypes::Schema;
+
+    #[test]
+    fn test_stats_verifier_valid_stats() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int64, false),
+            Field::new("value", DataType::Utf8, true),
+        ]));
+
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Int64Array::from(vec![1, 2, 3])),
+                Arc::new(StringArray::from(vec![Some("a"), None, Some("c")])),
+            ],
+        )
+        .unwrap();
+
+        let mut collector =
+            StatisticsCollector::new(schema, &["id".to_string(), "value".to_string()]);
+        collector.update(&batch, None).unwrap();
+        let stats = collector.finalize().unwrap();
+
+        let result =
+            StatsVerifier::verify(&stats, &["id".to_string(), "value".to_string()]).unwrap();
+
+        assert_eq!(result.num_records, 3);
+        assert!(result.tight_bounds);
+        assert!(result.has_all_null_counts());
+        assert_eq!(result.present_null_count_columns.len(), 2);
+    }
+
+    #[test]
+    fn test_stats_verifier_detailed_output() {
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
+
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int64Array::from(vec![1, 2, 3]))],
+        )
+        .unwrap();
+
+        let mut collector = StatisticsCollector::new(schema, &["id".to_string()]);
+        collector.update(&batch, None).unwrap();
+        let stats = collector.finalize().unwrap();
+
+        let detailed = StatsVerifier::verify_detailed(&stats, &["id".to_string()]).unwrap();
+        assert!(detailed.contains("numRecords=3"));
+        assert!(detailed.contains("tightBounds=true"));
+    }
+}

kernel/src/transaction/mod.rs

@@ -41,6 +41,65 @@ use crate::{
 };
 use delta_kernel_derive::internal_api;
 
+/// Visitor to validate statistics in add file metadata.
+/// Uses RowVisitor pattern to extract and validate stats from EngineData.
+struct StatsValidationVisitor {
+    rows_validated: usize,
+    rows_with_num_records: usize,
+    errors: Vec<String>,
+}
+
+impl StatsValidationVisitor {
+    fn new() -> Self {
+        Self {
+            rows_validated: 0,
+            rows_with_num_records: 0,
+            errors: Vec::new(),
+        }
+    }
+
+    fn validate(&self) -> DeltaResult<()> {
+        if self.rows_validated == 0 {
+            return Err(Error::generic("No rows to validate"));
+        }
+        if self.rows_with_num_records == 0 {
+            // This is a warning case, not an error - stats might be missing
+            // but we don't fail the commit for it
+        }
+        if !self.errors.is_empty() {
+            return Err(Error::generic(format!(
+                "Stats validation errors: {}",
+                self.errors.join("; ")
+            )));
+        }
+        Ok(())
+    }
+}
+
+impl RowVisitor for StatsValidationVisitor {
+    fn selected_column_names_and_types(&self) -> (&'static [ColumnName], &'static [DataType]) {
+        static NAMES: LazyLock<Vec<ColumnName>> = LazyLock::new(|| vec![column_name!("stats")]);
+        static TYPES: LazyLock<Vec<DataType>> = LazyLock::new(|| {
+            vec![DataType::STRING] // stats is a JSON string
+        });
+        (NAMES.as_slice(), TYPES.as_slice())
+    }
+
+    fn visit<'a>(&mut self, row_count: usize, getters: &[&'a dyn GetData<'a>]) -> DeltaResult<()> {
+        for row_index in 0..row_count {
+            self.rows_validated += 1;
+            if let Some(stats_str) = getters[0].get_opt(row_index, "stats")? {
+                let stats_str: String = stats_str;
+                // Check if stats has numRecords
+                if stats_str.contains("\"numRecords\"") {
+                    self.rows_with_num_records += 1;
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
 /// Type alias for an iterator of [`EngineData`] results.
 pub(crate) type EngineDataResultIterator<'a> =
     Box<dyn Iterator<Item = DeltaResult<Box<dyn EngineData>>> + Send + 'a>;
@@ -333,6 +392,14 @@ impl Transaction {
     ///   transaction in case of a conflict so the user can retry, etc.)
     /// - Err(Error) indicates a non-retryable error (e.g. logic/validation error).
     pub fn commit(self, engine: &dyn Engine) -> DeltaResult<CommitResult> {
+        // Step 0: Validate stats in add file metadata
+        // This ensures all files have valid stats before committing
+        for add_metadata in &self.add_files_metadata {
+            let mut validator = StatsValidationVisitor::new();
+            validator.visit_rows_of(add_metadata.as_ref())?;
+            validator.validate()?;
+        }
+
         // Step 1: Check for duplicate app_ids and generate set transactions (`txn`)
         // Note: The commit info must always be the first action in the commit but we generate it in
         // step 2 to fail early on duplicate transaction appIds
@@ -903,6 +970,37 @@ impl Transaction {
         self.add_files_metadata.push(add_metadata);
     }
 
+    /// Add files with statistics validation.
+    ///
+    /// Similar to [`add_files`], but validates that the metadata contains valid statistics
+    /// before adding. Returns an error if validation fails.
+    ///
+    /// [`add_files`]: Transaction::add_files
+    #[allow(unused)]
+    pub fn add_files_validated(&mut self, add_metadata: Box<dyn EngineData>) -> DeltaResult<()> {
+        let mut validator = StatsValidationVisitor::new();
+        validator.visit_rows_of(add_metadata.as_ref())?;
+        validator.validate()?;
+        self.add_files_metadata.push(add_metadata);
+        Ok(())
+    }
+
+    /// Validate statistics in add file metadata without modifying the transaction.
+    ///
+    /// Returns a summary string describing the validation results.
+    #[allow(unused)]
+    pub fn validate_add_files_stats(add_metadata: &dyn EngineData) -> DeltaResult<String> {
+        let mut validator = StatsValidationVisitor::new();
+        validator.visit_rows_of(add_metadata)?;
+        validator.validate()?;
+        Ok(format!(
+            "Validated {} rows, {} had numRecords. Errors: {}",
+            validator.rows_validated,
+            validator.rows_with_num_records,
+            validator.errors.join("; ")
+        ))
+    }
+
     /// Generate add actions, handling row tracking internally if needed
     fn generate_adds<'a>(
         &'a self,