diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/HiveDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/HiveDialect.scala
new file mode 100644
index 000000000000..ee098297e0a5
--- /dev/null
+++ b/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)
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
index a0c6d20f3645..5c6581efdcb9 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
+++ b/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.
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
index fd691f547eb0..2292b881d833 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
+++ b/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)
   }