Skip to content

[SPARK-40425][SQL] DROP TABLE does not need to do table lookup #37879

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
cloud-fan wants to merge 2 commits into from
Closed

[SPARK-40425][SQL] DROP TABLE does not need to do table lookup #37879

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
42b82bf
DROP TABLE does not need to do table lookup
cloud-fan Sep 13, 2022
4f6c4cb
address comments
cloud-fan Sep 16, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions core/src/main/resources/error/error-classes.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -538,6 +538,11 @@
"AES-<mode> with the padding <padding> by the <functionName> function."
]
},
"CATALOG_OPERATION" : {
"message" : [
"Catalog <catalogName> does not support <operation>."
]
},
"DESC_TABLE_COLUMN_PARTITION" : {
"message" : [
"DESC TABLE COLUMN for a specific partition."
Expand Down
11 changes: 8 additions & 3 deletions core/src/test/scala/org/apache/spark/SparkFunSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -299,10 +299,15 @@ abstract class SparkFunSuite
parameters: Map[String, String] = Map.empty,
matchPVals: Boolean = false,
queryContext: Array[QueryContext] = Array.empty): Unit = {
assert(exception.getErrorClass === errorClass)
val mainErrorClass :: tail = errorClass.split("\\.").toList
assert(tail.isEmpty || tail.length == 1)
// TODO: remove the `errorSubClass` parameter.
Copy link
Member

@dongjoon-hyun dongjoon-hyun Sep 17, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

just nit. If we use IDed TODO with JIRA id, some contributor can pick up the item more easily.

Copy link
Contributor Author

@cloud-fan cloud-fan Sep 19, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I didn't create a JIRA for this TODO because @MaxGekk will fix it shortly (we talked offline) :)

assert(tail.isEmpty || errorSubClass.isEmpty)
assert(exception.getErrorClass === mainErrorClass)
if (exception.getErrorSubClass != null) {
assert(errorSubClass.isDefined)
assert(exception.getErrorSubClass === errorSubClass.get)
val subClass = errorSubClass.orElse(tail.headOption)
assert(subClass.isDefined)
assert(exception.getErrorSubClass === subClass.get)
}
sqlState.foreach(state => assert(exception.getSqlState === state))
val expectedParameters = (exception.getParameterNames zip exception.getMessageParameters).toMap
Expand Down
9 changes: 7 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1090,7 +1090,7 @@ class Analyzer(override val catalogManager: CatalogManager)
}.getOrElse(u)

case u @ UnresolvedView(identifier, cmd, allowTemp, relationTypeMismatchHint) =>
lookupTableOrView(identifier).map {
lookupTableOrView(identifier, viewOnly = true).map {
case _: ResolvedTempView if !allowTemp =>
throw QueryCompilationErrors.expectViewNotTempViewError(identifier, cmd, u)
case t: ResolvedTable =>
Expand Down Expand Up @@ -1136,12 +1136,17 @@ class Analyzer(override val catalogManager: CatalogManager)
* Resolves relations to `ResolvedTable` or `Resolved[Temp/Persistent]View`. This is
* for resolving DDL and misc commands.
*/
private def lookupTableOrView(identifier: Seq[String]): Option[LogicalPlan] = {
private def lookupTableOrView(
identifier: Seq[String],
viewOnly: Boolean = false): Option[LogicalPlan] = {
lookupTempView(identifier).map { tempView =>
ResolvedTempView(identifier.asIdentifier, tempView.tableMeta.schema)
}.orElse {
expandIdentifier(identifier) match {
case CatalogAndIdentifier(catalog, ident) =>
if (viewOnly && !CatalogV2Util.isSessionCatalog(catalog)) {
throw QueryCompilationErrors.catalogOperationNotSupported(catalog, "views")
}
CatalogV2Util.loadTable(catalog, ident).map {
case v1Table: V1Table if CatalogV2Util.isSessionCatalog(catalog) &&
v1Table.v1Table.tableType == CatalogTableType.VIEW =>
Expand Down
6 changes: 0 additions & 6 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,12 +131,6 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
case u: UnresolvedTable =>
u.failAnalysis(s"Table not found: ${u.multipartIdentifier.quoted}")

case u @ UnresolvedView(NonSessionCatalogAndIdentifier(catalog, ident), cmd, _, _) =>
u.failAnalysis(
s"Cannot specify catalog `${catalog.name}` for view ${ident.quoted} " +
"because view support in v2 catalog has not been implemented yet. " +
s"$cmd expects a view.")

case u: UnresolvedView =>
u.failAnalysis(s"View not found: ${u.multipartIdentifier.quoted}")

Expand Down
4 changes: 4 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,10 @@ case class NoSuchTableException(
def this(tableIdent: Identifier) = {
this(s"Table ${tableIdent.quoted} not found")
}

def this(nameParts: Seq[String]) = {
this(s"Table ${nameParts.quoted} not found")
}
}

case class NoSuchPartitionException(
Expand Down
12 changes: 9 additions & 3 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ package org.apache.spark.sql.catalyst.analysis

import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.connector.catalog.{CatalogManager, LookupCatalog}
import org.apache.spark.sql.connector.catalog.{CatalogManager, Identifier, LookupCatalog}

/**
* Resolves the catalog of the name parts for table/view/function/namespace.
Expand All @@ -28,8 +28,14 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
extends Rule[LogicalPlan] with LookupCatalog {

override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
case UnresolvedIdentifier(CatalogAndIdentifier(catalog, identifier)) =>
ResolvedIdentifier(catalog, identifier)
case UnresolvedIdentifier(nameParts, allowTemp) =>
if (allowTemp && catalogManager.v1SessionCatalog.isTempView(nameParts)) {
val ident = Identifier.of(nameParts.dropRight(1).toArray, nameParts.last)
Copy link
Contributor

@ryan-johnson-databricks ryan-johnson-databricks Sep 19, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: nameParts.init is the counterpart to nameParts.last:
https://www.scala-lang.org/api/2.12.5/scala/collection/Seq.html#inits:Iterator[Repr]

Copy link
Contributor Author

@cloud-fan cloud-fan Sep 19, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good idea!

ResolvedIdentifier(FakeSystemCatalog, ident)
} else {
val CatalogAndIdentifier(catalog, identifier) = nameParts
ResolvedIdentifier(catalog, identifier)
}
case s @ ShowTables(UnresolvedNamespace(Seq()), _, _) =>
s.copy(namespace = ResolvedNamespace(currentCatalog, catalogManager.currentNamespace))
case s @ ShowTableExtended(UnresolvedNamespace(Seq()), _, _, _) =>
Expand Down
6 changes: 1 addition & 5 deletions ...t/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCommandsWithIfExists.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package org.apache.spark.sql.catalyst.analysis

import org.apache.spark.sql.catalyst.plans.logical.{DropFunction, DropTable, DropView, LogicalPlan, NoopCommand, UncacheTable}
import org.apache.spark.sql.catalyst.plans.logical.{DropFunction, LogicalPlan, NoopCommand, UncacheTable}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.COMMAND

Expand All @@ -29,10 +29,6 @@ import org.apache.spark.sql.catalyst.trees.TreePattern.COMMAND
object ResolveCommandsWithIfExists extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
_.containsPattern(COMMAND)) {
case DropTable(u: UnresolvedTableOrView, ifExists, _) if ifExists =>
NoopCommand("DROP TABLE", u.multipartIdentifier)
case DropView(u: UnresolvedView, ifExists) if ifExists =>
NoopCommand("DROP VIEW", u.multipartIdentifier)
case UncacheTable(u: UnresolvedRelation, ifExists, _) if ifExists =>
NoopCommand("UNCACHE TABLE", u.multipartIdentifier)
case DropFunction(u: UnresolvedFunc, ifExists) if ifExists =>
Expand Down
10 changes: 9 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/v2ResolutionPlans.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
import org.apache.spark.sql.connector.catalog.TableChange.ColumnPosition
import org.apache.spark.sql.connector.catalog.functions.UnboundFunction
import org.apache.spark.sql.types.{DataType, StructField, StructType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap

/**
* Holds the name of a namespace that has yet to be looked up in a catalog. It will be resolved to
Expand Down Expand Up @@ -135,7 +136,8 @@ case class UnresolvedFunc(
* Holds the name of a table/view/function identifier that we need to determine the catalog. It will
* be resolved to [[ResolvedIdentifier]] during analysis.
*/
case class UnresolvedIdentifier(nameParts: Seq[String]) extends LeafNode {
case class UnresolvedIdentifier(nameParts: Seq[String], allowTemp: Boolean = false)
extends LeafNode {
override lazy val resolved: Boolean = false
override def output: Seq[Attribute] = Nil
}
Expand Down Expand Up @@ -244,3 +246,9 @@ case class ResolvedIdentifier(
identifier: Identifier) extends LeafNodeWithoutStats {
override def output: Seq[Attribute] = Nil
}

// A fake v2 catalog to hold temp views.
object FakeSystemCatalog extends CatalogPlugin {
override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = {}
override def name(): String = "SYSTEM"
Copy link
Member

@viirya viirya Sep 15, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FAKE_SYSTEM?

Copy link
Contributor Author

@cloud-fan cloud-fan Sep 16, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the name doesn't matter. We won't show it or look it up for now. But later I think it's a good idea to add a system catalog officially, to host temp view, temp functions and builtin functions.

}
8 changes: 2 additions & 6 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3673,7 +3673,7 @@ class AstBuilder extends SqlBaseParserBaseVisitor[AnyRef] with SQLConfHelper wit
override def visitDropTable(ctx: DropTableContext): LogicalPlan = withOrigin(ctx) {
// DROP TABLE works with either a table or a temporary view.
DropTable(
createUnresolvedTableOrView(ctx.multipartIdentifier(), "DROP TABLE"),
UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()), allowTemp = true),
ctx.EXISTS != null,
ctx.PURGE != null)
}
Expand All @@ -3683,11 +3683,7 @@ class AstBuilder extends SqlBaseParserBaseVisitor[AnyRef] with SQLConfHelper wit
*/
override def visitDropView(ctx: DropViewContext): AnyRef = withOrigin(ctx) {
DropView(
createUnresolvedView(
ctx.multipartIdentifier(),
commandName = "DROP VIEW",
allowTemp = true,
relationTypeMismatchHint = Some("Please use DROP TABLE instead.")),
UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()), allowTemp = true),
ctx.EXISTS != null)
}

Expand Down
13 changes: 13 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -604,6 +604,15 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
"operation" -> operation))
}

def catalogOperationNotSupported(catalog: CatalogPlugin, operation: String): Throwable = {
new AnalysisException(
errorClass = "UNSUPPORTED_FEATURE",
errorSubClass = "CATALOG_OPERATION",
messageParameters = Map(
"catalogName" -> toSQLId(Seq(catalog.name())),
"operation" -> operation))
}

def alterColumnWithV1TableCannotSpecifyNotNullError(): Throwable = {
new AnalysisException("ALTER COLUMN with v1 tables cannot specify NOT NULL.")
}
Expand Down Expand Up @@ -958,6 +967,10 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
new NoSuchTableException(ident)
}

def noSuchTableError(nameParts: Seq[String]): Throwable = {
new NoSuchTableException(nameParts)
}

def noSuchNamespaceError(namespace: Array[String]): Throwable = {
new NoSuchNamespaceException(namespace)
}
Expand Down
2 changes: 0 additions & 2 deletions ...rc/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisExceptionPositionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,14 +34,12 @@ class AnalysisExceptionPositionSuite extends AnalysisTest {
}

test("SPARK-33918: UnresolvedView should retain sql text position") {
verifyViewPosition("DROP VIEW unknown", "unknown")
verifyViewPosition("ALTER VIEW unknown SET TBLPROPERTIES ('k'='v')", "unknown")
verifyViewPosition("ALTER VIEW unknown UNSET TBLPROPERTIES ('k')", "unknown")
verifyViewPosition("ALTER VIEW unknown AS SELECT 1", "unknown")
}

test("SPARK-34057: UnresolvedTableOrView should retain sql text position") {
verifyTableOrViewPosition("DROP TABLE unknown", "unknown")
verifyTableOrViewPosition("DESCRIBE TABLE unknown", "unknown")
verifyTableOrPermanentViewPosition("ANALYZE TABLE unknown COMPUTE STATISTICS", "unknown")
verifyTableOrViewPosition("ANALYZE TABLE unknown COMPUTE STATISTICS FOR COLUMNS col", "unknown")
Expand Down
10 changes: 5 additions & 5 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -685,15 +685,15 @@ class DDLParserSuite extends AnalysisTest {
val cmd = "DROP VIEW"
val hint = Some("Please use DROP TABLE instead.")
parseCompare(s"DROP VIEW testcat.db.view",
DropView(UnresolvedView(Seq("testcat", "db", "view"), cmd, true, hint), ifExists = false))
Copy link
Contributor

@ryan-johnson-databricks ryan-johnson-databricks Sep 19, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why does UnresolvedView even continue to exist, if it's not useful for dropping? Do we still use it for add/select/etc?

Copy link
Contributor Author

@cloud-fan cloud-fan Sep 19, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it's still used by commands like SetViewProperties

DropView(UnresolvedIdentifier(Seq("testcat", "db", "view"), true), ifExists = false))
parseCompare(s"DROP VIEW db.view",
DropView(UnresolvedView(Seq("db", "view"), cmd, true, hint), ifExists = false))
DropView(UnresolvedIdentifier(Seq("db", "view"), true), ifExists = false))
parseCompare(s"DROP VIEW IF EXISTS db.view",
DropView(UnresolvedView(Seq("db", "view"), cmd, true, hint), ifExists = true))
DropView(UnresolvedIdentifier(Seq("db", "view"), true), ifExists = true))
parseCompare(s"DROP VIEW view",
DropView(UnresolvedView(Seq("view"), cmd, true, hint), ifExists = false))
DropView(UnresolvedIdentifier(Seq("view"), true), ifExists = false))
parseCompare(s"DROP VIEW IF EXISTS view",
DropView(UnresolvedView(Seq("view"), cmd, true, hint), ifExists = true))
DropView(UnresolvedIdentifier(Seq("view"), true), ifExists = true))
}

