<NME> SharkRunner.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

import org.apache.hadoop.hive.metastore.MetaStoreUtils.DEFAULT_DATABASE_NAME

import shark.api.JavaSharkContext
import shark.memstore2.MemoryMetadataManager


object SharkRunner {

  val WAREHOUSE_PATH = TestUtils.getWarehousePath()
  val METASTORE_PATH = TestUtils.getMetastorePath()
  val MASTER = "local"

  var sharkMetastore: MemoryMetadataManager = _


  def init(): SharkContext = {
  	if (sc == null) {
      sc = SharkEnv.initWithSharkContext("shark-sql-suite-testing", MASTER)


      sc.runSql("set javax.jdo.option.ConnectionURL=jdbc:derby:;databaseName=" +
      METASTORE_PATH + ";create=true")
      sc.runSql("set hive.metastore.warehouse.dir=" + WAREHOUSE_PATH)
      sc.runSql("set shark.test.data.path=" + TestUtils.dataFilePath)

      // second db
      sc.sql("create database if not exists seconddb")

      loadTables()
    }
    sc
  }

  def initWithJava(): JavaSharkContext = synchronized {
    if (javaSc == null) {
   * Tables accessible by any test. Their properties should remain constant across
   * tests.
   */
  def loadTables() {
    require(sc != null, "call init() to instantiate a SharkContext first")

    // Use the default namespace
   * tests.
   */
  def loadTables() = synchronized {
    require(sc != null, "call init() to instantiate a SharkContext first")

    // Use the default namespace
    sc.runSql("USE " + DEFAULT_DATABASE_NAME)

    // test
    sc.runSql("drop table if exists test")
    sc.runSql("CREATE TABLE test (key INT, val STRING)")
    sc.runSql("LOAD DATA LOCAL INPATH '${hiveconf:shark.test.data.path}/kv1.txt' INTO TABLE test")
    sc.runSql("drop table if exists test_cached")
    sc.runSql("CREATE TABLE test_cached tblproperties('shark.cache'='memory') AS SELECT * FROM test")

    // test_null
    sc.runSql("drop table if exists test_null")
    sc.runSql("CREATE TABLE test_null (key INT, val STRING)")
    sc.runSql("""LOAD DATA LOCAL INPATH '${hiveconf:shark.test.data.path}/kv3.txt'
      INTO TABLE test_null""")
    sc.runSql("drop table if exists test_null_cached")
    sc.runSql("CREATE TABLE test_null_cached tblproperties('shark.cache'='memory') AS SELECT * FROM test_null")

    // clicks
    sc.runSql("drop table if exists clicks")
    sc.runSql("""create table clicks (id int, click int)
      row format delimited fields terminated by '\t'""")
    sc.runSql("""load data local inpath '${hiveconf:shark.test.data.path}/clicks.txt'
      OVERWRITE INTO TABLE clicks""")
    sc.runSql("drop table if exists clicks_cached")
    sc.runSql("create table clicks_cached tblproperties('shark.cache'='memory') as select * from clicks")

    // users
    sc.runSql("drop table if exists users")
    sc.runSql("""create table users (id int, name string)
      row format delimited fields terminated by '\t'""")
    sc.runSql("""load data local inpath '${hiveconf:shark.test.data.path}/users.txt'
      OVERWRITE INTO TABLE users""")
    sc.runSql("drop table if exists users_cached")
    sc.sql("LOAD DATA LOCAL INPATH '${hiveconf:shark.test.data.path}/test1.txt' INTO TABLE test1")
    sc.sql("drop table if exists test1_cached")
    sc.sql("CREATE TABLE test1_cached AS SELECT * FROM test1")
  }

  def expectSql(sql: String, expectedResults: Array[String], sort: Boolean = true) {
    sc.sql("drop table if exists test1_cached")
    sc.sql("CREATE TABLE test1_cached tblproperties('shark.cache'='memory') AS SELECT * FROM test1")
    Unit
  }

  def expectSql(sql: String, expectedResults: Array[String], sort: Boolean = true) {
    val sharkResults: Array[String] = sc.runSql(sql).results.map(_.mkString("\t")).toArray
    val results = if (sort) sharkResults.sortWith(_ < _) else sharkResults
    val expected = if (sort) expectedResults.sortWith(_ < _) else expectedResults
    assert(results.corresponds(expected)(_.equals(_)),
      "In SQL: " + sql + "\n" +
      "Expected: " + expected.mkString("\n") + "; got " + results.mkString("\n"))
  }

  // A shortcut for single row results.
  def expectSql(sql: String, expectedResult: String) {
    expectSql(sql, Array(expectedResult))
  }

}

<MSG> Some docs for MemoryTable, updates based on review.
<DFF> @@ -32,7 +32,7 @@ object SharkRunner {
   var sharkMetastore: MemoryMetadataManager = _
 
 
-  def init(): SharkContext = {
+  def init(): SharkContext = synchronized {
   	if (sc == null) {
       sc = SharkEnv.initWithSharkContext("shark-sql-suite-testing", MASTER)
 
@@ -55,7 +55,7 @@ object SharkRunner {
    * Tables accessible by any test. Their properties should remain constant across
    * tests.
    */
-  def loadTables() {
+  def loadTables() = synchronized {
     require(sc != null, "call init() to instantiate a SharkContext first")
 
     // Use the default namespace
@@ -101,6 +101,7 @@ object SharkRunner {
     sc.sql("LOAD DATA LOCAL INPATH '${hiveconf:shark.test.data.path}/test1.txt' INTO TABLE test1")
     sc.sql("drop table if exists test1_cached")
     sc.sql("CREATE TABLE test1_cached AS SELECT * FROM test1")
+    Unit
   }
 
   def expectSql(sql: String, expectedResults: Array[String], sort: Boolean = true) {

{'commit': 'diff --git a/src/main/scala/shark/execution/MapSplitPruning.scala b/src/main/scala/shark/execution/MapSplitPruning.scala\nindex 4de38fc..976ebf5 100644\n--- a/src/main/scala/shark/execution/MapSplitPruning.scala\n+++ b/src/main/scala/shark/execution/MapSplitPruning.scala\n@@ -181,16 +181,16 @@ object MapSplitPruning {\n     }\n   }\n \n-  private def getIDStructField(field: StructField): IDStructField = {\n-    field match {\n-      case myField: MyField => {\n-        MapSplitPruningHelper.getStructFieldFromUnionOIField(myField)\n-          .asInstanceOf[IDStructField]\n-      }\n-      case idStructField: IDStructField => idStructField\n-      case otherFieldType: Any => {\n-        throw new Exception("Unrecognized StructField: " + otherFieldType)\n-      }\n+  private def getIDStructField(field: StructField): IDStructField = field match {\n+    case myField: MyField => {\n+      // For partitioned tables, the ColumnarStruct\'s IDStructFields are enclosed inside\n+      // the Hive UnionStructObjectInspector\'s MyField objects.\n+      MapSplitPruningHelper.getStructFieldFromUnionOIField(myField)\n+        .asInstanceOf[IDStructField]\n+    }\n+    case idStructField: IDStructField => idStructField\n+    case otherFieldType: Any => {\n+      throw new Exception("Unrecognized StructField: " + otherFieldType)\n     }\n   }\n }\ndiff --git a/src/main/scala/shark/memstore2/MemoryTable.scala b/src/main/scala/shark/memstore2/MemoryTable.scala\nindex bfd5163..1a971d4 100644\n--- a/src/main/scala/shark/memstore2/MemoryTable.scala\n+++ b/src/main/scala/shark/memstore2/MemoryTable.scala\n@@ -33,32 +33,45 @@ private[shark] class MemoryTable(\n     cacheMode: CacheType.CacheType)\n   extends Table(databaseName, tableName, cacheMode) {\n \n-  var rddValueOpt: Option[RDDValue] = None\n+  private var _rddValueOpt: Option[RDDValue] = None\n \n+  /**\n+   * Sets the RDD and stats fields the `_rddValueOpt`. Used for INSERT/LOAD OVERWRITE.\n+   * @param newRDD The table\'s data.\n+   * @param newStats Stats for each TablePartition in `newRDD`.\n+   * @return The previous (RDD, stats) pair for this table.\n+   */\n   def put(\n   \t  newRDD: RDD[TablePartition],\n   \t  newStats: collection.Map[Int, TablePartitionStats] = new HashMap[Int, TablePartitionStats]()\n   \t): Option[(RDD[TablePartition], collection.Map[Int, TablePartitionStats])] = {\n-  \tval prevRDDAndStatsOpt = rddValueOpt.map(_.toTuple)\n-  \tif (rddValueOpt.isDefined) {\n-  \t  rddValueOpt.foreach { rddValue =>\n+  \tval prevRDDAndStatsOpt = _rddValueOpt.map(_.toTuple)\n+  \tif (_rddValueOpt.isDefined) {\n+  \t  _rddValueOpt.foreach { rddValue =>\n   \t  \trddValue.rdd = newRDD\n   \t  \trddValue.stats = newStats\n   \t  }\n   \t} else {\n-      rddValueOpt = Some(new RDDValue(newRDD, newStats))\n+      _rddValueOpt = Some(new RDDValue(newRDD, newStats))\n   \t}\n     prevRDDAndStatsOpt \n   }\n \n+  /**\n+   * Used for append operations, such as INSERT and LOAD INTO.\n+   *\n+   * @param newRDD Data to append to the table.\n+   * @param newStats Stats for each TablePartition in `newRDD`.\n+   * @return The previous (RDD, stats) pair for this table.\n+   */\n   def update(\n   \t  newRDD: RDD[TablePartition],\n   \t  newStats: Buffer[(Int, TablePartitionStats)]\n   \t): Option[(RDD[TablePartition], collection.Map[Int, TablePartitionStats])] = {\n-    val prevRDDAndStatsOpt = rddValueOpt.map(_.toTuple)\n-  \tif (rddValueOpt.isDefined) {\n+    val prevRDDAndStatsOpt = _rddValueOpt.map(_.toTuple)\n+  \tif (_rddValueOpt.isDefined) {\n       val (prevRDD, prevStats) = (prevRDDAndStatsOpt.get._1, prevRDDAndStatsOpt.get._2)\n-      val updatedRDDValue = rddValueOpt.get\n+      val updatedRDDValue = _rddValueOpt.get\n       updatedRDDValue.rdd = RDDUtils.unionAndFlatten(prevRDD, newRDD)\n       updatedRDDValue.stats = Table.mergeStats(newStats, prevStats).toMap\n     } else {\n@@ -67,8 +80,8 @@ private[shark] class MemoryTable(\n     prevRDDAndStatsOpt\n   }\n \n-  def getRDD = rddValueOpt.map(_.rdd)\n+  def getRDD = _rddValueOpt.map(_.rdd)\n \n-  def getStats = rddValueOpt.map(_.stats)\n+  def getStats = _rddValueOpt.map(_.stats)\n \n }\ndiff --git a/src/test/scala/shark/SharkRunner.scala b/src/test/scala/shark/SharkRunner.scala\nindex 5ea8047..754737b 100644\n--- a/src/test/scala/shark/SharkRunner.scala\n+++ b/src/test/scala/shark/SharkRunner.scala\n@@ -32,7 +32,7 @@ object SharkRunner {\n   var sharkMetastore: MemoryMetadataManager = _\n \n \n-  def init(): SharkContext = {\n+  def init(): SharkContext = synchronized {\n   \tif (sc == null) {\n       sc = SharkEnv.initWithSharkContext("shark-sql-suite-testing", MASTER)\n \n@@ -55,7 +55,7 @@ object SharkRunner {\n    * Tables accessible by any test. Their properties should remain constant across\n    * tests.\n    */\n-  def loadTables() {\n+  def loadTables() = synchronized {\n     require(sc != null, "call init() to instantiate a SharkContext first")\n \n     // Use the default namespace\n@@ -101,6 +101,7 @@ object SharkRunner {\n     sc.sql("LOAD DATA LOCAL INPATH \'${hiveconf:shark.test.data.path}/test1.txt\' INTO TABLE test1")\n     sc.sql("drop table if exists test1_cached")\n     sc.sql("CREATE TABLE test1_cached AS SELECT * FROM test1")\n+    Unit\n   }\n \n   def expectSql(sql: String, expectedResults: Array[String], sort: Boolean = true) {\n', 'commit_subject': 'Some docs for MemoryTable, updates based on review.', 'repo_name': 'amplab/shark', 'committer': {'name': 'Harvey Feng', 'email': '8e57c2dd36c1d38cf087beed5b323e05905bfe7f@databricks.com', 'time_sec': '1389776851', 'tz_offset': '-480', 'date': {'seconds': '1389776851'}}, 'commit_tree': '982f32b7fb9e4050e40628f8288e913057f9f52e', 'commit_message': 'Some docs for MemoryTable, updates based on review.', 'language_name': 'Java', 'license': 'apache-2.0', 'hunks': '@@ -32,7 +32,7 @@ object SharkRunner {\n   var sharkMetastore: MemoryMetadataManager = _\n \n \n-  def init(): SharkContext = {\n+  def init(): SharkContext = synchronized {\n   \tif (sc == null) {\n       sc = SharkEnv.initWithSharkContext("shark-sql-suite-testing", MASTER)\n \n@@ -55,7 +55,7 @@ object SharkRunner {\n    * Tables accessible by any test. Their properties should remain constant across\n    * tests.\n    */\n-  def loadTables() {\n+  def loadTables() = synchronized {\n     require(sc != null, "call init() to instantiate a SharkContext first")\n \n     // Use the default namespace\n@@ -101,6 +101,7 @@ object SharkRunner {\n     sc.sql("LOAD DATA LOCAL INPATH \'${hiveconf:shark.test.data.path}/test1.txt\' INTO TABLE test1")\n     sc.sql("drop table if exists test1_cached")\n     sc.sql("CREATE TABLE test1_cached AS SELECT * FROM test1")\n+    Unit\n   }\n \n   def expectSql(sql: String, expectedResults: Array[String], sort: Boolean = true) {\n', 'addition_count': 3, 'deletion_count': 2, 'src_file': 'a/src/test/scala/shark/SharkRunner.scala', 'tgt_file': 'b/src/test/scala/shark/SharkRunner.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\n\nimport org.apache.hadoop.hive.metastore.MetaStoreUtils.DEFAULT_DATABASE_NAME\n\nimport shark.api.JavaSharkContext\nimport shark.memstore2.MemoryMetadataManager\n\n\nobject SharkRunner {\n\n  val WAREHOUSE_PATH = TestUtils.getWarehousePath()\n  val METASTORE_PATH = TestUtils.getMetastorePath()\n  val MASTER = "local"\n\n  var sharkMetastore: MemoryMetadataManager = _\n\n\n  def init(): SharkContext = {\n  \tif (sc == null) {\n      sc = SharkEnv.initWithSharkContext("shark-sql-suite-testing", MASTER)\n\n\n      sc.runSql("set javax.jdo.option.ConnectionURL=jdbc:derby:;databaseName=" +\n      METASTORE_PATH + ";create=true")\n      sc.runSql("set hive.metastore.warehouse.dir=" + WAREHOUSE_PATH)\n      sc.runSql("set shark.test.data.path=" + TestUtils.dataFilePath)\n\n      // second db\n      sc.sql("create database if not exists seconddb")\n\n      loadTables()\n    }\n    sc\n  }\n\n  def initWithJava(): JavaSharkContext = synchronized {\n    if (javaSc == null) {\n   * Tables accessible by any test. Their properties should remain constant across\n   * tests.\n   */\n  def loadTables() {\n    require(sc != null, "call init() to instantiate a SharkContext first")\n\n    // Use the default namespace\n   * tests.\n   */\n  def loadTables() = synchronized {\n    require(sc != null, "call init() to instantiate a SharkContext first")\n\n    // Use the default namespace\n    sc.runSql("USE " + DEFAULT_DATABASE_NAME)\n\n    // test\n    sc.runSql("drop table if exists test")\n    sc.runSql("CREATE TABLE test (key INT, val STRING)")\n    sc.runSql("LOAD DATA LOCAL INPATH \'${hiveconf:shark.test.data.path}/kv1.txt\' INTO TABLE test")\n    sc.runSql("drop table if exists test_cached")\n    sc.runSql("CREATE TABLE test_cached tblproperties(\'shark.cache\'=\'memory\') AS SELECT * FROM test")\n\n    // test_null\n    sc.runSql("drop table if exists test_null")\n    sc.runSql("CREATE TABLE test_null (key INT, val STRING)")\n    sc.runSql("""LOAD DATA LOCAL INPATH \'${hiveconf:shark.test.data.path}/kv3.txt\'\n      INTO TABLE test_null""")\n    sc.runSql("drop table if exists test_null_cached")\n    sc.runSql("CREATE TABLE test_null_cached tblproperties(\'shark.cache\'=\'memory\') AS SELECT * FROM test_null")\n\n    // clicks\n    sc.runSql("drop table if exists clicks")\n    sc.runSql("""create table clicks (id int, click int)\n      row format delimited fields terminated by \'\\t\'""")\n    sc.runSql("""load data local inpath \'${hiveconf:shark.test.data.path}/clicks.txt\'\n      OVERWRITE INTO TABLE clicks""")\n    sc.runSql("drop table if exists clicks_cached")\n    sc.runSql("create table clicks_cached tblproperties(\'shark.cache\'=\'memory\') as select * from clicks")\n\n    // users\n    sc.runSql("drop table if exists users")\n    sc.runSql("""create table users (id int, name string)\n      row format delimited fields terminated by \'\\t\'""")\n    sc.runSql("""load data local inpath \'${hiveconf:shark.test.data.path}/users.txt\'\n      OVERWRITE INTO TABLE users""")\n    sc.runSql("drop table if exists users_cached")\n    sc.sql("LOAD DATA LOCAL INPATH \'${hiveconf:shark.test.data.path}/test1.txt\' INTO TABLE test1")\n    sc.sql("drop table if exists test1_cached")\n    sc.sql("CREATE TABLE test1_cached AS SELECT * FROM test1")\n  }\n\n  def expectSql(sql: String, expectedResults: Array[String], sort: Boolean = true) {\n    sc.sql("drop table if exists test1_cached")\n    sc.sql("CREATE TABLE test1_cached tblproperties(\'shark.cache\'=\'memory\') AS SELECT * FROM test1")\n    Unit\n  }\n\n  def expectSql(sql: String, expectedResults: Array[String], sort: Boolean = true) {\n    val sharkResults: Array[String] = sc.runSql(sql).results.map(_.mkString("\\t")).toArray\n    val results = if (sort) sharkResults.sortWith(_ < _) else sharkResults\n    val expected = if (sort) expectedResults.sortWith(_ < _) else expectedResults\n    assert(results.corresponds(expected)(_.equals(_)),\n      "In SQL: " + sql + "\\n" +\n      "Expected: " + expected.mkString("\\n") + "; got " + results.mkString("\\n"))\n  }\n\n  // A shortcut for single row results.\n  def expectSql(sql: String, expectedResult: String) {\n    expectSql(sql, Array(expectedResult))\n  }\n\n}\n', 'message': 'Some docs for MemoryTable, updates based on review.', 'author_name': 'Harvey Feng', 'lang': 'scala'}