Add explicit tests for aggregate hash maps and other minor fixs.

Author: ooq

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala

@@ -288,7 +288,7 @@ case class HashAggregateExec(
   private var isVectorizedHashMapEnabled: Boolean = false
   private var isRowBasedHashMapEnabled: Boolean = false
   // auxiliary flag, true if any of two above is true
-  private var isFastHashMapEnabled: Boolean = false
+  private var isFastHashMapEnabled: Boolean = isVectorizedHashMapEnabled || isRowBasedHashMapEnabled
 
   // The name for UnsafeRow HashMap
   private var hashMapTerm: String = _
@@ -517,18 +517,18 @@ case class HashAggregateExec(
     sqlContext.conf.enforceFastAggHashMapImpl match {
       case "rowbased" =>
         if (!enableRowBasedHashMap(ctx)) {
-          if (modes.forall(mode => mode == Partial || mode == PartialMerge)) {
-            logWarning("spark.sql.codegen.aggregate.map.enforce.impl is set to rowbased, but "
+          if (modes.forall(mode => mode == Partial || mode == PartialMerge) && !Utils.isTesting) {
+            logWarning("spark.sql.codegen.aggregate.map.enforce.impl is set to rowbased, but"
               + " current version of codegened row-based hashmap does not support this aggregate.")
           }
         } else {
           isRowBasedHashMapEnabled = true
         }
       case "vectorized" =>
         if (!enableVectorizedHashMap(ctx)) {
-          if (modes.forall(mode => mode == Partial || mode == PartialMerge)) {
-            logWarning("spark.sql.codegen.aggregate.map.enforce.impl is set to vectorized, but "
-              + " current version of codegened row-based hashmap does not support this aggregate.")
+          if (modes.forall(mode => mode == Partial || mode == PartialMerge) && !Utils.isTesting) {
+            logWarning("spark.sql.codegen.aggregate.map.enforce.impl is set to vectorized, but"
+              + " current version of codegened vectorized hashmap does not support this aggregate.")
           }
         } else {
           isVectorizedHashMapEnabled = true

sql/core/src/test/scala/org/apache/spark/sql/AggregateHashMapSuite.scala

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types.DecimalType
+
+abstract class AggregateHashMapSuite extends DataFrameAggregateSuite {
+  import testImplicits._
+
+  protected def setAggregateHashMapImpl(): Unit
+
+  protected override def beforeAll(): Unit = {
+      setAggregateHashMapImpl()
+      sparkConf.set("spark.sql.codegen.fallback", "false")
+      super.beforeAll()
+  }
+
+  test("SQL decimal test") {
+    checkAnswer(
+      decimalData.groupBy('a cast DecimalType(10, 2)).agg(avg('b cast DecimalType(10, 2))),
+      Seq(Row(new java.math.BigDecimal(1.0), new java.math.BigDecimal(1.5)),
+        Row(new java.math.BigDecimal(2.0), new java.math.BigDecimal(1.5)),
+        Row(new java.math.BigDecimal(3.0), new java.math.BigDecimal(1.5))))
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/RowBasedAggregateHashMapSuite.scala

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+class RowBasedAggregateHashMapSuite extends AggregateHashMapSuite {
+
+  protected def setAggregateHashMapImpl() {
+    sparkConf.set("spark.sql.codegen.aggregate.map.enforce.impl", "rowbased")
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/VectorizedAggregateHashMapSuite.scala

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+class VectorizedAggregateHashMapSuite extends AggregateHashMapSuite {
+
+  protected def setAggregateHashMapImpl() {
+    sparkConf.set("spark.sql.codegen.aggregate.map.enforce.impl", "vectorized")
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestData.scala

@@ -21,6 +21,7 @@ import java.nio.charset.StandardCharsets
 
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{DataFrame, SparkSession, SQLContext, SQLImplicits}
+import org.apache.spark.sql.types.Decimal
 
 /**
  * A collection of sample data used in SQL tests.