[spark] Fix drop temporary view

docs/content/spark/sql-ddl.md

@@ -295,20 +295,20 @@ And in this case, views are supported when the `metastore` type is `hive` or `re
 CREATE VIEW constructs a virtual table that has no physical data.
 
 ```sql
--- create a view.
-CREATE [TEMPORARY] VIEW v1 AS SELECT * FROM t1;
+-- create a view or a temporary view. (temporary view should not specify database name)
+CREATE [TEMPORARY] VIEW <mydb>.v1 AS SELECT * FROM t1;
 
--- create a view, if a view of same name already exists, it will be replaced.
-CREATE OR REPLACE [TEMPORARY] VIEW v1 AS SELECT * FROM t1;
+-- create a view or a temporary view, if a view of same name already exists, it will be replaced. (temporary view should not specify database name)
+CREATE OR REPLACE [TEMPORARY] VIEW <mydb>.v1 AS SELECT * FROM t1;
 ```
 
 ### Drop View
 
 DROP VIEW removes the metadata associated with a specified view from the catalog.
 
 ```sql
--- drop a view
-DROP VIEW v1;
+-- drop a view or a temporary view.
+DROP VIEW <mydb>.v1;
 ```
 
 ## Tag

docs/content/spark/sql-functions.md

@@ -100,8 +100,8 @@ CREATE [TEMPORARY] FUNCTION <mydb>.simple_udf
 AS 'com.example.SimpleUdf' 
 USING JAR '/tmp/SimpleUdf.jar' [, JAR '/tmp/SimpleUdfR.jar'];
 
--- Create or Replace Function
-CREATE OR REPLACE FUNCTION <mydb>.simple_udf 
+-- Create or Replace Temporary Function (Temporary function should not specify database name)
+CREATE OR REPLACE [TEMPORARY] FUNCTION <mydb>.simple_udf 
 AS 'com.example.SimpleUdf'
 USING JAR '/tmp/SimpleUdf.jar';
 

paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonViewCommands.scala

@@ -67,10 +67,18 @@ case class RewritePaimonViewCommands(spark: SparkSession)
         output)
   }
 
+  private def isTempView(nameParts: Seq[String]): Boolean = {
+    catalogManager.v1SessionCatalog.isTempView(nameParts)
+  }
+
   private object ResolvedIdent {
     def unapply(unresolved: Any): Option[ResolvedIdentifier] = unresolved match {
+      case nameParts: Seq[String] if isTempView(nameParts) =>
+        None
       case CatalogAndIdentifier(viewCatalog: SupportView, ident) =>
         Some(ResolvedIdentifier(viewCatalog, ident))
+      case UnresolvedView(ident, _, true, _) if isTempView(ident) =>
+        None
       case UnresolvedView(CatalogAndIdentifier(viewCatalog: SupportView, ident), _, _, _) =>
         Some(ResolvedIdentifier(viewCatalog, ident))
       case _ =>

paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonViewCommands.scala

@@ -66,10 +66,18 @@ case class RewritePaimonViewCommands(spark: SparkSession)
         output)
   }
 
+  private def isTempView(nameParts: Seq[String]): Boolean = {
+    catalogManager.v1SessionCatalog.isTempView(nameParts)
+  }
+
   private object ResolvedIdent {
     def unapply(unresolved: Any): Option[ResolvedIdentifier] = unresolved match {
+      case UnresolvedDBObjectName(ident, _) if isTempView(ident) =>
+        None
       case UnresolvedDBObjectName(CatalogAndIdentifier(viewCatalog: SupportView, ident), _) =>
         Some(ResolvedIdentifier(viewCatalog, ident))
+      case UnresolvedView(ident, _, true, _) if isTempView(ident) =>
+        None
       case UnresolvedView(CatalogAndIdentifier(viewCatalog: SupportView, ident), _, _, _) =>
         Some(ResolvedIdentifier(viewCatalog, ident))
       case _ =>

paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonViewCommands.scala

@@ -66,8 +66,14 @@ case class RewritePaimonViewCommands(spark: SparkSession)
         output)
   }
 
+  private def isTempView(nameParts: Seq[String]): Boolean = {
+    catalogManager.v1SessionCatalog.isTempView(nameParts)
+  }
+
   private object ResolvedIdent {
     def unapply(unresolved: Any): Option[ResolvedIdentifier] = unresolved match {
+      case UnresolvedIdentifier(nameParts, true) if isTempView(nameParts) =>
+        None
       case UnresolvedIdentifier(CatalogAndIdentifier(viewCatalog: SupportView, ident), _) =>
         Some(ResolvedIdentifier(viewCatalog, ident))
       case _ =>

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonViewTestBase.scala

@@ -176,4 +176,21 @@ abstract class PaimonViewTestBase extends PaimonHiveTestBase {
       }
     }
   }
+
+  test("Paimon View: drop temp view") {
+    sql(s"USE $paimonHiveCatalogName")
+    withDatabase("test_db") {
+      sql("CREATE DATABASE test_db")
+      sql("USE test_db")
+      withTable("t") {
+        withTempView("v1") {
+          sql("CREATE TABLE t (id INT) USING paimon")
+          sql("INSERT INTO t VALUES (1), (2)")
+
+          sql("CREATE TEMPORARY VIEW v1 AS SELECT * FROM t")
+          sql("DROP VIEW v1")
+        }
+      }
+    }
+  }
 }