Added isCached() predicate

Author: liancheng

sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala

@@ -188,6 +188,14 @@ class SQLContext(@transient val sparkContext: SparkContext)
     }
   }
 
+  def isCached(tableName: String): Boolean = {
+    val relation = catalog.lookupRelation(None, tableName)
+    EliminateAnalysisOperators(relation) match {
+      case SparkLogicalPlan(_: InMemoryColumnarTableScan) => true
+      case _ => false
+    }
+  }
+
   protected[sql] class SparkPlanner extends SparkStrategies {
     val sparkContext = self.sparkContext
 

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

@@ -77,11 +77,13 @@ class CachedTableSuite extends QueryTest {
       case _: InMemoryColumnarTableScan => // Found evidence of caching
       case _ => fail(s"Table 'testData' should be cached")
     }
+    assert(TestSQLContext.isCached("testData"), "Table 'testData' should be cached")
 
     TestSQLContext.sql("UNCACHE TABLE testData")
     TestSQLContext.table("testData").queryExecution.executedPlan match {
       case _: InMemoryColumnarTableScan => fail(s"Table 'testData' should not be cached")
       case _ => // Found evidence of uncaching
     }
+    assert(!TestSQLContext.isCached("testData"), "Table 'testData' should not be cached")
   }
 }

sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala

@@ -63,11 +63,13 @@ class CachedTableSuite extends HiveComparisonTest {
       case _: InMemoryColumnarTableScan => // Found evidence of caching
       case _ => fail(s"Table 'src' should be cached")
     }
+    assert(TestHive.isCached("src"), "Table 'src' should be cached")
 
     TestHive.hql("UNCACHE TABLE src")
     TestHive.table("src").queryExecution.executedPlan match {
       case _: InMemoryColumnarTableScan => fail(s"Table 'src' should not be cached")
       case _ => // Found evidence of uncaching
     }
+    assert(!TestHive.isCached("src"), "Table 'src' should not be cached")
   }
 }