[SPARK-54415][SQL] CREATE GLOBAL TEMPORARY VIEW / TEMPORARY VIEW supports IF NOT EXISTS clause

common/utils/src/main/resources/error/error-conditions.json

@@ -997,6 +997,12 @@
     ],
     "sqlState" : "0A000"
   },
+  "CREATE_OR_REPLACE_WITH_IF_NOT_EXISTS_IS_NOT_ALLOWED" : {
+    "message" : [
+      "CREATE OR REPLACE <resourceType> with IF NOT EXISTS is not allowed."
+    ],
+    "sqlState" : "42000"
+  },
   "CREATE_PERMANENT_VIEW_WITHOUT_ALIAS" : {
     "message" : [
       "Not allowed to create the permanent view <name> without explicitly assigning an alias for the expression <attr>."
@@ -6019,7 +6025,7 @@
   "TEMP_TABLE_OR_VIEW_ALREADY_EXISTS" : {
     "message" : [
       "Cannot create the temporary view <relationName> because it already exists.",
-      "Choose a different name, drop or replace the existing view,  or add the IF NOT EXISTS clause to tolerate pre-existing views."
+      "Choose a different name, drop or replace the existing view, or add the IF NOT EXISTS clause to tolerate pre-existing views."
     ],
     "sqlState" : "42P07"
   },
@@ -7567,16 +7573,6 @@
       "Empty set in <element> grouping sets is not supported."
     ]
   },
-  "_LEGACY_ERROR_TEMP_0052" : {
-    "message" : [
-      "CREATE VIEW with both IF NOT EXISTS and REPLACE is not allowed."
-    ]
-  },
-  "_LEGACY_ERROR_TEMP_0053" : {
-    "message" : [
-      "It is not allowed to define a TEMPORARY view with IF NOT EXISTS."
-    ]
-  },
   "_LEGACY_ERROR_TEMP_0056" : {
     "message" : [
       "Invalid time travel spec: <reason>."

core/src/test/scala/org/apache/spark/SparkThrowableSuite.scala

@@ -40,7 +40,7 @@ import org.apache.spark.util.Utils
  * To re-generate the error class file, run:
  * {{{
  *   SPARK_GENERATE_GOLDEN_FILES=1 build/sbt \
- *     "core/testOnly *SparkThrowableSuite -- -t \"Error conditions are correctly formatted\""
+ *     'core/testOnly *SparkThrowableSuite -- -t "Error conditions are correctly formatted"'
  * }}}
  */
 class SparkThrowableSuite extends SparkFunSuite {

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

@@ -323,7 +323,7 @@ statement
          (PARTITIONED ON identifierList) |
          (TBLPROPERTIES propertyList))*
         AS query                                                       #createView
-    | CREATE (OR REPLACE)? GLOBAL? TEMPORARY VIEW
+    | CREATE (OR REPLACE)? GLOBAL? TEMPORARY VIEW (IF errorCapturingNot EXISTS)?
         tableIdentifier (LEFT_PAREN colTypeList RIGHT_PAREN)? tableProvider
         (OPTIONS propertyList)?                                        #createTempViewUsing
     | ALTER VIEW identifierReference AS? query                         #alterViewQuery

sql/api/src/main/scala/org/apache/spark/sql/errors/CompilationErrors.scala

@@ -143,6 +143,12 @@ private[sql] trait CompilationErrors extends DataTypeErrorsBase {
       errorClass = "CANNOT_MODIFY_CONFIG",
       messageParameters = Map("key" -> toSQLConf(key), "docroot" -> docroot))
   }
+
+  def createViewWithBothIfNotExistsAndReplaceError(): Throwable = {
+    new AnalysisException(
+      errorClass = "CREATE_OR_REPLACE_WITH_IF_NOT_EXISTS_IS_NOT_ALLOWED",
+      messageParameters = Map("resourceType" -> "VIEW"))
+  }
 }
 
 private[sql] object CompilationErrors extends CompilationErrors

sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala

@@ -619,7 +619,10 @@ private[sql] object QueryParsingErrors extends DataTypeErrorsBase {
   }
 
   def createViewWithBothIfNotExistsAndReplaceError(ctx: CreateViewContext): Throwable = {
-    new ParseException(errorClass = "_LEGACY_ERROR_TEMP_0052", ctx)
+    new ParseException(
+      errorClass = "CREATE_OR_REPLACE_WITH_IF_NOT_EXISTS_IS_NOT_ALLOWED",
+      messageParameters = Map("resourceType" -> "VIEW"),
+      ctx)
   }
 
   def temporaryViewWithSchemaBindingMode(ctx: StatementContext): Throwable = {
@@ -637,10 +640,6 @@ private[sql] object QueryParsingErrors extends DataTypeErrorsBase {
       messageParameters = Map("statement" -> statement))
   }
 
