You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/05/22 14:22:02 UTC
(spark) branch master updated: [SPARK-50137][SQL] Avoid fallback to Hive-incompatible ways when table creation fails by thrift exception
This is an automated email from the ASF dual-hosted git repository.
dongjoon 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 de3d44d46fdc [SPARK-50137][SQL] Avoid fallback to Hive-incompatible ways when table creation fails by thrift exception
de3d44d46fdc is described below
commit de3d44d46fdc08f879922cce4b9c02cbc8eab030
Author: wecharyu <yu...@gmail.com>
AuthorDate: Thu May 22 07:21:48 2025 -0700
[SPARK-50137][SQL] Avoid fallback to Hive-incompatible ways when table creation fails by thrift exception
### What changes were proposed in this pull request?
Enhance the datasource table creation, do not fallback to hive incompatible way if failure was caused by thrift exception.
### Why are the changes needed?
Thrift exception is unrelated to hive compatibility of datasource table, so fallback is unnecessary in this case.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Add a unit test:
```bash
mvn test -Dtest=none -Dsuites="org.apache.spark.sql.hive.HiveExternalCatalogSuite SPARK-50137: Avoid fallback to Hive-incompatible ways on thrift exception" -pl :spark-hive_2.13
```
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #50985 from wecharyu/SPARK-50137_2.
Authored-by: wecharyu <yu...@gmail.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../spark/sql/hive/HiveExternalCatalog.scala | 2 +-
.../org/apache/spark/sql/hive/HiveUtils.scala | 15 +++++++++++
.../spark/sql/hive/client/HiveClientImpl.scala | 14 +---------
.../spark/sql/hive/HiveExternalCatalogSuite.scala | 30 ++++++++++++++++++++++
4 files changed, 47 insertions(+), 14 deletions(-)
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
index 97c93e725074..5c7a60151c49 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
@@ -417,7 +417,7 @@ private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configurat
logInfo(message)
saveTableIntoHive(table, ignoreIfExists)
} catch {
- case NonFatal(e) =>
+ case NonFatal(e) if !HiveUtils.causedByThrift(e) =>
val warningMessage =
log"Could not persist ${MDC(TABLE_NAME, table.identifier.quotedString)} in a Hive " +
log"compatible way. Persisting it into Hive metastore in Spark SQL specific format."
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
index c5d370f02e8f..11a1f1166e5e 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
@@ -521,4 +521,19 @@ private[spark] object HiveUtils extends Logging {
case PATTERN_FOR_KEY_EQ_VAL(_, v) => FileUtils.unescapePathName(v)
}
}
+
+ /**
+ * Determine if a Hive call exception is caused by thrift error.
+ */
+ def causedByThrift(e: Throwable): Boolean = {
+ var target = e
+ while (target != null) {
+ val msg = target.getMessage()
+ if (msg != null && msg.matches("(?s).*(TApplication|TProtocol|TTransport)Exception.*")) {
+ return true
+ }
+ target = target.getCause()
+ }
+ false
+ }
}
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index 57f6f999b6ad..8f7b892cf83b 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -236,7 +236,7 @@ private[hive] class HiveClientImpl(
try {
return f
} catch {
- case e: Exception if causedByThrift(e) =>
+ case e: Exception if HiveUtils.causedByThrift(e) =>
caughtException = e
logWarning(
log"HiveClient got thrift exception, destroying client and retrying " +
@@ -251,18 +251,6 @@ private[hive] class HiveClientImpl(
throw caughtException
}
- private def causedByThrift(e: Throwable): Boolean = {
- var target = e
- while (target != null) {
- val msg = target.getMessage()
- if (msg != null && msg.matches("(?s).*(TApplication|TProtocol|TTransport)Exception.*")) {
- return true
- }
- target = target.getCause()
- }
- false
- }
-
private def client: Hive = {
if (clientLoader.cachedHive != null) {
clientLoader.cachedHive.asInstanceOf[Hive]
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveExternalCatalogSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveExternalCatalogSuite.scala
index 8bb33e3383be..655491d24535 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveExternalCatalogSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveExternalCatalogSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.hive
import org.apache.hadoop.conf.Configuration
+import org.apache.logging.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.sql.catalyst.TableIdentifier
@@ -241,4 +242,33 @@ class HiveExternalCatalogSuite extends ExternalCatalogSuite {
val alteredTable = externalCatalog.getTable("db1", tableName)
assert(DataTypeUtils.sameType(alteredTable.schema, newSchema))
}
+
+ test("SPARK-50137: Avoid fallback to Hive-incompatible ways on thrift exception") {
+ val hadoopConf = new Configuration()
+ // Use an unavailable uri to mock client connection timeout.
+ hadoopConf.set("hive.metastore.uris", "thrift://192.0.2.1:9083")
+ hadoopConf.set("hive.metastore.client.connection.timeout", "1s")
+ // Dummy HiveExternalCatalog to mock that the hive client is still available
+ // when checking database and table.
+ val catalog = new HiveExternalCatalog(new SparkConf, hadoopConf) {
+ override def requireDbExists(db: String): Unit = {}
+ override def tableExists(db: String, table: String): Boolean = false
+ }
+ val logAppender = new LogAppender()
+ withLogAppender(logAppender, level = Some(Level.WARN)) {
+ val table = CatalogTable(
+ identifier = TableIdentifier("tbl", Some("default")),
+ tableType = CatalogTableType.EXTERNAL,
+ storage = storageFormat.copy(locationUri = Some(newUriForDatabase())),
+ schema = new StructType()
+ .add("col1", "string"),
+ provider = Some("parquet"))
+ intercept[Throwable] {
+ catalog.createTable(table, ignoreIfExists = false)
+ }
+ }
+ assert(!logAppender.loggingEvents.map(_.getMessage.getFormattedMessage).contains(
+ "Could not persist `default`.`tbl` in a Hive compatible way. " +
+ "Persisting it into Hive metastore in Spark SQL specific format."))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org