Merge pull request #2 from sdf-jkl/TLP

Add TLP Where Oracle

Author: 2010YOUY01

datafusion-fuzzer.toml

@@ -31,7 +31,7 @@ max_expr_level = 3
 max_table_count = 3
 max_insert_per_table = 20
 
-# Supported oracles: NoCrash, NestedQueries.
+# Supported oracles: NoCrash, NestedQueries, TlpWhere.
 # Randomly select one oracle from the configured set for each query.
 oracles = ["NoCrash"]
-# oracles = ["NoCrash", "NestedQueries"]
+# oracles = ["NoCrash", "NestedQueries", "TlpWhere"]

src/common/util.rs

@@ -1,9 +1,95 @@
-use datafusion::{prelude::Expr, sql::unparser::expr_to_sql};
+use datafusion::{
+    arrow::array::RecordBatch, common::utils::get_row_at_idx, prelude::Expr, scalar::ScalarValue,
+    sql::unparser::expr_to_sql,
+};
+use std::collections::HashMap;
 
-use super::Result;
+use super::{Result, fuzzer_err};
 
 /// Convert a DataFusion `Expr` into a SQL string using DataFusion's unparser.
 pub fn to_sql_string(expr: &Expr) -> Result<String> {
     let unparsed = expr_to_sql(expr)?;
     Ok(unparsed.to_string())
 }
+
+pub(crate) type RowMultiset = HashMap<Vec<ScalarValue>, usize>;
+
+pub(crate) fn batches_to_row_multiset(batches: &[RecordBatch]) -> Result<RowMultiset> {
+    let mut multiset: RowMultiset = HashMap::new();
+    let mut expected_num_cols: Option<usize> = None;
+
+    for batch in batches {
+        if let Some(expected) = expected_num_cols {
+            if batch.num_columns() != expected {
+                return Err(fuzzer_err(&format!(
+                    "Mismatched column count across batches: expected {}, got {}",
+                    expected,
+                    batch.num_columns()
+                )));
+            }
+        } else {
+            expected_num_cols = Some(batch.num_columns());
+        }
+
+        for row_idx in 0..batch.num_rows() {
+            let mut row_key = get_row_at_idx(batch.columns(), row_idx)
+                .map_err(|e| fuzzer_err(&format!("Failed to extract row {}: {}", row_idx, e)))?;
+            row_key.iter_mut().for_each(|v| *v = v.clone().compacted());
+            *multiset.entry(row_key).or_insert(0) += 1;
+        }
+    }
+
+    Ok(multiset)
+}
+
+pub(crate) fn format_row_multiset_diff(left: &RowMultiset, right: &RowMultiset) -> String {
+    let mut lines = Vec::new();
+    for (row, left_count) in left {
+        let right_count = right.get(row).copied().unwrap_or(0);
+        if *left_count != right_count {
+            lines.push(format!(
+                "row={:?}, left_count={}, right_count={}",
+                row, left_count, right_count
+            ));
+        }
+    }
+    for (row, right_count) in right {
+        if !left.contains_key(row) {
+            lines.push(format!(
+                "row={:?}, left_count=0, right_count={}",
+                row, right_count
+            ));
+        }
+    }
+
+    lines.sort();
+    let preview = lines.into_iter().take(20).collect::<Vec<_>>();
+    if preview.is_empty() {
+        "no row differences".to_string()
+    } else {
+        preview.join("\n")
+    }
+}
+
+pub(crate) fn count_total_rows(batches: &[RecordBatch]) -> usize {
+    batches.iter().map(RecordBatch::num_rows).sum()
+}
+
+pub(crate) fn validate_batches_value_equivalence(
+    left_batches: &[RecordBatch],
+    right_batches: &[RecordBatch],
+    oracle_name: &str,
+) -> Result<()> {
+    let left_multiset = batches_to_row_multiset(left_batches)?;
+    let right_multiset = batches_to_row_multiset(right_batches)?;
+
+    if left_multiset != right_multiset {
+        return Err(fuzzer_err(&format!(
+            "{} value equivalence violated:\n{}",
+            oracle_name,
+            format_row_multiset_diff(&left_multiset, &right_multiset)
+        )));
+    }
+
+    Ok(())
+}

src/fuzz_context/runner_config.rs

@@ -169,14 +169,18 @@ max_row_count = 100
 max_expr_level = 3
 max_table_count = 3
 max_insert_per_table = 20
-oracles = ["NoCrash", "NestedQueries"]
+oracles = ["NoCrash", "NestedQueries", "TlpWhere"]
 "#,
         )
         .unwrap();
 
         assert_eq!(
             config.oracles,
-            vec![ConfiguredOracle::NoCrash, ConfiguredOracle::NestedQueries]
+            vec![
+                ConfiguredOracle::NoCrash,
+                ConfiguredOracle::NestedQueries,
+                ConfiguredOracle::TlpWhere
+            ]
         );
     }
 

src/oracle/mod.rs

@@ -1,7 +1,9 @@
 // Oracle module - provides testing oracles for query consistency and correctness
 
+pub(crate) mod oracle_common;
 pub mod oracle_impl_nested_queries;
 pub mod oracle_impl_no_crash;
+pub mod oracle_impl_tlp_where;
 pub mod oracle_trait;
 
 use std::sync::Arc;
@@ -13,6 +15,7 @@ use crate::fuzz_context::GlobalContext;
 // Re-export main types and traits
 pub use oracle_impl_nested_queries::NestedQueriesOracle;
 pub use oracle_impl_no_crash::NoCrashOracle;
+pub use oracle_impl_tlp_where::TlpWhereOracle;
 pub use oracle_trait::{Oracle, QueryContext, QueryExecutionResult};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -21,13 +24,16 @@ pub enum ConfiguredOracle {
     NoCrash,
     #[serde(rename = "NestedQueries", alias = "NestedQueriesOracle")]
     NestedQueries,
+    #[serde(rename = "TlpWhere", alias = "TlpWhereOracle")]
+    TlpWhere,
 }
 
 impl ConfiguredOracle {
     pub fn build(self, seed: u64, ctx: Arc<GlobalContext>) -> Box<dyn Oracle + Send> {
         match self {
             Self::NoCrash => Box::new(NoCrashOracle::new(seed, ctx)),
             Self::NestedQueries => Box::new(NestedQueriesOracle::new(seed, ctx)),
+            Self::TlpWhere => Box::new(TlpWhereOracle::new(seed, ctx)),
         }
     }
 }

src/oracle/oracle_common.rs

@@ -0,0 +1,27 @@
+use crate::common::{Result, fuzzer_err, util};
+use crate::oracle::QueryExecutionResult;
+
+pub(crate) fn validate_value_equivalence(
+    results: &[QueryExecutionResult],
+    left_idx: usize,
+    right_idx: usize,
+    oracle_name: &str,
+) -> Result<()> {
+    let left_result = results
+        .get(left_idx)
+        .ok_or_else(|| fuzzer_err(&format!("Missing result at index {}", left_idx)))?;
+    let right_result = results
+        .get(right_idx)
+        .ok_or_else(|| fuzzer_err(&format!("Missing result at index {}", right_idx)))?;
+
+    let left_batches = left_result
+        .result
+        .as_ref()
+        .map_err(|e| fuzzer_err(&e.to_string()))?;
+    let right_batches = right_result
+        .result
+        .as_ref()
+        .map_err(|e| fuzzer_err(&e.to_string()))?;
+
+    util::validate_batches_value_equivalence(left_batches, right_batches, oracle_name)
+}