You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ur...@apache.org on 2026/10/22 07:48:11 UTC
(spark) branch master updated: [SPARK-60422][SQL] Extract shared catalog identifier and DROP TABLE helpers
This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 04328dee2b44 [SPARK-60422][SQL] Extract shared catalog identifier and DROP TABLE helpers
04328dee2b44 is described below
commit 04328dee2b44a99dc78110c44d75e401a8d2fb82
Author: Vladan Vasić <vl...@databricks.com>
AuthorDate: Thu Oct 22 07:47:55 2026 +0000
[SPARK-60422][SQL] Extract shared catalog identifier and DROP TABLE helpers
### What changes were proposed in this pull request?
Jira: [SPARK-60422](https://issues.apache.org/jira/browse/SPARK-60422)
Extract existing analyzer logic into shared helpers:
- `CatalogIdentifierResolution` handles command-target catalog lookup and temporary-view selection for `ResolveCatalogs`.
- `DropTableHelper` handles session-catalog `DROP TABLE` lowering for `ResolveSessionCatalog`.
- `V1CommandUtils` contains catalog eligibility checks for V1 commands, and `ResolvedV1CommandIdentifier` provides their standalone identifier extractor. Both live in Catalyst and are used directly by the analyzer helpers.
### Why are the changes needed?
The helper extraction makes catalog identifier resolution and DROP TABLE command lowering reusable while preserving the existing analyzer's rule ordering, catalog selection, and command construction.
### Does this PR introduce _any_ user-facing change?
No. This is a refactor-only change; analyzed plans and command behavior are unchanged.
### How was this patch tested?
All 120 existing tests passed across `PlanResolutionSuite`, V1/V2 `DropTableSuite`, and Hive `DropTableSuite`.
Catalyst and SQL Scala style checks passed.
Independent review verified behavior equivalence and the helper-only scope.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6)
Closes #59646 from vladanvasi-db/vladanvasi-db/drop-table-resolver.
Authored-by: Vladan Vasić <vl...@databricks.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../analysis/CatalogIdentifierResolution.scala | 42 ++++++++++++++++++++++
.../sql/catalyst/analysis/ResolveCatalogs.scala | 10 ++----
.../analysis/ResolvedV1CommandIdentifier.scala | 32 +++++++++++++++++
.../sql/catalyst/analysis/V1CommandUtils.scala | 31 ++++++++++++++++
.../sql/catalyst/analysis/DropTableHelper.scala | 38 ++++++++++++++++++++
.../catalyst/analysis/ResolveSessionCatalog.scala | 35 +++++-------------
6 files changed, 155 insertions(+), 33 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CatalogIdentifierResolution.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CatalogIdentifierResolution.scala
new file mode 100644
index 000000000000..aba685b7c13a
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CatalogIdentifierResolution.scala
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.{CatalogManager, Identifier, LookupCatalog}
+
+/**
+ * Resolves command target names to a [[ResolvedIdentifier]] for [[ResolveCatalogs]].
+ */
+class CatalogIdentifierResolution(override val catalogManager: CatalogManager)
+ extends LookupCatalog {
+
+ /** A temporary view resolves to [[FakeSystemCatalog]] when `allowTemp` is set. */
+ def resolveIdentifier(
+ nameParts: Seq[String],
+ allowTemp: Boolean,
+ columnOutput: Seq[Attribute]): ResolvedIdentifier = {
+ if (allowTemp && catalogManager.v1SessionCatalog.isTempView(nameParts)) {
+ val identifier = Identifier.of(nameParts.dropRight(1).toArray, nameParts.last)
+ ResolvedIdentifier(FakeSystemCatalog, identifier, columnOutput)
+ } else {
+ val CatalogAndIdentifier(catalog, identifier) = nameParts
+ ResolvedIdentifier(catalog, identifier, columnOutput)
+ }
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala
index 6fc196774a04..6e73dc317baa 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala
@@ -39,6 +39,8 @@ import org.apache.spark.util.ArrayImplicits._
class ResolveCatalogs(val catalogManager: CatalogManager)
extends Rule[LogicalPlan] with LookupCatalog {
+ private val catalogIdentifierResolution = new CatalogIdentifierResolution(catalogManager)
+
override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperatorsDown {
// We only support temp variables for now and the system catalog is not properly implemented
// yet. We need to resolve `UnresolvedIdentifier` for variable commands specially.
@@ -163,13 +165,7 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
}
AttributeReference(col.name, dataType, col.nullable, col.metadata)()
}
- if (allowTemp && catalogManager.v1SessionCatalog.isTempView(nameParts)) {
- val ident = Identifier.of(nameParts.dropRight(1).toArray, nameParts.last)
- ResolvedIdentifier(FakeSystemCatalog, ident, columnOutput)
- } else {
- val CatalogAndIdentifier(catalog, identifier) = nameParts
- ResolvedIdentifier(catalog, identifier, columnOutput)
- }
+ catalogIdentifierResolution.resolveIdentifier(nameParts, allowTemp, columnOutput)
}
private def rejectTimeTravelOptionsForWrite(
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolvedV1CommandIdentifier.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolvedV1CommandIdentifier.scala
new file mode 100644
index 000000000000..e46ee3d7adc7
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolvedV1CommandIdentifier.scala
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.V1CommandUtils.supportsV1Command
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+
+/** Extracts a V1 command identifier, rejecting namespaces that cannot be represented in V1. */
+object ResolvedV1CommandIdentifier {
+ def unapply(resolved: LogicalPlan): Option[TableIdentifier] = resolved match {
+ case ResolvedIdentifier(catalog, ident) if supportsV1Command(catalog) =>
+ Some(ident.asTableIdentifier.copy(catalog = Some(catalog.name)))
+ case _ => None
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/V1CommandUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/V1CommandUtils.scala
new file mode 100644
index 000000000000..f65c18dbc6e0
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/V1CommandUtils.scala
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.connector.catalog.{CatalogExtension, CatalogPlugin, CatalogV2Util}
+import org.apache.spark.sql.internal.SQLConf
+
+/** Utilities for determining whether catalogs support V1 command handling. */
+object V1CommandUtils {
+ /** Whether commands on `catalog` can be converted to V1 session-catalog commands. */
+ def supportsV1Command(catalog: CatalogPlugin): Boolean = {
+ CatalogV2Util.isSessionCatalog(catalog) && (
+ SQLConf.get.getConf(SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION) == "builtin" ||
+ catalog.isInstanceOf[CatalogExtension])
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/DropTableHelper.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/DropTableHelper.scala
new file mode 100644
index 000000000000..76a6a4c0495f
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/DropTableHelper.scala
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.plans.logical.{DropTable, LogicalPlan}
+import org.apache.spark.sql.execution.command.{DropTableCommand, DropTempViewCommand}
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Rewrites a resolved [[DropTable]] to its session-catalog command in [[ResolveSessionCatalog]].
+ * Other catalogs keep the V2 command.
+ */
+object DropTableHelper {
+ def rewrite(dropTable: DropTable): LogicalPlan = dropTable match {
+ case DropTable(ResolvedV1CommandIdentifier(identifier), ifExists, purge)
+ if SQLConf.get.useV1Command =>
+ DropTableCommand(identifier, ifExists, isView = false, purge = purge)
+ case DropTable(ResolvedIdentifier(FakeSystemCatalog, identifier), _, _) =>
+ DropTempViewCommand(identifier)
+ case _ =>
+ dropTable
+ }
+}
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 dd1a26fa1a6b..7f4d824dd8c4 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
@@ -21,6 +21,7 @@ import org.apache.spark.SparkException
import org.apache.spark.internal.LogKeys.CONFIG
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
+import org.apache.spark.sql.catalyst.analysis.V1CommandUtils.supportsV1Command
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType, CatalogUtils, ClusterBySpec, HiveTableRelation}
import org.apache.spark.sql.catalyst.expressions.{Alias, AliasHelper, Attribute}
import org.apache.spark.sql.catalyst.optimizer.ConstantFolding
@@ -28,7 +29,7 @@ import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.{quoteIfNeeded, toPrettySQL, CharVarcharUtils, ResolveDefaultColumns => DefaultCols}
import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
-import org.apache.spark.sql.connector.catalog.{CatalogExtension, CatalogManager, CatalogPlugin, CatalogV2Util, LookupCatalog, SupportsNamespaces, V1Table, ViewCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, LookupCatalog, SupportsNamespaces, V1Table, ViewCatalog}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
import org.apache.spark.sql.execution.command._
@@ -268,7 +269,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
// For CREATE TABLE [AS SELECT], we should use the v1 command if the catalog is resolved to the
// session catalog and the table provider is not v2.
- case c @ CreateTable(ResolvedV1Identifier(ident), _, _, tableSpec: TableSpec, _)
+ case c @ CreateTable(ResolvedV1CommandIdentifier(ident), _, _, tableSpec: TableSpec, _)
if c.resolved && c.columns.forall(_.isDefaultValueTypeCoerced) =>
val (storageFormat, provider) = getStorageFormatAndProvider(
c.tableSpec.provider, tableSpec.options, c.tableSpec.location, c.tableSpec.serde,
@@ -299,7 +300,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
}
case c @ CreateTableAsSelect(
- ResolvedV1Identifier(ident), _, _, tableSpec: TableSpec, writeOptions, _, _) =>
+ ResolvedV1CommandIdentifier(ident), _, _, tableSpec: TableSpec, writeOptions, _, _) =>
val (storageFormat, provider) = getStorageFormatAndProvider(
c.tableSpec.provider,
tableSpec.options ++ writeOptions,
@@ -323,7 +324,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
// For REPLACE TABLE [AS SELECT], we should fail if the catalog is resolved to the
// session catalog and the table provider is not v2.
- case c @ ReplaceTable(ResolvedV1Identifier(ident), _, _, _, _) if c.resolved =>
+ case c @ ReplaceTable(ResolvedV1CommandIdentifier(ident), _, _, _, _) if c.resolved =>
val provider = c.tableSpec.provider.getOrElse(conf.defaultDataSourceName)
if (!isV2Provider(provider)) {
throw QueryCompilationErrors.unsupportedTableOperationError(
@@ -332,7 +333,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
c
}
- case c @ ReplaceTableAsSelect(ResolvedV1Identifier(ident), _, _, _, _, _, _) =>
+ case c @ ReplaceTableAsSelect(ResolvedV1CommandIdentifier(ident), _, _, _, _, _, _) =>
val provider = c.tableSpec.provider.getOrElse(conf.defaultDataSourceName)
if (!isV2Provider(provider)) {
throw QueryCompilationErrors.unsupportedTableOperationError(
@@ -345,19 +346,15 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
// catalog (or a V1-compatible catalog extension). If source is in a different catalog, fall
// through to the V2 execution path (CreateTableLikeExec via DataSourceV2Strategy).
case CreateTableLike(
- ResolvedV1Identifier(targetIdent),
+ ResolvedV1CommandIdentifier(targetIdent),
ResolvedV1TableOrViewIdentifier(sourceIdent),
location, provider, serdeInfo, properties, ifNotExists) =>
val fileFormat = buildStorageFormatFromSerdeInfo(location, serdeInfo)
CreateTableLikeCommand(
targetIdent, sourceIdent, fileFormat, provider, properties, ifNotExists)
- case DropTable(ResolvedV1Identifier(ident), ifExists, purge) if conf.useV1Command =>
- DropTableCommand(ident, ifExists, isView = false, purge = purge)
-
- // v1 DROP TABLE supports temp view.
- case DropTable(ResolvedIdentifier(FakeSystemCatalog, ident), _, _) =>
- DropTempViewCommand(ident)
+ case dropTable @ DropTable(_: ResolvedIdentifier, _, _) =>
+ DropTableHelper.rewrite(dropTable)
// Session-qualified temp view: same execution path as FakeSystemCatalog DROP VIEW.
case DropView(ResolvedTempView(ident, _), ifExists) =>
@@ -893,14 +890,6 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
}
}
- object ResolvedV1Identifier {
- def unapply(resolved: LogicalPlan): Option[TableIdentifier] = resolved match {
- case ResolvedIdentifier(catalog, ident) if supportsV1Command(catalog) =>
- Some(ident.asTableIdentifier.copy(catalog = Some(catalog.name)))
- case _ => None
- }
- }
-
private object CreateViewInSessionCatalog
extends ResolvedIdentifierInSessionCatalog("CREATE", "VIEW")
private object DropViewInSessionCatalog
@@ -1013,12 +1002,6 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
}
}
- private def supportsV1Command(catalog: CatalogPlugin): Boolean = {
- isSessionCatalog(catalog) && (
- SQLConf.get.getConf(SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION) == "builtin" ||
- catalog.isInstanceOf[CatalogExtension])
- }
-
// True when the ALTER COLUMN specs only change comments (set or drop, no type / nullability /
// position / default changes) and, taken together, touch multiple columns or a nested field.
// Such changes come from `COMMENT ON TABLE ... COLUMN (...)` and cannot be expressed by the
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org