fix!: incorrect coercion when comparing with string literals

datafusion/core/tests/expr_api/mod.rs

@@ -330,12 +330,12 @@ async fn test_create_physical_expr_coercion() {
     create_expr_test(lit(1i32).eq(col("id")), "CAST(1 AS Utf8) = id@0");
     // compare int col to string literal `i = '202410'`
     // Note this casts the column (not the field)
-    create_expr_test(col("i").eq(lit("202410")), "CAST(i@1 AS Utf8) = 202410");
-    create_expr_test(lit("202410").eq(col("i")), "202410 = CAST(i@1 AS Utf8)");
+    create_expr_test(col("i").eq(lit("202410")), "i@1 = 202410");
+    create_expr_test(lit("202410").eq(col("i")), "202410 = i@1");
     // however, when simplified the casts on i should removed
     // https://github.com/apache/datafusion/issues/14944
-    create_simplified_expr_test(col("i").eq(lit("202410")), "CAST(i@1 AS Utf8) = 202410");
-    create_simplified_expr_test(lit("202410").eq(col("i")), "CAST(i@1 AS Utf8) = 202410");
+    create_simplified_expr_test(col("i").eq(lit("202410")), "i@1 = 202410");
+    create_simplified_expr_test(lit("202410").eq(col("i")), "i@1 = 202410");
 }
 
 /// Evaluates the specified expr as an aggregate and compares the result to the

datafusion/optimizer/src/analyzer/type_coercion.rs

@@ -290,12 +290,31 @@ impl<'a> TypeCoercionRewriter<'a> {
         right: Expr,
         right_schema: &DFSchema,
     ) -> Result<(Expr, Expr)> {
-        let (left_type, right_type) = BinaryTypeCoercer::new(
-            &left.get_type(left_schema)?,
-            &op,
-            &right.get_type(right_schema)?,
-        )
-        .get_input_types()?;
+        let left_type = left.get_type(left_schema)?;
+        let right_type = right.get_type(right_schema)?;
+
+        match (&left, &right) {
+            (left, Expr::Literal(ref lit_value)) => {
+                if let Some(casted) =
+                    try_cast_literal_to_type_with_op(&left_type, op, lit_value)?
+                {
+                    return Ok((left.clone(), Expr::Literal(casted)));
+                }
+            }
+            (Expr::Literal(ref lit_value), right) if op.swap().is_some() => {
+                if let Some(casted) = try_cast_literal_to_type_with_op(
+                    &right_type,
+                    op.swap().unwrap(),
+                    lit_value,
+                )? {
+                    return Ok((Expr::Literal(casted), right.clone()));
+                }
+            }
+            _ => {}
+        }
+
+        let (left_type, right_type) =
+            BinaryTypeCoercer::new(&left_type, &op, &right_type).get_input_types()?;
 
         Ok((
             left.cast_to(&left_type, left_schema)?,
@@ -304,6 +323,33 @@ impl<'a> TypeCoercionRewriter<'a> {
     }
 }
 
+fn try_cast_literal_to_type_with_op(
+    target_type: &DataType,
+    op: Operator,
+    lit_value: &ScalarValue,
+) -> Result<Option<ScalarValue>> {
+    match (op, lit_value) {
+        (
+            op,
+            ScalarValue::Utf8(_) | ScalarValue::Utf8View(_) | ScalarValue::LargeUtf8(_),
+        ) if op.supports_propagation() => {
+            if target_type.is_numeric() {
+                let Ok(casted) = lit_value.cast_to(target_type) else {
+                    return plan_err!(
+                        "Cannot coerce '{}' to type '{}'",
+                        lit_value,
+                        target_type
+                    );
+                };
+                return Ok(Some(casted));
+            }
+
+            Ok(None)
+        }
+        _ => Ok(None),
+    }
+}
+
 impl TreeNodeRewriter for TypeCoercionRewriter<'_> {
     type Node = Expr;
 

datafusion/sqllogictest/test_files/push_down_filter.slt

@@ -230,19 +230,19 @@ logical_plan TableScan: t projection=[a], full_filters=[t.a != Int32(100)]
 query TT
 explain select a from t where a = '99999999999';
 ----
-logical_plan TableScan: t projection=[a], full_filters=[CAST(t.a AS Utf8) = Utf8("99999999999")]
+
 
 # The predicate should still have the column cast when the value is a NOT valid i32
 query TT
 explain select a from t where a = '99.99';
 ----
-logical_plan TableScan: t projection=[a], full_filters=[CAST(t.a AS Utf8) = Utf8("99.99")]
+
 
 # The predicate should still have the column cast when the value is a NOT valid i32
 query TT
 explain select a from t where a = '';
 ----
-logical_plan TableScan: t projection=[a], full_filters=[CAST(t.a AS Utf8) = Utf8("")]
+
 
 # The predicate should not have a column cast when the operator is = or != and the literal can be round-trip casted without losing information.
 query TT