<NME> ReduceSinkOperator.scala
<BEF> /*
 * Copyright (C) 2012 The Regents of The University California.
 * All rights reserved.
 *
 * Licensed 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 shark.execution

import java.util.{ArrayList, Arrays, List => JList, Random}

import scala.collection.Iterator
import scala.collection.JavaConversions._
import scala.reflect.BeanProperty

import org.apache.hadoop.hive.ql.exec.{ExprNodeEvaluator, ExprNodeEvaluatorFactory}
import org.apache.hadoop.hive.ql.plan.ReduceSinkDesc
import org.apache.hadoop.hive.serde2.SerDe
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils
import org.apache.hadoop.hive.serde2.objectinspector.StandardUnionObjectInspector.StandardUnion
import org.apache.hadoop.io.BytesWritable


/**
 * Converts a collection of rows into key, value pairs. This is the
 * upstream operator for joins and groupbys.
 */
class ReduceSinkOperator extends UnaryOperator[ReduceSinkDesc] {

  @BeanProperty var conf: ReduceSinkDesc = _


    // Note that we get input object inspector from hiveOp rather than Shark's
    // objectInspector because initializeMasterOnAll() hasn't been invoked yet.
    val vars = ReduceSinkOperator.initializeOisAndSers(conf, hiveOp.getInputObjInspectors().head)
    keyEval = vars._1
    valueEval = vars._2
    keySer = vars._3
    valueSer = vars._4
    keyObjInspector = vars._5
    valObjInspector = vars._6

    // Determine output object inspector (a struct of KEY, VALUE).
    val ois = new ArrayList[ObjectInspector]

  // The evaluator for the partition columns (used in CLUSTER BY and
  // DISTRIBUTE BY in Hive). Partition columns decide the reducer that the
  // current row goes to. Partition columns are not passed to reducers.
  @transient var partitionEval: Array[ExprNodeEvaluator] = _

  @transient var keySer: SerDe = _
  @transient var valueSer: SerDe = _
  @transient var keyObjInspector: ObjectInspector = _
  @transient var keyFieldObjInspectors: Array[ObjectInspector] = _
  @transient var valObjInspector: ObjectInspector = _
  @transient var valFieldObjInspectors: Array[ObjectInspector] = _
  @transient var partitionObjInspectors: Array[ObjectInspector] = _

  override def getTag() = conf.getTag()

  override def initializeOnMaster() {
    super.initializeOnMaster()
    
    conf = desc
  }

  override def initializeOnSlave() {
    }
  }

  override def initializeOnMaster() {
    conf = hiveOp.getConf()
  }

  override def initializeOnSlave() {
    val vars = ReduceSinkOperator.initializeOisAndSers(conf, objectInspector)
    keyEval = vars._1
    valueEval = vars._2
    keySer = vars._3
    valueSer = vars._4
    keyObjInspector = vars._5
    valObjInspector = vars._6
    partitionEval = vars._7
    partitionObjInspectors = vars._8
  }

  override def processPartition[T](iter: Iterator[T]) = {
      distinctColIndices,
      conf.getOutputKeyColumnNames,
      numDistributionKeys,
      rowInspector)

    // Initialize object inspector for value columns.
    valFieldObjInspectors = valueEval.map(eval => eval.initialize(rowInspector))
    valObjInspector = ObjectInspectorFactory.getStandardStructObjectInspector(
      conf.getOutputValueColumnNames(), valFieldObjInspectors.toList)

    // Initialize evaluator and object inspector for partition columns.
    partitionEval = conf.getPartitionCols.map(ExprNodeEvaluatorFactory.get(_)).toArray
    partitionObjInspectors = partitionEval.map(_.initialize(rowInspector)).toArray
  }

  /**
   * Process a partition when there is NO distinct key aggregations.
   */
  def processPartitionNoDistinct(iter: Iterator[_]) = {
    val numDistributionKeys = conf.getNumDistributionKeys
    // Buffer for key, value evaluation to avoid frequent object allocation.
    val evaluatedKey = new Array[Object](keyEval.length)
    val evaluatedValue = new Array[Object](valueEval.length)
    val reduceKey = new ReduceKeyMapSide

    // A random number generator, used to distribute keys evenly if there is
    // no partition columns specified. Use a constant seed to make the code
    // deterministic as did Hive.
    val rand = new Random(13)

    iter.map { row =>
      // TODO: we don't need partition code for group-by or join

      // Determine the partition code (Hive calls it keyHashCode), used for
      // DISTRIBUTED BY / CLUSTER BY.
      var partitionCode = 0
      if (partitionEval.length == 0) {
        // If there is no partition columns, we randomly distribute the data
        partitionCode = rand.nextInt()
      } else {
        // With partition columns, determine a partitionCode to distribute.
        var i = 0
        while (i < partitionEval.length) {
          val o = partitionEval(i).evaluate(row)
          partitionCode = partitionCode * 31 +
            ObjectInspectorUtils.hashCode(o, partitionObjInspectors(i))
          i += 1
        }
      }

      // Evaluate the key columns.
      var i = 0
      while (i < numDistributionKeys) {
        evaluatedKey(i) = keyEval(i).evaluate(row)
        i += 1
      }

      // Evaluate the value columns.
      i = 0
      while (i < valueEval.length) {
        evaluatedValue(i) = valueEval(i).evaluate(row)
        i += 1
      }

      val key = keySer.serialize(evaluatedKey, keyObjInspector).asInstanceOf[BytesWritable]
      val value = valueSer.serialize(evaluatedValue, valObjInspector).asInstanceOf[BytesWritable]

      reduceKey.bytesWritable = key
      reduceKey.partitionCode = partitionCode
      (reduceKey, value)
    }
    }
  }
}


object ReduceSinkOperator {

  def initializeOisAndSers(conf: ReduceSinkDesc, rowInspector: ObjectInspector) = {
    val keyEval = conf.getKeyCols.map(ExprNodeEvaluatorFactory.get(_)).toArray
    val numDistributionKeys = conf.getNumDistributionKeys()
    val distinctColIndices = conf.getDistinctColumnIndices()
    val numDistinctExprs = distinctColIndices.size()
    val valueEval = conf.getValueCols.map(ExprNodeEvaluatorFactory.get(_)).toArray

    val keyTableDesc = conf.getKeySerializeInfo()
    val keySer = keyTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]
    keySer.initialize(null, keyTableDesc.getProperties())
    val keyIsText = keySer.getSerializedClass().equals(classOf[Text])

