add missing lazys and null handling

Author: peter-toth

sql/core/src/main/scala/org/apache/spark/sql/execution/LocalTableScanExec.scala

@@ -59,7 +59,7 @@ case class LocalTableScanExec(
   }
 
   override protected def stringArgs: Iterator[Any] = {
-    if (rows.isEmpty) {
+    if (rows == null || rows.isEmpty) {
       Iterator("<empty>", output)
     } else {
       Iterator(output)

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

@@ -92,7 +92,7 @@ case class HashAggregateExec(
 
   // This is for testing. We force TungstenAggregationIterator to fall back to the unsafe row hash
   // map and/or the sort-based aggregation once it has processed a given number of input rows.
-  private val testFallbackStartsAt: Option[(Int, Int)] = {
+  private lazy val testFallbackStartsAt: Option[(Int, Int)] = {
     sqlContext.getConf("spark.sql.TungstenAggregate.testFallbackStartsAt", null) match {
       case null | "" => None
       case fallbackStartsAt =>

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala

@@ -59,7 +59,7 @@ case class ShuffleExchangeExec(
 
   override def nodeName: String = "Exchange"
 
-  private val serializer: Serializer =
+  private lazy val serializer: Serializer =
     new UnsafeRowSerializer(child.output.size, longMetric("dataSize"))
 
   @transient lazy val inputRDD: RDD[InternalRow] = child.execute()