[SNAP-1034] Optimizations at Spark layer as seen in profiling (#10)

 - added a aggBufferAttributeForGroup to aggregates to be used to avoid nullable
   checks in generated code in aggregate buffers used in HashAggregateExec (if aggregate
       is on zero rows, then there will be no row in the map); accompanying "initialValuesForGroup"
   added for initial aggregation buffer values
 - use OpenHashMap in DictionaryEncoding which is faster than normal hash map;
   added clear methods to OpenHashMap/OpenHashSet for reuse
 - minor correction in the string in HiveUtils

core/src/main/scala/org/apache/spark/util/collection/OpenHashMap.scala

@@ -149,6 +149,23 @@ class OpenHashMap[K : ClassTag, @specialized(Long, Int, Double) V: ClassTag](
     }
   }
 
+  def clear() {
+    // first clear the values array and value for null key
+    val bitSet = _keySet.getBitSet
+    val nullV = null.asInstanceOf[V]
+    val values = _values
+    var pos = bitSet.nextSetBit(0)
+    while (pos >= 0) {
+      values(pos) = nullV
+      pos = bitSet.nextSetBit(pos + 1)
+    }
+    haveNullValue = false
+    nullValue = nullV
+    _oldValues = null
+    // next clear the key set
+    _keySet.clear()
+  }
+
   // The following member variables are declared as protected instead of private for the
   // specialization to work (specialized class extends the non-specialized one and needs access
   // to the "private" variables).

core/src/main/scala/org/apache/spark/util/collection/OpenHashSet.scala

