[SPARK-31457][SQL]spark jdbc read hive created the wrong PreparedStatementadd

sql/core/src/main/scala/org/apache/spark/sql/jdbc/HiveDialect.scala

@@ -0,0 +1,26 @@
+package org.apache.spark.sql.jdbc
+
+import java.util.Locale
+
+import org.apache.spark.sql.types.{DataType, MetadataBuilder}
+
+private case object HiveDialect extends JdbcDialect {
+
+  override def canHandle(url : String): Boolean =
+    url.toLowerCase(Locale.ROOT).startsWith("jdbc:hive")
+
+  override def getCatalystType(
+      sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = {
+    None
+  }
+
+  override def quoteIdentifier(colName: String): String = {
+    colName
+  }
+
+  override def getTableExistsQuery(table: String): String = {
+    s"SELECT 1 FROM $table LIMIT 1"
+  }
+
+  override def isCascadingTruncateTable(): Option[Boolean] = Some(false)
+}

sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala

@@ -224,6 +224,7 @@ object JdbcDialects {
   registerDialect(DerbyDialect)
   registerDialect(OracleDialect)
   registerDialect(TeradataDialect)
+  registerDialect(HiveDialect)
 
   /**
    * Fetch the JdbcDialect class corresponding to a given database url.

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -716,6 +716,7 @@ class JDBCSuite extends QueryTest
     assert(JdbcDialects.get("jdbc:db2://127.0.0.1/db") == DB2Dialect)
     assert(JdbcDialects.get("jdbc:sqlserver://127.0.0.1/db") == MsSqlServerDialect)
     assert(JdbcDialects.get("jdbc:derby:db") == DerbyDialect)
+    assert(JdbcDialects.get("jdbc:hive2://127.0.0.1/db") == HiveDialect)
     assert(JdbcDialects.get("test.invalid") == NoopDialect)
   }