[SPARK-37266][SQL] View text can only be SELECT queries

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -36,7 +36,7 @@ import org.apache.spark.sql.catalyst._
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.analysis.FunctionRegistry.FunctionBuilder
 import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, ExpressionInfo, UpCast}
-import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParserInterface}
+import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException, ParserInterface}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, SubqueryAlias, View}
 import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, StringUtils}
 import org.apache.spark.sql.connector.catalog.CatalogManager
@@ -877,7 +877,12 @@ class SessionCatalog(
     }
     val viewConfigs = metadata.viewSQLConfigs
     val parsedPlan = SQLConf.withExistingConf(View.effectiveSQLConf(viewConfigs, isTempView)) {
-      parser.parsePlan(viewText)
+      try {
+        parser.parseQuery(viewText)
+      } catch {
+        case _: ParseException =>
+          throw QueryCompilationErrors.invalidViewText(viewText, metadata.qualifiedName)
+      }
     }
     val projectList = if (!isHiveCreatedView(metadata)) {
       val viewColumnNames = if (metadata.viewQueryColumnNames.isEmpty) {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseDriver.scala

@@ -73,6 +73,11 @@ abstract class AbstractSqlParser extends ParserInterface with SQLConfHelper with
     astBuilder.visitSingleTableSchema(parser.singleTableSchema())
   }
 
+  /** Creates LogicalPlan for a given SQL string of query. */
+  override def parseQuery(sqlText: String): LogicalPlan = parse(sqlText) { parser =>
+    astBuilder.visitQuery(parser.query())
+  }
+
   /** Creates LogicalPlan for a given SQL string. */
   override def parsePlan(sqlText: String): LogicalPlan = parse(sqlText) { parser =>
     astBuilder.visitSingleStatement(parser.singleStatement()) match {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserInterface.scala

@@ -70,4 +70,10 @@ trait ParserInterface {
    */
   @throws[ParseException]("Text cannot be parsed to a DataType")
   def parseDataType(sqlText: String): DataType
+
+  /**
+   * Parse a query string to a [[LogicalPlan]].
+   */
+  @throws[ParseException]("Text cannot be parsed to a LogicalPlan")
+  def parseQuery(sqlText: String): LogicalPlan
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -2366,4 +2366,9 @@ object QueryCompilationErrors {
   def tableIndexNotSupportedError(errorMessage: String): Throwable = {
     new AnalysisException(errorMessage)
   }
+
+  def invalidViewText(viewText: String, tableName: String): Throwable = {
+    new AnalysisException(
+      s"Invalid view text: $viewText. The view $tableName may have been tampered with")
+  }
 }

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

@@ -431,6 +431,9 @@ case class MyParser(spark: SparkSession, delegate: ParserInterface) extends Pars
 
   override def parseDataType(sqlText: String): DataType =
     delegate.parseDataType(sqlText)
+
+  override def parseQuery(sqlText: String): LogicalPlan =
+    delegate.parseQuery(sqlText)
 }
 
 object MyExtensions {

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala

@@ -324,4 +324,23 @@ class HiveCommandSuite extends QueryTest with SQLTestUtils with TestHiveSingleto
       FsConstants.LOCAL_FS_URI, workingDir, new Path("kv1.txt"))
     assert(r === new Path(s"$workingDir/kv1.txt"))
   }
+
+  test("SPARK-37266: Optimize the analysis for view text of persistent view and" +
+    " fix security vulnerabilities caused by sql tampering") {
+    sql("CREATE VIEW parquet_view2 as select * from parquet_tab4")
+    val table = hiveContext.sessionState.catalog.getTableMetadata(TableIdentifier("parquet_view2"))
+    try {
+      // Simulate the behavior of hackers
+      val tamperedViewText = "drop view parquet_view2"
+      val tamperedTable = table.copy(viewText = Some(tamperedViewText))
+      hiveContext.sessionState.catalog.alterTable(tamperedTable)
+      val message = intercept[AnalysisException] {
+        sql("SELECT * FROM parquet_view2")
+      }.getMessage
+      assert(message.contains(s"Invalid view text: $tamperedViewText." +
+        s" The view ${table.qualifiedName} may have been tampered with"))
+    } finally {
+      sql("DROP VIEW parquet_view2")
+    }
+  }
 }