    val valueTableDesc = conf.getValueSerializeInfo()
    val valueSer = valueTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]
    valueSer.initialize(null, valueTableDesc.getProperties())
    
    val keyObjInspector = SharkEnv.objectInspectorLock.synchronized {
      ReduceSinkOperatorHelper.initEvaluatorsAndReturnStruct(
        keyEval,
        distinctColIndices,
        conf.getOutputKeyColumnNames,
        numDistributionKeys,
        rowInspector)
    }
    val valFieldInspectors = valueEval.map(eval => eval.initialize(rowInspector)).toList
    val valObjInspector = SharkEnv.objectInspectorLock.synchronized {
      ObjectInspectorFactory.getStandardStructObjectInspector(
        conf.getOutputValueColumnNames(), valFieldInspectors)
    }

    val partitionEval = conf.getPartitionCols.map(ExprNodeEvaluatorFactory.get(_)).toArray
    val partitionObjInspectors = partitionEval.map(_.initialize(rowInspector)).toArray
    (keyEval, valueEval, keySer, valueSer, keyObjInspector, valObjInspector, partitionEval, partitionObjInspectors)
  }

}
   * Process a partition when there is distinct key aggregations. One row per
   * distinct column is emitted here.
   */
  def processPartitionDistinct(iter: Iterator[_]) = {
    // The following is copied from an email from Cliff Engle:
    //
    // First, I want to clarify that the UnionObjectInspector has nothing to do with a SQL union.
    // It's just an unfortunate naming coincidence.
    //
    // What a union does is allows a field to have one of multiple types. For instance, if you had
    // a UnionObjectInspector that unioned {boolean, int, string}, its field could be only one of
    // either of those three. The UnionObjectInspector identifies which of the possible types that
    // field is based on a byte tag  that is stored before that actual data, e.g. a field could be
    // like {0:true}, {1:4}, or {2:"abc"} for the previous example.
    //
    // In terms of its usage, I believe that the UnionObjectInspector is only used for group-by's
    // with more than one distinct, e.g.
    //   select c, sum(distinct a), count(distinct b) from table group by c;
    //
    // (Note that a and b can be different types). The purpose of the tag is so that
    // multiple-distincts can occur in a single shuffle. There is an example in this link:
    // https://issues.apache.org/jira/browse/HIVE-537, but I'll walk through this one too.
    //
    // First, we must send all records with the same group-by key (aka distribution key here) to
    // the same reducer to get a complete grouping. Next, the shuffle phase needs to sort the data
    // such that the reducer can stream through the rows and produce both distinct aggregates
    // for a given group-by key. This means that rows must be sorted by the distribution key,
    // along with their respective distinct aggregation key. The UnionObjectInspector is what
    // allows the reducer to sort by each distinct key independently in a single reduce.
    //
    // Here's an example with some trivial data:
    //  table (a: int, b: string, c: int)
    //  1, "x", 1
    //  4, "z", 2
    //  2, "x", 1
    //  3, "y", 1
    //
    // When sent to the ReduceSink, each row produces two output rows (because of the two distincts)
    // whose keys look like this:
    //  (1, {0:1})
    //  (1, {1:"x"})
    //  (2, {0:4})
    //  (2, {1:"z"})
    //  (1, {0:2})
    //  (1, {1:"x"})
    //  (1, {0:3})
    //  (1, {1:"y"})
    //
    // Note that the squiggles represent {tag:field} for the Union object. After the data is sorted
    // by this key, the reducer can perform the distinct aggregations using fixed size memory since
    // distinct values will always be read sequentially by the reducer.

    val numDistributionKeys = conf.getNumDistributionKeys
    val numValues = valueEval.length
    val valueBuffer = new Array[Object](numValues)

    // keyBuffer holds the group by keys plus an extra distinct column.
    val allKeys = new Array[Object](keyEval.length)
    val keyBuffer = new Array[Object](numDistributionKeys + 1)

    // Populate the partition code of the reduce key using hash code of the group by keys.
    val reduceKey = new ReduceKeyMapSide

    // We output one row per distinct column
    val distinctColumnIndices: Array[Array[Int]] = {
      conf.getDistinctColumnIndices().map { indices: JList[java.lang.Integer] =>
        indices.map(_.intValue()).toArray
      }.toArray
    }

    val distinctBuffer = Array.tabulate[Array[Object]](distinctColumnIndices.length) { i =>
      new Array[Object](distinctColumnIndices(i).length)
    }

    val emptyValue = new BytesWritable

    iter.flatMap { row =>
      // Evaluate the values and serialize them into a BytesWritable.
      var i = 0
      while (i < numValues) {
        valueBuffer(i) = valueEval(i).evaluate(row)
        i += 1
      }
      val value = valueSer.serialize(valueBuffer, valObjInspector).asInstanceOf[BytesWritable]

      // Evaluate the distribution key. This is the group by key (i.e. no distinct columns).
      i = 0
      while (i < keyEval.length) {
        allKeys(i) = keyEval(i).evaluate(row)
        i += 1
      }
      System.arraycopy(allKeys, 0, keyBuffer, 0, numDistributionKeys)
      keyBuffer(numDistributionKeys) = null

      // The partition code determines where we are sending the rows.
      reduceKey.partitionCode = Arrays.hashCode(keyBuffer)

      Iterator.tabulate(distinctColumnIndices.length) { distinctI =>
        var i = 0
        val distinctParameters: Array[Object] = distinctBuffer(distinctI)
        while (i < distinctColumnIndices(distinctI).length) {
          val aggPos = distinctColumnIndices(distinctI)(i)
          distinctParameters(i) = allKeys(aggPos)
          i += 1
        }

        keyBuffer(numDistributionKeys) = new StandardUnion(distinctI.toByte, distinctParameters)
        reduceKey.bytesWritable = keySer.serialize(
          keyBuffer, keyObjInspector).asInstanceOf[BytesWritable]

        // Only send the value if this is the first distinct expression.
        // Type is (ReduceKeyMapSide, BytesWritable)
        if (distinctI == 0) {
          (reduceKey, value)
        } else {
          (reduceKey, emptyValue)
        }
      }
    }
  }

}

<MSG> Some cleanups.

<DFF> @@ -44,13 +44,7 @@ class ReduceSinkOperator extends UnaryOperator[HiveReduceSinkOperator] {
 
     // Note that we get input object inspector from hiveOp rather than Shark's
     // objectInspector because initializeMasterOnAll() hasn't been invoked yet.
-    val vars = ReduceSinkOperator.initializeOisAndSers(conf, hiveOp.getInputObjInspectors().head)
-    keyEval = vars._1
-    valueEval = vars._2
-    keySer = vars._3
-    valueSer = vars._4
-    keyObjInspector = vars._5
-    valObjInspector = vars._6
+    initializeOisAndSers(conf, hiveOp.getInputObjInspectors().head)
 
     // Determine output object inspector (a struct of KEY, VALUE).
     val ois = new ArrayList[ObjectInspector]
@@ -80,20 +74,50 @@ class ReduceSinkOperator extends UnaryOperator[HiveReduceSinkOperator] {
     }
   }
 
+  /**
+   * Initialize the object inspectors and serializers. Used on both the master
+   * and the slave.
+   */
+  def initializeOisAndSers(conf: ReduceSinkDesc, rowInspector: ObjectInspector) {
+    keyEval = conf.getKeyCols.map(ExprNodeEvaluatorFactory.get(_)).toArray
+    val numDistributionKeys = conf.getNumDistributionKeys()
+    val distinctColIndices = conf.getDistinctColumnIndices()
+    val numDistinctExprs = distinctColIndices.size()
+    valueEval = conf.getValueCols.map(ExprNodeEvaluatorFactory.get(_)).toArray
+
+    val keyTableDesc = conf.getKeySerializeInfo()
+    keySer = keyTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]
+    keySer.initialize(null, keyTableDesc.getProperties())
+    val keyIsText = keySer.getSerializedClass().equals(classOf[Text])
+
+    val valueTableDesc = conf.getValueSerializeInfo()
+    valueSer = valueTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]
+    valueSer.initialize(null, valueTableDesc.getProperties())
+    
+    keyObjInspector = SharkEnv.objectInspectorLock.synchronized {
+      ReduceSinkOperatorHelper.initEvaluatorsAndReturnStruct(
+        keyEval,
+        distinctColIndices,
+        conf.getOutputKeyColumnNames,
+        numDistributionKeys,
+        rowInspector)
+    }
+    val valFieldInspectors = valueEval.map(eval => eval.initialize(rowInspector)).toList
+    valObjInspector = SharkEnv.objectInspectorLock.synchronized {
+      ObjectInspectorFactory.getStandardStructObjectInspector(
+        conf.getOutputValueColumnNames(), valFieldInspectors)
+    }
+
+    partitionEval = conf.getPartitionCols.map(ExprNodeEvaluatorFactory.get(_)).toArray
+    partitionObjInspectors = partitionEval.map(_.initialize(rowInspector)).toArray
+  }
+
   override def initializeOnMaster() {
     conf = hiveOp.getConf()
   }
 
   override def initializeOnSlave() {
-    val vars = ReduceSinkOperator.initializeOisAndSers(conf, objectInspector)
-    keyEval = vars._1
-    valueEval = vars._2
-    keySer = vars._3
-    valueSer = vars._4
-    keyObjInspector = vars._5
-    valObjInspector = vars._6
-    partitionEval = vars._7
-    partitionObjInspectors = vars._8
+    initializeOisAndSers(conf, objectInspector)
   }
 
   override def processPartition[T](iter: Iterator[T]) = {
@@ -171,43 +195,3 @@ class ReduceSinkOperator extends UnaryOperator[HiveReduceSinkOperator] {
     }
   }
 }
