[SPARK-33095][SQL] Support ALTER TABLE in JDBC v2 Table Catalog: add, update type and nullability of columns (MySQL dialect)

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

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.jdbc.v2
+
+import java.sql.Connection
+
+import org.scalatest.time.SpanSugar._
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog
+import org.apache.spark.sql.jdbc.{DatabaseOnDocker, DockerJDBCIntegrationSuite}
+import org.apache.spark.sql.types._
+import org.apache.spark.tags.DockerTest
+
+/**
+ *
+ * To run this test suite for a specific version (e.g., mysql:5.7.31):
+ * {{{
+ * MYSQL_DOCKER_IMAGE_NAME=mysql:5.7.31
+ *         ./build/sbt -Pdocker-integration-tests "testOnly *v2*MySQLIntegrationSuite"
+ *
+ * }}}
+ *
+ */
+@DockerTest
+class MySQLIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTest {
+  override val catalogName: String = "mysql"
+  override val db = new DatabaseOnDocker {
+    override val imageName = sys.env.getOrElse("MYSQL_DOCKER_IMAGE_NAME", "mysql:5.7.31")
+    override val env = Map(
+      "MYSQL_ROOT_PASSWORD" -> "rootpass"
+    )
+    override val usesIpc = false
+    override val jdbcPort: Int = 3306
+
+    override def getJdbcUrl(ip: String, port: Int): String =
+      s"jdbc:mysql://$ip:$port/mysql?user=root&password=rootpass"
+  }
+
+  override def sparkConf: SparkConf = super.sparkConf
+    .set("spark.sql.catalog.mysql", classOf[JDBCTableCatalog].getName)
+    .set("spark.sql.catalog.mysql.url", db.getJdbcUrl(dockerIp, externalPort))
+
+  override val connectionTimeout = timeout(7.minutes)
+
+  override def dataPreparation(conn: Connection): Unit = {}
+
+  override def testUpdateColumnType(tbl: String): Unit = {
+    sql(s"CREATE TABLE $tbl (ID INTEGER) USING _")
+    var t = spark.table(tbl)
+    var expectedSchema = new StructType().add("ID", IntegerType)
+    assert(t.schema === expectedSchema)
+    sql(s"ALTER TABLE $tbl ALTER COLUMN id TYPE STRING")
+    t = spark.table(tbl)
+    expectedSchema = new StructType().add("ID", StringType)
+    assert(t.schema === expectedSchema)
+    // Update column type from STRING to INTEGER
+    val msg1 = intercept[AnalysisException] {
+      sql(s"ALTER TABLE $tbl ALTER COLUMN id TYPE INTEGER")
+    }.getMessage
+    assert(msg1.contains("Cannot update alt_table field ID: string cannot be cast to int"))
+  }
+
+}

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala

