[SPARK-52461] [SQL] Collapse inner Cast from DecimalType to DecimalType

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecisionTypeCoercion.scala

@@ -138,11 +138,28 @@ object DecimalPrecisionTypeCoercion extends SQLConfHelper {
     case b @ BinaryComparison(e1 @ DecimalExpression(p1, s1), e2 @ DecimalExpression(p2, s2))
         if p1 != p2 || s1 != s2 =>
       val resultType = widerDecimalType(p1, s1, p2, s2)
-      val newE1 = if (e1.dataType == resultType) e1 else Cast(e1, resultType)
-      val newE2 = if (e2.dataType == resultType) e2 else Cast(e2, resultType)
+      val newE1 = if (e1.dataType == resultType) e1 else tryCollapseInnerCast(e1, resultType)
+      val newE2 = if (e2.dataType == resultType) e2 else tryCollapseInnerCast(e2, resultType)
       b.withNewChildren(Seq(newE1, newE2))
   }
 
+  /**
+   * In case `SQLConf.COLLAPSE_INNER_CAST_TO_DECIMAL` is true and inner cast is redundant (if the
+   * inner one is not a `USER_SPECIFIED_CAST` has lower precision and scale than the outer one),
+   * remove it.
+   */
+  private def tryCollapseInnerCast(expression: Expression, decimalType: DecimalType): Cast = {
+    expression match {
+      case innerCast @ Cast(_, innerDecimalType: DecimalType, _, _)
+          if conf.getConf(SQLConf.COLLAPSE_INNER_CAST_TO_DECIMAL) &&
+          decimalType.precision >= innerDecimalType.precision &&
+          decimalType.scale >= innerDecimalType.scale &&
+          innerCast.getTagValue(Cast.USER_SPECIFIED_CAST).isEmpty =>
+        Cast(innerCast.child, decimalType)
+      case other => Cast(other, decimalType)
+    }
+  }
+
   /**
    * Type coercion for BinaryOperator in which one side is a non-decimal numeric, and the other
    * side is a decimal.

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -5949,6 +5949,17 @@ object SQLConf {
       .createWithDefault(2)
   }
 
+  val COLLAPSE_INNER_CAST_TO_DECIMAL =
+    buildConf("spark.sql.collapseInnerCastToDecimal.enabled")
+      .internal()
+      .doc(
+        "If true, collapse inner cast to DecimalType if it is redundant (if the inner one has " +
+        "lower precision and scale than the outer one)."
+      )
+      .version("4.1.0")
+      .booleanConf
+      .createWithDefault(true)
+
   /**
    * Holds information about keys that have been deprecated.
    *

sql/core/src/test/resources/sql-tests/analyzer-results/group-by-all-mosha.sql.out

@@ -91,7 +91,7 @@ SELECT s AS s, COUNT(*) c FROM stuff GROUP BY ALL HAVING SUM(f) > 0 ORDER BY s
 -- !query analysis
 Sort [s#x ASC NULLS FIRST], true
 +- Project [s#x, c#xL]
-   +- Filter (sum(f)#x > cast(cast(0 as decimal(1,0)) as decimal(16,4)))
+   +- Filter (sum(f)#x > cast(0 as decimal(16,4)))
       +- Aggregate [s#x], [s#x AS s#x, count(1) AS c#xL, sum(f#x) AS sum(f)#x]
          +- SubqueryAlias stuff
             +- View (`stuff`, [i#x, f#x, s#x, t#x, d#x, a#x])