use only for partial aggregation

Author: sameeragarwal

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

@@ -265,7 +265,8 @@ case class TungstenAggregate(
 
   // The name for Vectorized HashMap
   private var vectorizedHashMapTerm: String = _
-  private var isVectorizedHashMapEnabled: Boolean = sqlContext.conf.columnarAggregateMapEnabled
+  private var isVectorizedHashMapEnabled: Boolean = sqlContext.conf.columnarAggregateMapEnabled &&
+    (modes.contains(Partial) || modes.contains(PartialMerge))
 
   // The name for UnsafeRow HashMap
   private var hashMapTerm: String = _

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

@@ -437,11 +437,12 @@ object SQLConf {
     .stringConf
     .createOptional
 
+  // TODO: This is still WIP and shouldn't be turned on without extensive test coverage
   val COLUMNAR_AGGREGATE_MAP_ENABLED = SQLConfigBuilder("spark.sql.codegen.aggregate.map.enabled")
     .internal()
     .doc("When true, aggregate with keys use an in-memory columnar map to speed up execution.")
     .booleanConf
-    .createWithDefault(true)
+    .createWithDefault(false)
 
   object Deprecated {
     val MAPRED_REDUCE_TASKS = "mapred.reduce.tasks"