private def testCreateOrReplaceDdl(
Expand Down
18 changes: 11 additions & 7 deletions sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,19 +216,23 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
c
}

case DropTable(ResolvedV1TableIdentifier(ident), ifExists, purge) =>
case DropTable(ResolvedV1Identifier(ident), ifExists, purge) =>
Copy link
Contributor

@aokolnychyi aokolnychyi Apr 18, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am afraid this breaks the session catalog delegation. Previously, we checked the table was V1Table. Right now, we simply check the identifier looks like a V1 table identifier, which still may point to a valid V2 table. If I have a custom session catalog, it may be able to load both V1 and V2 tables. After this change, the V1 drop code is invoked for V2 tables in custom session catalogs. That means I can't drop tables correctly in custom session catalogs.

Copy link
Contributor

@aokolnychyi aokolnychyi Apr 18, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@cloud-fan @viirya @dongjoon-hyun, could you double check if I missed anything?

Copy link
Member

@viirya viirya Apr 18, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I checked the difference between ResolvedV1TableIdentifier and ResolvedV1Identifier. So do you mean ResolvedV1Identifier could wrongly apply on a V2 table? I.e.,

case ResolvedIdentifier(catalog, ident) if isSessionCatalog(catalog)

If catalog is a custom session catalog which is capable for V1 and V2 tables?

