[SPARK-28552][SQL] Case-insensitive database URLs in JdbcDialect

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

@@ -18,12 +18,14 @@
 package org.apache.spark.sql.jdbc
 
 import java.sql.Types
+import java.util.Locale
 
 import org.apache.spark.sql.types._
 
 private object DB2Dialect extends JdbcDialect {
 
-  override def canHandle(url: String): Boolean = url.startsWith("jdbc:db2")
+  override def canHandle(url: String): Boolean =
+    url.toLowerCase(Locale.ROOT).startsWith("jdbc:db2")
 
   override def getCatalystType(
       sqlType: Int,

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

@@ -18,13 +18,15 @@
 package org.apache.spark.sql.jdbc
 
 import java.sql.Types
+import java.util.Locale
 
 import org.apache.spark.sql.types._
 
 
 private object DerbyDialect extends JdbcDialect {
 
-  override def canHandle(url: String): Boolean = url.startsWith("jdbc:derby")
+  override def canHandle(url: String): Boolean =
+    url.toLowerCase(Locale.ROOT).startsWith("jdbc:derby")
 
   override def getCatalystType(
       sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = {

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

@@ -17,12 +17,15 @@
 
 package org.apache.spark.sql.jdbc
 
+import java.util.Locale
+
 import org.apache.spark.sql.types._
 
 
 private object MsSqlServerDialect extends JdbcDialect {
 
-  override def canHandle(url: String): Boolean = url.startsWith("jdbc:sqlserver")
+  override def canHandle(url: String): Boolean =
+    url.toLowerCase(Locale.ROOT).startsWith("jdbc:sqlserver")
 
   override def getCatalystType(
       sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = {

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

@@ -18,12 +18,14 @@
 package org.apache.spark.sql.jdbc
 
 import java.sql.Types
+import java.util.Locale
 
 import org.apache.spark.sql.types.{BooleanType, DataType, LongType, MetadataBuilder}
 
 private case object MySQLDialect extends JdbcDialect {
 
-  override def canHandle(url : String): Boolean = url.startsWith("jdbc:mysql")
+  override def canHandle(url : String): Boolean =
+    url.toLowerCase(Locale.ROOT).startsWith("jdbc:mysql")
 
   override def getCatalystType(
       sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = {

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

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.jdbc
 
 import java.sql.{Date, Timestamp, Types}
-import java.util.TimeZone
+import java.util.{Locale, TimeZone}
 
 import org.apache.spark.sql.catalyst.util.DateTimeUtils
 import org.apache.spark.sql.internal.SQLConf
@@ -30,7 +30,8 @@ private case object OracleDialect extends JdbcDialect {
   private[jdbc] val BINARY_DOUBLE = 101
   private[jdbc] val TIMESTAMPTZ = -101
 
-  override def canHandle(url: String): Boolean = url.startsWith("jdbc:oracle")
+  override def canHandle(url: String): Boolean =
+    url.toLowerCase(Locale.ROOT).startsWith("jdbc:oracle")
 
   private def supportTimeZoneTypes: Boolean = {
     val timeZone = DateTimeUtils.getTimeZone(SQLConf.get.sessionLocalTimeZone)

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

@@ -18,14 +18,16 @@
 package org.apache.spark.sql.jdbc
 
 import java.sql.{Connection, Types}
+import java.util.Locale
 
 import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils}
 import org.apache.spark.sql.types._
 
 
 private object PostgresDialect extends JdbcDialect {
 
-  override def canHandle(url: String): Boolean = url.startsWith("jdbc:postgresql")
+  override def canHandle(url: String): Boolean =
+    url.toLowerCase(Locale.ROOT).startsWith("jdbc:postgresql")
 
   override def getCatalystType(
       sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = {

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

@@ -17,14 +17,15 @@
 
 package org.apache.spark.sql.jdbc
 
-import java.sql.Types
+import java.util.Locale
 
 import org.apache.spark.sql.types._
 
 
 private case object TeradataDialect extends JdbcDialect {
 
-  override def canHandle(url: String): Boolean = { url.startsWith("jdbc:teradata") }
+  override def canHandle(url: String): Boolean =
+    url.toLowerCase(Locale.ROOT).startsWith("jdbc:teradata")
 
   override def getJDBCType(dt: DataType): Option[JdbcType] = dt match {
     case StringType => Some(JdbcType("VARCHAR(255)", java.sql.Types.VARCHAR))

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

@@ -58,7 +58,7 @@ class JDBCSuite extends QueryTest
   }
 
   val testH2DialectTinyInt = new JdbcDialect {
-    override def canHandle(url: String): Boolean = url.startsWith("jdbc:h2")
+    override def canHandle(url: String) : Boolean = url.startsWith("jdbc:h2")
     override def getCatalystType(
         sqlType: Int,
         typeName: String,
@@ -1662,4 +1662,50 @@ class JDBCSuite extends QueryTest
       "Invalid value `test` for parameter `isolationLevel`. This can be " +
       "`NONE`, `READ_UNCOMMITTED`, `READ_COMMITTED`, `REPEATABLE_READ` or `SERIALIZABLE`."))
   }
+
+  test("SPARK-28552: Check whether a dialect instance can be applied on the given jdbc url") {
+    var dialects = List[JdbcDialect]()
+
+    def registerDialect(dialect: JdbcDialect) : Unit = {
+      dialects = dialect :: dialects.filterNot(_ == dialect)
+    }
+
+    registerDialect(MySQLDialect)
+    registerDialect(PostgresDialect)
+    registerDialect(DB2Dialect)
+    registerDialect(MsSqlServerDialect)
+    registerDialect(DerbyDialect)
+    registerDialect(OracleDialect)
+    registerDialect(TeradataDialect)
+
+    def get(url: String): JdbcDialect = {
+      val matchingDialects = dialects.filter(_.canHandle(url))
+      matchingDialects.length match {
+        case 0 => NoopDialect
+        case 1 => matchingDialects.head
+        case _ => new AggregatedDialect(matchingDialects)
+      }
+    }
+
+    assert(get("jdbc:mysql://localhost/db") == MySQLDialect)
+    assert(get("jdbc:MySQL://localhost/db") == MySQLDialect)
+
+    assert(get("jdbc:postgresql://localhost/db") == PostgresDialect)
+    assert(get("jdbc:postGresql://localhost/db") == PostgresDialect)
+
+    assert(get("jdbc:db2://localhost/db") == DB2Dialect)
+    assert(get("jdbc:DB2://localhost/db") == DB2Dialect)
+
+    assert(get("jdbc:sqlserver://localhost/db") == MsSqlServerDialect)
+    assert(get("jdbc:sqlServer://localhost/db") == MsSqlServerDialect)
+
+    assert(get("jdbc:derby://localhost/db") == DerbyDialect)
+    assert(get("jdbc:derBy://localhost/db") == DerbyDialect)
+
+    assert(get("jdbc:oracle://localhost/db") == OracleDialect)
+    assert(get("jdbc:Oracle://localhost/db") == OracleDialect)
+
+    assert(get("jdbc:teradata://localhost/db") == TeradataDialect)
+    assert(get("jdbc:Teradata://localhost/db") == TeradataDialect)
+  }
 }