fix(spark): remove internal functions MakeDecimal and UnscaledValue (#386)

These two functions are inserted by the catalyst optimizer for
queries that involve aggregation (sum & average) of decimal
values.

Approx 50% of the TPC-DS tests rely on these internal functions
which doesn’t make them interchangeable with other query processors.

This commit reverses this particular optimisation before
conversion to substrait, and removes MakeDecimal and
UnscaledValue from the `spark.yaml` file.

Author: andrew-coleman

spark/src/main/resources/spark.yml

@@ -15,25 +15,6 @@
 %YAML 1.2
 ---
 scalar_functions:
-  -
-    name: unscaled
-    description:  >-
-      Return the unscaled Long value of a Decimal, assuming it fits in a Long.
-      Note: this expression is internal and created only by the optimizer,
-      we don't need to do type check for it.
-    impls:
-      - args:
-          - value: DECIMAL<P,S>
-        return: i64
-  -
-    name: make_decimal
-    description:  >-
-      Return the Decimal value of an unscaled Long.
-      Note: this expression is internal and created only by the optimizer,
-    impls:
-      - args:
-          - value: i64
-        return: DECIMAL<P,S>
   - name: add
     description: >-
       Adds days to a date

spark/src/main/scala/io/substrait/spark/expression/FunctionMappings.scala

@@ -148,11 +148,7 @@ class FunctionMappings {
     ss[Year]("extract"),
     ss[Quarter]("extract"),
     ss[Month]("extract"),
-    ss[DayOfMonth]("extract"),
-
-    // internal
-    s[MakeDecimal]("make_decimal"),
-    s[UnscaledValue]("unscaled")
+    ss[DayOfMonth]("extract")
   )
 
   val AGGREGATE_SIGS: Seq[Sig] = Seq(

spark/src/main/scala/io/substrait/spark/expression/ToSparkExpression.scala

@@ -235,34 +235,22 @@ class ToSparkExpression(
         arg.accept(expr.declaration(), i, this)
     }.toList
 
-    expr.declaration.name match {
-      case "make_decimal" if expr.declaration.uri == SparkExtension.uri =>
-        expr.outputType match {
-          // Need special case handing of this internal function.
-          // Because the precision and scale arguments are extracted from the output type,
-          // we can't use the generic scalar function conversion mechanism here.
-          case d: Type.Decimal => MakeDecimal(args.head, d.precision, d.scale)
-          case _ =>
-            throw new IllegalArgumentException("Output type of MakeDecimal must be a decimal type")
-        }
-      case _ =>
-        scalarFunctionConverter
-          .getSparkExpressionFromSubstraitFunc(expr.declaration.key, args)
-          .getOrElse({
-            val msg = String.format(
-              "Unable to convert scalar function %s(%s).",
-              expr.declaration.name,
-              expr.arguments.asScala
-                .map {
-                  case ea: exp.EnumArg => ea.value.toString
-                  case e: SExpression => e.getType.accept(new StringTypeVisitor)
-                  case t: Type => t.accept(new StringTypeVisitor)
-                  case a => throw new IllegalStateException("Unexpected value: " + a)
-                }
-                .mkString(", ")
-            )
-            throw new IllegalArgumentException(msg)
-          })
-    }
+    scalarFunctionConverter
+      .getSparkExpressionFromSubstraitFunc(expr.declaration.key, args)
+      .getOrElse({
+        val msg = String.format(
+          "Unable to convert scalar function %s(%s).",
+          expr.declaration.name,
+          expr.arguments.asScala
+            .map {
+              case ea: exp.EnumArg => ea.value.toString
+              case e: SExpression => e.getType.accept(new StringTypeVisitor)
+              case t: Type => t.accept(new StringTypeVisitor)
+              case a => throw new IllegalStateException("Unexpected value: " + a)
+            }
+            .mkString(", ")
+        )
+        throw new IllegalArgumentException(msg)
+      })
   }
 }

spark/src/main/scala/io/substrait/spark/expression/ToSubstraitExpression.scala

@@ -20,6 +20,7 @@ import io.substrait.spark.{HasOutputStack, ToSubstraitType}
 
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Project}
+import org.apache.spark.sql.types.LongType
 import org.apache.spark.substrait.SparkTypeUtil
 
 import io.substrait.expression.{EnumArg, Expression => SExpression, ExpressionCreator, FieldReference, ImmutableEnumArg, ImmutableExpression}
@@ -172,6 +173,7 @@ abstract class ToSubstraitExpression extends HasOutputStack[Seq[Attribute]] {
         translateUp(child)
           .map(ExpressionCreator
             .cast(ToSubstraitType.apply(dataType, c.nullable), _, FailureBehavior.THROW_EXCEPTION))
+      case UnscaledValue(value) => translateUp(Cast(value, LongType))
       case c @ CheckOverflow(child, dataType, _) =>
         // CheckOverflow similar with cast
         translateUp(child)

spark/src/main/scala/io/substrait/spark/logical/ToSubstraitRel.scala

@@ -22,7 +22,7 @@ import io.substrait.spark.expression._
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Average, Sum}
 import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation}
@@ -133,7 +133,30 @@ class ToSubstraitRel extends AbstractLogicalPlanVisitor with Logging {
    */
   override def visitAggregate(agg: Aggregate): relation.Rel = {
     val input = visit(agg.child)
-    val actualResultExprs = agg.aggregateExpressions
+    val actualResultExprs = agg.aggregateExpressions.map {
+      // eliminate the internal MakeDecimal and UnscaledValue functions by undoing the spark optimisation:
+      // https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala#L2223
+      case Alias(expr, name) =>
+        Alias(
+          expr.transform {
+            case MakeDecimal(
+                  ae @ AggregateExpression(Sum(UnscaledValue(e), _), _, _, _, _),
+                  _,
+                  _,
+                  _) =>
+              ae.copy(aggregateFunction = Sum(e))
+            case Cast(
+                  Divide(ae @ AggregateExpression(Average(UnscaledValue(e), _), _, _, _, _), _, _),
+                  _,
+                  _,
+                  _) =>
+              ae.copy(aggregateFunction = Average(e))
+            case e => e
+          },
+          name
+        )()
+      case e => e
+    }
     val actualGroupExprs = agg.groupingExpressions
 
     val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
@@ -198,8 +221,7 @@ class ToSubstraitRel extends AbstractLogicalPlanVisitor with Logging {
   override def visitWindow(window: Window): relation.Rel = {
     val windowExpressions = window.windowExpressions.map {
       case w: WindowExpression => fromWindowCall(w, window.child.output)
-      case a: Alias if a.child.isInstanceOf[WindowExpression] =>
-        fromWindowCall(a.child.asInstanceOf[WindowExpression], window.child.output)
+      case Alias(w: WindowExpression, _) => fromWindowCall(w, window.child.output)
       case other =>
         throw new UnsupportedOperationException(s"Unsupported window expression: $other")
     }.asJava

spark/src/test/scala/io/substrait/spark/NumericSuite.scala

@@ -42,4 +42,9 @@ class NumericSuite extends SparkFunSuite with SharedSparkSession with SubstraitP
       "select floor(num), ceil(num), round(num, 0), round(num, 1) from (values (0.5), (-0.5)) as table(num)"
     )
   }
+
+  test("decimal aggregation") {
+    assertSqlSubstraitRelRoundTrip(
+      "select sum(num), avg(num)from (values (2.5), (-0.5)) as table(num)")
+  }
 }