Copy link
Member

@viirya viirya Apr 18, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I saw for many commands, there is a isV2Provider check, but DropTable doesn't. So seems we need it?

Copy link
Contributor

@aokolnychyi aokolnychyi Apr 18, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think ResolvedV1Identifier simply means it is an identifier in the session catalog that has only db and table name (in other words it is a valid V1 identifier). In custom session catalogs, it may point to a valid V2 table.

Copy link
Contributor Author

@cloud-fan cloud-fan Apr 20, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shall we switch to V2 DROP path for all cases to fix SPARK-43203?

Yea we should. Can you create a PR? thanks!

Copy link
Member

@viirya viirya Apr 20, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds good to me to switch to V2 DROP.

Copy link
Contributor

@aokolnychyi aokolnychyi Apr 20, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll have time, probably, on Monday. I'll do that then unless someone gets there earlier.

Copy link
Member

@Hisoka-X Hisoka-X May 26, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @aokolnychyi Any update for this? If you don't mind I can finish it this weekend.😄

Copy link

@gaecoli gaecoli Aug 20, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you create a management table in spark 3.5.1, you cannot delete the path when dropping the table. I think this is a bug. It can be deleted correctly in spark 3.3.

DropTableCommand(ident, ifExists, isView = false, purge = purge)