-  def defineTempViewWithIfNotExistsError(ctx: CreateViewContext): Throwable = {
-    new ParseException(errorClass = "_LEGACY_ERROR_TEMP_0053", ctx)
-  }
-
   def notAllowedToAddDBPrefixForTempViewError(
       nameParts: Seq[String],
       ctx: CreateViewContext): Throwable = {

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

@@ -55,9 +55,17 @@ class GlobalTempViewManager(database: String) {
   def create(
       name: String,
       viewDefinition: TemporaryViewRelation,
+      ignoreIfExists: Boolean,
       overrideIfExists: Boolean): Unit = synchronized {
-    if (!overrideIfExists && viewDefinitions.contains(name)) {
-      throw new TempTableAlreadyExistsException(name)
+    if (ignoreIfExists && overrideIfExists) {
+      throw QueryCompilationErrors.createViewWithBothIfNotExistsAndReplaceError()
+    }
+    if (viewDefinitions.contains(name)) {
+      if (ignoreIfExists) {
+        return
+      } else if (!overrideIfExists) {
+        throw new TempTableAlreadyExistsException(name)
+      }
     }
     viewDefinitions.put(name, viewDefinition)
   }

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

@@ -669,10 +669,18 @@ class SessionCatalog(
   def createTempView(
       name: String,
       viewDefinition: TemporaryViewRelation,
+      ignoreIfExists: Boolean,
       overrideIfExists: Boolean): Unit = synchronized {
+    if (ignoreIfExists && overrideIfExists) {
+      throw QueryCompilationErrors.createViewWithBothIfNotExistsAndReplaceError()
+    }
     val normalized = format(name)
-    if (tempViews.contains(normalized) && !overrideIfExists) {
-      throw new TempTableAlreadyExistsException(name)
+    if (tempViews.contains(normalized)) {
+      if (ignoreIfExists) {
+        return
+      } else if (!overrideIfExists) {
+        throw new TempTableAlreadyExistsException(name)
+      }
     }
     tempViews.put(normalized, viewDefinition)
   }
@@ -683,8 +691,10 @@ class SessionCatalog(
   def createGlobalTempView(
       name: String,
       viewDefinition: TemporaryViewRelation,
+      ignoreIfExists: Boolean,
       overrideIfExists: Boolean): Unit = {
-    globalTempViewManager.create(format(name), viewDefinition, overrideIfExists)
+    globalTempViewManager.create(
+      format(name), viewDefinition, ignoreIfExists, overrideIfExists)
   }
 
   /**
@@ -697,7 +707,8 @@ class SessionCatalog(
     val viewName = format(name.table)
     if (name.database.isEmpty) {
       if (tempViews.contains(viewName)) {
-        createTempView(viewName, viewDefinition, overrideIfExists = true)
+        createTempView(
+          viewName, viewDefinition, ignoreIfExists = false, overrideIfExists = true)
         true
       } else {
         false

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisTest.scala

@@ -43,23 +43,25 @@ trait AnalysisTest extends PlanTest {
       catalog: SessionCatalog,
       name: String,
       plan: LogicalPlan,
+      ignoreIfExists: Boolean,
       overrideIfExists: Boolean): Unit = {
     val identifier = TableIdentifier(name)
     val metadata = createTempViewMetadata(identifier, plan.schema)
     val viewDefinition = TemporaryViewRelation(metadata, Some(plan))
-    catalog.createTempView(name, viewDefinition, overrideIfExists)
+    catalog.createTempView(name, viewDefinition, ignoreIfExists, overrideIfExists)
   }
 
   protected def createGlobalTempView(
       catalog: SessionCatalog,
       name: String,
       plan: LogicalPlan,
+      ignoreIfExists: Boolean,
       overrideIfExists: Boolean): Unit = {
     val globalDb = Some(SQLConf.get.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE))
     val identifier = TableIdentifier(name, globalDb)
     val metadata = createTempViewMetadata(identifier, plan.schema)
     val viewDefinition = TemporaryViewRelation(metadata, Some(plan))
-    catalog.createGlobalTempView(name, viewDefinition, overrideIfExists)
+    catalog.createGlobalTempView(name, viewDefinition, ignoreIfExists, overrideIfExists)
   }
 
   private def createTempViewMetadata(
@@ -79,13 +81,18 @@ trait AnalysisTest extends PlanTest {
     catalog.createDatabase(
       CatalogDatabase("default", "", new URI("loc"), Map.empty),
       ignoreIfExists = false)
-    createTempView(catalog, "TaBlE", TestRelations.testRelation, overrideIfExists = true)
-    createTempView(catalog, "TaBlE2", TestRelations.testRelation2, overrideIfExists = true)
-    createTempView(catalog, "TaBlE3", TestRelations.testRelation3, overrideIfExists = true)
-    createGlobalTempView(catalog, "TaBlE4", TestRelations.testRelation4, overrideIfExists = true)
-    createGlobalTempView(catalog, "TaBlE5", TestRelations.testRelation5, overrideIfExists = true)
+    createTempView(catalog, "TaBlE", TestRelations.testRelation,
+      ignoreIfExists = false, overrideIfExists = true)
+    createTempView(catalog, "TaBlE2", TestRelations.testRelation2,
+      ignoreIfExists = false, overrideIfExists = true)
+    createTempView(catalog, "TaBlE3", TestRelations.testRelation3,
+      ignoreIfExists = false, overrideIfExists = true)
+    createGlobalTempView(catalog, "TaBlE4", TestRelations.testRelation4,
+      ignoreIfExists = false, overrideIfExists = true)
+    createGlobalTempView(catalog, "TaBlE5", TestRelations.testRelation5,
+      ignoreIfExists = false, overrideIfExists = true)
     createTempView(catalog, "streamingTable", TestRelations.streamingRelation,
-      overrideIfExists = true)
+      ignoreIfExists = false, overrideIfExists = true)
     new Analyzer(catalog) {
       catalogManager.tempVariableManager.create(
         Seq("testA", "testVarA"),