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

* add a failing test case for #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