case DropTable(_: ResolvedPersistentView, ifExists, purge) =>
throw QueryCompilationErrors.cannotDropViewWithDropTableError

// v1 DROP TABLE supports temp view.
case DropTable(ResolvedTempView(ident, _), ifExists, purge) =>
DropTableCommand(ident.asTableIdentifier, ifExists, isView = false, purge = purge)
case DropTable(ResolvedIdentifier(FakeSystemCatalog, ident), _, _) =>
DropTempViewCommand(ident)

case DropView(ResolvedViewIdentifier(ident), ifExists) =>
case DropView(ResolvedV1Identifier(ident), ifExists) =>
DropTableCommand(ident, ifExists, isView = true, purge = false)

case DropView(r @ ResolvedIdentifier(catalog, ident), _) =>
if (catalog == FakeSystemCatalog) {
DropTempViewCommand(ident)
} else {
throw QueryCompilationErrors.catalogOperationNotSupported(catalog, "views")
}
Copy link
Contributor

@ryan-johnson-databricks ryan-johnson-databricks Sep 19, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: r not used? And if you make FakeSystemCatalog a case object it can participate directly in matching here:

case DropView(ResolvedIdentifier(FakeSystemCatalog, ident), _) => 
  DropTempViewCommand(ident)
case DropView(ResolvedIdentifier(catalog, _), _) => 
  throw ...


