feat: warn on NULL equality predicates (#22948)

## Which issue does this PR close?

- Closes #14434.

## Rationale for this change

SQL comparisons such as `expr = NULL` and `expr <> NULL` evaluate to
`NULL` under SQL three-valued logic, not to `true` or `false`. When
those comparisons appear in predicate contexts such as `WHERE`, `JOIN
ON`, or `HAVING`, they are almost always a user mistake where `IS NULL`
or `IS NOT NULL` was intended.

This PR emits non-fatal diagnostic warnings for those cases so callers
that surface `Diagnostic` information can help users find and fix the
query without changing planning success or query semantics.

## What changes are included in this PR?

- Adds warning collection to `SqlToRel`, with
`SqlToRel::take_warnings()` to drain non-fatal planning diagnostics
after planning.
- Adds predicate-scoped detection for `= NULL` and `<> NULL`
comparisons.
- Wires detection into `WHERE`, `JOIN ON`, and `HAVING` planning paths.
- Recursively checks predicate expression structure such as nested `AND`
/ `OR` binary predicates and `CASE WHEN` conditions, without warning for
projection-only expressions like `SELECT col = NULL`.
- Creates `Diagnostic::new_warning` entries with a primary span on the
comparison expression and help pointing at the `NULL` literal when spans
are available.

## Are these changes tested?

Yes. Added regression coverage in
`datafusion/sql/tests/cases/diagnostic.rs` for:

- `WHERE col = NULL`
- `WHERE NULL = col`
- `WHERE col <> NULL`
- `JOIN ... ON col = NULL`
- `HAVING ... = NULL`
- nested `CASE WHEN col = NULL` inside a predicate
- no warning for `IS NULL`
- no warning for projection-only `SELECT col = NULL`
- multiple warnings in one predicate

Validation run locally:

```shell
cargo fmt --all
cargo test -p datafusion-sql --test sql_integration diagnostic
cargo test -p datafusion-sql
cargo clippy -p datafusion-sql --all-targets --all-features -- -D warnings
cargo clippy --all-targets --all-features -- -D warnings
git diff main..HEAD --check
```

## Are there any user-facing changes?

Yes, but non-breaking. SQL planning can now collect warning diagnostics
for likely mistaken `NULL` equality predicates. Consumers can retrieve
them with `SqlToRel::take_warnings()` and decide how to present them.
Planning still succeeds and query semantics are unchanged.

Author: ametel01

datafusion/sql/src/expr/mod.rs

@@ -15,20 +15,23 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::ops::ControlFlow;
+
 use arrow::datatypes::{DataType, TimeUnit};
 use datafusion_expr::planner::{
     PlannerResult, RawBinaryExpr, RawDictionaryExpr, RawFieldAccessExpr,
 };
 use sqlparser::ast::{
     AccessExpr, BinaryOperator, CastFormat, CastKind, CeilFloorKind,
     DataType as SQLDataType, DateTimeField, DictionaryField, Expr as SQLExpr,
-    ExprWithAlias as SQLExprWithAlias, JsonPath, MapEntry, StructField, Subscript,
-    TrimWhereField, TypedString, Value, ValueWithSpan,
+    ExprWithAlias as SQLExprWithAlias, JsonPath, MapEntry, Spanned, StructField,
+    Subscript, TrimWhereField, TypedString, Value, ValueWithSpan,
 };
+use sqlparser::ast::{Query, Visit, Visitor};
 
 use datafusion_common::{
-    DFSchema, Result, ScalarValue, internal_datafusion_err, internal_err, not_impl_err,
-    plan_err,
+    DFSchema, Diagnostic, Result, ScalarValue, Span, internal_datafusion_err,
+    internal_err, not_impl_err, plan_err,
 };
 
 use datafusion_expr::expr::ScalarFunction;
@@ -54,7 +57,86 @@ mod substring;
 mod unary_op;
 mod value;
 
+fn null_value_span(expr: &SQLExpr) -> Option<Option<Span>> {
+    if let SQLExpr::Value(ValueWithSpan {
+        value: Value::Null,
+        span,
+    }) = expr
+    {
+        Some(Span::try_from_sqlparser_span(*span))
+    } else {
+        None
+    }
+}
+
+fn null_equality_warning(expr: &SQLExpr) -> Option<Diagnostic> {
+    let SQLExpr::BinaryOp { left, op, right } = expr else {
+        return None;
+    };
+
+    let null_span = null_value_span(left).or_else(|| null_value_span(right))?;
+
+    let (message, help) = match op {
+        BinaryOperator::Eq => (
+            "comparison with NULL using `=` always evaluates to NULL",
+            "use `IS NULL` to check for NULL values",
+        ),
+        BinaryOperator::NotEq => (
+            "comparison with NULL using `<>` always evaluates to NULL",
+            "use `IS NOT NULL` to check for non-NULL values",
+        ),
+        _ => return None,
+    };
+
+    Some(
+        Diagnostic::new_warning(message, Span::try_from_sqlparser_span(expr.span()))
+            .with_help(help, null_span),
+    )
+}
+
+struct NullEqualityPredicateVisitor<'a, 'b, S: ContextProvider> {
+    sql_to_rel: &'a SqlToRel<'b, S>,
+    subquery_depth: usize,
+}
+
+impl<'a, 'b, S: ContextProvider> NullEqualityPredicateVisitor<'a, 'b, S> {
+    fn new(sql_to_rel: &'a SqlToRel<'b, S>) -> Self {
+        Self {
+            sql_to_rel,
+            subquery_depth: 0,
+        }
+    }
+}
+
+impl<S: ContextProvider> Visitor for NullEqualityPredicateVisitor<'_, '_, S> {
+    type Break = ();
+
+    fn pre_visit_query(&mut self, _query: &Query) -> ControlFlow<Self::Break> {
+        self.subquery_depth += 1;
+        ControlFlow::Continue(())
+    }
+
+    fn post_visit_query(&mut self, _query: &Query) -> ControlFlow<Self::Break> {
+        self.subquery_depth -= 1;
+        ControlFlow::Continue(())
+    }
+
+    fn pre_visit_expr(&mut self, expr: &SQLExpr) -> ControlFlow<Self::Break> {
+        if self.subquery_depth == 0
+            && let Some(warning) = null_equality_warning(expr)
+        {
+            self.sql_to_rel.add_warning(warning);
+        }
+        ControlFlow::Continue(())
+    }
+}
+
 impl<S: ContextProvider> SqlToRel<'_, S> {
+    pub(crate) fn warn_on_null_equality_predicate(&self, predicate: &SQLExpr) {
+        let mut visitor = NullEqualityPredicateVisitor::new(self);
+        let _ = predicate.visit(&mut visitor);
+    }
+
     pub(crate) fn sql_expr_to_logical_expr_with_alias(
         &self,
         sql: SQLExprWithAlias,

datafusion/sql/src/planner.rs

@@ -18,7 +18,7 @@
 //! [`SqlToRel`]: SQL Query Planner (produces [`LogicalPlan`] from SQL AST)
 use std::collections::HashMap;
 use std::str::FromStr;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::vec;
 
 use crate::utils::make_decimal_type;
@@ -455,6 +455,7 @@ pub struct SqlToRel<'a, S: ContextProvider> {
     pub(crate) context_provider: &'a S,
     pub(crate) options: ParserOptions,
     pub(crate) ident_normalizer: IdentNormalizer,
+    warnings: Mutex<Vec<Diagnostic>>,
 }
 
 impl<'a, S: ContextProvider> SqlToRel<'a, S> {
@@ -477,9 +478,27 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
             context_provider,
             options,
             ident_normalizer: IdentNormalizer::new(ident_normalize),
+            warnings: Mutex::new(vec![]),
         }
     }
 
