fix(substrait): fix regressed edge case in renaming inner struct fields (apache#15634)

* add a failing test case for apache#15239 (comment)

* fix invariants to use logical schema equality instead, and mark utf8 and utf8view as logically equivalent

* fix logical equivalence to be strictly superset of semantic equivalence, including ignoring decimal precision/scale

Author: Blizzara

datafusion/common/src/dfschema.rs

@@ -641,7 +641,7 @@ impl DFSchema {
                         || (!DFSchema::datatype_is_semantically_equal(
                             f1.data_type(),
                             f2.data_type(),
-                        ) && !can_cast_types(f2.data_type(), f1.data_type()))
+                        ))
                     {
                         _plan_err!(
                             "Schema mismatch: Expected field '{}' with type {:?}, \
@@ -659,9 +659,12 @@ impl DFSchema {
     }
 
     /// Checks if two [`DataType`]s are logically equal. This is a notably weaker constraint
-    /// than datatype_is_semantically_equal in that a Dictionary<K,V> type is logically
-    /// equal to a plain V type, but not semantically equal. Dictionary<K1, V1> is also
-    /// logically equal to Dictionary<K2, V1>.
+    /// than datatype_is_semantically_equal in that different representations of same data can be
+    /// logically but not semantically equivalent. Semantically equivalent types are always also
+    /// logically equivalent. For example:
+    /// - a Dictionary<K,V> type is logically equal to a plain V type
+    /// - a Dictionary<K1, V1> is also logically equal to Dictionary<K2, V1>
+    /// - Utf8 and Utf8View are logically equal
     pub fn datatype_is_logically_equal(dt1: &DataType, dt2: &DataType) -> bool {
         // check nested fields
         match (dt1, dt2) {
@@ -711,12 +714,15 @@ impl DFSchema {
                         .zip(iter2)
                         .all(|((t1, f1), (t2, f2))| t1 == t2 && Self::field_is_logically_equal(f1, f2))
             }
-            _ => dt1 == dt2,
+            // Utf8 and Utf8View are logically equivalent
+            (DataType::Utf8, DataType::Utf8View) => true,
+            (DataType::Utf8View, DataType::Utf8) => true,
+            _ => Self::datatype_is_semantically_equal(dt1, dt2),
         }
     }
 
     /// Returns true of two [`DataType`]s are semantically equal (same
-    /// name and type), ignoring both metadata and nullability.
+    /// name and type), ignoring both metadata and nullability, and decimal precision/scale.
     ///
     /// request to upstream: <https://github.com/apache/arrow-rs/issues/3199>
     pub fn datatype_is_semantically_equal(dt1: &DataType, dt2: &DataType) -> bool {

datafusion/expr/src/logical_plan/invariants.rs

@@ -112,11 +112,11 @@ fn assert_valid_semantic_plan(plan: &LogicalPlan) -> Result<()> {
 /// Returns an error if the plan does not have the expected schema.
 /// Ignores metadata and nullability.
 pub fn assert_expected_schema(schema: &DFSchemaRef, plan: &LogicalPlan) -> Result<()> {
-    let compatible = plan.schema().has_equivalent_names_and_types(schema);
+    let compatible = plan.schema().logically_equivalent_names_and_types(schema);
 
-    if let Err(e) = compatible {
+    if !compatible {
         internal_err!(
-            "Failed due to a difference in schemas: {e}, original schema: {:?}, new schema: {:?}",
+            "Failed due to a difference in schemas: original schema: {:?}, new schema: {:?}",
             schema,
             plan.schema()
         )

datafusion/optimizer/src/optimizer.rs

@@ -506,8 +506,11 @@ mod tests {
         });
         let err = opt.optimize(plan, &config, &observe).unwrap_err();
 
-        // Simplify assert to check the error message contains the expected message, which is only the schema length mismatch
-        assert_contains!(err.strip_backtrace(), "Schema mismatch: the schema length are not same Expected schema length: 3, got: 0");
+        // Simplify assert to check the error message contains the expected message
+        assert_contains!(
+            err.strip_backtrace(),
+            "Failed due to a difference in schemas: original schema: DFSchema"
+        );
     }
 
     #[test]

datafusion/substrait/tests/cases/roundtrip_logical_plan.rs

@@ -1061,7 +1061,7 @@ async fn roundtrip_literal_list() -> Result<()> {
 async fn roundtrip_literal_struct() -> Result<()> {
     let plan = generate_plan_from_sql(
         "SELECT STRUCT(1, true, CAST(NULL AS STRING)) FROM data",
-        false,
+        true,
         true,
     )
     .await?;
@@ -1076,6 +1076,46 @@ async fn roundtrip_literal_struct() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn roundtrip_literal_named_struct() -> Result<()> {
+    let plan = generate_plan_from_sql(
+        "SELECT STRUCT(1 as int_field, true as boolean_field, CAST(NULL AS STRING) as string_field) FROM data",
+        true,
+        true,
+    )
+        .await?;
+
+    assert_snapshot!(
+    plan,
+    @r#"
+    Projection: Struct({int_field:1,boolean_field:true,string_field:}) AS named_struct(Utf8("int_field"),Int64(1),Utf8("boolean_field"),Boolean(true),Utf8("string_field"),NULL)
+      TableScan: data projection=[]
+    "#
+            );
+    Ok(())
+}
+
+#[tokio::test]
+async fn roundtrip_literal_renamed_struct() -> Result<()> {
+    // This test aims to hit a case where the struct column itself has the expected name, but its
+    // inner field needs to be renamed.
+    let plan = generate_plan_from_sql(
+        "SELECT CAST((STRUCT(1)) AS Struct<\"int_field\"Int>) AS 'Struct({c0:1})' FROM data",
+        true,
+        true,
+    )
+    .await?;
+
+    assert_snapshot!(
+    plan,
+    @r#"
+    Projection: Struct({int_field:1}) AS Struct({c0:1})
+      TableScan: data projection=[]
+    "#
+            );
+    Ok(())
+}
+
 #[tokio::test]
 async fn roundtrip_values() -> Result<()> {
     // TODO: would be nice to have a struct inside the LargeList, but arrow_cast doesn't support that currently