This is an automated email from the ASF dual-hosted git repository. dongjoon pushed a commit to branch branch-3.4 in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push: new d50a3ce4a42 [SPARK-43203][SQL][3.4] Move all Drop Table case to DataSource V2 d50a3ce4a42 is described below commit d50a3ce4a42ce95d3a7eb4c1a4db4a77a070da13 Author: Jia Fan <fanjiaemi...@qq.com> AuthorDate: Fri Sep 8 08:22:17 2023 -0700 [SPARK-43203][SQL][3.4] Move all Drop Table case to DataSource V2 ### What changes were proposed in this pull request? cherry pick #41348 and #42056 , this a bug fixed should be included in 3.4.2 ### Why are the changes needed? Fix DROP table behavior in session catalog ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Tested by: - V2 table catalog tests: `org.apache.spark.sql.execution.command.v2.DropTableSuite` - V1 table catalog tests: `org.apache.spark.sql.execution.command.v1.DropTableSuiteBase` Closes #41765 from Hisoka-X/move_drop_table_v2_to_3.4.2. Lead-authored-by: Jia Fan <fanjiaemi...@qq.com> Co-authored-by: Wenchen Fan <wenc...@databricks.com> Signed-off-by: Dongjoon Hyun <dh...@apple.com> --- .../catalyst/analysis/ResolveSessionCatalog.scala | 2 +- .../apache/spark/sql/execution/CacheManager.scala | 6 ++++ .../datasources/v2/V2SessionCatalog.scala | 42 +++++++++++++++++----- .../execution/command/PlanResolutionSuite.scala | 17 ++++----- .../hive/execution/command/DropTableSuite.scala | 2 +- 5 files changed, 50 insertions(+), 19 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala index b2b35b40492..e5a36394a44 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala @@ -210,7 +210,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager) c } - case DropTable(ResolvedV1Identifier(ident), ifExists, purge) => + case DropTable(ResolvedV1Identifier(ident), ifExists, purge) if conf.useV1Command => DropTableCommand(ident, ifExists, isView = false, purge = purge) // v1 DROP TABLE supports temp view. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala index d41611439f0..b1153d7a1e8 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala @@ -24,6 +24,7 @@ import org.apache.hadoop.fs.{FileSystem, Path} import org.apache.spark.internal.Logging import org.apache.spark.internal.config.ConfigEntry import org.apache.spark.sql.{Dataset, SparkSession} +import org.apache.spark.sql.catalyst.catalog.HiveTableRelation import org.apache.spark.sql.catalyst.expressions.{Attribute, SubqueryExpression} import org.apache.spark.sql.catalyst.optimizer.EliminateResolvedHint import org.apache.spark.sql.catalyst.plans.logical.{IgnoreCachedData, LogicalPlan, ResolvedHint, SubqueryAlias, View} @@ -190,6 +191,11 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper { isSameName(ident.qualifier :+ ident.name) && isSameName(v1Ident.catalog.toSeq ++ v1Ident.database :+ v1Ident.table) + case SubqueryAlias(ident, HiveTableRelation(catalogTable, _, _, _, _)) => + val v1Ident = catalogTable.identifier + isSameName(ident.qualifier :+ ident.name) && + isSameName(v1Ident.catalog.toSeq ++ v1Ident.database :+ v1Ident.table) + case _ => false } } 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 b4789c98df9..d48f771b9db 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 @@ -70,7 +70,12 @@ class V2SessionCatalog(catalog: SessionCatalog) } override def loadTable(ident: Identifier): Table = { - V1Table(catalog.getTableMetadata(ident.asTableIdentifier)) + try { + V1Table(catalog.getTableMetadata(ident.asTableIdentifier)) + } catch { + case _: NoSuchDatabaseException => + throw QueryCompilationErrors.noSuchTableError(ident) + } } override def loadTable(ident: Identifier, timestamp: Long): Table = { @@ -189,16 +194,35 @@ class V2SessionCatalog(catalog: SessionCatalog) loadTable(ident) } + override def purgeTable(ident: Identifier): Boolean = { + dropTableInternal(ident, purge = true) + } + override def dropTable(ident: Identifier): Boolean = { + dropTableInternal(ident) + } + + private def dropTableInternal(ident: Identifier, purge: Boolean = false): Boolean = { try { - if (loadTable(ident) != null) { - catalog.dropTable( - ident.asTableIdentifier, - ignoreIfNotExists = true, - purge = true /* skip HDFS trash */) - true - } else { - false + loadTable(ident) match { + case V1Table(v1Table) if v1Table.tableType == CatalogTableType.VIEW => + throw QueryCompilationErrors.wrongCommandForObjectTypeError( + operation = "DROP TABLE", + requiredType = s"${CatalogTableType.EXTERNAL.name} or" + + s" ${CatalogTableType.MANAGED.name}", + objectName = v1Table.qualifiedName, + foundType = v1Table.tableType.name, + alternative = "DROP VIEW" + ) + case null => + false + case _ => + catalog.invalidateCachedTable(ident.asTableIdentifier) + catalog.dropTable( + ident.asTableIdentifier, + ignoreIfNotExists = true, + purge = purge) + true } } catch { case _: NoSuchTableException => diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala index 2cf4792b8c1..dc69acc6f91 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala @@ -691,22 +691,23 @@ class PlanResolutionSuite extends AnalysisTest { test("drop table") { val tableName1 = "db.v1Table" - val tableIdent1 = TableIdentifier("v1Table", Option("db"), Some(SESSION_CATALOG_NAME)) + val tableIdent1 = ResolvedIdentifier(v2SessionCatalog, Identifier.of(Array("db"), "v1Table")) val tableName2 = "v1Table" - val tableIdent2 = TableIdentifier("v1Table", Some("default"), Some(SESSION_CATALOG_NAME)) + val tableIdent2 = ResolvedIdentifier(v2SessionCatalog, Identifier.of(Array("default"), + "v1Table")) parseResolveCompare(s"DROP TABLE $tableName1", - DropTableCommand(tableIdent1, ifExists = false, isView = false, purge = false)) + DropTable(tableIdent1, ifExists = false, purge = false)) parseResolveCompare(s"DROP TABLE IF EXISTS $tableName1", - DropTableCommand(tableIdent1, ifExists = true, isView = false, purge = false)) + DropTable(tableIdent1, ifExists = true, purge = false)) parseResolveCompare(s"DROP TABLE $tableName2", - DropTableCommand(tableIdent2, ifExists = false, isView = false, purge = false)) + DropTable(tableIdent2, ifExists = false, purge = false)) parseResolveCompare(s"DROP TABLE IF EXISTS $tableName2", - DropTableCommand(tableIdent2, ifExists = true, isView = false, purge = false)) + DropTable(tableIdent2, ifExists = true, purge = false)) parseResolveCompare(s"DROP TABLE $tableName2 PURGE", - DropTableCommand(tableIdent2, ifExists = false, isView = false, purge = true)) + DropTable(tableIdent2, ifExists = false, purge = true)) parseResolveCompare(s"DROP TABLE IF EXISTS $tableName2 PURGE", - DropTableCommand(tableIdent2, ifExists = true, isView = false, purge = true)) + DropTable(tableIdent2, ifExists = true, purge = true)) } test("drop table in v2 catalog") { diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropTableSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropTableSuite.scala index 8c6d718f18a..e847e92c4ce 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropTableSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropTableSuite.scala @@ -26,7 +26,7 @@ class DropTableSuite extends v1.DropTableSuiteBase with CommandSuiteBase { test("hive client calls") { withNamespaceAndTable("ns", "tbl") { t => sql(s"CREATE TABLE $t (id int) $defaultUsing") - checkHiveClientCalls(expected = 11) { + checkHiveClientCalls(expected = 10) { sql(s"DROP TABLE $t") } } --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org