[SPARK-22954][SQL] Fix the problem for other versions of analyze comm…

…ands - AnalyzeColum and AnalyzePartition. Fix tests to check if the error messages are proper.

sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeColumnCommand.scala

@@ -39,9 +39,9 @@ case class AnalyzeColumnCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val sessionState = sparkSession.sessionState
-    val db = tableIdent.database.getOrElse(sessionState.catalog.getCurrentDatabase)
-    val tableIdentWithDB = TableIdentifier(tableIdent.table, Some(db))
-    val tableMeta = sessionState.catalog.getTableMetadata(tableIdentWithDB)
+    val db = tableIdent.database
+    val tableIdentWithDB = TableIdentifier(tableIdent.table, db)
+    val tableMeta = sessionState.catalog.getTempViewOrPermanentTableMetadata(tableIdentWithDB)
     if (tableMeta.tableType == CatalogTableType.VIEW) {
       throw new AnalysisException("ANALYZE TABLE is not supported on views.")
     }

sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzePartitionCommand.scala

@@ -75,9 +75,9 @@ case class AnalyzePartitionCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val sessionState = sparkSession.sessionState
-    val db = tableIdent.database.getOrElse(sessionState.catalog.getCurrentDatabase)
-    val tableIdentWithDB = TableIdentifier(tableIdent.table, Some(db))
-    val tableMeta = sessionState.catalog.getTableMetadata(tableIdentWithDB)
+    val db = tableIdent.database
+    val tableIdentWithDB = TableIdentifier(tableIdent.table, db)
+    val tableMeta = sessionState.catalog.getTempViewOrPermanentTableMetadata(tableIdentWithDB)
     if (tableMeta.tableType == CatalogTableType.VIEW) {
       throw new AnalysisException("ANALYZE TABLE is not supported on views.")
     }
@@ -88,7 +88,8 @@ case class AnalyzePartitionCommand(
 
     if (partitions.isEmpty) {
       if (partitionValueSpec.isDefined) {
-        throw new NoSuchPartitionException(db, tableIdent.table, partitionValueSpec.get)
+        throw new NoSuchPartitionException(db.getOrElse(sessionState.catalog.getCurrentDatabase),
+          tableIdent.table, partitionValueSpec.get)
       } else {
         // the user requested to analyze all partitions for a table which has no partitions
         // return normally, since there is nothing to do

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

@@ -155,14 +155,22 @@ abstract class SQLViewSuite extends QueryTest with SQLTestUtils {
       assertNoSuchTable(s"TRUNCATE TABLE $viewName")
       assertNoSuchTable(s"SHOW CREATE TABLE $viewName")
       assertNoSuchTable(s"SHOW PARTITIONS $viewName")
-      assertAnalysisException(s"ANALYZE TABLE $viewName COMPUTE STATISTICS")
-      assertAnalysisException(s"ANALYZE TABLE $viewName COMPUTE STATISTICS FOR COLUMNS id")
     }
   }
-
-  private def assertAnalysisException(query: String): Unit = {
-    intercept[AnalysisException] {
-      sql(query)
+
+  test("SPARK-22954 - Issue AnalysisException when analysis is run on view") {
+    val viewName = "testView"
+    val analyzeNotSupportedOnViewsMsg = "ANALYZE TABLE is not supported on views."
+    withTempView(viewName) {
+      spark.range(10).createTempView(viewName)
+
+      assert(intercept[AnalysisException] {
+        sql(s"ANALYZE TABLE $viewName COMPUTE STATISTICS")
+      }.getMessage.contains(analyzeNotSupportedOnViewsMsg))
+
+      assert(intercept[AnalysisException] {
+        sql(s"ANALYZE TABLE $viewName COMPUTE STATISTICS FOR COLUMNS id")
+      }.getMessage.contains(analyzeNotSupportedOnViewsMsg))
     }
   }