Elide same-type field-aware casts in CastExpr

Restore no-op cast canonicalization in
cast_with_target_field_and_options(...). Update
CastExpr::nullable() to align with runtime nullability,
while maintaining logical target field semantics via
return_field(). Adjust tests to ensure type-changing
casts preserve logical target field semantics,
and validate cast nullability against runtime semantics.

Author: kosiew

datafusion/physical-expr/src/expressions/cast.rs

@@ -243,14 +243,7 @@ impl PhysicalExpr for CastExpr {
     }
 
     fn nullable(&self, input_schema: &Schema) -> Result<bool> {
-        // A cast is nullable if **either** the child is nullable or the
-        // target field allows nulls.  This conservative rule prevents
-        // optimizers from assuming a non-null result when a null input could
-        // still propagate.  `return_field()` continues to expose the exact
-        // target metadata separately.
-        let child_nullable = self.expr.nullable(input_schema)?;
-        let target_nullable = self.resolved_target_field(input_schema)?.is_nullable();
-        Ok(child_nullable || target_nullable)
+        self.expr.nullable(input_schema)
     }
 
     fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
@@ -338,8 +331,11 @@ pub(crate) fn cast_with_target_field_and_options(
 ) -> Result<Arc<dyn PhysicalExpr>> {
     let expr_type = expr.data_type(input_schema)?;
     let cast_type = target_field.data_type();
-    let is_valid_cast = expr_type == *cast_type
-        || match (&expr_type, cast_type) {
+    if expr_type == *cast_type {
+        return Ok(Arc::clone(&expr));
+    }
+
+    let is_valid_cast = match (&expr_type, cast_type) {
             // Allow struct-to-struct casts that pass name-based compatibility validation.
             // This validation is applied at planning time (now) to fail fast, rather than
             // deferring errors to execution time. The name-based casting logic will be
@@ -954,16 +950,16 @@ mod tests {
 
     #[test]
     fn field_aware_cast_nullable_child_nonnullable_targets_nullable() -> Result<()> {
-        // child is non-nullable but the target field is marked nullable; the
-        // nullable() result should still be true because the field allows nulls.
+        // Runtime cast nullability follows the child expression, while
+        // return_field() preserves the logical target field nullability.
         let schema = Schema::new(vec![Field::new("a", Int32, false)]);
         let expr = CastExpr::new_with_target_field(
             col("a", &schema)?,
             Arc::new(Field::new("cast_target", Int64, true)),
             None,
         );
 
-        assert!(expr.nullable(&schema)?);
+        assert!(!expr.nullable(&schema)?);
         assert!(expr.return_field(&schema)?.is_nullable());
 
         Ok(())

datafusion/physical-expr/src/planner.rs

@@ -491,7 +491,7 @@ mod tests {
         let field = physical_expr.return_field(&schema)?;
         assert_eq!(field.data_type(), extension_field_type.data_type());
         assert_eq!(field.metadata(), extension_field_type.metadata());
-        assert!(physical_expr.nullable(&schema)?);
+        assert!(!physical_expr.nullable(&schema)?);
 
         let try_cast_expr = Expr::TryCast(TryCast::new_from_field(
             Box::new(expr.clone()),
@@ -536,13 +536,13 @@ mod tests {
             field.metadata().get("semantic").map(String::as_str),
             Some("target")
         );
-        assert!(physical_expr.nullable(&schema)?);
+        assert!(!physical_expr.nullable(&schema)?);
 
         Ok(())
     }
 
     #[test]
-    fn test_create_physical_expr_same_type_cast_keeps_target_field() -> Result<()> {
+    fn test_create_physical_expr_same_type_cast_is_elided() -> Result<()> {
         let schema = Schema::new(vec![Field::new("value", DataType::Int32, false)]);
         let df_schema = DFSchema::try_from(schema.clone())?;
         let target_field = Arc::new(
@@ -558,17 +558,9 @@ mod tests {
         let physical_expr =
             create_physical_expr(&expr, &df_schema, &ExecutionProps::new())?;
 
-        assert!(physical_expr.as_any().downcast_ref::<Column>().is_none());
-        assert!(physical_expr.as_any().downcast_ref::<CastExpr>().is_some());
-
-        let field = physical_expr.return_field(&schema)?;
-        assert_eq!(field.name(), "same_type_cast");
-        assert_eq!(field.data_type(), &DataType::Int32);
-        assert!(field.is_nullable());
-        assert_eq!(
-            field.metadata().get("semantic").map(String::as_str),
-            Some("same_type")
-        );
+        assert!(physical_expr.as_any().downcast_ref::<Column>().is_some());
+        assert!(physical_expr.as_any().downcast_ref::<CastExpr>().is_none());
+        assert_eq!(physical_expr.return_field(&schema)?.name(), "value");
 
         Ok(())
     }