@@ -893,10 +893,11 @@ object JdbcUtils extends Logging {
       conn: Connection,
       tableName: String,
       changes: Seq[TableChange],
-      options: JDBCOptions): Unit = {
+      options: JDBCOptions,
+      tableSchema: StructType): Unit = {
     val dialect = JdbcDialects.get(options.url)
     if (changes.length == 1) {
-      executeStatement(conn, options, dialect.alterTable(tableName, changes)(0))
+      executeStatement(conn, options, dialect.alterTable(tableName, changes, tableSchema)(0))
     } else {
       val metadata = conn.getMetaData
       if (!metadata.supportsTransactions) {
@@ -907,7 +908,7 @@ object JdbcUtils extends Logging {
         val statement = conn.createStatement
         try {
           statement.setQueryTimeout(options.queryTimeout)
-          for (sql <- dialect.alterTable(tableName, changes)) {
+          for (sql <- dialect.alterTable(tableName, changes, tableSchema)) {
             statement.executeUpdate(sql)
           }
           conn.commit()

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala

@@ -139,7 +139,10 @@ class JDBCTableCatalog extends TableCatalog with Logging {
     checkNamespace(ident.namespace())
     withConnection { conn =>
       classifyException(s"Failed table altering: $ident") {
-        JdbcUtils.alterTable(conn, getTableName(ident), changes, options)
+        val optionsWithTableName = new JDBCOptions(
+          options.parameters + (JDBCOptions.JDBC_TABLE_NAME -> getTableName(ident)))
+        val tableSchema: StructType = JdbcUtils.getSchemaOption(conn, optionsWithTableName).get
+        JdbcUtils.alterTable(conn, getTableName(ident), changes, options, tableSchema)
       }
       loadTable(ident)
     }

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

@@ -75,7 +75,8 @@ private object DB2Dialect extends JdbcDialect {
   override def getUpdateColumnNullabilityQuery(
       tableName: String,
       columnName: String,
-      isNullable: Boolean): String = {
+      isNullable: Boolean,
+      dataType: String): String = {
     val nullable = if (isNullable) "DROP NOT NULL" else "SET NOT NULL"
     s"ALTER TABLE $tableName ALTER COLUMN ${quoteIdentifier(columnName)} $nullable"
   }

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

@@ -25,9 +25,11 @@ import org.apache.commons.lang3.StringUtils
 
 import org.apache.spark.annotation.{DeveloperApi, Since}
 import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.analysis.{caseInsensitiveResolution, caseSensitiveResolution}
 import org.apache.spark.sql.connector.catalog.TableChange
 import org.apache.spark.sql.connector.catalog.TableChange._
 import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
 /**
@@ -200,12 +202,17 @@ abstract class JdbcDialect extends Serializable {
 
   /**
    * Alter an existing table.
+   * TODO (SPARK-32523): Override this method in the dialects that have different syntax.
    *
    * @param tableName The name of the table to be altered.
    * @param changes Changes to apply to the table.
+   * @param tableSchema Schema of the table to be altered.
    * @return The SQL statements to use for altering the table.
    */
-  def alterTable(tableName: String, changes: Seq[TableChange]): Array[String] = {
+  def alterTable(
+      tableName: String,
+      changes: Seq[TableChange],
+      tableSchema: StructType): Array[String] = {
     val updateClause = ArrayBuilder.make[String]
     for (change <- changes) {
       change match {
@@ -226,7 +233,16 @@ abstract class JdbcDialect extends Serializable {
           updateClause += getUpdateColumnTypeQuery(tableName, name(0), dataType)
         case updateNull: UpdateColumnNullability if updateNull.fieldNames.length == 1 =>
           val name = updateNull.fieldNames
-          updateClause += getUpdateColumnNullabilityQuery(tableName, name(0), updateNull.nullable())
+          val columnNameEquality = if (SQLConf.get.caseSensitiveAnalysis) {
+            caseSensitiveResolution
+          } else {
+            caseInsensitiveResolution
+          }
+          val columnDataType =
+            tableSchema.filter(x => columnNameEquality(x.name, name(0))).head.dataType
+          val jdbcDataType = JdbcUtils.getJdbcType(columnDataType, this).databaseTypeDefinition
+          updateClause +=
+            getUpdateColumnNullabilityQuery(tableName, name(0), updateNull.nullable(), jdbcDataType)
         case _ =>
           throw new SQLFeatureNotSupportedException(s"Unsupported TableChange $change")
       }
@@ -253,7 +269,8 @@ abstract class JdbcDialect extends Serializable {
   def getUpdateColumnNullabilityQuery(
       tableName: String,
       columnName: String,
-      isNullable: Boolean): String = {
+      isNullable: Boolean,
+      dataType: String): String = {
     val nullable = if (isNullable) "NULL" else "NOT NULL"
     s"ALTER TABLE $tableName ALTER COLUMN ${quoteIdentifier(columnName)} SET $nullable"
   }

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

@@ -48,4 +48,21 @@ private case object MySQLDialect extends JdbcDialect {
   }
 
   override def isCascadingTruncateTable(): Option[Boolean] = Some(false)
+
+  // See https://dev.mysql.com/doc/refman/8.0/en/alter-table.html
+  override def getUpdateColumnTypeQuery(
+      tableName: String,
+      columnName: String,
+      newDataType: String): String = {
+    s"ALTER TABLE $tableName MODIFY COLUMN ${quoteIdentifier(columnName)} $newDataType"
+  }
+
+  override def getUpdateColumnNullabilityQuery(
+      tableName: String,
+      columnName: String,
+      isNullable: Boolean,
+      dataType: String): String = {
+    val nullable = if (isNullable) "NULL" else "NOT NULL"
+    s"ALTER TABLE $tableName MODIFY COLUMN ${quoteIdentifier(columnName)} $dataType $nullable"
+  }
 }

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

@@ -136,7 +136,8 @@ private case object OracleDialect extends JdbcDialect {
   override def getUpdateColumnNullabilityQuery(
     tableName: String,
     columnName: String,
-    isNullable: Boolean): String = {
+    isNullable: Boolean,
+    dataType: String): String = {
     val nullable = if (isNullable) "NULL" else "NOT NULL"
     s"ALTER TABLE $tableName MODIFY ${quoteIdentifier(columnName)} $nullable"
   }