-
-
-object ReduceSinkOperator {
-
-  def initializeOisAndSers(conf: ReduceSinkDesc, rowInspector: ObjectInspector) = {
-    val keyEval = conf.getKeyCols.map(ExprNodeEvaluatorFactory.get(_)).toArray
-    val numDistributionKeys = conf.getNumDistributionKeys()
-    val distinctColIndices = conf.getDistinctColumnIndices()
-    val numDistinctExprs = distinctColIndices.size()
-    val valueEval = conf.getValueCols.map(ExprNodeEvaluatorFactory.get(_)).toArray
-
-    val keyTableDesc = conf.getKeySerializeInfo()
-    val keySer = keyTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]
-    keySer.initialize(null, keyTableDesc.getProperties())
-    val keyIsText = keySer.getSerializedClass().equals(classOf[Text])
-
-    val valueTableDesc = conf.getValueSerializeInfo()
-    val valueSer = valueTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]
-    valueSer.initialize(null, valueTableDesc.getProperties())
-    
-    val keyObjInspector = SharkEnv.objectInspectorLock.synchronized {
-      ReduceSinkOperatorHelper.initEvaluatorsAndReturnStruct(
-        keyEval,
-        distinctColIndices,
-        conf.getOutputKeyColumnNames,
-        numDistributionKeys,
-        rowInspector)
-    }
-    val valFieldInspectors = valueEval.map(eval => eval.initialize(rowInspector)).toList
-    val valObjInspector = SharkEnv.objectInspectorLock.synchronized {
-      ObjectInspectorFactory.getStandardStructObjectInspector(
-        conf.getOutputValueColumnNames(), valFieldInspectors)
-    }
-
-    val partitionEval = conf.getPartitionCols.map(ExprNodeEvaluatorFactory.get(_)).toArray
-    val partitionObjInspectors = partitionEval.map(_.initialize(rowInspector)).toArray
-    (keyEval, valueEval, keySer, valueSer, keyObjInspector, valObjInspector, partitionEval, partitionObjInspectors)
-  }
-
-}

