Fix DataSourceTest et al.

Test suites that extend DataSourceTest used to have this weird
implicit SQLContext that was created in the constructor. This was
failing tests because the base SQLContext is not ready until after
the first test is run. A minor refactor was required to fix the
resulting NPEs.

This commit also fixes test suites that need to materialize the
test data. These suites were materializing them in the constructor
before the SQLContext was ready.

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

@@ -27,7 +27,7 @@ import org.apache.spark.sql.test.SQLTestUtils
 class JoinSuite extends QueryTest with BeforeAndAfterEach with SQLTestUtils {
   import testImplicits._
 
-  loadTestData()
+  setupTestData()
 
   test("equi-join is hash-join") {
     val x = testData2.as("x")

sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala

@@ -71,12 +71,6 @@ class QueryTest extends PlanTest {
     checkAnswer(df, expectedAnswer.collect())
   }
 
-  def sqlTest(sqlString: String, expectedAnswer: Seq[Row])(implicit sqlContext: SQLContext) {
-    test(sqlString) {
-      checkAnswer(sqlContext.sql(sqlString), expectedAnswer)
-    }
-  }
-
   /**
    * Asserts that a given [[DataFrame]] will be executed using the given number of cached results.
    */

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -36,7 +36,7 @@ class MyDialect extends DefaultParserDialect
 class SQLQuerySuite extends QueryTest with SQLTestUtils {
   import testImplicits._
 
-  loadTestData()
+  setupTestData()
 
   test("having clause") {
     Seq(("one", 1), ("two", 2), ("three", 3), ("one", 5)).toDF("k", "v").registerTempTable("hav")

sql/core/src/test/scala/org/apache/spark/sql/columnar/InMemoryColumnarQuerySuite.scala

@@ -28,7 +28,7 @@ import org.apache.spark.storage.StorageLevel.MEMORY_ONLY
 class InMemoryColumnarQuerySuite extends QueryTest with SQLTestUtils {
   import testImplicits._
 
-  loadTestData()
+  setupTestData()
 
   test("simple columnar query") {
     val plan = ctx.executePlan(testData.logicalPlan).executedPlan

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

@@ -30,7 +30,7 @@ import org.apache.spark.sql.{execution, Row, SQLConf}
 class PlannerSuite extends SparkFunSuite with SQLTestUtils {
   import testImplicits._
 
-  loadTestData()
+  setupTestData()
 
   private def testPartialAggregationPlan(query: LogicalPlan): Unit = {
     val _ctx = ctx

sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlSerializer2Suite.scala

@@ -73,6 +73,7 @@ abstract class SparkSqlSerializer2Suite extends QueryTest with SQLTestUtils {
   var useSerializer2: Boolean = _
 
   override def beforeAll(): Unit = {
+    super.beforeAll()
     numShufflePartitions = ctx.conf.numShufflePartitions
     useSerializer2 = ctx.conf.useSqlSerializer2
 
@@ -111,8 +112,6 @@ abstract class SparkSqlSerializer2Suite extends QueryTest with SQLTestUtils {
       }
 
     ctx.createDataFrame(rdd, schema).registerTempTable("shuffle")
-
-    super.beforeAll()
   }
 
   override def afterAll(): Unit = {

sql/core/src/test/scala/org/apache/spark/sql/execution/TungstenSortSuite.scala

@@ -30,11 +30,13 @@ import org.apache.spark.sql.types._
 class TungstenSortSuite extends SparkPlanTest {
 
   override def beforeAll(): Unit = {
+    super.beforeAll()
     ctx.conf.setConf(SQLConf.CODEGEN_ENABLED, true)
   }
 
   override def afterAll(): Unit = {
     ctx.conf.setConf(SQLConf.CODEGEN_ENABLED, SQLConf.CODEGEN_ENABLED.defaultValue.get)
+    super.afterAll()
   }
 
   test("sort followed by limit") {

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

@@ -24,22 +24,21 @@ import org.apache.spark.sql.catalyst.plans.{FullOuter, LeftOuter, RightOuter}
 import org.apache.spark.sql.execution.{SparkPlan, SparkPlanTest}
 
 class OuterJoinSuite extends SparkPlanTest {
-
-  val left = Seq(
+  private lazy val left = Seq(
     (1, 2.0),
     (2, 1.0),
     (3, 3.0)
   ).toDF("a", "b")
 
-  val right = Seq(
+  private lazy val right = Seq(
     (2, 3.0),
     (3, 2.0),
     (4, 1.0)
   ).toDF("c", "d")
 
-  val leftKeys: List[Expression] = 'a :: Nil
-  val rightKeys: List[Expression] = 'c :: Nil
-  val condition = Some(LessThan('b, 'd))
+  private val leftKeys: List[Expression] = 'a :: Nil
+  private val rightKeys: List[Expression] = 'c :: Nil
+  private val condition = Some(LessThan('b, 'd))
 
   test("shuffled hash outer join") {
     checkAnswer2(left, right, (left: SparkPlan, right: SparkPlan) =>

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

@@ -24,24 +24,24 @@ import org.apache.spark.sql.execution.{SparkPlan, SparkPlanTest}
 
 
 class SemiJoinSuite extends SparkPlanTest {
-  val left = Seq(
+  private lazy val left = Seq(
     (1, 2.0),
     (1, 2.0),
     (2, 1.0),
     (2, 1.0),
     (3, 3.0)
   ).toDF("a", "b")
 
-  val right = Seq(
+  private lazy val right = Seq(
     (2, 3.0),
     (2, 3.0),
     (3, 2.0),
     (4, 1.0)
   ).toDF("c", "d")
 
-  val leftKeys: List[Expression] = 'a :: Nil
-  val rightKeys: List[Expression] = 'c :: Nil
-  val condition = Some(LessThan('b, 'd))
+  private val leftKeys: List[Expression] = 'a :: Nil
+  private val rightKeys: List[Expression] = 'c :: Nil
+  private val condition = Some(LessThan('b, 'd))
 
   test("left semi join hash") {
     checkAnswer2(left, right, (left: SparkPlan, right: SparkPlan) =>

sql/core/src/test/scala/org/apache/spark/sql/sources/CreateTableAsSelectSuite.scala

@@ -26,29 +26,27 @@ import org.apache.spark.sql.execution.datasources.DDLException
 import org.apache.spark.util.Utils
 
 class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
-
-  import caseInsensitiveContext.sql
-
   private lazy val sparkContext = caseInsensitiveContext.sparkContext
-
-  var path: File = null
+  private var path: File = null
 
   override def beforeAll(): Unit = {
+    super.beforeAll()
     path = Utils.createTempDir()
     val rdd = sparkContext.parallelize((1 to 10).map(i => s"""{"a":$i, "b":"str${i}"}"""))
     caseInsensitiveContext.read.json(rdd).registerTempTable("jt")
   }
 
   override def afterAll(): Unit = {
     caseInsensitiveContext.dropTempTable("jt")
+    super.afterAll()
   }
 
   after {
     Utils.deleteRecursively(path)
   }
 
   test("CREATE TEMPORARY TABLE AS SELECT") {
-    sql(
+    caseInsensitiveContext.sql(
       s"""
         |CREATE TEMPORARY TABLE jsonTable
         |USING org.apache.spark.sql.json.DefaultSource
@@ -59,8 +57,8 @@ class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
       """.stripMargin)
 
     checkAnswer(
-      sql("SELECT a, b FROM jsonTable"),
-      sql("SELECT a, b FROM jt").collect())
+      caseInsensitiveContext.sql("SELECT a, b FROM jsonTable"),
+      caseInsensitiveContext.sql("SELECT a, b FROM jt").collect())
 
     caseInsensitiveContext.dropTempTable("jsonTable")
   }
@@ -72,7 +70,7 @@ class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
     path.setWritable(false)
 
     val e = intercept[IOException] {
-      sql(
+      caseInsensitiveContext.sql(
         s"""
            |CREATE TEMPORARY TABLE jsonTable
            |USING org.apache.spark.sql.json.DefaultSource
@@ -81,15 +79,15 @@ class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
            |) AS
            |SELECT a, b FROM jt
         """.stripMargin)
-      sql("SELECT a, b FROM jsonTable").collect()
+      caseInsensitiveContext.sql("SELECT a, b FROM jsonTable").collect()
     }
     assert(e.getMessage().contains("Unable to clear output directory"))
 
     path.setWritable(true)
   }
 
   test("create a table, drop it and create another one with the same name") {
-    sql(
+    caseInsensitiveContext.sql(
       s"""
         |CREATE TEMPORARY TABLE jsonTable
         |USING org.apache.spark.sql.json.DefaultSource
@@ -100,11 +98,11 @@ class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
       """.stripMargin)
 
     checkAnswer(
-      sql("SELECT a, b FROM jsonTable"),
-      sql("SELECT a, b FROM jt").collect())
+      caseInsensitiveContext.sql("SELECT a, b FROM jsonTable"),
+      caseInsensitiveContext.sql("SELECT a, b FROM jt").collect())
 
     val message = intercept[DDLException]{
-      sql(
+      caseInsensitiveContext.sql(
         s"""
         |CREATE TEMPORARY TABLE IF NOT EXISTS jsonTable
         |USING org.apache.spark.sql.json.DefaultSource
@@ -119,7 +117,7 @@ class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
       "CREATE TEMPORARY TABLE IF NOT EXISTS should not be allowed.")
 
     // Overwrite the temporary table.
-    sql(
+    caseInsensitiveContext.sql(
       s"""
         |CREATE TEMPORARY TABLE jsonTable
         |USING org.apache.spark.sql.json.DefaultSource
@@ -129,14 +127,14 @@ class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
         |SELECT a * 4 FROM jt
       """.stripMargin)
     checkAnswer(
-      sql("SELECT * FROM jsonTable"),
-      sql("SELECT a * 4 FROM jt").collect())
+      caseInsensitiveContext.sql("SELECT * FROM jsonTable"),
+      caseInsensitiveContext.sql("SELECT a * 4 FROM jt").collect())
 
     caseInsensitiveContext.dropTempTable("jsonTable")
     // Explicitly delete the data.
     if (path.exists()) Utils.deleteRecursively(path)
 
-    sql(
+    caseInsensitiveContext.sql(
       s"""
         |CREATE TEMPORARY TABLE jsonTable
         |USING org.apache.spark.sql.json.DefaultSource
@@ -147,15 +145,15 @@ class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
       """.stripMargin)
 
     checkAnswer(
-      sql("SELECT * FROM jsonTable"),
-      sql("SELECT b FROM jt").collect())
+      caseInsensitiveContext.sql("SELECT * FROM jsonTable"),
+      caseInsensitiveContext.sql("SELECT b FROM jt").collect())
 
     caseInsensitiveContext.dropTempTable("jsonTable")
   }
 
   test("CREATE TEMPORARY TABLE AS SELECT with IF NOT EXISTS is not allowed") {
     val message = intercept[DDLException]{
-      sql(
+      caseInsensitiveContext.sql(
         s"""
         |CREATE TEMPORARY TABLE IF NOT EXISTS jsonTable
         |USING org.apache.spark.sql.json.DefaultSource
@@ -172,7 +170,7 @@ class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
 
   test("a CTAS statement with column definitions is not allowed") {
     intercept[DDLException]{
-      sql(
+      caseInsensitiveContext.sql(
         s"""
         |CREATE TEMPORARY TABLE jsonTable (a int, b string)
         |USING org.apache.spark.sql.json.DefaultSource
@@ -185,7 +183,7 @@ class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
   }
 
   test("it is not allowed to write to a table while querying it.") {
-    sql(
+    caseInsensitiveContext.sql(
       s"""
         |CREATE TEMPORARY TABLE jsonTable
         |USING org.apache.spark.sql.json.DefaultSource
@@ -196,7 +194,7 @@ class CreateTableAsSelectSuite extends DataSourceTest with BeforeAndAfterAll {
       """.stripMargin)
 
     val message = intercept[AnalysisException] {
-      sql(
+      caseInsensitiveContext.sql(
         s"""
         |CREATE TEMPORARY TABLE jsonTable
         |USING org.apache.spark.sql.json.DefaultSource

sql/core/src/test/scala/org/apache/spark/sql/sources/DataSourceTest.scala

@@ -24,11 +24,18 @@ import org.apache.spark.sql.test.SQLTestUtils
 
 
 abstract class DataSourceTest extends QueryTest with BeforeAndAfter with SQLTestUtils {
+
   // We want to test some edge cases.
-  protected implicit lazy val caseInsensitiveContext = {
+  protected lazy val caseInsensitiveContext: SQLContext = {
     val ctx = new SQLContext(sqlContext.sparkContext)
     ctx.setConf(SQLConf.CASE_SENSITIVE, false)
     ctx
   }
 
+  protected def sqlTest(sqlString: String, expectedAnswer: Seq[Row]) {
+    test(sqlString) {
+      checkAnswer(caseInsensitiveContext.sql(sqlString), expectedAnswer)
+    }
+  }
+
 }

sql/core/src/test/scala/org/apache/spark/sql/sources/FilteredScanSuite.scala

@@ -97,10 +97,8 @@ object FiltersPushed {
 
 class FilteredScanSuite extends DataSourceTest {
 
-  import caseInsensitiveContext.sql
-
   before {
-    sql(
+    caseInsensitiveContext.sql(
       """
         |CREATE TEMPORARY TABLE oneToTenFiltered
         |USING org.apache.spark.sql.sources.FilteredScanSource
@@ -237,7 +235,7 @@ class FilteredScanSuite extends DataSourceTest {
 
   def testPushDown(sqlString: String, expectedCount: Int): Unit = {
     test(s"PushDown Returns $expectedCount: $sqlString") {
-      val queryExecution = sql(sqlString).queryExecution
+      val queryExecution = caseInsensitiveContext.sql(sqlString).queryExecution
       val rawPlan = queryExecution.executedPlan.collect {
         case p: execution.PhysicalRDD => p
       } match {