From a73ab3890b2c1cf14a29a0ac2b89626a8b5dcf90 Mon Sep 17 00:00:00 2001
From: Wenchen Fan <wenchen@databricks.com>
Date: Tue, 18 Jul 2023 14:48:44 +0800
Subject: [PATCH] V2SessionCatalog.dropTable should handle null table

---
 .../datasources/v2/V2SessionCatalog.scala     | 22 ++++++++++---------
 1 file changed, 12 insertions(+), 10 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala
index e5496a4676015..f311ccbb6309d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala
@@ -115,10 +115,6 @@ class V2SessionCatalog(catalog: SessionCatalog)
     createTable(ident, CatalogV2Util.v2ColumnsToStructType(columns), partitions, properties)
   }
 
-  override def purgeTable(ident: Identifier): Boolean = {
-    dropTableInternal(ident, purge = true)
-  }
-
   // TODO: remove it when no tests calling this deprecated method.
   override def createTable(
       ident: Identifier,
@@ -202,6 +198,10 @@ class V2SessionCatalog(catalog: SessionCatalog)
     loadTable(ident)
   }
 
+  override def purgeTable(ident: Identifier): Boolean = {
+    dropTableInternal(ident, purge = true)
+  }
+
   override def dropTable(ident: Identifier): Boolean = {
     dropTableInternal(ident)
   }
@@ -218,14 +218,16 @@ class V2SessionCatalog(catalog: SessionCatalog)
             foundType = v1Table.tableType.name,
             alternative = "DROP VIEW"
           )
+        case null =>
+          false
         case _ =>
+          catalog.invalidateCachedTable(ident.asTableIdentifier)
+          catalog.dropTable(
+            ident.asTableIdentifier,
+            ignoreIfNotExists = true,
+            purge = purge)
+          true
       }
-      catalog.invalidateCachedTable(ident.asTableIdentifier)
-      catalog.dropTable(
-        ident.asTableIdentifier,
-        ignoreIfNotExists = true,
-        purge = purge)
-      true
     } catch {
       case _: NoSuchTableException =>
         false