{'commit': 'diff --git a/src/main/scala/org/apache/hadoop/hive/ql/exec/GroupByPreShuffleOperator.scala b/src/main/scala/org/apache/hadoop/hive/ql/exec/GroupByPreShuffleOperator.scala\nindex e8071dd..32bc983 100755\n--- a/src/main/scala/org/apache/hadoop/hive/ql/exec/GroupByPreShuffleOperator.scala\n+++ b/src/main/scala/org/apache/hadoop/hive/ql/exec/GroupByPreShuffleOperator.scala\n@@ -1,14 +1,14 @@\n package org.apache.hadoop.hive.ql.exec\n \n-import java.util.ArrayList\n+import java.util.{ArrayList, HashMap => JHashMap}\n \n import org.apache.hadoop.hive.ql.exec.{GroupByOperator => HiveGroupByOperator}\n import org.apache.hadoop.hive.ql.plan.{AggregationDesc, ExprNodeDesc, ExprNodeColumnDesc, GroupByDesc}\n import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator\n import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.AggregationBuffer\n-import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspector, ObjectInspectorFactory, ObjectInspectorUtils}\n+import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspector, ObjectInspectorFactory,\n+    ObjectInspectorUtils, StandardStructObjectInspector, StructObjectInspector}\n import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils.ObjectInspectorCopyOption\n-import org.apache.hadoop.hive.serde2.objectinspector.{StandardStructObjectInspector, StructObjectInspector}\n \n import scala.collection.JavaConversions._\n import scala.reflect.BeanProperty\n@@ -19,6 +19,9 @@ import spark.RDD\n import spark.SparkContext._\n \n \n+/**\n+ * The pre-shuffle group by operator responsible for map side aggregations.\n+ */\n class GroupByPreShuffleOperator extends UnaryOperator[HiveGroupByOperator] {\n \n   @BeanProperty var conf: GroupByDesc = _\n@@ -71,16 +74,13 @@ class GroupByPreShuffleOperator extends UnaryOperator[HiveGroupByOperator] {\n         aggregationParameterObjectInspectors(pair._2))\n     }\n \n-    val fieldNames = conf.getOutputColumnNames    \n-    val keyFieldNames = fieldNames.slice(0,keyFields.length)\n-    \n+    val keyFieldNames = conf.getOutputColumnNames.slice(0, keyFields.length)\n     val totalFields = keyFields.length + aggregationEvals.length\n     val keyois = new ArrayList[ObjectInspector](totalFields)\n     keyObjectInspectors.foreach(keyois.add(_))\n \n     keyObjectInspector = SharkEnv.objectInspectorLock.synchronized {\n-      ObjectInspectorFactory.getStandardStructObjectInspector(\n-        keyFieldNames, keyois)\n+      ObjectInspectorFactory.getStandardStructObjectInspector(keyFieldNames, keyois)\n     }\n \n     keyFactory = new KeyWrapperFactory(keyFields, keyObjectInspectors, currentKeyObjectInspectors)\n@@ -88,30 +88,39 @@ class GroupByPreShuffleOperator extends UnaryOperator[HiveGroupByOperator] {\n \n   override def processPartition[T](iter: Iterator[T]) = {\n     logInfo("Running Pre-Shuffle Group-By")\n-    val hashAggregations = new java.util.HashMap[KeyWrapper, Array[AggregationBuffer]]()\n+\n+    // Do aggregation on map side using hashAggregations hash table.\n+    val hashAggregations = new JHashMap[KeyWrapper, Array[AggregationBuffer]]()\n     val newKeys: KeyWrapper = keyFactory.getKeyWrapper()\n     iter.foreach { case row: AnyRef =>\n-      \n       newKeys.getNewKey(row, rowInspector)\n       newKeys.setHashKey()\n \n       var aggs = hashAggregations.get(newKeys)\n-      var newKey = false\n+      var isNewKey = false\n       if (aggs == null) {\n-        newKey = true\n+        isNewKey = true\n         val newKeyProber = newKeys.copyKey()\n         aggs = newAggregations()\n         hashAggregations.put(newKeyProber, aggs)\n       }\n-      aggregate(row, aggs, newKey)\n-      Unit\n+      aggregate(row, aggs, isNewKey)\n     }\n-    \n+\n+    // Generate an iterator for the aggregation output from hashAggregations.\n     val outputCache = new Array[Object](keyFields.length + aggregationEvals.length)\n     hashAggregations.toIterator.map { case(key, aggrs) =>\n       val arr = key.getKeyArray()\n-      arr.zipWithIndex foreach { case(key, i) => outputCache(i) = key }\n-      aggrs.zipWithIndex foreach { case(aggr, i) => outputCache(i + arr.length) = aggregationEvals(i).evaluate(aggr) }\n+      var i = 0\n+      while (i < arr.length) {\n+        outputCache(i) = arr(i)\n+        i += 1\n+      }\n+      i = 0\n+      while (i < aggrs.length) {\n+        outputCache(i + arr.length) = aggregationEvals(i).evaluate(aggrs(i))\n+        i += 1\n+      }\n       outputCache\n     }\n   }\n@@ -120,9 +129,8 @@ class GroupByPreShuffleOperator extends UnaryOperator[HiveGroupByOperator] {\n     var i = 0\n     while (i < aggregations.length) {\n       if (!aggregationIsDistinct(i) || isNewKey) {\n-        //TODO: remove .map and reuse aggregationParameterFields array\n-        aggregationEvals(i).aggregate(aggregations(i), \n-                                      aggregationParameterFields(i).map(_.evaluate(row)))\n+        aggregationEvals(i).aggregate(\n+          aggregations(i), aggregationParameterFields(i).map(_.evaluate(row)))\n       }\n       i += 1\n     }\ndiff --git a/src/main/scala/shark/Column.scala b/src/main/scala/shark/Column.scala\nindex 229abdc..97cbb17 100755\n--- a/src/main/scala/shark/Column.scala\n+++ b/src/main/scala/shark/Column.scala\n@@ -22,8 +22,44 @@ import it.unimi.dsi.fastutil.doubles.DoubleArrayList\n import javaewah.EWAHCompressedBitmap\n import javaewah.IntIterator\n \n-// Columns hold primitive arrays of data\n-// They provide methods to add objects given their objectInspectors and retrieve writable objects\n+\n+/**\n+ * A column of data, backed by some primitive array(s). These columns provide\n+ * methods to add objects using their object inspectors and return Hadoop\n+ * writable objects.\n+ */\n+abstract class Column {\n+\n+  /**\n+   *  Initial number of elements in a column.\n+   *  TODO(rxin): Choose this more intelligently. See \n+   *  https://github.com/amplab/shark/issues/43\n+   */\n+  val initialSize = 1000000 // Number of rows\n+\n+  /**\n+   * A compressed bitmap to indicate whether an element is NULL.\n+   */\n+  val nulls = new EWAHCompressedBitmap()\n+  var nullsIter: IntIterator = null\n+  var nextNullIndex = -1\n+\n+  /**\n+   * Return a hadoop writable object for the field value in this column at\n+   * index position i.\n+   */\n+  def apply(i: Int): Object\n+\n+  /**\n+   * Use the object inspector to extract the field value from the object and\n+   * add it to the columnar collection.\n+   */\n+  def add(o: Object, oi: ObjectInspector)\n+\n+  def close(): Unit \n+}\n+\n+\n object Column {\n   def apply(oi: ObjectInspector) = {\n     oi.getCategory match {\n@@ -47,20 +83,6 @@ object Column {\n }\n \n \n-abstract class Column {\n-  val initialSize = 1000000 // Number of rows\n-  val nulls = new EWAHCompressedBitmap()\n-  var nullsIter: IntIterator = null\n-  var nextNullIndex = -1\n-\n-  def apply(i: Int): Object\n-\n-  def add(o: Object, oi: ObjectInspector)\n-\n-  def close(): Unit \n-}\n-\n-\n class BooleanColumn extends Column {\n \n   val arr = new BooleanArrayList(initialSize)\n@@ -285,6 +307,11 @@ class DoubleColumn extends Column {\n }\n \n \n+/**\n+ * A string column. Store a collection of strings using a single byte array by\n+ * concatenating them together. An additional int array is used to index the\n+ * starting position of each string.\n+ */\n class StringColumn extends Column {\n \n   val arr = new ByteArrayList(initialSize)\n@@ -332,7 +359,10 @@ class VoidColumn extends Column {\n }\n \n \n-// Holds non-primitive column in serialized form\n+/**\n+ * For non-primitive columns, serialize the value and store them as a single\n+ * byte array.\n+ */\n class LazyColumn(outputOI: ObjectInspector) extends Column {\n \n   val arr = new ByteArrayList(initialSize) // all serialized bytes\ndiff --git a/src/main/scala/shark/ColumnarSerDe.scala b/src/main/scala/shark/ColumnarSerDe.scala\nindex 77f332b..e6f8fb2 100755\n--- a/src/main/scala/shark/ColumnarSerDe.scala\n+++ b/src/main/scala/shark/ColumnarSerDe.scala\n@@ -1,11 +1,9 @@\n package shark\n \n-import java.io.DataOutput\n-import java.io.DataInput\n+import java.io.{DataInput, DataOutput}\n import java.util.Properties\n \n import org.apache.hadoop.conf.Configuration\n-import org.apache.hadoop.io.Writable\n import org.apache.hadoop.hive.serde2.{ByteStream, SerDe, SerDeException}\n import org.apache.hadoop.hive.serde2.`lazy`.{LazyFactory, LazySimpleSerDe}\n import org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe.SerDeParameters\n@@ -16,6 +14,7 @@ import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector\n import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category\n import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory\n import org.apache.hadoop.hive.serde2.typeinfo.{TypeInfo, PrimitiveTypeInfo}\n+import org.apache.hadoop.io.Writable\n \n \n class ColumnarSerDe extends SerDe {\ndiff --git a/src/main/scala/shark/RDDUtils.scala b/src/main/scala/shark/RDDUtils.scala\nindex de03c08..b308606 100755\n--- a/src/main/scala/shark/RDDUtils.scala\n+++ b/src/main/scala/shark/RDDUtils.scala\n@@ -26,8 +26,6 @@ import spark.SparkContext._\n  * to Spark\'s built-in abstractions.\n  */\n object RDDUtils {\n-\n-  val INITIAL_BUFFER_SIZE = 100000 // 10MB\n   \n   /**\n    * Used to serialize an RDD using a Hive SerDe before caching in a\n@@ -38,14 +36,14 @@ object RDDUtils {\n     val serializer = td.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n     serializer.initialize(hconf, td.getProperties())\n     var v: Object = null\n-    var i: Int = 0\n+    var numRows: Int = 0\n     iter.foreach { row =>\n       v = serializer.serialize(row, oi)\n-      i += 1\n+      numRows += 1\n     }\n     if (v != null) {\n       v.asInstanceOf[ColumnarWritable].close()\n-      Iterator(i, v)\n+      Iterator(numRows, v)\n     } else {\n       Iterator()\n     }\ndiff --git a/src/main/scala/shark/dashboard/CacheTrackerDashboardHandler.scala b/src/main/scala/shark/dashboard/CacheTrackerDashboardHandler.scala\ndeleted file mode 100644\nindex d7357f1..0000000\n--- a/src/main/scala/shark/dashboard/CacheTrackerDashboardHandler.scala\n+++ /dev/null\n@@ -1,65 +0,0 @@\n-package shark.dashboard\n-\n-import javax.servlet.http.{HttpServletRequest, HttpServletResponse}\n-\n-import org.eclipse.jetty.server.handler.AbstractHandler\n-import org.eclipse.jetty.server.Request\n-\n-import spark.{CacheTracker, SparkEnv, Utils}\n-\n-/**\n- * A dashboard to view the cache status on each slave nodes.\n- */\n-class CacheTrackerDashboardHandler(\n-  val cacheTracker: CacheTracker = SparkEnv.get.cacheTracker)\n-  extends AbstractHandler {\n-\n-  override def handle(\n-    target: String,\n-    baseRequest: Request,\n-    request: HttpServletRequest,\n-    response: HttpServletResponse)\n-  {\n-    // Get the cache status from tracker.\n-    val cacheStatus = cacheTracker.getCacheStatus()\n-\n-    // Serve the HTTP content.\n-    response.setContentType("text/html;charset=utf-8")\n-    response.setStatus(HttpServletResponse.SC_OK)\n-    baseRequest.setHandled(true)\n-    response.getWriter().println(\n-      <html>\n-        <head>\n-          <title>Shark Dashboard - Cache Status</title>\n-          <link rel="stylesheet" type="text/css" href="/static/dashboard.css" />\n-        </head>\n-        <body>\n-          <h1>Shark Cache Status</h1>\n-          <p>Master node: { System.getProperty("spark.master.host") }</p>\n-          <table border="1" cellspacing="0">\n-          <tr><th>Host</th><th>Capacity</th><th>Used</th><th>Used (%)</th></tr>\n-          {\n-            for ((host, capacity, use) <- cacheStatus) yield {\n-              val percent = (use * 100 / capacity).toInt\n-\n-              <tr>\n-                <td class="node">{ host }</td>\n-                <td class="node">{ Utils.memoryBytesToString(capacity) }</td>\n-                <td class="node">{ Utils.memoryBytesToString(use) }</td>\n-                <td class="node">\n-                  <table border="1px" class="percent_bar"><tbody><tr>\n-                  <td cellspacing="0" class="percent_used" width={ percent + "%" }></td>\n-                  <td cellspacing="0" class="percent_unused" width={ (100 - percent) + "%" }></td>\n-                  </tr></tbody></table>\n-                </td>\n-              </tr>\n-            }\n-          }\n-          </table>\n-        </body>\n-      </html>\n-    )\n-  }\n-\n-}\n-\ndiff --git a/src/main/scala/shark/dashboard/Dashboard.scala b/src/main/scala/shark/dashboard/Dashboard.scala\ndeleted file mode 100644\nindex 72f987f..0000000\n--- a/src/main/scala/shark/dashboard/Dashboard.scala\n+++ /dev/null\n@@ -1,53 +0,0 @@\n-package shark.dashboard\n-\n-import org.eclipse.jetty.server.handler.{AbstractHandler, ContextHandler, HandlerList, ResourceHandler}\n-import org.eclipse.jetty.server.Server\n-\n-/**\n- * The basic object for the Shark HTTP UI. Call start() to launch the HTTP\n- * service. This is backed by Jetty.\n- */\n-object Dashboard extends shark.LogHelper {\n-\n-  private var server: Server = null\n-  private var port: Int = -1\n-\n-  def start() {\n-\n-    val staticFileBaseUrl = getClass.getResource("/dashboard")\n-    println(staticFileBaseUrl)\n-    val staticFileBase = staticFileBaseUrl.toString\n-    logInfo("Starting Jetty with static file base: %s".format(staticFileBase))\n-\n-    port = System.getProperty("spark.dashboard.port", "8088").toInt\n-    server = new Server(port)\n-\n-    // A resource handler for static files.\n-    val staticFileHandler = new ResourceHandler\n-    staticFileHandler.setResourceBase(staticFileBase)\n-\n-    // Handlers to register. Wildcard handlers must go last.\n-    val handlers = Array[(String, AbstractHandler)](\n-      ("/static", staticFileHandler),\n-      ("/queryPlan", new QueryPlanDashboardHandler),\n-      ("*", new CacheTrackerDashboardHandler)\n-    )\n-\n-    val handlersToRegister = handlers.map { case(path, handler) =>\n-      if (path == "*") {\n-        handler\n-      } else {\n-        val contextHandler = new ContextHandler(path)\n-        contextHandler.setHandler(handler)\n-        contextHandler.asInstanceOf[org.eclipse.jetty.server.Handler]\n-      }\n-    }\n-\n-    // Set the handlers and start the http server.\n-    val handlerList = new HandlerList\n-    handlerList.setHandlers(handlersToRegister)\n-    server.setHandler(handlerList)\n-    server.start()\n-  }\n-}\n-\ndiff --git a/src/main/scala/shark/dashboard/QueryPlanDashboardHandler.scala b/src/main/scala/shark/dashboard/QueryPlanDashboardHandler.scala\ndeleted file mode 100644\nindex 3e62ffb..0000000\n--- a/src/main/scala/shark/dashboard/QueryPlanDashboardHandler.scala\n+++ /dev/null\n@@ -1,75 +0,0 @@\n-package shark.dashboard\n-\n-import javax.servlet.http.{HttpServletRequest, HttpServletResponse}\n-\n-import org.eclipse.jetty.server.handler.AbstractHandler\n-import org.eclipse.jetty.server.Request\n-\n-import shark.execution.{ExplainTaskHelper, HiveOperator, Operator, TerminalAbstractOperator}\n-import spark.{CacheTracker, Utils}\n-\n-\n-/**\n- * A dashboard for visualizing the query plan. It outputs the query plan in JSON\n- * and enables the client side HTML page to illustrate the query plan (e.g. in\n- * d3.js).\n- */\n-class QueryPlanDashboardHandler extends AbstractHandler with shark.LogHelper {\n-\n-  override def handle(\n-    target: String,\n-    baseRequest: Request,\n-    request: HttpServletRequest,\n-    response: HttpServletResponse)\n-  {\n-    val json = QueryPlanDashboardHandler.generateJson(\n-      QueryPlanDashboardHandler.terminalOperator)\n-\n-    // Serve the HTTP content.\n-    response.setContentType("application/json;charset=utf-8")\n-    response.setStatus(HttpServletResponse.SC_OK)\n-    baseRequest.setHandled(true)\n-    response.getWriter().println(json)\n-  }\n-}\n-\n-\n-object QueryPlanDashboardHandler {\n-\n-  /**\n-   * A static variable for communicating the query plan between the parser and\n-   * this Dashboard. This is not the best design, but oh well .... :)\n-   */\n-  var terminalOperator: TerminalAbstractOperator[_] = _\n-\n-  /**\n-   * Generates a JSON representation of the query plan.\n-   */\n-  def generateJson(op: Operator[_], indent: String = ""): String = {\n-\n-    val parentJson = \n-      if (op.parentOperators.size > 0) {\n-        op.parentOperators.map { p =>\n-          generateJson(p, indent + "    ")\n-        }.reduce(_ + ",\\n" + _)\n-      } else {\n-        ""\n-      }\n-\n-    val os = new java.io.ByteArrayOutputStream\n-    val ps = new java.io.PrintStream(os)\n-    ExplainTaskHelper.outputWork(\n-      op.hiveOp.asInstanceOf[HiveOperator].getConf.asInstanceOf[java.io.Serializable],\n-      ps,\n-      true,\n-      0)\n-    val details = os.toString().replace("\\n", "\\\\n").replace("\\"", "\\\\\\"")\n-\n-    indent + "{\\n" + \n-    indent + "  \\"name\\": \\"" + op.getClass.getSimpleName + "\\",\\n" +\n-    indent + "  \\"details\\": \\"" + details + "\\",\\n" +\n-    indent + "  \\"contents\\": [\\n" + parentJson +\n-    indent + "  ]\\n" +\n-    indent + "}\\n"\n-  }\n-}\n\\ No newline at end of file\ndiff --git a/src/main/scala/shark/execution/ReduceSinkOperator.scala b/src/main/scala/shark/execution/ReduceSinkOperator.scala\nindex 379cf30..9ddd72d 100755\n--- a/src/main/scala/shark/execution/ReduceSinkOperator.scala\n+++ b/src/main/scala/shark/execution/ReduceSinkOperator.scala\n@@ -44,13 +44,7 @@ class ReduceSinkOperator extends UnaryOperator[HiveReduceSinkOperator] {\n \n     // Note that we get input object inspector from hiveOp rather than Shark\'s\n     // objectInspector because initializeMasterOnAll() hasn\'t been invoked yet.\n-    val vars = ReduceSinkOperator.initializeOisAndSers(conf, hiveOp.getInputObjInspectors().head)\n-    keyEval = vars._1\n-    valueEval = vars._2\n-    keySer = vars._3\n-    valueSer = vars._4\n-    keyObjInspector = vars._5\n-    valObjInspector = vars._6\n+    initializeOisAndSers(conf, hiveOp.getInputObjInspectors().head)\n \n     // Determine output object inspector (a struct of KEY, VALUE).\n     val ois = new ArrayList[ObjectInspector]\n@@ -80,20 +74,50 @@ class ReduceSinkOperator extends UnaryOperator[HiveReduceSinkOperator] {\n     }\n   }\n \n+  /**\n+   * Initialize the object inspectors and serializers. Used on both the master\n+   * and the slave.\n+   */\n+  def initializeOisAndSers(conf: ReduceSinkDesc, rowInspector: ObjectInspector) {\n+    keyEval = conf.getKeyCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n+    val numDistributionKeys = conf.getNumDistributionKeys()\n+    val distinctColIndices = conf.getDistinctColumnIndices()\n+    val numDistinctExprs = distinctColIndices.size()\n+    valueEval = conf.getValueCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n+\n+    val keyTableDesc = conf.getKeySerializeInfo()\n+    keySer = keyTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n+    keySer.initialize(null, keyTableDesc.getProperties())\n+    val keyIsText = keySer.getSerializedClass().equals(classOf[Text])\n+\n+    val valueTableDesc = conf.getValueSerializeInfo()\n+    valueSer = valueTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n+    valueSer.initialize(null, valueTableDesc.getProperties())\n+    \n+    keyObjInspector = SharkEnv.objectInspectorLock.synchronized {\n+      ReduceSinkOperatorHelper.initEvaluatorsAndReturnStruct(\n+        keyEval,\n+        distinctColIndices,\n+        conf.getOutputKeyColumnNames,\n+        numDistributionKeys,\n+        rowInspector)\n+    }\n+    val valFieldInspectors = valueEval.map(eval => eval.initialize(rowInspector)).toList\n+    valObjInspector = SharkEnv.objectInspectorLock.synchronized {\n+      ObjectInspectorFactory.getStandardStructObjectInspector(\n+        conf.getOutputValueColumnNames(), valFieldInspectors)\n+    }\n+\n+    partitionEval = conf.getPartitionCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n+    partitionObjInspectors = partitionEval.map(_.initialize(rowInspector)).toArray\n+  }\n+\n   override def initializeOnMaster() {\n     conf = hiveOp.getConf()\n   }\n \n   override def initializeOnSlave() {\n-    val vars = ReduceSinkOperator.initializeOisAndSers(conf, objectInspector)\n-    keyEval = vars._1\n-    valueEval = vars._2\n-    keySer = vars._3\n-    valueSer = vars._4\n-    keyObjInspector = vars._5\n-    valObjInspector = vars._6\n-    partitionEval = vars._7\n-    partitionObjInspectors = vars._8\n+    initializeOisAndSers(conf, objectInspector)\n   }\n \n   override def processPartition[T](iter: Iterator[T]) = {\n@@ -171,43 +195,3 @@ class ReduceSinkOperator extends UnaryOperator[HiveReduceSinkOperator] {\n     }\n   }\n }\n-\n-\n-object ReduceSinkOperator {\n-\n-  def initializeOisAndSers(conf: ReduceSinkDesc, rowInspector: ObjectInspector) = {\n-    val keyEval = conf.getKeyCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n-    val numDistributionKeys = conf.getNumDistributionKeys()\n-    val distinctColIndices = conf.getDistinctColumnIndices()\n-    val numDistinctExprs = distinctColIndices.size()\n-    val valueEval = conf.getValueCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n-\n-    val keyTableDesc = conf.getKeySerializeInfo()\n-    val keySer = keyTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n-    keySer.initialize(null, keyTableDesc.getProperties())\n-    val keyIsText = keySer.getSerializedClass().equals(classOf[Text])\n-\n-    val valueTableDesc = conf.getValueSerializeInfo()\n-    val valueSer = valueTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n-    valueSer.initialize(null, valueTableDesc.getProperties())\n-    \n-    val keyObjInspector = SharkEnv.objectInspectorLock.synchronized {\n-      ReduceSinkOperatorHelper.initEvaluatorsAndReturnStruct(\n-        keyEval,\n-        distinctColIndices,\n-        conf.getOutputKeyColumnNames,\n-        numDistributionKeys,\n-        rowInspector)\n-    }\n-    val valFieldInspectors = valueEval.map(eval => eval.initialize(rowInspector)).toList\n-    val valObjInspector = SharkEnv.objectInspectorLock.synchronized {\n-      ObjectInspectorFactory.getStandardStructObjectInspector(\n-        conf.getOutputValueColumnNames(), valFieldInspectors)\n-    }\n-\n-    val partitionEval = conf.getPartitionCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n-    val partitionObjInspectors = partitionEval.map(_.initialize(rowInspector)).toArray\n-    (keyEval, valueEval, keySer, valueSer, keyObjInspector, valObjInspector, partitionEval, partitionObjInspectors)\n-  }\n-\n-}\n', 'commit_subject': 'Some cleanups.', 'repo_name': 'amplab/shark', 'committer': {'name': 'Reynold Xin', 'email': 'c9d35a384bc579d1c6c8c055f2a2b5c04e11b521@cs.berkeley.edu', 'time_sec': '1346888049', 'tz_offset': '-420', 'date': {'seconds': '1346888049'}}, 'commit_tree': 'a34e17736cdc59595098488285c88b21b92af78c', 'commit_message': 'Some cleanups.\n', 'language_name': 'Shell', 'license': 'apache-2.0', 'hunks': "@@ -44,13 +44,7 @@ class ReduceSinkOperator extends UnaryOperator[HiveReduceSinkOperator] {\n \n     // Note that we get input object inspector from hiveOp rather than Shark's\n     // objectInspector because initializeMasterOnAll() hasn't been invoked yet.\n-    val vars = ReduceSinkOperator.initializeOisAndSers(conf, hiveOp.getInputObjInspectors().head)\n-    keyEval = vars._1\n-    valueEval = vars._2\n-    keySer = vars._3\n-    valueSer = vars._4\n-    keyObjInspector = vars._5\n-    valObjInspector = vars._6\n+    initializeOisAndSers(conf, hiveOp.getInputObjInspectors().head)\n \n     // Determine output object inspector (a struct of KEY, VALUE).\n     val ois = new ArrayList[ObjectInspector]\n@@ -80,20 +74,50 @@ class ReduceSinkOperator extends UnaryOperator[HiveReduceSinkOperator] {\n     }\n   }\n \n+  /**\n+   * Initialize the object inspectors and serializers. Used on both the master\n+   * and the slave.\n+   */\n+  def initializeOisAndSers(conf: ReduceSinkDesc, rowInspector: ObjectInspector) {\n+    keyEval = conf.getKeyCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n+    val numDistributionKeys = conf.getNumDistributionKeys()\n+    val distinctColIndices = conf.getDistinctColumnIndices()\n+    val numDistinctExprs = distinctColIndices.size()\n+    valueEval = conf.getValueCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n+\n+    val keyTableDesc = conf.getKeySerializeInfo()\n+    keySer = keyTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n+    keySer.initialize(null, keyTableDesc.getProperties())\n+    val keyIsText = keySer.getSerializedClass().equals(classOf[Text])\n+\n+    val valueTableDesc = conf.getValueSerializeInfo()\n+    valueSer = valueTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n+    valueSer.initialize(null, valueTableDesc.getProperties())\n+    \n+    keyObjInspector = SharkEnv.objectInspectorLock.synchronized {\n+      ReduceSinkOperatorHelper.initEvaluatorsAndReturnStruct(\n+        keyEval,\n+        distinctColIndices,\n+        conf.getOutputKeyColumnNames,\n+        numDistributionKeys,\n+        rowInspector)\n+    }\n+    val valFieldInspectors = valueEval.map(eval => eval.initialize(rowInspector)).toList\n+    valObjInspector = SharkEnv.objectInspectorLock.synchronized {\n+      ObjectInspectorFactory.getStandardStructObjectInspector(\n+        conf.getOutputValueColumnNames(), valFieldInspectors)\n+    }\n+\n+    partitionEval = conf.getPartitionCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n+    partitionObjInspectors = partitionEval.map(_.initialize(rowInspector)).toArray\n+  }\n+\n   override def initializeOnMaster() {\n     conf = hiveOp.getConf()\n   }\n \n   override def initializeOnSlave() {\n-    val vars = ReduceSinkOperator.initializeOisAndSers(conf, objectInspector)\n-    keyEval = vars._1\n-    valueEval = vars._2\n-    keySer = vars._3\n-    valueSer = vars._4\n-    keyObjInspector = vars._5\n-    valObjInspector = vars._6\n-    partitionEval = vars._7\n-    partitionObjInspectors = vars._8\n+    initializeOisAndSers(conf, objectInspector)\n   }\n \n   override def processPartition[T](iter: Iterator[T]) = {\n@@ -171,43 +195,3 @@ class ReduceSinkOperator extends UnaryOperator[HiveReduceSinkOperator] {\n     }\n   }\n }\n-\n-\n-object ReduceSinkOperator {\n-\n-  def initializeOisAndSers(conf: ReduceSinkDesc, rowInspector: ObjectInspector) = {\n-    val keyEval = conf.getKeyCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n-    val numDistributionKeys = conf.getNumDistributionKeys()\n-    val distinctColIndices = conf.getDistinctColumnIndices()\n-    val numDistinctExprs = distinctColIndices.size()\n-    val valueEval = conf.getValueCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n-\n-    val keyTableDesc = conf.getKeySerializeInfo()\n-    val keySer = keyTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n-    keySer.initialize(null, keyTableDesc.getProperties())\n-    val keyIsText = keySer.getSerializedClass().equals(classOf[Text])\n-\n-    val valueTableDesc = conf.getValueSerializeInfo()\n-    val valueSer = valueTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n-    valueSer.initialize(null, valueTableDesc.getProperties())\n-    \n-    val keyObjInspector = SharkEnv.objectInspectorLock.synchronized {\n-      ReduceSinkOperatorHelper.initEvaluatorsAndReturnStruct(\n-        keyEval,\n-        distinctColIndices,\n-        conf.getOutputKeyColumnNames,\n-        numDistributionKeys,\n-        rowInspector)\n-    }\n-    val valFieldInspectors = valueEval.map(eval => eval.initialize(rowInspector)).toList\n-    val valObjInspector = SharkEnv.objectInspectorLock.synchronized {\n-      ObjectInspectorFactory.getStandardStructObjectInspector(\n-        conf.getOutputValueColumnNames(), valFieldInspectors)\n-    }\n-\n-    val partitionEval = conf.getPartitionCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n-    val partitionObjInspectors = partitionEval.map(_.initialize(rowInspector)).toArray\n-    (keyEval, valueEval, keySer, valueSer, keyObjInspector, valObjInspector, partitionEval, partitionObjInspectors)\n-  }\n-\n-}\n", 'addition_count': 40, 'deletion_count': 56, 'src_file': 'a/src/main/scala/shark/execution/ReduceSinkOperator.scala', 'tgt_file': 'b/src/main/scala/shark/execution/ReduceSinkOperator.scala', 'file_extension': '.scala', 'before_file': '/*\n * Copyright (C) 2012 The Regents of The University California.\n * All rights reserved.\n *\n * Licensed under the Apache License, Version 2.0 (the "License");\n * you may not use this file except in compliance with the License.\n * You may obtain a copy of the License at\n *\n *      http://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an "AS IS" BASIS,\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n * See the License for the specific language governing permissions and\n * limitations under the License.\n */\n\npackage shark.execution\n\nimport java.util.{ArrayList, Arrays, List => JList, Random}\n\nimport scala.collection.Iterator\nimport scala.collection.JavaConversions._\nimport scala.reflect.BeanProperty\n\nimport org.apache.hadoop.hive.ql.exec.{ExprNodeEvaluator, ExprNodeEvaluatorFactory}\nimport org.apache.hadoop.hive.ql.plan.ReduceSinkDesc\nimport org.apache.hadoop.hive.serde2.SerDe\nimport org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector\nimport org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory\nimport org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils\nimport org.apache.hadoop.hive.serde2.objectinspector.StandardUnionObjectInspector.StandardUnion\nimport org.apache.hadoop.io.BytesWritable\n\n\n/**\n * Converts a collection of rows into key, value pairs. This is the\n * upstream operator for joins and groupbys.\n */\nclass ReduceSinkOperator extends UnaryOperator[ReduceSinkDesc] {\n\n  @BeanProperty var conf: ReduceSinkDesc = _\n\n\n    // Note that we get input object inspector from hiveOp rather than Shark\'s\n    // objectInspector because initializeMasterOnAll() hasn\'t been invoked yet.\n    val vars = ReduceSinkOperator.initializeOisAndSers(conf, hiveOp.getInputObjInspectors().head)\n    keyEval = vars._1\n    valueEval = vars._2\n    keySer = vars._3\n    valueSer = vars._4\n    keyObjInspector = vars._5\n    valObjInspector = vars._6\n\n    // Determine output object inspector (a struct of KEY, VALUE).\n    val ois = new ArrayList[ObjectInspector]\n\n  // The evaluator for the partition columns (used in CLUSTER BY and\n  // DISTRIBUTE BY in Hive). Partition columns decide the reducer that the\n  // current row goes to. Partition columns are not passed to reducers.\n  @transient var partitionEval: Array[ExprNodeEvaluator] = _\n\n  @transient var keySer: SerDe = _\n  @transient var valueSer: SerDe = _\n  @transient var keyObjInspector: ObjectInspector = _\n  @transient var keyFieldObjInspectors: Array[ObjectInspector] = _\n  @transient var valObjInspector: ObjectInspector = _\n  @transient var valFieldObjInspectors: Array[ObjectInspector] = _\n  @transient var partitionObjInspectors: Array[ObjectInspector] = _\n\n  override def getTag() = conf.getTag()\n\n  override def initializeOnMaster() {\n    super.initializeOnMaster()\n    \n    conf = desc\n  }\n\n  override def initializeOnSlave() {\n    }\n  }\n\n  override def initializeOnMaster() {\n    conf = hiveOp.getConf()\n  }\n\n  override def initializeOnSlave() {\n    val vars = ReduceSinkOperator.initializeOisAndSers(conf, objectInspector)\n    keyEval = vars._1\n    valueEval = vars._2\n    keySer = vars._3\n    valueSer = vars._4\n    keyObjInspector = vars._5\n    valObjInspector = vars._6\n    partitionEval = vars._7\n    partitionObjInspectors = vars._8\n  }\n\n  override def processPartition[T](iter: Iterator[T]) = {\n      distinctColIndices,\n      conf.getOutputKeyColumnNames,\n      numDistributionKeys,\n      rowInspector)\n\n    // Initialize object inspector for value columns.\n    valFieldObjInspectors = valueEval.map(eval => eval.initialize(rowInspector))\n    valObjInspector = ObjectInspectorFactory.getStandardStructObjectInspector(\n      conf.getOutputValueColumnNames(), valFieldObjInspectors.toList)\n\n    // Initialize evaluator and object inspector for partition columns.\n    partitionEval = conf.getPartitionCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n    partitionObjInspectors = partitionEval.map(_.initialize(rowInspector)).toArray\n  }\n\n  /**\n   * Process a partition when there is NO distinct key aggregations.\n   */\n  def processPartitionNoDistinct(iter: Iterator[_]) = {\n    val numDistributionKeys = conf.getNumDistributionKeys\n    // Buffer for key, value evaluation to avoid frequent object allocation.\n    val evaluatedKey = new Array[Object](keyEval.length)\n    val evaluatedValue = new Array[Object](valueEval.length)\n    val reduceKey = new ReduceKeyMapSide\n\n    // A random number generator, used to distribute keys evenly if there is\n    // no partition columns specified. Use a constant seed to make the code\n    // deterministic as did Hive.\n    val rand = new Random(13)\n\n    iter.map { row =>\n      // TODO: we don\'t need partition code for group-by or join\n\n      // Determine the partition code (Hive calls it keyHashCode), used for\n      // DISTRIBUTED BY / CLUSTER BY.\n      var partitionCode = 0\n      if (partitionEval.length == 0) {\n        // If there is no partition columns, we randomly distribute the data\n        partitionCode = rand.nextInt()\n      } else {\n        // With partition columns, determine a partitionCode to distribute.\n        var i = 0\n        while (i < partitionEval.length) {\n          val o = partitionEval(i).evaluate(row)\n          partitionCode = partitionCode * 31 +\n            ObjectInspectorUtils.hashCode(o, partitionObjInspectors(i))\n          i += 1\n        }\n      }\n\n      // Evaluate the key columns.\n      var i = 0\n      while (i < numDistributionKeys) {\n        evaluatedKey(i) = keyEval(i).evaluate(row)\n        i += 1\n      }\n\n      // Evaluate the value columns.\n      i = 0\n      while (i < valueEval.length) {\n        evaluatedValue(i) = valueEval(i).evaluate(row)\n        i += 1\n      }\n\n      val key = keySer.serialize(evaluatedKey, keyObjInspector).asInstanceOf[BytesWritable]\n      val value = valueSer.serialize(evaluatedValue, valObjInspector).asInstanceOf[BytesWritable]\n\n      reduceKey.bytesWritable = key\n      reduceKey.partitionCode = partitionCode\n      (reduceKey, value)\n    }\n    }\n  }\n}\n\n\nobject ReduceSinkOperator {\n\n  def initializeOisAndSers(conf: ReduceSinkDesc, rowInspector: ObjectInspector) = {\n    val keyEval = conf.getKeyCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n    val numDistributionKeys = conf.getNumDistributionKeys()\n    val distinctColIndices = conf.getDistinctColumnIndices()\n    val numDistinctExprs = distinctColIndices.size()\n    val valueEval = conf.getValueCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n\n    val keyTableDesc = conf.getKeySerializeInfo()\n    val keySer = keyTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n    keySer.initialize(null, keyTableDesc.getProperties())\n    val keyIsText = keySer.getSerializedClass().equals(classOf[Text])\n\n    val valueTableDesc = conf.getValueSerializeInfo()\n    val valueSer = valueTableDesc.getDeserializerClass().newInstance().asInstanceOf[SerDe]\n    valueSer.initialize(null, valueTableDesc.getProperties())\n    \n    val keyObjInspector = SharkEnv.objectInspectorLock.synchronized {\n      ReduceSinkOperatorHelper.initEvaluatorsAndReturnStruct(\n        keyEval,\n        distinctColIndices,\n        conf.getOutputKeyColumnNames,\n        numDistributionKeys,\n        rowInspector)\n    }\n    val valFieldInspectors = valueEval.map(eval => eval.initialize(rowInspector)).toList\n    val valObjInspector = SharkEnv.objectInspectorLock.synchronized {\n      ObjectInspectorFactory.getStandardStructObjectInspector(\n        conf.getOutputValueColumnNames(), valFieldInspectors)\n    }\n\n    val partitionEval = conf.getPartitionCols.map(ExprNodeEvaluatorFactory.get(_)).toArray\n    val partitionObjInspectors = partitionEval.map(_.initialize(rowInspector)).toArray\n    (keyEval, valueEval, keySer, valueSer, keyObjInspector, valObjInspector, partitionEval, partitionObjInspectors)\n  }\n\n}\n   * Process a partition when there is distinct key aggregations. One row per\n   * distinct column is emitted here.\n   */\n  def processPartitionDistinct(iter: Iterator[_]) = {\n    // The following is copied from an email from Cliff Engle:\n    //\n    // First, I want to clarify that the UnionObjectInspector has nothing to do with a SQL union.\n    // It\'s just an unfortunate naming coincidence.\n    //\n    // What a union does is allows a field to have one of multiple types. For instance, if you had\n    // a UnionObjectInspector that unioned {boolean, int, string}, its field could be only one of\n    // either of those three. The UnionObjectInspector identifies which of the possible types that\n    // field is based on a byte tag  that is stored before that actual data, e.g. a field could be\n    // like {0:true}, {1:4}, or {2:"abc"} for the previous example.\n    //\n    // In terms of its usage, I believe that the UnionObjectInspector is only used for group-by\'s\n    // with more than one distinct, e.g.\n    //   select c, sum(distinct a), count(distinct b) from table group by c;\n    //\n    // (Note that a and b can be different types). The purpose of the tag is so that\n    // multiple-distincts can occur in a single shuffle. There is an example in this link:\n    // https://issues.apache.org/jira/browse/HIVE-537, but I\'ll walk through this one too.\n    //\n    // First, we must send all records with the same group-by key (aka distribution key here) to\n    // the same reducer to get a complete grouping. Next, the shuffle phase needs to sort the data\n    // such that the reducer can stream through the rows and produce both distinct aggregates\n    // for a given group-by key. This means that rows must be sorted by the distribution key,\n    // along with their respective distinct aggregation key. The UnionObjectInspector is what\n    // allows the reducer to sort by each distinct key independently in a single reduce.\n    //\n    // Here\'s an example with some trivial data:\n    //  table (a: int, b: string, c: int)\n    //  1, "x", 1\n    //  4, "z", 2\n    //  2, "x", 1\n    //  3, "y", 1\n    //\n    // When sent to the ReduceSink, each row produces two output rows (because of the two distincts)\n    // whose keys look like this:\n    //  (1, {0:1})\n    //  (1, {1:"x"})\n    //  (2, {0:4})\n    //  (2, {1:"z"})\n    //  (1, {0:2})\n    //  (1, {1:"x"})\n    //  (1, {0:3})\n    //  (1, {1:"y"})\n    //\n    // Note that the squiggles represent {tag:field} for the Union object. After the data is sorted\n    // by this key, the reducer can perform the distinct aggregations using fixed size memory since\n    // distinct values will always be read sequentially by the reducer.\n\n    val numDistributionKeys = conf.getNumDistributionKeys\n    val numValues = valueEval.length\n    val valueBuffer = new Array[Object](numValues)\n\n    // keyBuffer holds the group by keys plus an extra distinct column.\n    val allKeys = new Array[Object](keyEval.length)\n    val keyBuffer = new Array[Object](numDistributionKeys + 1)\n\n    // Populate the partition code of the reduce key using hash code of the group by keys.\n    val reduceKey = new ReduceKeyMapSide\n\n    // We output one row per distinct column\n    val distinctColumnIndices: Array[Array[Int]] = {\n      conf.getDistinctColumnIndices().map { indices: JList[java.lang.Integer] =>\n        indices.map(_.intValue()).toArray\n      }.toArray\n    }\n\n    val distinctBuffer = Array.tabulate[Array[Object]](distinctColumnIndices.length) { i =>\n      new Array[Object](distinctColumnIndices(i).length)\n    }\n\n    val emptyValue = new BytesWritable\n\n    iter.flatMap { row =>\n      // Evaluate the values and serialize them into a BytesWritable.\n      var i = 0\n      while (i < numValues) {\n        valueBuffer(i) = valueEval(i).evaluate(row)\n        i += 1\n      }\n      val value = valueSer.serialize(valueBuffer, valObjInspector).asInstanceOf[BytesWritable]\n\n      // Evaluate the distribution key. This is the group by key (i.e. no distinct columns).\n      i = 0\n      while (i < keyEval.length) {\n        allKeys(i) = keyEval(i).evaluate(row)\n        i += 1\n      }\n      System.arraycopy(allKeys, 0, keyBuffer, 0, numDistributionKeys)\n      keyBuffer(numDistributionKeys) = null\n\n      // The partition code determines where we are sending the rows.\n      reduceKey.partitionCode = Arrays.hashCode(keyBuffer)\n\n      Iterator.tabulate(distinctColumnIndices.length) { distinctI =>\n        var i = 0\n        val distinctParameters: Array[Object] = distinctBuffer(distinctI)\n        while (i < distinctColumnIndices(distinctI).length) {\n          val aggPos = distinctColumnIndices(distinctI)(i)\n          distinctParameters(i) = allKeys(aggPos)\n          i += 1\n        }\n\n        keyBuffer(numDistributionKeys) = new StandardUnion(distinctI.toByte, distinctParameters)\n        reduceKey.bytesWritable = keySer.serialize(\n          keyBuffer, keyObjInspector).asInstanceOf[BytesWritable]\n\n        // Only send the value if this is the first distinct expression.\n        // Type is (ReduceKeyMapSide, BytesWritable)\n        if (distinctI == 0) {\n          (reduceKey, value)\n        } else {\n          (reduceKey, emptyValue)\n        }\n      }\n    }\n  }\n\n}\n', 'message': 'Some cleanups.\n', 'author_name': 'Reynold Xin', 'lang': 'scala'}