fix: if field contains space in constraint expression, the check will fail

Signed-off-by: Alexander Falk <[email protected]>

Author: Nordalf

crates/core/src/delta_datafusion/mod.rs

@@ -1415,15 +1415,16 @@ impl DeltaDataChecker {
             return Ok(());
         }
         let table = MemTable::try_new(record_batch.schema(), vec![vec![record_batch.clone()]])?;
-
+        let schema = table.schema();
         // Use a random table name to avoid clashes when running multiple parallel tasks, e.g. when using a partitioned table
         let table_name: String = uuid::Uuid::new_v4().to_string();
         self.ctx.register_table(&table_name, Arc::new(table))?;
 
         let mut violations: Vec<String> = Vec::new();
 
         for check in checks {
-            if check.get_name().contains('.') {
+            let check_name = check.get_name();
+            if check_name.contains('.') {
                 return Err(DeltaTableError::Generic(
                     "Support for nested columns is not supported.".to_string(),
                 ));
@@ -1432,12 +1433,23 @@ impl DeltaDataChecker {
             let field_to_select = if check.as_any().is::<Constraint>() {
                 "*"
             } else {
-                check.get_name()
+                check_name
             };
+
+            // Loop through schema to find the matching field. If the field has a whitespace, we
+            // need to backtick it, since the expression is an unquoted string
+            let mut expression = check.get_expression().to_string();
+            for field in schema.fields() {
+                if expression.contains(field.name()) {
+                    expression =
+                        expression.replace(field.name(), format!("`{}` ", field.name()).as_str());
+                    break;
+                }
+            }
+
             let sql = format!(
                 "SELECT {} FROM `{table_name}` WHERE NOT ({}) LIMIT 1",
-                field_to_select,
-                check.get_expression()
+                field_to_select, expression
             );
 
             let dfs: Vec<RecordBatch> = self.ctx.sql(&sql).await?.collect().await?;