(spark) branch branch-4.0 updated: [SPARK-46934][SQL][FOLLOWUP] Handle Special Characters for HMS

[do...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dongjoon pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.0 by this push:
     new 603f53b27800 [SPARK-46934][SQL][FOLLOWUP] Handle Special Characters for HMS
603f53b27800 is described below

commit 603f53b278004c997eafac90948eb90f92d2c2de
Author: Kent Yao <ya...@apache.org>
AuthorDate: Thu Jan 16 07:36:57 2025 -0800

    [SPARK-46934][SQL][FOLLOWUP] Handle Special Characters for HMS
    
    ### What changes were proposed in this pull request?
    
    This PR explicitly marks struct type with special characters as a hive-incompatible type to `newSparkSQLSpecificMetastoreTable()` instead of depending on a regex and parse failure to fall back.
    
    ### Why are the changes needed?
    
    regex is flaky
    
    ### Does this PR introduce _any_ user-facing change?
    
    no
    
    ### How was this patch tested?
    
    We have tests in the original PR
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    no
    
    Closes #49506 from yaooqinn/SPARK-46934-FF.
    
    Authored-by: Kent Yao <ya...@apache.org>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
    (cherry picked from commit 523ff15b2891f8521aaa25c8683940a6f2f34c00)
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../spark/sql/catalyst/util/QuotingUtils.scala     | 10 +++----
 .../spark/sql/hive/HiveExternalCatalog.scala       | 12 ++++++--
 .../spark/sql/hive/client/HiveClientImpl.scala     |  7 -----
 .../spark/sql/hive/execution/HiveDDLSuite.scala    | 33 ++++------------------
 4 files changed, 21 insertions(+), 41 deletions(-)

diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/QuotingUtils.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/QuotingUtils.scala
index 198b4d84b86e..81ad0fd67a4d 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/QuotingUtils.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/QuotingUtils.scala
@@ -45,12 +45,12 @@ object QuotingUtils {
 
   private val validIdentPattern = Pattern.compile("^[a-zA-Z_][a-zA-Z0-9_]*")
 
+  def needQuote(part: String): Boolean = {
+    !validIdentPattern.matcher(part).matches()
+  }
+
   def quoteIfNeeded(part: String): String = {
-    if (validIdentPattern.matcher(part).matches()) {
-      part
-    } else {
-      quoteIdentifier(part)
-    }
+    if (needQuote(part)) quoteIdentifier(part) else part
   }
 
   def quoted(namespace: Array[String]): String = {
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 db2835263577..ca8b5369a9cb 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
@@ -42,7 +42,7 @@ import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.catalog.ExternalCatalogUtils._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
-import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils, QuotingUtils}
 import org.apache.spark.sql.execution.command.DDLUtils
 import org.apache.spark.sql.execution.datasources.{PartitioningUtils, SourceOptions}
 import org.apache.spark.sql.hive.client.HiveClient
@@ -1445,7 +1445,15 @@ object HiveExternalCatalog {
     case _: AnsiIntervalType => false
     case _: TimestampNTZType => false
     case _: VariantType => false
-    case s: StructType => s.forall(f => isHiveCompatibleDataType(f.dataType))
+    case s: StructType =>
+      s.forall { f =>
+        // Because HMS thrift API passes StructType's catalog string in a whole, we can not
+        // ensure the HMS server side can parse this string correctly if the names of the fields
+        // contain special characters.
+        // Also, the HMS thrift APIs are kind of internal to Hive, it's hard for us to ensure
+        // which special character is valid or not. So here we just allow regular characters.
+        !QuotingUtils.needQuote(f.name) && isHiveCompatibleDataType(f.dataType)
+      }
     case a: ArrayType => isHiveCompatibleDataType(a.elementType)
     case m: MapType =>
       isHiveCompatibleDataType(m.keyType) && isHiveCompatibleDataType(m.valueType)
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 00407f0ecc17..74f938e18179 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
@@ -584,7 +584,6 @@ private[hive] class HiveClientImpl(
   }
 
   override def createTable(table: CatalogTable, ignoreIfExists: Boolean): Unit = withHiveState {
-    verifyColumnDataType(table.dataSchema)
     shim.createTable(client, toHiveTable(table, Some(userName)), ignoreIfExists)
   }
 
@@ -604,7 +603,6 @@ private[hive] class HiveClientImpl(
     // these properties are still available to the others that share the same Hive metastore.
     // If users explicitly alter these Hive-specific properties through ALTER TABLE DDL, we respect
     // these user-specified values.
-    verifyColumnDataType(table.dataSchema)
     val hiveTable = toHiveTable(
       table.copy(properties = table.ignoredProperties ++ table.properties), Some(userName))
     // Do not use `table.qualifiedName` here because this may be a rename
@@ -628,7 +626,6 @@ private[hive] class HiveClientImpl(
       newDataSchema: StructType,
       schemaProps: Map[String, String]): Unit = withHiveState {
     val oldTable = shim.getTable(client, dbName, tableName)
-    verifyColumnDataType(newDataSchema)
     val hiveCols = newDataSchema.map(toHiveColumn)
     oldTable.setFields(hiveCols.asJava)
 
@@ -1134,10 +1131,6 @@ private[hive] object HiveClientImpl extends Logging {
     Option(hc.getComment).map(field.withComment).getOrElse(field)
   }
 
-  private def verifyColumnDataType(schema: StructType): Unit = {
-    schema.foreach(col => getSparkSQLDataType(toHiveColumn(col)))
-  }
-
   private def toInputFormat(name: String) =
     Utils.classForName[org.apache.hadoop.mapred.InputFormat[_, _]](name)
 
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index a58adbce7ec5..9d474379db64 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -34,7 +34,6 @@ import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
 import org.apache.spark.sql.connector.catalog.CatalogManager
 import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME
 import org.apache.spark.sql.connector.catalog.SupportsNamespaces.PROP_OWNER
-import org.apache.spark.sql.errors.DataTypeErrors.toSQLType
 import org.apache.spark.sql.execution.command.{DDLSuite, DDLUtils}
 import org.apache.spark.sql.execution.datasources.orc.OrcCompressionCodec
 import org.apache.spark.sql.execution.datasources.parquet.{ParquetCompressionCodec, ParquetFooterReader}
@@ -2849,38 +2848,18 @@ class HiveDDLSuite
   }
 
   test("SPARK-47101 checks if nested column names do not include invalid characters") {
-    // delimiter characters
-    Seq(",", ":").foreach { c =>
+    Seq(",", ":", ";", "^", "\\", "/", "%").foreach { c =>
       val typ = s"array<struct<`abc${c}xyz`:int>>"
-      // The regex is from HiveClientImpl.getSparkSQLDataType, please keep them in sync.
-      val replaced = typ.replaceAll("`", "").replaceAll("(?<=struct<|,)([^,<:]+)(?=:)", "`$1`")
-      withTable("t") {
-        checkError(
-          exception = intercept[SparkException] {
-            sql(s"CREATE TABLE t (a $typ) USING hive")
-          },
-          condition = "CANNOT_RECOGNIZE_HIVE_TYPE",
-          parameters = Map(
-            "fieldType" -> toSQLType(replaced),
-            "fieldName" -> "`a`")
-        )
+      val e = intercept[AnalysisException] {
+        sql(s"CREATE TABLE t (a $typ) USING hive")
       }
-    }
-    // other special characters
-    Seq(";", "^", "\\", "/", "%").foreach { c =>
-      val typ = s"array<struct<`abc${c}xyz`:int>>"
-      val replaced = typ.replaceAll("`", "")
-      val msg = s"java.lang.IllegalArgumentException: Error: : expected at the position " +
-        s"16 of '$replaced' but '$c' is found."
       withTable("t") {
         checkError(
-          exception = intercept[AnalysisException] {
-            sql(s"CREATE TABLE t (a $typ) USING hive")
-          },
+          exception = e,
           condition = "_LEGACY_ERROR_TEMP_3065",
           parameters = Map(
-            "clazz" -> "org.apache.hadoop.hive.ql.metadata.HiveException",
-            "msg" -> msg)
+            "clazz" -> e.getCause.getClass.getName,
+            "msg" -> e.getCause.getMessage)
         )
       }
     }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org