[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 <yuwq1996@gmail.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: wecharyu

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."

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
+  }
 }

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]

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."))
+  }
 }