This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 2274217b217e [SPARK-57518][SQL] Make ThriftServer JDBC metadata
operations DataSource V2 catalog-aware
2274217b217e is described below
commit 2274217b217e699b552c9d5d398fa1405260215b
Author: Anupam Yadav <[email protected]>
AuthorDate: Wed Jul 15 01:46:55 2026 +0800
[SPARK-57518][SQL] Make ThriftServer JDBC metadata operations DataSource V2
catalog-aware
### What changes were proposed in this pull request?
Route ThriftServer JDBC metadata operations (getCatalogs, getSchemas,
getTables, getColumns) through CatalogManager so they honor DataSource V2
catalogs and the default catalog. Populate TABLE_CAT with the real catalog name
where the operation actually resolves through that catalog. Introduce a new
conf `spark.sql.thriftServer.catalogMetadata.enabled` (default true) with
legacy fallback when disabled.
### Why are the changes needed?
With `spark.sql.catalog.*` and `spark.sql.defaultCatalog` set, JDBC/BI
clients get inconsistent metadata because the metadata operations used the V1
SessionCatalog directly, ignoring any configured DSv2 catalogs.
### Design notes
(a) A null `catalogName` resolves to the CURRENT catalog (consistent with
Spark's own unspecified-to-current resolution and with Trino/Snowflake
behavior), not all catalogs.
(b) getCatalogs returns CatalogManager.listCatalogs() (including
spark_catalog, sorted alphabetically).
(c) getSchemas is fully DSv2-aware: it routes to the V1 SessionCatalog for
spark_catalog and to `SupportsNamespaces.listNamespaces()` (top-level
namespaces) for a DSv2 current catalog, populating TABLE_CATALOG with the real
catalog name.
(d) getTables/getColumns enumerate from the V1 SessionCatalog only.
TABLE_CAT is therefore populated with the real catalog name only when the
current catalog IS spark_catalog; otherwise it stays the legacy empty/null
value, so V1-listed rows are never mislabeled with an unrelated DSv2 catalog
name. The conf defaults to ON with an escape hatch for clients that relied on
parsing empty TABLE_CAT.
(e) KNOWN LIMITATION: listCatalogs() returns only ALREADY-LOADED catalogs,
so catalogs that are configured but never accessed will not be listed. This is
documented; we do not eagerly load catalogs.
(f) KNOWN LIMITATION / FOLLOWUP: getTables/getColumns do not yet route
through DSv2 catalogs. When the current catalog is a DSv2 catalog they return
spark_catalog's tables/columns with empty/null TABLE_CAT (not the DSv2
catalog's objects). Routing these two operations through DSv2 catalogs is a
follow-up.
(g) V2-specific metadata authorization is deferred to a follow-up. Existing
Hive auth hooks are unchanged and getCatalogs/getSchemas already pass null priv
objects.
### Does this PR introduce _any_ user-facing change?
Yes. getCatalogs now returns loaded catalogs, getSchemas reflects the
current DSv2 catalog's namespaces, and TABLE_CAT/TABLE_CATALOG are populated
with the real catalog name where applicable (gated by the new conf). New conf:
`spark.sql.thriftServer.catalogMetadata.enabled`.
### How was this patch tested?
SparkMetadataOperationSuite covers: default spark_catalog path; configured
in-memory DSv2 catalog path for getCatalogs and getSchemas; the conf-disabled
legacy path for all four operations; and a test asserting getTables/getColumns
do not label their V1-listed rows with a DSv2 current catalog's name.
### Was this patch authored or co-authored using generative AI tooling?
Authored with assistance from Claude Opus 4.8
Closes #56627 from yadavay-amzn/fix/SPARK-57518-thriftserver-dsv2-metadata.
Authored-by: Anupam Yadav <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 15 ++
.../thriftserver/SparkGetCatalogsOperation.scala | 8 +
.../thriftserver/SparkGetColumnsOperation.scala | 2 +-
.../thriftserver/SparkGetSchemasOperation.scala | 55 +++++-
.../thriftserver/SparkGetTablesOperation.scala | 2 +-
.../sql/hive/thriftserver/SparkOperation.scala | 21 +++
.../thriftserver/SparkMetadataOperationSuite.scala | 206 ++++++++++++++++++++-
7 files changed, 292 insertions(+), 17 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 22f5c8416e05..aca325a43e85 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -2097,6 +2097,21 @@ object SQLConf {
.intConf
.createWithDefault(200)
+ val THRIFTSERVER_CATALOG_METADATA_ENABLED =
+ buildConf("spark.sql.thriftServer.catalogMetadata.enabled")
+ .doc("When true, JDBC metadata operations (getCatalogs, getSchemas,
getTables, " +
+ "getColumns) are DataSource V2 catalog-aware: getCatalogs returns all
loaded catalogs, " +
+ "and TABLE_CAT is populated with the real catalog name. When false,
the legacy behavior " +
+ "is preserved (empty TABLE_CAT and getCatalogs returns no rows). " +
+ "Known limitation: when the current catalog is a DSv2 catalog and this
conf is enabled, " +
+ "getSchemas lists the DSv2 catalog's namespaces while
getTables/getColumns still " +
+ "enumerate session catalog (spark_catalog) objects until DSv2 routing
for those " +
+ "operations lands in a follow-up.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(true)
+
val DATA_SOURCE_DONT_ASSERT_ON_PREDICATE =
buildConf("spark.sql.dataSource.skipAssertOnPredicatePushdown")
.internal()
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetCatalogsOperation.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetCatalogsOperation.scala
index a02b2ca8966d..aecb5fee580e 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetCatalogsOperation.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetCatalogsOperation.scala
@@ -54,6 +54,14 @@ private[hive] class SparkGetCatalogsOperation(
if (isAuthV2Enabled) {
authorizeMetaGets(HiveOperationType.GET_CATALOGS, null)
}
+ if (isCatalogMetadataEnabled) {
+ // Note: listCatalogs() returns only ALREADY-LOADED catalogs (catalogs
configured via
+ // spark.sql.catalog.* but never accessed are not returned). We do NOT
eagerly load
+ // all configured catalogs to avoid unexpected side effects and
performance overhead.
+ catalogManager.listCatalogs(None).foreach { catalogName =>
+ rowSet.addRow(Array[AnyRef](catalogName))
+ }
+ }
setState(OperationState.FINISHED)
} catch onError()
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetColumnsOperation.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetColumnsOperation.scala
index 01b94c1e9cba..b429086db22f 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetColumnsOperation.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetColumnsOperation.scala
@@ -207,7 +207,7 @@ private[hive] class SparkGetColumnsOperation(
pos + 1
}
val rowData = Array[AnyRef](
- null, // TABLE_CAT
+ sessionCatalogTableCat(null), // TABLE_CAT
dbName, // TABLE_SCHEM
tableName, // TABLE_NAME
column.name, // COLUMN_NAME
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetSchemasOperation.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetSchemasOperation.scala
index 4b8b603eede5..9800de8706f6 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetSchemasOperation.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetSchemasOperation.scala
@@ -28,6 +28,7 @@ import org.apache.hive.service.cli.session.HiveSession
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys._
import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.connector.catalog.{CatalogManager,
SupportsNamespaces}
/**
* Spark's own GetSchemasOperation
@@ -71,15 +72,51 @@ private[hive] class SparkGetSchemasOperation(
try {
val schemaPattern = convertSchemaPattern(schemaName)
- catalog.listDatabases(schemaPattern).foreach { dbName =>
- rowSet.addRow(Array[AnyRef](dbName, DEFAULT_HIVE_CATALOG))
- }
-
- val globalTempViewDb = catalog.globalTempDatabase
- val databasePattern =
Pattern.compile(CLIServiceUtils.patternToRegex(schemaName))
- if (schemaName == null || schemaName.isEmpty ||
- databasePattern.matcher(globalTempViewDb).matches()) {
- rowSet.addRow(Array[AnyRef](globalTempViewDb, DEFAULT_HIVE_CATALOG))
+ if (isCatalogMetadataEnabled) {
+ // The JDBC catalogName parameter is intentionally not used for
filtering;
+ // resolution always uses the current catalog (SPARK-57518; per-catalog
+ // filtering deferred).
+ val resolvedCatalog = catalogManager.currentCatalog
+ val catalogNameValue = resolvedCatalog.name()
+ if (catalogNameValue == CatalogManager.SESSION_CATALOG_NAME) {
+ // For spark_catalog, use the V1 SessionCatalog directly
(transparent delegation)
+ catalog.listDatabases(schemaPattern).foreach { dbName =>
+ rowSet.addRow(Array[AnyRef](dbName, catalogNameValue))
+ }
+ // Global temp view database is only relevant for spark_catalog
+ val globalTempViewDb = catalog.globalTempDatabase
+ val databasePattern =
Pattern.compile(CLIServiceUtils.patternToRegex(schemaName))
+ if (schemaName == null || schemaName.isEmpty ||
+ databasePattern.matcher(globalTempViewDb).matches()) {
+ rowSet.addRow(Array[AnyRef](globalTempViewDb, catalogNameValue))
+ }
+ } else {
+ resolvedCatalog match {
+ case nsCatalog: SupportsNamespaces =>
+ val databasePattern = Pattern.compile(
+ CLIServiceUtils.patternToRegex(schemaName))
+ nsCatalog.listNamespaces().foreach { ns =>
+ // Only top-level namespaces (depth=1) are exposed as JDBC
schemas.
+ val nsName = ns.head
+ if (schemaName == null || schemaName.isEmpty ||
+ databasePattern.matcher(nsName).matches()) {
+ rowSet.addRow(Array[AnyRef](nsName, catalogNameValue))
+ }
+ }
+ case _ =>
+ // Catalog doesn't support namespaces -- return empty
+ }
+ }
+ } else {
+ catalog.listDatabases(schemaPattern).foreach { dbName =>
+ rowSet.addRow(Array[AnyRef](dbName, DEFAULT_HIVE_CATALOG))
+ }
+ val globalTempViewDb = catalog.globalTempDatabase
+ val databasePattern =
Pattern.compile(CLIServiceUtils.patternToRegex(schemaName))
+ if (schemaName == null || schemaName.isEmpty ||
+ databasePattern.matcher(globalTempViewDb).matches()) {
+ rowSet.addRow(Array[AnyRef](globalTempViewDb, DEFAULT_HIVE_CATALOG))
+ }
}
setState(OperationState.FINISHED)
} catch onError()
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetTablesOperation.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetTablesOperation.scala
index 0579d567d022..d722ea8ae50f 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetTablesOperation.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetTablesOperation.scala
@@ -122,7 +122,7 @@ private[hive] class SparkGetTablesOperation(
tableType: String,
comment: Option[String]): Unit = {
val rowData = Array[AnyRef](
- "",
+ sessionCatalogTableCat(""),
dbName,
tableName,
tableType,
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkOperation.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkOperation.scala
index 10f520314865..419b48c6ab26 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkOperation.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkOperation.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.CurrentUserContext.CURRENT_USER
import org.apache.spark.sql.catalyst.catalog.{CatalogTableType, SessionCatalog}
import org.apache.spark.sql.catalyst.catalog.CatalogTableType.{EXTERNAL,
MANAGED, METRIC_VIEW, VIEW}
+import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.internal.{SessionState, SharedState, SQLConf}
import org.apache.spark.util.Utils
@@ -51,8 +52,28 @@ private[hive] trait SparkOperation extends Operation with
Logging {
final protected def catalog: SessionCatalog = sessionState.catalog
+ final protected def catalogManager: CatalogManager =
sessionState.catalogManager
+
final protected def conf: SQLConf = sessionState.conf
+ final protected def isCatalogMetadataEnabled: Boolean =
+ conf.getConf(SQLConf.THRIFTSERVER_CATALOG_METADATA_ENABLED)
+
+ /**
+ * The TABLE_CAT value for rows listed from the V1 SessionCatalog
(getTables, getColumns).
+ * These operations always list from `spark_catalog`, so we only populate
the real catalog
+ * name when the current catalog IS the session catalog; otherwise we keep
the legacy value
+ * to avoid labeling V1-sourced rows with an unrelated DSv2 catalog name.
Routing
+ * getTables/getColumns through DSv2 catalogs is a follow-up (SPARK-57518).
+ */
+ final protected def sessionCatalogTableCat(legacyValue: AnyRef): AnyRef =
+ if (isCatalogMetadataEnabled &&
+ catalogManager.currentCatalog.name() ==
CatalogManager.SESSION_CATALOG_NAME) {
+ CatalogManager.SESSION_CATALOG_NAME
+ } else {
+ legacyValue
+ }
+
final protected def sparkContext: SparkContext = session.sparkContext
final protected def withClassLoader(f: ClassLoader => Unit): Unit = {
diff --git
a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkMetadataOperationSuite.scala
b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkMetadataOperationSuite.scala
index 2374865b1a0d..b3194cfdef3f 100644
---
a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkMetadataOperationSuite.scala
+++
b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkMetadataOperationSuite.scala
@@ -38,7 +38,7 @@ class SparkMetadataOperationSuite extends
HiveThriftServer2TestBase {
val expected = dbNames.iterator
while (rs.next() || expected.hasNext) {
assert(rs.getString("TABLE_SCHEM") === expected.next())
- assert(rs.getString("TABLE_CATALOG").isEmpty)
+ assert(rs.getString("TABLE_CATALOG") === "spark_catalog")
}
// Make sure there are no more elements
assert(!rs.next())
@@ -250,6 +250,20 @@ class SparkMetadataOperationSuite extends
HiveThriftServer2TestBase {
withJdbcStatement() { statement =>
val metaData = statement.getConnection.getMetaData
val rs = metaData.getCatalogs
+ // With catalog metadata enabled (default), getCatalogs returns loaded
catalogs
+ assert(rs.next())
+ assert(rs.getString("TABLE_CAT") === "spark_catalog")
+ assert(!rs.next())
+ }
+ }
+
+ test("GetCatalogsOperation with catalog metadata disabled") {
+ withJdbcStatement() { statement =>
+ statement.execute(
+ "SET spark.sql.thriftServer.catalogMetadata.enabled=false")
+ val metaData = statement.getConnection.getMetaData
+ val rs = metaData.getCatalogs
+ // Legacy behavior: empty result set
assert(!rs.next())
}
}
@@ -315,7 +329,7 @@ class SparkMetadataOperationSuite extends
HiveThriftServer2TestBase {
var pos = 0
while (rowSet.next()) {
- assert(rowSet.getString("TABLE_CAT") === null)
+ assert(rowSet.getString("TABLE_CAT") === "spark_catalog")
assert(rowSet.getString("TABLE_SCHEM") === schemaName)
assert(rowSet.getString("TABLE_NAME") === tableName)
assert(rowSet.getString("COLUMN_NAME") === schema(pos).name)
@@ -368,7 +382,7 @@ class SparkMetadataOperationSuite extends
HiveThriftServer2TestBase {
val data = statement.getConnection.getMetaData
val rowSet = data.getColumns("", "global_temp", viewName, null)
while (rowSet.next()) {
- assert(rowSet.getString("TABLE_CAT") === null)
+ assert(rowSet.getString("TABLE_CAT") === "spark_catalog")
assert(rowSet.getString("TABLE_SCHEM") === "global_temp")
assert(rowSet.getString("TABLE_NAME") === viewName)
assert(rowSet.getString("COLUMN_NAME") === "i")
@@ -396,7 +410,7 @@ class SparkMetadataOperationSuite extends
HiveThriftServer2TestBase {
val data = statement.getConnection.getMetaData
val rowSet = data.getColumns("", "global_temp", viewName1, null)
while (rowSet.next()) {
- assert(rowSet.getString("TABLE_CAT") === null)
+ assert(rowSet.getString("TABLE_CAT") === "spark_catalog")
assert(rowSet.getString("TABLE_SCHEM") === "global_temp")
assert(rowSet.getString("TABLE_NAME") === viewName1)
assert(rowSet.getString("COLUMN_NAME") === "i")
@@ -422,7 +436,7 @@ class SparkMetadataOperationSuite extends
HiveThriftServer2TestBase {
val data = statement.getConnection.getMetaData
val rowSet = data.getColumns("", "global_temp", viewName2, null)
while (rowSet.next()) {
- assert(rowSet.getString("TABLE_CAT") === null)
+ assert(rowSet.getString("TABLE_CAT") === "spark_catalog")
assert(rowSet.getString("TABLE_SCHEM") === "global_temp")
assert(rowSet.getString("TABLE_NAME") === viewName2)
assert(rowSet.getString("COLUMN_NAME") === "i")
@@ -449,7 +463,7 @@ class SparkMetadataOperationSuite extends
HiveThriftServer2TestBase {
val data = statement.getConnection.getMetaData
val rowSet = data.getColumns("", "global_temp", viewName, "n")
while (rowSet.next()) {
- assert(rowSet.getString("TABLE_CAT") === null)
+ assert(rowSet.getString("TABLE_CAT") === "spark_catalog")
assert(rowSet.getString("TABLE_SCHEM") === "global_temp")
assert(rowSet.getString("TABLE_NAME") === viewName)
assert(rowSet.getString("COLUMN_NAME") === "n")
@@ -719,4 +733,184 @@ class SparkMetadataOperationSuite extends
HiveThriftServer2TestBase {
}
}
}
+
+ test("SPARK-57518: getCatalogs with DSv2 catalog returns all loaded catalogs
sorted") {
+ withJdbcStatement() { statement =>
+ // Configure and load a DSv2 catalog
+ statement.execute(
+ "SET spark.sql.catalog.testcat=" +
+ "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog")
+ // Trigger catalog loading by accessing it
+ statement.execute("USE testcat")
+ // Switch back to spark_catalog for subsequent operations
+ statement.execute("USE spark_catalog")
+
+ val metaData = statement.getConnection.getMetaData
+ val rs = metaData.getCatalogs
+ // Should return both catalogs sorted alphabetically
+ assert(rs.next())
+ assert(rs.getString("TABLE_CAT") === "spark_catalog")
+ assert(rs.next())
+ assert(rs.getString("TABLE_CAT") === "testcat")
+ assert(!rs.next())
+ }
+ }
+
+ test("SPARK-57518: getSchemas with null catalog returns current catalog
schemas only") {
+ withJdbcStatement() { statement =>
+ // Configure and load a DSv2 catalog
+ statement.execute(
+ "SET spark.sql.catalog.testcat=" +
+ "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog")
+ statement.execute("USE testcat")
+ statement.execute("CREATE NAMESPACE testcat.testns")
+ // Switch to spark_catalog
+ statement.execute("USE spark_catalog")
+
+ val metaData = statement.getConnection.getMetaData
+ // null catalog -> current catalog only (spark_catalog)
+ val rs = metaData.getSchemas(null, "%")
+ val schemas = scala.collection.mutable.ArrayBuffer.empty[String]
+ while (rs.next()) {
+ schemas += rs.getString("TABLE_SCHEM")
+ assert(rs.getString("TABLE_CATALOG") === "spark_catalog")
+ }
+ // Should contain default and global_temp at minimum; should NOT contain
testns
+ assert(schemas.contains("default"))
+ assert(schemas.contains("global_temp"))
+ assert(!schemas.contains("testns"))
+ }
+ }
+
+ test("SPARK-57518: getSchemas returns schemas from current catalog when set
to DSv2 catalog") {
+ withJdbcStatement() { statement =>
+ statement.execute(
+ "SET spark.sql.catalog.testcat=" +
+ "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog")
+ statement.execute("USE testcat")
+ statement.execute("CREATE NAMESPACE testcat.ns1")
+ statement.execute("CREATE NAMESPACE testcat.ns2")
+
+ val metaData = statement.getConnection.getMetaData
+ // null catalog, current catalog is testcat
+ val rs = metaData.getSchemas(null, "%")
+ val schemas = scala.collection.mutable.ArrayBuffer.empty[String]
+ while (rs.next()) {
+ schemas += rs.getString("TABLE_SCHEM")
+ assert(rs.getString("TABLE_CATALOG") === "testcat")
+ }
+ assert(schemas.contains("ns1"))
+ assert(schemas.contains("ns2"))
+ }
+ }
+
+ test("SPARK-57518: getTables returns TABLE_CAT with current catalog name") {
+ withJdbcStatement("dsv2_table") { statement =>
+ statement.execute("CREATE TABLE dsv2_table(id INT, name STRING)")
+
+ val metaData = statement.getConnection.getMetaData
+ val rs = metaData.getTables(null, "default", "dsv2_table", null)
+ assert(rs.next())
+ assert(rs.getString("TABLE_CAT") === "spark_catalog")
+ assert(rs.getString("TABLE_SCHEM") === "default")
+ assert(rs.getString("TABLE_NAME") === "dsv2_table")
+ assert(!rs.next())
+ }
+ }
+
+ test("SPARK-57518: getColumns returns TABLE_CAT with current catalog name") {
+ withJdbcStatement("dsv2_col_table") { statement =>
+ statement.execute("CREATE TABLE dsv2_col_table(id INT, name STRING)")
+
+ val metaData = statement.getConnection.getMetaData
+ val rs = metaData.getColumns(null, "default", "dsv2_col_table", null)
+ assert(rs.next())
+ assert(rs.getString("TABLE_CAT") === "spark_catalog")
+ assert(rs.getString("TABLE_SCHEM") === "default")
+ assert(rs.getString("TABLE_NAME") === "dsv2_col_table")
+ assert(rs.getString("COLUMN_NAME") === "id")
+ assert(rs.next())
+ assert(rs.getString("TABLE_CAT") === "spark_catalog")
+ assert(rs.getString("COLUMN_NAME") === "name")
+ assert(!rs.next())
+ }
+ }
+
+ test("SPARK-57518: getSchemas with catalog metadata disabled returns empty
TABLE_CATALOG") {
+ withJdbcStatement() { statement =>
+ statement.execute("SET
spark.sql.thriftServer.catalogMetadata.enabled=false")
+ val metaData = statement.getConnection.getMetaData
+ val rs = metaData.getSchemas(null, "default")
+ assert(rs.next())
+ assert(rs.getString("TABLE_SCHEM") === "default")
+ assert(rs.getString("TABLE_CATALOG").isEmpty)
+ }
+ }
+
+ test("SPARK-57518: getTables/getColumns with catalog metadata disabled
return legacy " +
+ "empty/null TABLE_CAT") {
+ withJdbcStatement("legacy_t") { statement =>
+ statement.execute("SET
spark.sql.thriftServer.catalogMetadata.enabled=false")
+ statement.execute("CREATE TABLE legacy_t(id INT, name STRING)")
+
+ val metaData = statement.getConnection.getMetaData
+
+ // getTables should return empty string or null for TABLE_CAT
+ val tablesRs = metaData.getTables(null, "default", "legacy_t", null)
+ assert(tablesRs.next())
+ val tableCat = tablesRs.getString("TABLE_CAT")
+ assert(tableCat == null || tableCat.isEmpty,
+ s"Expected empty/null TABLE_CAT but got: $tableCat")
+ assert(tablesRs.getString("TABLE_NAME") === "legacy_t")
+ assert(!tablesRs.next())
+
+ // getColumns should return empty string or null for TABLE_CAT
+ val colsRs = metaData.getColumns(null, "default", "legacy_t", null)
+ assert(colsRs.next())
+ val colCat = colsRs.getString("TABLE_CAT")
+ assert(colCat == null || colCat.isEmpty,
+ s"Expected empty/null TABLE_CAT but got: $colCat")
+ assert(colsRs.getString("COLUMN_NAME") === "id")
+ assert(colsRs.next())
+ assert(colsRs.getString("COLUMN_NAME") === "name")
+ assert(!colsRs.next())
+ }
+ }
+
+ test("SPARK-57518: getTables/getColumns do not label V1 rows with a DSv2
current catalog") {
+ // getTables/getColumns list from the V1 SessionCatalog (spark_catalog).
When the current
+ // catalog is a DSv2 catalog, TABLE_CAT must NOT be stamped with that
catalog's name -- the
+ // listed rows belong to spark_catalog, not the DSv2 catalog. They stay
legacy empty/null
+ // here; DSv2 routing for getTables/getColumns is a follow-up.
+ withJdbcStatement("v1_table") { statement =>
+ statement.execute("CREATE TABLE v1_table(id INT, name STRING)")
+ statement.execute(
+ "SET spark.sql.catalog.testcat=" +
+ "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog")
+ // Make the DSv2 catalog the current catalog.
+ statement.execute("USE testcat")
+
+ val metaData = statement.getConnection.getMetaData
+
+ // getTables still lists the V1 spark_catalog table; TABLE_CAT must be
legacy empty/null,
+ // never "testcat".
+ val tablesRs = metaData.getTables(null, "default", "v1_table", null)
+ assert(tablesRs.next())
+ val tableCat = tablesRs.getString("TABLE_CAT")
+ assert(tableCat == null || tableCat.isEmpty,
+ s"V1-listed table must not be labeled with the DSv2 current catalog,
got: $tableCat")
+ assert(tablesRs.getString("TABLE_NAME") === "v1_table")
+
+ // Same for getColumns.
+ val colsRs = metaData.getColumns(null, "default", "v1_table", null)
+ assert(colsRs.next())
+ val colCat = colsRs.getString("TABLE_CAT")
+ assert(colCat == null || colCat.isEmpty,
+ s"V1-listed column must not be labeled with the DSv2 current catalog,
got: $colCat")
+ assert(colsRs.getString("COLUMN_NAME") === "id")
+
+ // Switch back so withJdbcStatement's DROP TABLE cleanup targets
spark_catalog.
+ statement.execute("USE spark_catalog")
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
