[SPARK-48385][SQL] Migrate the jdbc driver of mariadb from 2.x to 3.x

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MariaDBKrbIntegrationSuite.scala

@@ -25,9 +25,9 @@ import org.apache.spark.sql.execution.datasources.jdbc.connection.SecureConnecti
 import org.apache.spark.tags.DockerTest
 
 /**
- * To run this test suite for a specific version (e.g., mariadb:10.5.12):
+ * To run this test suite for a specific version (e.g., mariadb:10.5.24):
  * {{{
- *   ENABLE_DOCKER_INTEGRATION_TESTS=1 MARIADB_DOCKER_IMAGE_NAME=mariadb:10.5.12
+ *   ENABLE_DOCKER_INTEGRATION_TESTS=1 MARIADB_DOCKER_IMAGE_NAME=mariadb:10.5.24
  *     ./build/sbt -Pdocker-integration-tests
  *     "docker-integration-tests/testOnly org.apache.spark.sql.jdbc.MariaDBKrbIntegrationSuite"
  * }}}
@@ -38,15 +38,15 @@ class MariaDBKrbIntegrationSuite extends DockerKrbJDBCIntegrationSuite {
   override protected val keytabFileName = "mariadb.keytab"
 
   override val db = new DatabaseOnDocker {
-    override val imageName = sys.env.getOrElse("MARIADB_DOCKER_IMAGE_NAME", "mariadb:10.5.12")
+    override val imageName = sys.env.getOrElse("MARIADB_DOCKER_IMAGE_NAME", "mariadb:10.5.24")
     override val env = Map(
       "MYSQL_ROOT_PASSWORD" -> "rootpass"
     )
     override val usesIpc = false
     override val jdbcPort = 3306
 
     override def getJdbcUrl(ip: String, port: Int): String =
-      s"jdbc:mysql://$ip:$port/mysql?user=$principal"
+      s"jdbc:mysql://$ip:$port/mysql?permitMysqlScheme&user=$principal"
 
     override def getEntryPoint: Option[String] =
       Some("/docker-entrypoint/mariadb_docker_entrypoint.sh")

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala

@@ -154,30 +154,43 @@ class MySQLIntegrationSuite extends DockerJDBCIntegrationSuite {
   test("SPARK-47462: Unsigned numeric types") {
     val df = sqlContext.read.jdbc(jdbcUrl, "unsigned_numbers", new Properties)
     val rows = df.head()
-    assert(rows.get(0).isInstanceOf[Short])
-    assert(rows.get(1).isInstanceOf[Integer])
-    assert(rows.get(2).isInstanceOf[Integer])
-    assert(rows.get(3).isInstanceOf[Long])
-    assert(rows.get(4).isInstanceOf[BigDecimal])
-    assert(rows.get(5).isInstanceOf[BigDecimal])
-    assert(rows.get(6).isInstanceOf[Double])
-    // Unlike MySQL, MariaDB seems not to distinguish signed and unsigned tinyint(1).
     val isMaria = jdbcUrl.indexOf("disableMariaDbDriver") == -1
     if (isMaria) {
+      assert(rows.get(0).isInstanceOf[Integer])
+      assert(rows.get(1).isInstanceOf[Long])
+      assert(rows.get(2).isInstanceOf[Integer])
+      assert(rows.get(3).isInstanceOf[BigDecimal])
+      assert(rows.get(4).isInstanceOf[BigDecimal])
+      assert(rows.get(5).isInstanceOf[BigDecimal])
+      assert(rows.get(6).isInstanceOf[Double])
+      // Unlike MySQL, MariaDB seems not to distinguish signed and unsigned tinyint(1).
       assert(rows.get(7).isInstanceOf[Boolean])
-    } else {
-      assert(rows.get(7).isInstanceOf[Short])
-    }
-    assert(rows.getShort(0) === 255)
-    assert(rows.getInt(1) === 65535)
-    assert(rows.getInt(2) === 16777215)
-    assert(rows.getLong(3) === 4294967295L)
-    assert(rows.getAs[BigDecimal](4).equals(new BigDecimal("9223372036854775808")))
-    assert(rows.getAs[BigDecimal](5).equals(new BigDecimal("123456789012345.123456789012345000")))
-    assert(rows.getDouble(6) === 1.0000000000000002)
-    if (isMaria) {
+
+      assert(rows.getInt(0) === 255)
+      assert(rows.getLong(1) === 65535L)
+      assert(rows.getInt(2) === 16777215)
+      assert(rows.getAs[BigDecimal](3).equals(new BigDecimal("4294967295")))
+      assert(rows.getAs[BigDecimal](4).equals(new BigDecimal("9223372036854775808")))
+      assert(rows.getAs[BigDecimal](5).equals(new BigDecimal("123456789012345.123456789012345000")))
+      assert(rows.getDouble(6) === 1.0000000000000002)
       assert(rows.getBoolean(7) === false)
     } else {
+      assert(rows.get(0).isInstanceOf[Short])
+      assert(rows.get(1).isInstanceOf[Integer])
+      assert(rows.get(2).isInstanceOf[Integer])
+      assert(rows.get(3).isInstanceOf[Long])
+      assert(rows.get(4).isInstanceOf[BigDecimal])
+      assert(rows.get(5).isInstanceOf[BigDecimal])
+      assert(rows.get(6).isInstanceOf[Double])
+      assert(rows.get(7).isInstanceOf[Short])
+
+      assert(rows.getShort(0) === 255)
+      assert(rows.getInt(1) === 65535)
+      assert(rows.getInt(2) === 16777215)
+      assert(rows.getLong(3) === 4294967295L)
+      assert(rows.getAs[BigDecimal](4).equals(new BigDecimal("9223372036854775808")))
+      assert(rows.getAs[BigDecimal](5).equals(new BigDecimal("123456789012345.123456789012345000")))
+      assert(rows.getDouble(6) === 1.0000000000000002)
       assert(rows.getShort(7) === 0)
     }
   }
@@ -367,13 +380,13 @@ class MySQLOverMariaConnectorIntegrationSuite extends MySQLIntegrationSuite {
 
   override val db = new MySQLDatabaseOnDocker {
     override def getJdbcUrl(ip: String, port: Int): String =
-      s"jdbc:mysql://$ip:$port/mysql?user=root&password=rootpass&allowPublicKeyRetrieval=true" +
-        s"&useSSL=false"
+      s"jdbc:mysql://$ip:$port/mysql?permitMysqlScheme&user=root&password=rootpass" +
+        s"&allowPublicKeyRetrieval=true&useSSL=false"
   }
 
   override def testConnection(): Unit = {
     Using.resource(getConnection()) { conn =>
-      assert(conn.getClass.getName === "org.mariadb.jdbc.MariaDbConnection")
+      assert(conn.getClass.getName === "org.mariadb.jdbc.Connection")
     }
   }
 }

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala

@@ -177,7 +177,7 @@ class MySQLOverMariaConnectorIntegrationSuite extends MySQLIntegrationSuite {
 
   override val db = new MySQLDatabaseOnDocker {
     override def getJdbcUrl(ip: String, port: Int): String =
-      s"jdbc:mysql://$ip:$port/mysql?user=root&password=rootpass&allowPublicKeyRetrieval=true" +
-        s"&useSSL=false"
+      s"jdbc:mysql://$ip:$port/mysql?permitMysqlScheme&user=root&password=rootpass" +
+        s"&allowPublicKeyRetrieval=true&useSSL=false"
   }
 }

pom.xml

@@ -322,7 +322,7 @@
       -Djdk.reflect.useDirectMethodHandle=false
       -Dio.netty.tryReflectionSetAccessible=true
     </extraJavaTestArgs>
-    <mariadb.java.client.version>2.7.12</mariadb.java.client.version>
+    <mariadb.java.client.version>3.4.0</mariadb.java.client.version>
     <mysql.connector.version>8.4.0</mysql.connector.version>
     <postgresql.version>42.7.3</postgresql.version>
     <db2.jcc.version>11.5.9.0</db2.jcc.version>

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/MariaDBConnectionProviderSuite.scala

@@ -22,6 +22,7 @@ class MariaDBConnectionProviderSuite extends ConnectionProviderSuiteBase {
     val provider = new MariaDBConnectionProvider()
     val driver = registerDriver(provider.driverClass)
 
-    testSecureConnectionProvider(provider, driver, options("jdbc:mysql://localhost/mysql"))
+    testSecureConnectionProvider(provider, driver,
+      options("jdbc:mysql://localhost/mysql?permitMysqlScheme"))
   }
 }

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

@@ -1400,7 +1400,7 @@ class JDBCSuite extends QueryTest with SharedSparkSession {
 
   test("SPARK-18419: Fix `asConnectionProperties` to filter case-insensitively") {
     val parameters = Map(
-      "url" -> "jdbc:mysql://localhost:3306/temp",
+      "url" -> "jdbc:mysql://localhost:3306/temp?permitMysqlScheme",
       "dbtable" -> "t1",
       "numPartitions" -> "10")
     assert(new JDBCOptions(parameters).asConnectionProperties.isEmpty)