This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new 1040657ecbce [SPARK-49211][SQL][FOLLOW-UP] Support catalog in
QualifiedTableName
1040657ecbce is described below
commit 1040657ecbce001d7233f4607009d1bcc3fecc18
Author: Rui Wang <[email protected]>
AuthorDate: Fri Sep 27 08:26:24 2024 +0800
[SPARK-49211][SQL][FOLLOW-UP] Support catalog in QualifiedTableName
Support catalog in QualifiedTableName and remove `FullQualifiedTableName`.
Consolidate and remove duplicate code.
No
Existing UT
No
Closes #48255 from amaliujia/qualifedtablename.
Authored-by: Rui Wang <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit fc9d421a2345987105aa97947c867ac80ba48a05)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../spark/sql/catalyst/catalog/SessionCatalog.scala | 18 +++++++++---------
.../org/apache/spark/sql/catalyst/identifiers.scala | 21 +++++++++++++++++----
.../sql/catalyst/catalog/SessionCatalogSuite.scala | 8 ++++----
.../execution/datasources/DataSourceStrategy.scala | 4 ++--
.../spark/sql/StatisticsCollectionTestBase.scala | 4 ++--
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 10 +++++-----
.../spark/sql/execution/command/DDLSuite.scala | 6 +++---
.../execution/command/v1/TruncateTableSuite.scala | 4 ++--
.../spark/sql/hive/HiveMetastoreCatalog.scala | 10 +++++-----
9 files changed, 49 insertions(+), 36 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index 99074b859a7f..0de9673a5f96 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -194,7 +194,7 @@ class SessionCatalog(
}
}
- private val tableRelationCache: Cache[FullQualifiedTableName, LogicalPlan] =
{
+ private val tableRelationCache: Cache[QualifiedTableName, LogicalPlan] = {
var builder = CacheBuilder.newBuilder()
.maximumSize(cacheSize)
@@ -202,33 +202,33 @@ class SessionCatalog(
builder = builder.expireAfterWrite(cacheTTL, TimeUnit.SECONDS)
}
- builder.build[FullQualifiedTableName, LogicalPlan]()
+ builder.build[QualifiedTableName, LogicalPlan]()
}
/** This method provides a way to get a cached plan. */
- def getCachedPlan(t: FullQualifiedTableName, c: Callable[LogicalPlan]):
LogicalPlan = {
+ def getCachedPlan(t: QualifiedTableName, c: Callable[LogicalPlan]):
LogicalPlan = {
tableRelationCache.get(t, c)
}
/** This method provides a way to get a cached plan if the key exists. */
- def getCachedTable(key: FullQualifiedTableName): LogicalPlan = {
+ def getCachedTable(key: QualifiedTableName): LogicalPlan = {
tableRelationCache.getIfPresent(key)
}
/** This method provides a way to cache a plan. */
- def cacheTable(t: FullQualifiedTableName, l: LogicalPlan): Unit = {
+ def cacheTable(t: QualifiedTableName, l: LogicalPlan): Unit = {
tableRelationCache.put(t, l)
}
/** This method provides a way to invalidate a cached plan. */
- def invalidateCachedTable(key: FullQualifiedTableName): Unit = {
+ def invalidateCachedTable(key: QualifiedTableName): Unit = {
tableRelationCache.invalidate(key)
}
/** This method discards any cached table relation plans for the given table
identifier. */
def invalidateCachedTable(name: TableIdentifier): Unit = {
val qualified = qualifyIdentifier(name)
- invalidateCachedTable(FullQualifiedTableName(
+ invalidateCachedTable(QualifiedTableName(
qualified.catalog.get, qualified.database.get, qualified.table))
}
@@ -297,7 +297,7 @@ class SessionCatalog(
}
if (cascade && databaseExists(dbName)) {
listTables(dbName).foreach { t =>
- invalidateCachedTable(FullQualifiedTableName(SESSION_CATALOG_NAME,
dbName, t.table))
+ invalidateCachedTable(QualifiedTableName(SESSION_CATALOG_NAME, dbName,
t.table))
}
}
externalCatalog.dropDatabase(dbName, ignoreIfNotExists, cascade)
@@ -1132,7 +1132,7 @@ class SessionCatalog(
def refreshTable(name: TableIdentifier): Unit = synchronized {
getLocalOrGlobalTempView(name).map(_.refresh).getOrElse {
val qualifiedIdent = qualifyIdentifier(name)
- val qualifiedTableName = FullQualifiedTableName(
+ val qualifiedTableName = QualifiedTableName(
qualifiedIdent.catalog.get, qualifiedIdent.database.get,
qualifiedIdent.table)
tableRelationCache.invalidate(qualifiedTableName)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/identifiers.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/identifiers.scala
index cc881539002b..ceced9313940 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/identifiers.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/identifiers.scala
@@ -17,6 +17,8 @@
package org.apache.spark.sql.catalyst
+import org.apache.spark.sql.connector.catalog.CatalogManager
+
/**
* An identifier that optionally specifies a database.
*
@@ -107,14 +109,25 @@ case class TableIdentifier(table: String, database:
Option[String], catalog: Opt
}
/** A fully qualified identifier for a table (i.e., database.tableName) */
-case class QualifiedTableName(database: String, name: String) {
- override def toString: String = s"$database.$name"
-}
+case class QualifiedTableName(catalog: String, database: String, name: String)
{
+ /** Two argument ctor for backward compatibility. */
+ def this(database: String, name: String) = this(
+ catalog = CatalogManager.SESSION_CATALOG_NAME,
+ database = database,
+ name = name)
-case class FullQualifiedTableName(catalog: String, database: String, name:
String) {
override def toString: String = s"$catalog.$database.$name"
}
+object QualifiedTableName {
+ def apply(catalog: String, database: String, name: String):
QualifiedTableName = {
+ new QualifiedTableName(catalog, database, name)
+ }
+
+ def apply(database: String, name: String): QualifiedTableName =
+ new QualifiedTableName(database = database, name = name)
+}
+
object TableIdentifier {
def apply(tableName: String): TableIdentifier = new
TableIdentifier(tableName)
def apply(table: String, database: Option[String]): TableIdentifier =
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
index 447a85fe4a41..05c1c33520da 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
@@ -22,7 +22,7 @@ import scala.concurrent.duration._
import org.scalatest.concurrent.Eventually
import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.{AliasIdentifier, FullQualifiedTableName,
FunctionIdentifier, TableIdentifier}
+import org.apache.spark.sql.catalyst.{AliasIdentifier, FunctionIdentifier,
QualifiedTableName, TableIdentifier}
import org.apache.spark.sql.catalyst.analysis._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
@@ -1880,7 +1880,7 @@ abstract class SessionCatalogSuite extends AnalysisTest
with Eventually {
conf.setConf(StaticSQLConf.METADATA_CACHE_TTL_SECONDS, 1L)
withConfAndEmptyCatalog(conf) { catalog =>
- val table = FullQualifiedTableName(
+ val table = QualifiedTableName(
CatalogManager.SESSION_CATALOG_NAME, catalog.getCurrentDatabase,
"test")
// First, make sure the test table is not cached.
@@ -1900,14 +1900,14 @@ abstract class SessionCatalogSuite extends AnalysisTest
with Eventually {
test("SPARK-34197: refreshTable should not invalidate the relation cache for
temporary views") {
withBasicCatalog { catalog =>
createTempView(catalog, "tbl1", Range(1, 10, 1, 10), false)
- val qualifiedName1 = FullQualifiedTableName(SESSION_CATALOG_NAME,
"default", "tbl1")
+ val qualifiedName1 = QualifiedTableName(SESSION_CATALOG_NAME, "default",
"tbl1")
catalog.cacheTable(qualifiedName1, Range(1, 10, 1, 10))
catalog.refreshTable(TableIdentifier("tbl1"))
assert(catalog.getCachedTable(qualifiedName1) != null)
createGlobalTempView(catalog, "tbl2", Range(2, 10, 1, 10), false)
val qualifiedName2 =
- FullQualifiedTableName(SESSION_CATALOG_NAME,
catalog.globalTempViewManager.database, "tbl2")
+ QualifiedTableName(SESSION_CATALOG_NAME,
catalog.globalTempViewManager.database, "tbl2")
catalog.cacheTable(qualifiedName2, Range(2, 10, 1, 10))
catalog.refreshTable(TableIdentifier("tbl2",
Some(catalog.globalTempViewManager.database)))
assert(catalog.getCachedTable(qualifiedName2) != null)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
index 548a5734da06..431480bb2edf 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
@@ -27,7 +27,7 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._
-import org.apache.spark.sql.catalyst.{expressions, CatalystTypeConverters,
FullQualifiedTableName, InternalRow, SQLConfHelper}
+import org.apache.spark.sql.catalyst.{expressions, CatalystTypeConverters,
InternalRow, QualifiedTableName, SQLConfHelper}
import org.apache.spark.sql.catalyst.CatalystTypeConverters.convertToScala
import org.apache.spark.sql.catalyst.analysis._
import org.apache.spark.sql.catalyst.catalog._
@@ -242,7 +242,7 @@ class FindDataSourceTable(sparkSession: SparkSession)
extends Rule[LogicalPlan]
private def readDataSourceTable(
table: CatalogTable, extraOptions: CaseInsensitiveStringMap):
LogicalPlan = {
val qualifiedTableName =
- FullQualifiedTableName(table.identifier.catalog.get, table.database,
table.identifier.table)
+ QualifiedTableName(table.identifier.catalog.get, table.database,
table.identifier.table)
val catalog = sparkSession.sessionState.catalog
val dsOptions = DataSourceUtils.generateDatasourceOptions(extraOptions,
table)
catalog.getCachedPlan(qualifiedTableName, () => {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionTestBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionTestBase.scala
index 2e237a9b9ea1..4468edc48373 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionTestBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionTestBase.scala
@@ -25,7 +25,7 @@ import java.time.LocalDateTime
import scala.collection.mutable
import scala.util.Random
-import org.apache.spark.sql.catalyst.{FullQualifiedTableName, TableIdentifier}
+import org.apache.spark.sql.catalyst.{QualifiedTableName, TableIdentifier}
import org.apache.spark.sql.catalyst.catalog.{CatalogColumnStat,
CatalogStatistics, CatalogTable, HiveTableRelation}
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions.AttributeMap
@@ -270,7 +270,7 @@ abstract class StatisticsCollectionTestBase extends
QueryTest with SQLTestUtils
def getTableFromCatalogCache(tableName: String): LogicalPlan = {
val catalog = spark.sessionState.catalog
- val qualifiedTableName = FullQualifiedTableName(
+ val qualifiedTableName = QualifiedTableName(
CatalogManager.SESSION_CATALOG_NAME, catalog.getCurrentDatabase,
tableName)
catalog.getCachedTable(qualifiedTableName)
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index f60e81d3a4c6..9df4b0932f25 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -27,7 +27,7 @@ import scala.concurrent.duration.MICROSECONDS
import org.apache.spark.{SparkException, SparkUnsupportedOperationException}
import org.apache.spark.sql._
-import org.apache.spark.sql.catalyst.{FullQualifiedTableName, InternalRow,
TableIdentifier}
+import org.apache.spark.sql.catalyst.{InternalRow, QualifiedTableName,
TableIdentifier}
import
org.apache.spark.sql.catalyst.analysis.{CannotReplaceMissingTableException,
NoSuchDatabaseException, NoSuchNamespaceException, TableAlreadyExistsException}
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType, CatalogUtils}
import org.apache.spark.sql.catalyst.parser.ParseException
@@ -3404,7 +3404,7 @@ class DataSourceV2SQLSuiteV1Filter
// Reset CatalogManager to clear the materialized `spark_catalog`
instance, so that we can
// configure a new implementation.
- val table1 = FullQualifiedTableName(SESSION_CATALOG_NAME, "default", "t")
+ val table1 = QualifiedTableName(SESSION_CATALOG_NAME, "default", "t")
spark.sessionState.catalogManager.reset()
withSQLConf(
V2_SESSION_CATALOG_IMPLEMENTATION.key ->
@@ -3413,7 +3413,7 @@ class DataSourceV2SQLSuiteV1Filter
checkParquet(table1.toString, path.getAbsolutePath)
}
}
- val table2 = FullQualifiedTableName("testcat3", "default", "t")
+ val table2 = QualifiedTableName("testcat3", "default", "t")
withSQLConf(
"spark.sql.catalog.testcat3" ->
classOf[V2CatalogSupportBuiltinDataSource].getName) {
withTempPath { path =>
@@ -3432,7 +3432,7 @@ class DataSourceV2SQLSuiteV1Filter
// Reset CatalogManager to clear the materialized `spark_catalog`
instance, so that we can
// configure a new implementation.
spark.sessionState.catalogManager.reset()
- val table1 = FullQualifiedTableName(SESSION_CATALOG_NAME, "default", "t")
+ val table1 = QualifiedTableName(SESSION_CATALOG_NAME, "default", "t")
withSQLConf(
V2_SESSION_CATALOG_IMPLEMENTATION.key ->
classOf[V2CatalogSupportBuiltinDataSource].getName) {
@@ -3441,7 +3441,7 @@ class DataSourceV2SQLSuiteV1Filter
}
}
- val table2 = FullQualifiedTableName("testcat3", "default", "t")
+ val table2 = QualifiedTableName("testcat3", "default", "t")
withSQLConf(
"spark.sql.catalog.testcat3" ->
classOf[V2CatalogSupportBuiltinDataSource].getName) {
withTempPath { path =>
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index 0466781c119a..1124184dded7 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -27,7 +27,7 @@ import org.apache.hadoop.fs.permission.{AclEntry, AclStatus}
import org.apache.spark.{SparkClassNotFoundException, SparkException,
SparkFiles, SparkRuntimeException}
import org.apache.spark.internal.config
import org.apache.spark.sql.{AnalysisException, QueryTest, Row, SaveMode}
-import org.apache.spark.sql.catalyst.{FullQualifiedTableName,
FunctionIdentifier, TableIdentifier}
+import org.apache.spark.sql.catalyst.{FunctionIdentifier, QualifiedTableName,
TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.TempTableAlreadyExistsException
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
@@ -220,7 +220,7 @@ class InMemoryCatalogedDDLSuite extends DDLSuite with
SharedSparkSession {
test("SPARK-25403 refresh the table after inserting data") {
withTable("t") {
val catalog = spark.sessionState.catalog
- val table = FullQualifiedTableName(
+ val table = QualifiedTableName(
CatalogManager.SESSION_CATALOG_NAME, catalog.getCurrentDatabase, "t")
sql("CREATE TABLE t (a INT) USING parquet")
sql("INSERT INTO TABLE t VALUES (1)")
@@ -234,7 +234,7 @@ class InMemoryCatalogedDDLSuite extends DDLSuite with
SharedSparkSession {
withTable("t") {
withTempDir { dir =>
val catalog = spark.sessionState.catalog
- val table = FullQualifiedTableName(
+ val table = QualifiedTableName(
CatalogManager.SESSION_CATALOG_NAME, catalog.getCurrentDatabase, "t")
val p1 = s"${dir.getCanonicalPath}/p1"
val p2 = s"${dir.getCanonicalPath}/p2"
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/TruncateTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/TruncateTableSuite.scala
index 5810a35ddcf8..747a37827501 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/TruncateTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/TruncateTableSuite.scala
@@ -23,7 +23,7 @@ import org.apache.hadoop.fs.Path
import org.apache.hadoop.fs.permission.{AclEntry, AclEntryScope, AclEntryType,
FsAction, FsPermission}
import org.apache.spark.sql.{AnalysisException, Row}
-import org.apache.spark.sql.catalyst.{FullQualifiedTableName, TableIdentifier}
+import org.apache.spark.sql.catalyst.{QualifiedTableName, TableIdentifier}
import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.execution.command
import org.apache.spark.sql.execution.command.FakeLocalFsFileSystem
@@ -148,7 +148,7 @@ trait TruncateTableSuiteBase extends
command.TruncateTableSuiteBase {
val catalog = spark.sessionState.catalog
val qualifiedTableName =
- FullQualifiedTableName(CatalogManager.SESSION_CATALOG_NAME, "ns",
"tbl")
+ QualifiedTableName(CatalogManager.SESSION_CATALOG_NAME, "ns",
"tbl")
val cachedPlan = catalog.getCachedTable(qualifiedTableName)
assert(cachedPlan.stats.sizeInBytes == 0)
}
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
index 00536135a1b1..54b9db967d2d 100644
---
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
+++
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
@@ -27,7 +27,7 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.SparkException
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{AnalysisException, SparkSession}
-import org.apache.spark.sql.catalyst.{FullQualifiedTableName, TableIdentifier}
+import org.apache.spark.sql.catalyst.{QualifiedTableName, TableIdentifier}
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.types.DataTypeUtils
@@ -55,7 +55,7 @@ private[hive] class HiveMetastoreCatalog(sparkSession:
SparkSession) extends Log
private val tableCreationLocks = Striped.lazyWeakLock(100)
/** Acquires a lock on the table cache for the duration of `f`. */
- private def withTableCreationLock[A](tableName: FullQualifiedTableName, f:
=> A): A = {
+ private def withTableCreationLock[A](tableName: QualifiedTableName, f: =>
A): A = {
val lock = tableCreationLocks.get(tableName)
lock.lock()
try f finally {
@@ -65,7 +65,7 @@ private[hive] class HiveMetastoreCatalog(sparkSession:
SparkSession) extends Log
// For testing only
private[hive] def getCachedDataSourceTable(table: TableIdentifier):
LogicalPlan = {
- val key = FullQualifiedTableName(
+ val key = QualifiedTableName(
// scalastyle:off caselocale
table.catalog.getOrElse(CatalogManager.SESSION_CATALOG_NAME).toLowerCase,
table.database.getOrElse(sessionState.catalog.getCurrentDatabase).toLowerCase,
@@ -75,7 +75,7 @@ private[hive] class HiveMetastoreCatalog(sparkSession:
SparkSession) extends Log
}
private def getCached(
- tableIdentifier: FullQualifiedTableName,
+ tableIdentifier: QualifiedTableName,
pathsInMetastore: Seq[Path],
schemaInMetastore: StructType,
expectedFileFormat: Class[_ <: FileFormat],
@@ -193,7 +193,7 @@ private[hive] class HiveMetastoreCatalog(sparkSession:
SparkSession) extends Log
fileType: String,
isWrite: Boolean): LogicalRelation = {
val metastoreSchema = relation.tableMeta.schema
- val tableIdentifier =
FullQualifiedTableName(relation.tableMeta.identifier.catalog.get,
+ val tableIdentifier =
QualifiedTableName(relation.tableMeta.identifier.catalog.get,
relation.tableMeta.database, relation.tableMeta.identifier.table)
val lazyPruningEnabled =
sparkSession.sqlContext.conf.manageFilesourcePartitions
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