case c @ CreateNamespace(DatabaseNameInSessionCatalog(name), _, _) if conf.useV1Command =>
val comment = c.properties.get(SupportsNamespaces.PROP_COMMENT)
val location = c.properties.get(SupportsNamespaces.PROP_LOCATION)
Expand Down
48 changes: 44 additions & 4 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import org.apache.spark.internal.config.ConfigEntry
import org.apache.spark.sql.{Dataset, SparkSession}
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}
import org.apache.spark.sql.catalyst.plans.logical.{IgnoreCachedData, LogicalPlan, ResolvedHint, SubqueryAlias, View}
import org.apache.spark.sql.catalyst.trees.TreePattern.PLAN_EXPRESSION
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.columnar.InMemoryRelation
Expand Down Expand Up @@ -159,11 +159,51 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
plan: LogicalPlan,
cascade: Boolean,
blocking: Boolean = false): Unit = {
uncacheQuery(spark, _.sameResult(plan), cascade, blocking)
}

def uncacheTableOrView(spark: SparkSession, name: Seq[String], cascade: Boolean): Unit = {
uncacheQuery(
spark,
isMatchedTableOrView(_, name, spark.sessionState.conf),
cascade,
blocking = false)
}

private def isMatchedTableOrView(plan: LogicalPlan, name: Seq[String], conf: SQLConf): Boolean = {
def isSameName(nameInCache: Seq[String]): Boolean = {
nameInCache.length == name.length && nameInCache.zip(name).forall(conf.resolver.tupled)
}

plan match {
case SubqueryAlias(ident, LogicalRelation(_, _, Some(catalogTable), _)) =>
val v1Ident = catalogTable.identifier
isSameName(ident.qualifier :+ ident.name) &&
isSameName(v1Ident.catalog.toSeq ++ v1Ident.database :+ v1Ident.table)

case SubqueryAlias(ident, DataSourceV2Relation(_, _, Some(catalog), Some(v2Ident), _)) =>
isSameName(ident.qualifier :+ ident.name) &&
isSameName(catalog.name() +: v2Ident.namespace() :+ v2Ident.name())
Comment on lines +184 to +186
Copy link
Member

@viirya viirya Sep 15, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does SubqueryAlias have same name as the underlying relation?

Copy link
Contributor Author

@cloud-fan cloud-fan Sep 16, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, see ResolveRelations.createRelation


case SubqueryAlias(ident, View(catalogTable, _, _)) =>
val v1Ident = catalogTable.identifier
isSameName(ident.qualifier :+ ident.name) &&
isSameName(v1Ident.catalog.toSeq ++ v1Ident.database :+ v1Ident.table)

case _ => false
}
}

def uncacheQuery(
spark: SparkSession,
isMatchedPlan: LogicalPlan => Boolean,
cascade: Boolean,
blocking: Boolean): Unit = {
val shouldRemove: LogicalPlan => Boolean =
if (cascade) {
_.exists(_.sameResult(plan))
Copy link
Member

@viirya viirya Sep 15, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sameResult doesn't work?

Copy link
Member

@viirya viirya Sep 16, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nvm

_.exists(isMatchedPlan)
} else {
_.sameResult(plan)
isMatchedPlan
}
val plansToUncache = cachedData.filter(cd => shouldRemove(cd.plan))
this.synchronized {
Expand All @@ -187,7 +227,7 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
// will keep it as it is. It means the physical plan has been re-compiled already in the
// other thread.
val cacheAlreadyLoaded = cd.cachedRepresentation.cacheBuilder.isCachedColumnBuffersLoaded
cd.plan.exists(_.sameResult(plan)) && !cacheAlreadyLoaded
cd.plan.exists(isMatchedPlan) && !cacheAlreadyLoaded
})
}
}
Expand Down
Loading