Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,8 @@ class DatabaseAlreadyExistsException(db: String)
class TableAlreadyExistsException(db: String, table: String)
extends AnalysisException(s"Table or view '$table' already exists in database '$db'")

class TempTableAlreadyExistsException(table: String)
extends AnalysisException(s"Temporary table '$table' already exists")
class TempViewAlreadyExistsException(name: String)
extends AnalysisException(s"Temporary view '$name' already exists")

class PartitionAlreadyExistsException(db: String, table: String, spec: TablePartitionSpec)
extends AnalysisException(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -439,7 +439,7 @@ class Analyzer(
object ResolveRelations extends Rule[LogicalPlan] {
private def lookupTableFromCatalog(u: UnresolvedRelation): LogicalPlan = {
try {
catalog.lookupRelation(u.tableIdentifier, u.alias)
catalog.lookupTempViewOrRelation(u.tableIdentifier, u.alias)
Copy link
Contributor

Choose a reason for hiding this comment

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

This is also for view, right? Should we just keep the old name?

} catch {
case _: NoSuchTableException =>
u.failAnalysis(s"Table or view not found: ${u.tableName}")
Expand Down

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
/*
* 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.catalog

import javax.annotation.concurrent.GuardedBy

import scala.collection.mutable

import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.analysis.TempViewAlreadyExistsException
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.util.StringUtils


/**
* A thread-safe manager for a list of temp views, providing atomic operations to manage temp views.
Copy link
Member

Choose a reason for hiding this comment

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

In the description of TempViewManager, could we mention the name of temp view is always case sensitive? The caller is responsible for handling case-related issues.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

yea good idea

* Note that, the temp view name is always case-sensitive here, callers are responsible to format
* the view name w.r.t. case-sensitivity config.
*/
class TempViewManager {

/** List of temporary views, mapping from view name to logical plan. */
@GuardedBy("this")
private val tempViews = new mutable.HashMap[String, LogicalPlan]

def get(name: String): Option[LogicalPlan] = synchronized {
tempViews.get(name)
}

def create(
name: String,
viewDefinition: LogicalPlan,
overrideIfExists: Boolean): Unit = synchronized {
if (!overrideIfExists && tempViews.contains(name)) {
throw new TempViewAlreadyExistsException(name)
}
tempViews.put(name, viewDefinition)
}

def update(
name: String,
viewDefinition: LogicalPlan): Boolean = synchronized {
// Only update it when the view with the given name exits.
if (tempViews.contains(name)) {
tempViews.put(name, viewDefinition)
true
} else {
false
}
}

def remove(name: String): Boolean = synchronized {
tempViews.remove(name).isDefined
}

def rename(oldName: String, newName: String): Boolean = synchronized {
if (tempViews.contains(oldName)) {
if (tempViews.contains(newName)) {
throw new AnalysisException(
s"rename temporary view from '$oldName' to '$newName': destination view already exists")
}

val viewDefinition = tempViews(oldName)
tempViews.remove(oldName)
tempViews.put(newName, viewDefinition)
true
} else {
false
}
}

def listNames(pattern: String): Seq[String] = synchronized {
StringUtils.filterPattern(tempViews.keys.toSeq, pattern)
}

def clear(): Unit = synchronized {
tempViews.clear()
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -201,16 +201,16 @@ class SessionCatalogSuite extends SparkFunSuite {
val tempTable2 = Range(1, 20, 2, 10)
catalog.createTempView("tbl1", tempTable1, overrideIfExists = false)
catalog.createTempView("tbl2", tempTable2, overrideIfExists = false)
assert(catalog.getTempTable("tbl1") == Option(tempTable1))
assert(catalog.getTempTable("tbl2") == Option(tempTable2))
assert(catalog.getTempTable("tbl3").isEmpty)
assert(catalog.lookupTempView("tbl1") == Option(tempTable1))
assert(catalog.lookupTempView("tbl2") == Option(tempTable2))
assert(catalog.lookupTempView("tbl3").isEmpty)
// Temporary table already exists
intercept[TempTableAlreadyExistsException] {
intercept[TempViewAlreadyExistsException] {
catalog.createTempView("tbl1", tempTable1, overrideIfExists = false)
}
// Temporary table already exists but we override it
catalog.createTempView("tbl1", tempTable2, overrideIfExists = true)
assert(catalog.getTempTable("tbl1") == Option(tempTable2))
assert(catalog.lookupTempView("tbl1") == Option(tempTable2))
}

test("drop table") {
Expand Down Expand Up @@ -245,28 +245,17 @@ class SessionCatalogSuite extends SparkFunSuite {
purge = false)
}

test("drop temp table") {
test("drop temp view") {
val externalCatalog = newBasicCatalog()
val sessionCatalog = new SessionCatalog(externalCatalog)
val tempTable = Range(1, 10, 2, 10)
sessionCatalog.createTempView("tbl1", tempTable, overrideIfExists = false)
sessionCatalog.setCurrentDatabase("db2")
assert(sessionCatalog.getTempTable("tbl1") == Some(tempTable))
assert(sessionCatalog.lookupTempView("tbl1") == Some(tempTable))
assert(externalCatalog.listTables("db2").toSet == Set("tbl1", "tbl2"))
// If database is not specified, temp table should be dropped first
sessionCatalog.dropTable(TableIdentifier("tbl1"), ignoreIfNotExists = false, purge = false)
assert(sessionCatalog.getTempTable("tbl1") == None)
sessionCatalog.dropTempView("tbl1")
assert(sessionCatalog.lookupTempView("tbl1").isEmpty)
assert(externalCatalog.listTables("db2").toSet == Set("tbl1", "tbl2"))
// If temp table does not exist, the table in the current database should be dropped
sessionCatalog.dropTable(TableIdentifier("tbl1"), ignoreIfNotExists = false, purge = false)
assert(externalCatalog.listTables("db2").toSet == Set("tbl2"))
// If database is specified, temp tables are never dropped
sessionCatalog.createTempView("tbl1", tempTable, overrideIfExists = false)
sessionCatalog.createTable(newTable("tbl1", "db2"), ignoreIfExists = false)
sessionCatalog.dropTable(TableIdentifier("tbl1", Some("db2")), ignoreIfNotExists = false,
purge = false)
assert(sessionCatalog.getTempTable("tbl1") == Some(tempTable))
assert(externalCatalog.listTables("db2").toSet == Set("tbl2"))
}

test("rename table") {
Expand Down Expand Up @@ -297,24 +286,18 @@ class SessionCatalogSuite extends SparkFunSuite {
}
}

test("rename temp table") {
test("rename temp view") {
val externalCatalog = newBasicCatalog()
val sessionCatalog = new SessionCatalog(externalCatalog)
val tempTable = Range(1, 10, 2, 10)
sessionCatalog.createTempView("tbl1", tempTable, overrideIfExists = false)
sessionCatalog.setCurrentDatabase("db2")
assert(sessionCatalog.getTempTable("tbl1") == Option(tempTable))
assert(sessionCatalog.lookupTempView("tbl1") == Option(tempTable))
assert(externalCatalog.listTables("db2").toSet == Set("tbl1", "tbl2"))
// If database is not specified, temp table should be renamed first
sessionCatalog.renameTable(TableIdentifier("tbl1"), "tbl3")
assert(sessionCatalog.getTempTable("tbl1").isEmpty)
assert(sessionCatalog.getTempTable("tbl3") == Option(tempTable))
sessionCatalog.renameTempView("tbl1", "tbl3")
assert(sessionCatalog.lookupTempView("tbl1").isEmpty)
assert(sessionCatalog.lookupTempView("tbl3") == Option(tempTable))
assert(externalCatalog.listTables("db2").toSet == Set("tbl1", "tbl2"))
// If database is specified, temp tables are never renamed
sessionCatalog.renameTable(TableIdentifier("tbl2", Some("db2")), "tbl4")
assert(sessionCatalog.getTempTable("tbl3") == Option(tempTable))
assert(sessionCatalog.getTempTable("tbl4").isEmpty)
assert(externalCatalog.listTables("db2").toSet == Set("tbl1", "tbl4"))
}

test("alter table") {
Expand Down Expand Up @@ -375,22 +358,22 @@ class SessionCatalogSuite extends SparkFunSuite {
}
}

test("lookup table relation") {
test("lookupTempViewOrRelation") {
val externalCatalog = newBasicCatalog()
val sessionCatalog = new SessionCatalog(externalCatalog)
val tempTable1 = Range(1, 10, 1, 10)
val metastoreTable1 = externalCatalog.getTable("db2", "tbl1")
sessionCatalog.createTempView("tbl1", tempTable1, overrideIfExists = false)
sessionCatalog.setCurrentDatabase("db2")
// If we explicitly specify the database, we'll look up the relation in that database
assert(sessionCatalog.lookupRelation(TableIdentifier("tbl1", Some("db2")))
assert(sessionCatalog.lookupTempViewOrRelation(TableIdentifier("tbl1", Some("db2")))
== SubqueryAlias("tbl1", SimpleCatalogRelation("db2", metastoreTable1), None))
// Otherwise, we'll first look up a temporary table with the same name
assert(sessionCatalog.lookupRelation(TableIdentifier("tbl1"))
assert(sessionCatalog.lookupTempViewOrRelation(TableIdentifier("tbl1"))
== SubqueryAlias("tbl1", tempTable1, Some(TableIdentifier("tbl1"))))
// Then, if that does not exist, look up the relation in the current database
sessionCatalog.dropTable(TableIdentifier("tbl1"), ignoreIfNotExists = false, purge = false)
assert(sessionCatalog.lookupRelation(TableIdentifier("tbl1"))
sessionCatalog.dropTempView("tbl1")
assert(sessionCatalog.lookupTempViewOrRelation(TableIdentifier("tbl1"))
== SubqueryAlias("tbl1", SimpleCatalogRelation("db2", metastoreTable1), None))
}

Expand All @@ -412,7 +395,7 @@ class SessionCatalogSuite extends SparkFunSuite {
val catalog = new SessionCatalog(newBasicCatalog())
val tmpView = Range(1, 10, 2, 10)
catalog.createTempView("vw1", tmpView, overrideIfExists = false)
val plan = catalog.lookupRelation(TableIdentifier("vw1"), Option("range"))
val plan = catalog.lookupTempViewOrRelation(TableIdentifier("vw1"), Option("range"))
assert(plan == SubqueryAlias("range", tmpView, Option(TableIdentifier("vw1"))))
}

Expand All @@ -423,48 +406,15 @@ class SessionCatalogSuite extends SparkFunSuite {
assert(!catalog.tableExists(TableIdentifier("tbl3", Some("db2"))))
assert(!catalog.tableExists(TableIdentifier("tbl1", Some("db1"))))
assert(!catalog.tableExists(TableIdentifier("tbl2", Some("db1"))))
// If database is explicitly specified, do not check temporary tables
val tempTable = Range(1, 10, 1, 10)
catalog.createTempView("tbl3", tempTable, overrideIfExists = false)
assert(!catalog.tableExists(TableIdentifier("tbl3", Some("db2"))))
// If database is not explicitly specified, check the current database
catalog.setCurrentDatabase("db2")
assert(catalog.tableExists(TableIdentifier("tbl1")))
assert(catalog.tableExists(TableIdentifier("tbl2")))
assert(catalog.tableExists(TableIdentifier("tbl3")))
}

test("tableExists on temporary views") {
val catalog = new SessionCatalog(newBasicCatalog())
val tempTable = Range(1, 10, 2, 10)
assert(!catalog.tableExists(TableIdentifier("view1")))
assert(!catalog.tableExists(TableIdentifier("view1", Some("default"))))
catalog.createTempView("view1", tempTable, overrideIfExists = false)
assert(catalog.tableExists(TableIdentifier("view1")))
assert(!catalog.tableExists(TableIdentifier("view1", Some("default"))))
}

test("getTableMetadata on temporary views") {
val catalog = new SessionCatalog(newBasicCatalog())
val tempTable = Range(1, 10, 2, 10)
val m = intercept[AnalysisException] {
catalog.getTableMetadata(TableIdentifier("view1"))
}.getMessage
assert(m.contains("Table or view 'view1' not found in database 'default'"))

val m2 = intercept[AnalysisException] {
catalog.getTableMetadata(TableIdentifier("view1", Some("default")))
}.getMessage
assert(m2.contains("Table or view 'view1' not found in database 'default'"))

catalog.createTempView("view1", tempTable, overrideIfExists = false)
assert(catalog.getTableMetadata(TableIdentifier("view1")).identifier.table == "view1")
assert(catalog.getTableMetadata(TableIdentifier("view1")).schema(0).name == "id")

val m3 = intercept[AnalysisException] {
catalog.getTableMetadata(TableIdentifier("view1", Some("default")))
}.getMessage
assert(m3.contains("Table or view 'view1' not found in database 'default'"))
// Even if database is not specified and there exists a same-name temp view, `tableExists`
// should ignore temp views and only check metastore tables/views.
catalog.createTempView("tbl3", Range(1, 10, 1, 10), overrideIfExists = false)
assert(!catalog.tableExists(TableIdentifier("tbl3")))
}

test("list tables without pattern") {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -463,9 +463,7 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
* @since 1.4.0
*/
def table(tableName: String): DataFrame = {
Dataset.ofRows(sparkSession,
sparkSession.sessionState.catalog.lookupRelation(
sparkSession.sessionState.sqlParser.parseTableIdentifier(tableName)))
sparkSession.table(tableName)
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -556,7 +556,7 @@ class SparkSession private(
}

private[sql] def table(tableIdent: TableIdentifier): DataFrame = {
Dataset.ofRows(self, sessionState.catalog.lookupRelation(tableIdent))
Dataset.ofRows(self, sessionState.catalog.lookupTempViewOrRelation(tableIdent))
}

/* ----------------- *
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -189,31 +189,39 @@ case class DropTableCommand(

override def run(sparkSession: SparkSession): Seq[Row] = {
val catalog = sparkSession.sessionState.catalog
if (!catalog.tableExists(tableName)) {
if (!ifExists) {
val objectName = if (isView) "View" else "Table"
throw new AnalysisException(s"$objectName to drop '$tableName' does not exist")
}
} else {
// If the command DROP VIEW is to drop a table or DROP TABLE is to drop a view
// issue an exception.
catalog.getTableMetadataOption(tableName).map(_.tableType match {
case CatalogTableType.VIEW if !isView =>
throw new AnalysisException(
"Cannot drop a view with DROP TABLE. Please use DROP VIEW instead")
case o if o != CatalogTableType.VIEW && isView =>
throw new AnalysisException(
s"Cannot drop a table with DROP VIEW. Please use DROP TABLE instead")
case _ =>
})
try {
sparkSession.sharedState.cacheManager.uncacheQuery(
sparkSession.table(tableName.quotedString))
} catch {
case NonFatal(e) => log.warn(e.toString, e)

// If the table name contains database part, we should drop a metastore table directly,
// otherwise, try to drop a temp view first, if that not exist, drop metastore table.
val dropMetastoreTable =
tableName.database.isDefined || !catalog.dropTempView(tableName.table)
Copy link
Member

Choose a reason for hiding this comment

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

Drop Table is unable to drop a temp view, right?

    spark.range(10).createTempView("tempView")
    sql("DESC tempView").show()
    sql("DROP TABLE tempView")
    sql("DESC tempView").show()

Copy link
Contributor Author

@cloud-fan cloud-fan Sep 7, 2016

Choose a reason for hiding this comment

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

Actually I noticed this and fixed it before, but it breaks a lot of tests, because we call "temp view" as "temp table" before. I'd like to keep this behaviour as it was, we can discuss how to fix it in follow-ups.

Copy link
Member

Choose a reason for hiding this comment

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

I see. Thanks!


if (dropMetastoreTable) {
if (!catalog.tableExists(tableName)) {
if (!ifExists) {
val objectName = if (isView) "View" else "Table"
throw new AnalysisException(s"$objectName to drop '$tableName' does not exist")
}
} else {
// If the command DROP VIEW is to drop a table or DROP TABLE is to drop a view
// issue an exception.
catalog.getTableMetadataOption(tableName).map(_.tableType match {
case CatalogTableType.VIEW if !isView =>
throw new AnalysisException(
"Cannot drop a view with DROP TABLE. Please use DROP VIEW instead")
case o if o != CatalogTableType.VIEW && isView =>
throw new AnalysisException(
s"Cannot drop a table with DROP VIEW. Please use DROP TABLE instead")
case _ =>
})
try {
sparkSession.sharedState.cacheManager.uncacheQuery(
sparkSession.table(tableName.quotedString))
} catch {
case NonFatal(e) => log.warn(e.toString, e)
}
catalog.refreshTable(tableName)
catalog.dropTable(tableName, ifExists, purge)
}
catalog.refreshTable(tableName)
catalog.dropTable(tableName, ifExists, purge)
}
Seq.empty[Row]
}
Expand Down Expand Up @@ -470,10 +478,6 @@ case class AlterTableRecoverPartitionsCommand(
if (!catalog.tableExists(tableName)) {
throw new AnalysisException(s"Table $tableName in $cmd does not exist.")
}
if (catalog.isTemporaryTable(tableName)) {
throw new AnalysisException(
s"Operation not allowed: $cmd on temporary tables: $tableName")
}
val table = catalog.getTableMetadata(tableName)
if (DDLUtils.isDatasourceTable(table)) {
throw new AnalysisException(
Expand Down
Loading