@@ -212,6 +212,12 @@ class OpenHashSet[@specialized(Long, Int) T: ClassTag](
    */
   def nextPos(fromPos: Int): Int = _bitset.nextSetBit(fromPos)
 
+  def clear() {
+    _data = new Array[T](_capacity)
+    _bitset.clear()
+    _size = 0
+  }
+
   /**
    * Double the table's size and re-hash everything. We are not really using k, but it is declared
    * so Scala compiler can specialize this method (which leads to calling the specialized version

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Average.scala

@@ -53,10 +53,16 @@ case class Average(child: Expression) extends DeclarativeAggregate {
   }
 
   private lazy val sum = AttributeReference("sum", sumDataType)()
-  private lazy val count = AttributeReference("count", LongType)()
+  private lazy val count = AttributeReference("count", LongType, nullable = false)()
 
   override lazy val aggBufferAttributes = sum :: count :: Nil
 
+  override lazy val aggBufferAttributesForGroup: Seq[AttributeReference] = {
+    if (child.nullable) aggBufferAttributes
+    else sum.copy(nullable = false)(sum.exprId, sum.qualifier,
+      sum.isGenerated) :: count :: Nil
+  }
+
   override lazy val initialValues = Seq(
     /* sum = */ Cast(Literal(0), sumDataType),
     /* count = */ Literal(0L)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Sum.scala

@@ -53,10 +53,22 @@ case class Sum(child: Expression) extends DeclarativeAggregate {
 
   override lazy val aggBufferAttributes = sum :: Nil
 
+  override lazy val aggBufferAttributesForGroup: Seq[AttributeReference] = {
+    if (child.nullable) aggBufferAttributes
+    else sum.copy(nullable = false)(sum.exprId, sum.qualifier,
+      sum.isGenerated) :: Nil
+  }
+
   override lazy val initialValues: Seq[Expression] = Seq(
     /* sum = */ Literal.create(null, sumDataType)
   )
 
+  override lazy val initialValuesForGroup: Seq[Expression] = Seq(
+    /* sum = */
+    if (child.nullable) Literal.create(null, sumDataType)
+    else Cast(Literal(0), sumDataType)
+  )
+
   override lazy val updateExpressions: Seq[Expression] = {
     if (child.nullable) {
       Seq(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/interfaces.scala

@@ -184,6 +184,9 @@ sealed abstract class AggregateFunction extends Expression with ImplicitCastInpu
   /** Attributes of fields in aggBufferSchema. */
   def aggBufferAttributes: Seq[AttributeReference]
 
+  /** Attributes of fields in aggBufferSchema used for group by. */
+  def aggBufferAttributesForGroup: Seq[AttributeReference] = aggBufferAttributes
+
   /**
    * Attributes of fields in input aggregation buffers (immutable aggregation buffers that are
    * merged with mutable aggregation buffers in the merge() function or merge expressions).
@@ -367,6 +370,11 @@ abstract class DeclarativeAggregate
    */
   val initialValues: Seq[Expression]
 
+  /**
+   * Expressions for initializing empty aggregation buffers for group by.
+   */
+  def initialValuesForGroup: Seq[Expression] = initialValues
+
   /**
    * Expressions for updating the mutable aggregation buffer based on an input row.
    */

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

@@ -68,6 +68,10 @@ case class HashAggregateExec(
     aggregateExpressions.flatMap(_.aggregateFunction.aggBufferAttributes)
   }
 
+  @transient lazy private[this] val aggregateBufferAttributesForGroup = {
+    aggregateExpressions.flatMap(_.aggregateFunction.aggBufferAttributesForGroup)
+  }
+
   require(HashAggregateExec.supportsAggregate(aggregateBufferAttributes))
 
   override lazy val allAttributes: AttributeSeq =
@@ -297,7 +301,7 @@ case class HashAggregateExec(
   private val declFunctions = aggregateExpressions.map(_.aggregateFunction)
     .filter(_.isInstanceOf[DeclarativeAggregate])
     .map(_.asInstanceOf[DeclarativeAggregate])
-  private val bufferSchema = StructType.fromAttributes(aggregateBufferAttributes)
+  private val bufferSchema = StructType.fromAttributes(aggregateBufferAttributesForGroup)
 
   // The name for Vectorized HashMap
   private var vectorizedHashMapTerm: String = _
@@ -312,7 +316,7 @@ case class HashAggregateExec(
    */
   def createHashMap(): UnsafeFixedWidthAggregationMap = {
     // create initialized aggregate buffer
-    val initExpr = declFunctions.flatMap(f => f.initialValues)
+    val initExpr = declFunctions.flatMap(_.initialValuesForGroup)
     val initialBuffer = UnsafeProjection.create(initExpr)(EmptyRow)
 
     // create hashMap
@@ -368,7 +372,7 @@ case class HashAggregateExec(
       val mergeExpr = declFunctions.flatMap(_.mergeExpressions)
       val mergeProjection = newMutableProjection(
         mergeExpr,
-        aggregateBufferAttributes ++ declFunctions.flatMap(_.inputAggBufferAttributes),
+        aggregateBufferAttributesForGroup ++ declFunctions.flatMap(_.inputAggBufferAttributes),
         subexpressionEliminationEnabled)
       val joinedRow = new JoinedRow()
 
@@ -433,14 +437,14 @@ case class HashAggregateExec(
       }
       val evaluateKeyVars = evaluateVariables(keyVars)
       ctx.INPUT_ROW = bufferTerm
-      val bufferVars = aggregateBufferAttributes.zipWithIndex.map { case (e, i) =>
+      val bufferVars = aggregateBufferAttributesForGroup.zipWithIndex.map { case (e, i) =>
         BoundReference(i, e.dataType, e.nullable).genCode(ctx)
       }
       val evaluateBufferVars = evaluateVariables(bufferVars)
       // evaluate the aggregation result
       ctx.currentVars = bufferVars
       val aggResults = declFunctions.map(_.evaluateExpression).map { e =>
-        BindReferences.bindReference(e, aggregateBufferAttributes).genCode(ctx)
+        BindReferences.bindReference(e, aggregateBufferAttributesForGroup).genCode(ctx)
       }
       val evaluateAggResults = evaluateVariables(aggResults)
       // generate the final result
@@ -648,8 +652,8 @@ case class HashAggregateExec(
     ctx.currentVars = input
     val hashEval = BindReferences.bindReference(hashExpr, child.output).genCode(ctx)
 
-    val inputAttr = aggregateBufferAttributes ++ child.output
-    ctx.currentVars = new Array[ExprCode](aggregateBufferAttributes.length) ++ input
+    val inputAttr = aggregateBufferAttributesForGroup ++ child.output
+    ctx.currentVars = new Array[ExprCode](aggregateBufferAttributesForGroup.length) ++ input
 
     val (checkFallbackForGeneratedHashMap, checkFallbackForBytesToBytesMap, resetCounter,
     incCounter) = if (testFallbackStartsAt.isDefined) {

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/compression/compressionSchemes.scala

@@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{MutableRow, SpecificMutableRow}
 import org.apache.spark.sql.execution.columnar._
 import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
 
 
 private[columnar] case object PassThrough extends CompressionScheme {
@@ -208,7 +209,7 @@ private[columnar] case object DictionaryEncoding extends CompressionScheme {
     private var values = new mutable.ArrayBuffer[T#InternalType](1024)
 
     // The dictionary that maps a value to the encoded short integer.
-    private val dictionary = mutable.HashMap.empty[Any, Short]
+    private val dictionary = new OpenHashMap[Any, Short]
 
     // Size of the serialized dictionary in bytes. Initialized to 4 since we need at least an `Int`
     // to store dictionary element count.

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala

@@ -134,9 +134,9 @@ trait HashJoin {
       joinRow.withLeft(srow)
       val matches = hashedRelation.get(joinKeys(srow))
       if (matches != null) {
-        matches.map(joinRow.withRight(_)).filter(boundCondition)
+        matches.map(joinRow.withRight).filter(boundCondition)
       } else {
-        Seq.empty
+        Iterator.empty
       }
     }
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala

@@ -134,10 +134,17 @@ private[joins] class UnsafeHashedRelation(
   // re-used in get()/getValue()
   var resultRow = new UnsafeRow(numFields)
 
+  private var mapLoc = initMapLoc()
+
+  private def initMapLoc(): BytesToBytesMap#Location = {
+    val map = binaryMap
+    new map.Location
+  }
+
   override def get(key: InternalRow): Iterator[InternalRow] = {
     val unsafeKey = key.asInstanceOf[UnsafeRow]
     val map = binaryMap  // avoid the compiler error
-    val loc = new map.Location  // this could be allocated in stack
+    val loc = mapLoc
     binaryMap.safeLookup(unsafeKey.getBaseObject, unsafeKey.getBaseOffset,
       unsafeKey.getSizeInBytes, loc, unsafeKey.hashCode())
     if (loc.isDefined) {
@@ -243,6 +250,7 @@ private[joins] class UnsafeHashedRelation(
       taskMemoryManager,
       (nKeys * 1.5 + 1).toInt, // reduce hash collision
       pageSizeBytes)
+    mapLoc = initMapLoc()
 
     var i = 0
     var keyBuffer = new Array[Byte](1024)

sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLMetrics.scala

@@ -24,7 +24,8 @@ import org.apache.spark.scheduler.AccumulableInfo
 import org.apache.spark.util.{AccumulatorContext, AccumulatorV2, Utils}
 
 
-class SQLMetric(val metricType: String, initValue: Long = 0L) extends AccumulatorV2[Long, Long] {
+final class SQLMetric(val metricType: String, initValue: Long = 0L)
+    extends AccumulatorV2[Long, Long] {
   // This is a workaround for SPARK-11013.
   // We may use -1 as initial value of the accumulator, if the accumulator is valid, we will
   // update it at the end of task and the value will be at least 0. Then we can filter out the -1

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala

@@ -280,7 +280,7 @@ private[spark] object HiveUtils extends Logging {
         throw new IllegalArgumentException(
           "Builtin jars can only be used when hive execution version == hive metastore version. " +
             s"Execution: $hiveExecutionVersion != Metastore: $hiveMetastoreVersion. " +
-            "Specify a vaild path to the correct hive jars using $HIVE_METASTORE_JARS " +
+            s"Specify a vaild path to the correct hive jars using $HIVE_METASTORE_JARS " +
             s"or change ${HIVE_METASTORE_VERSION.key} to $hiveExecutionVersion.")
       }