+    pub(crate) fn add_warning(&self, warning: Diagnostic) {
+        self.warnings
+            .lock()
+            .expect("warning diagnostic lock poisoned")
+            .push(warning);
+    }
+
+    /// Drain and return non-fatal warnings collected during SQL planning.
+    pub fn take_warnings(&self) -> Vec<Diagnostic> {
+        std::mem::take(
+            &mut self
+                .warnings
+                .lock()
+                .expect("warning diagnostic lock poisoned"),
+        )
+    }
+
     pub fn build_schema(&self, columns: Vec<SQLColumnDef>) -> Result<Schema> {
         let mut fields = Vec::with_capacity(columns.len());
 

datafusion/sql/src/relation/join.rs

@@ -122,6 +122,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
             JoinConstraint::On(sql_expr) => {
                 let join_schema = left.schema().join(right.schema())?;
                 // parse ON expression
+                self.warn_on_null_equality_predicate(&sql_expr);
                 let expr = self.sql_to_expr(sql_expr, &join_schema, planner_context)?;
                 LogicalPlanBuilder::from(left)
                     .join_on(right, join_type, Some(expr))?

datafusion/sql/src/select.rs

@@ -198,6 +198,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         let having_expr_opt = select
             .having
             .map::<Result<Expr>, _>(|having_expr| {
+                self.warn_on_null_equality_predicate(&having_expr);
                 let having_expr = self.sql_expr_to_logical_expr(
                     having_expr,
                     &combined_schema,
@@ -865,6 +866,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
             Some(predicate_expr) => {
                 let fallback_schemas = plan.fallback_normalize_schemas();
 
+                self.warn_on_null_equality_predicate(&predicate_expr);
                 let filter_expr =
                     self.sql_to_expr(predicate_expr, plan.schema(), planner_context)?;