Mark LongSQLMetric private[metric] to avoid using incorrectly and leak memory

Author: zsxwing

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

@@ -81,7 +81,7 @@ private[sql] class IntSQLMetricValue(private var _value: Int) extends SQLMetricV
  * A specialized long Accumulable to avoid boxing and unboxing when using Accumulator's
  * `+=` and `add`.
  */
-private[sql] class LongSQLMetric(name: String)
+private[sql] class LongSQLMetric private[metric](name: String)
   extends SQLMetric[LongSQLMetricValue, Long](name, LongSQLMetricParam) {
 
   override def +=(term: Long): Unit = {

sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala

@@ -22,7 +22,8 @@ import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream,
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.execution.metric.LongSQLMetric
+import org.apache.spark.sql.execution.metric.SQLMetrics
+import org.apache.spark.sql.test.TestSQLContext
 import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
 import org.apache.spark.util.collection.CompactBuffer
 
@@ -36,7 +37,7 @@ class HashedRelationSuite extends SparkFunSuite {
 
   test("GeneralHashedRelation") {
     val data = Array(InternalRow(0), InternalRow(1), InternalRow(2), InternalRow(2))
-    val numDataRows = new LongSQLMetric("data")
+    val numDataRows = SQLMetrics.createLongMetric(TestSQLContext.sparkContext, "data")
     val hashed = HashedRelation(data.iterator, numDataRows, keyProjection)
     assert(hashed.isInstanceOf[GeneralHashedRelation])
 
@@ -52,7 +53,7 @@ class HashedRelationSuite extends SparkFunSuite {
 
   test("UniqueKeyHashedRelation") {
     val data = Array(InternalRow(0), InternalRow(1), InternalRow(2))
-    val numDataRows = new LongSQLMetric("data")
+    val numDataRows = SQLMetrics.createLongMetric(TestSQLContext.sparkContext, "data")
     val hashed = HashedRelation(data.iterator, numDataRows, keyProjection)
     assert(hashed.isInstanceOf[UniqueKeyHashedRelation])
 
@@ -72,7 +73,7 @@ class HashedRelationSuite extends SparkFunSuite {
   test("UnsafeHashedRelation") {
     val schema = StructType(StructField("a", IntegerType, true) :: Nil)
     val data = Array(InternalRow(0), InternalRow(1), InternalRow(2), InternalRow(2))
-    val numDataRows = new LongSQLMetric("data")
+    val numDataRows = SQLMetrics.createLongMetric(TestSQLContext.sparkContext, "data")
     val toUnsafe = UnsafeProjection.create(schema)
     val unsafeData = data.map(toUnsafe(_).copy()).toArray