Fix test bug where instanceWeights weren't properly passed to update methods

Author: smurching

mllib/src/main/scala/org/apache/spark/ml/tree/impl/AggUpdateUtils.scala

@@ -52,7 +52,7 @@ private[impl] object AggUpdateUtils {
       featureIndex: Int,
       featureIndexIdx: Int,
       splits: Array[Array[Split]],
-      instanceWeight: Double = 1.0): Unit = {
+      instanceWeight: Double): Unit = {
     val leftNodeFeatureOffset = agg.getFeatureOffset(featureIndexIdx)
     // Each unordered split has a corresponding bin for impurity stats of data points that fall
     // onto the left side of the split. For each unordered split, update left-side bin if applicable
@@ -75,7 +75,7 @@ private[impl] object AggUpdateUtils {
       label: Double,
       featureIndex: Int,
       featureIndexIdx: Int,
-      instanceWeight: Double = 1.0): Unit = {
+      instanceWeight: Double): Unit = {
     // The bin index of an ordered feature is just the feature value itself
     val binIndex = featureValue
     agg.update(featureIndexIdx, binIndex, label, instanceWeight)

mllib/src/main/scala/org/apache/spark/ml/tree/impl/LocalDecisionTree.scala

@@ -51,8 +51,9 @@ private[ml] object LocalDecisionTree {
     = LocalDecisionTreeUtils.rowToColumnStoreDense(input.map(_.datum.binnedFeatures))
     val labels = input.map(_.datum.label)
 
-    // Train classifier if numClasses is between 1 and 32, otherwise fit a regression model
-    // on the dataset
+    // Fit a regression model on the dataset, throwing an error if metadata indicates that
+    // we should train a classifier.
+    // TODO: Add support for training classifiers
     if (metadata.numClasses > 1 && metadata.numClasses <= 32) {
       throw new UnsupportedOperationException("Local training of a decision tree classifier is " +
         "unsupported; currently, only regression is supported")
@@ -137,13 +138,13 @@ private[ml] object LocalDecisionTree {
       from.until(to).foreach { idx =>
         val rowIndex = col.indices(idx)
         AggUpdateUtils.updateUnorderedFeature(statsAggregator, col.values(idx), labels(rowIndex),
-          featureIndex = col.featureIndex, featureIndexIdx, splits)
+          featureIndex = col.featureIndex, featureIndexIdx, splits, instanceWeight = 1.0)
       }
     } else {
       from.until(to).foreach { idx =>
         val rowIndex = col.indices(idx)
         AggUpdateUtils.updateOrderedFeature(statsAggregator, col.values(idx), labels(rowIndex),
-          featureIndex = col.featureIndex, featureIndexIdx)
+          featureIndex = col.featureIndex, featureIndexIdx, instanceWeight = 1.0)
       }
     }
   }

mllib/src/main/scala/org/apache/spark/ml/tree/impl/RandomForest.scala

@@ -281,11 +281,13 @@ private[spark] object RandomForest extends Logging {
       if (unorderedFeatures.contains(featureIndex)) {
         AggUpdateUtils.updateUnorderedFeature(agg,
           featureValue = treePoint.binnedFeatures(featureIndex), label = treePoint.label,
-          featureIndex = featureIndex, featureIndexIdx = featureIndexIdx, splits = splits)
+          featureIndex = featureIndex, featureIndexIdx = featureIndexIdx, splits = splits,
+          instanceWeight = instanceWeight)
       } else {
         AggUpdateUtils.updateOrderedFeature(agg,
           featureValue = treePoint.binnedFeatures(featureIndex), label = treePoint.label,
-          featureIndex = featureIndex, featureIndexIdx = featureIndexIdx)
+          featureIndex = featureIndex, featureIndexIdx = featureIndexIdx,
+          instanceWeight = instanceWeight)
       }
       featureIndexIdx += 1
     }