[SPARK-36895][SQL][FOLLOWUP] Use property to specify index type

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

@@ -127,9 +127,10 @@ class MySQLIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTest {
     val properties = new util.HashMap[String, String]();
     properties.put("KEY_BLOCK_SIZE", "10")
     properties.put("COMMENT", "'this is a comment'")
+    properties.put("indexType", "BTREE")
     // MySQL doesn't allow property set on individual column, so use empty Array for
     // column properties
-    jdbcTable.createIndex("i1", "BTREE", Array(FieldReference("col1")),
+    jdbcTable.createIndex("i1", Array(FieldReference("col1")),
       new util.HashMap[NamedReference, util.Map[String, String]](), properties)
 
     var index = jdbcTable.listIndexes()

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

@@ -208,27 +208,30 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
         assert(jdbcTable.indexExists("i1") == false)
         assert(jdbcTable.indexExists("i2") == false)
 
-        val properties = new util.HashMap[String, String]();
+        var properties = new util.HashMap[String, String]();
         val indexType = "DUMMY"
+        properties.put("indexType", indexType)
         var m = intercept[UnsupportedOperationException] {
-          jdbcTable.createIndex("i1", indexType, Array(FieldReference("col1")),
+          jdbcTable.createIndex("i1", Array(FieldReference("col1")),
             new util.HashMap[NamedReference, util.Map[String, String]](), properties)
         }.getMessage
         assert(m.contains(s"Index Type $indexType is not supported." +
           s" The supported Index Types are: BTREE and HASH"))
 
-        jdbcTable.createIndex("i1", "BTREE", Array(FieldReference("col1")),
+        properties.put("indexType", "BTREE")
+        jdbcTable.createIndex("i1", Array(FieldReference("col1")),
           new util.HashMap[NamedReference, util.Map[String, String]](), properties)
 
-        jdbcTable.createIndex("i2", "",
-          Array(FieldReference("col2"), FieldReference("col3"), FieldReference("col5")),
+        properties = new util.HashMap[String, String]();
+        jdbcTable.createIndex(
+          "i2", Array(FieldReference("col2"), FieldReference("col3"), FieldReference("col5")),
           new util.HashMap[NamedReference, util.Map[String, String]](), properties)
 
         assert(jdbcTable.indexExists("i1") == true)
         assert(jdbcTable.indexExists("i2") == true)
 
         m = intercept[IndexAlreadyExistsException] {
-          jdbcTable.createIndex("i1", "", Array(FieldReference("col1")),
+          jdbcTable.createIndex("i1", Array(FieldReference("col1")),
             new util.HashMap[NamedReference, util.Map[String, String]](), properties)
         }.getMessage
         assert(m.contains("Failed to create index i1 in new_table"))

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/index/SupportsIndex.java

@@ -37,15 +37,14 @@ public interface SupportsIndex extends Table {
    * Creates an index.
    *
    * @param indexName the name of the index to be created
-   * @param indexType the type of the index to be created. If this is not specified, Spark
-   *                  will use empty String.
    * @param columns the columns on which index to be created
    * @param columnsProperties the properties of the columns on which index to be created
-   * @param properties the properties of the index to be created
+   * @param properties the properties of the index to be created. indexType is a
+   *                   special property that is used to specify the type of the index to be
+   *                   created.
    * @throws IndexAlreadyExistsException If the index already exists.
    */
   void createIndex(String indexName,
-      String indexType,
       NamedReference[] columns,
       Map<NamedReference, Map<String, String>> columnsProperties,
       Map<String, String> properties)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -4432,12 +4432,14 @@ class AstBuilder extends SqlBaseBaseVisitor[AnyRef] with SQLConfHelper with Logg
       .map(_.multipartIdentifier).map(typedVisit[Seq[String]]).toSeq
     val columnsProperties = ctx.columns.multipartIdentifierProperty.asScala
       .map(x => (Option(x.options).map(visitPropertyKeyValues).getOrElse(Map.empty))).toSeq
-    val options = Option(ctx.options).map(visitPropertyKeyValues).getOrElse(Map.empty)
+    var options = Option(ctx.options).map(visitPropertyKeyValues).getOrElse(Map.empty)
+    if (indexType.nonEmpty) {
+      options = options + ("indexType" -> indexType)
+    }
 
     CreateIndex(
       createUnresolvedTable(ctx.multipartIdentifier(), "CREATE INDEX"),
       indexName,
-      indexType,
       ctx.EXISTS != null,
       columns.map(UnresolvedFieldName(_)).zip(columnsProperties),
       options)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala

@@ -1063,7 +1063,6 @@ case class UncacheTable(
 case class CreateIndex(
     table: LogicalPlan,
     indexName: String,
-    indexType: String,
     ignoreIfExists: Boolean,
     columns: Seq[(FieldName, Map[String, String])],
     properties: Map[String, String]) extends UnaryCommand {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala

@@ -2275,18 +2275,19 @@ class DDLParserSuite extends AnalysisTest {
 
   test("CREATE INDEX") {
     parseCompare("CREATE index i1 ON a.b.c USING BTREE (col1)",
-      CreateIndex(UnresolvedTable(Seq("a", "b", "c"), "CREATE INDEX", None), "i1", "BTREE", false,
-        Seq(UnresolvedFieldName(Seq("col1"))).zip(Seq(Map.empty[String, String])), Map.empty))
+      CreateIndex(UnresolvedTable(Seq("a", "b", "c"), "CREATE INDEX", None), "i1", false,
+        Seq(UnresolvedFieldName(Seq("col1"))).zip(Seq(Map.empty[String, String])),
+        Map("indexType" -> "BTREE")))
 
     parseCompare("CREATE index IF NOT EXISTS i1 ON TABLE a.b.c USING BTREE" +
       " (col1 OPTIONS ('k1'='v1'), col2 OPTIONS ('k2'='v2')) ",
-      CreateIndex(UnresolvedTable(Seq("a", "b", "c"), "CREATE INDEX", None), "i1", "BTREE", true,
+      CreateIndex(UnresolvedTable(Seq("a", "b", "c"), "CREATE INDEX", None), "i1", true,
         Seq(UnresolvedFieldName(Seq("col1")), UnresolvedFieldName(Seq("col2")))
-          .zip(Seq(Map("k1" -> "v1"), Map("k2" -> "v2"))), Map.empty))
+          .zip(Seq(Map("k1" -> "v1"), Map("k2" -> "v2"))), Map("indexType" -> "BTREE")))
 
     parseCompare("CREATE index i1 ON a.b.c" +
       " (col1 OPTIONS ('k1'='v1'), col2 OPTIONS ('k2'='v2')) OPTIONS ('k3'='v3', 'k4'='v4')",
-      CreateIndex(UnresolvedTable(Seq("a", "b", "c"), "CREATE INDEX", None), "i1", "", false,
+      CreateIndex(UnresolvedTable(Seq("a", "b", "c"), "CREATE INDEX", None), "i1", false,
         Seq(UnresolvedFieldName(Seq("col1")), UnresolvedFieldName(Seq("col2")))
           .zip(Seq(Map("k1" -> "v1"), Map("k2" -> "v2"))), Map("k3" -> "v3", "k4" -> "v4")))
   }

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

@@ -1018,15 +1018,14 @@ object JdbcUtils extends Logging with SQLConfHelper {
   def createIndex(
       conn: Connection,
       indexName: String,
-      indexType: String,
       tableName: String,
       columns: Array[NamedReference],
       columnsProperties: util.Map[NamedReference, util.Map[String, String]],
       properties: util.Map[String, String],
       options: JDBCOptions): Unit = {
     val dialect = JdbcDialects.get(options.url)
     executeStatement(conn, options,
-      dialect.createIndex(indexName, indexType, tableName, columns, columnsProperties, properties))
+      dialect.createIndex(indexName, tableName, columns, columnsProperties, properties))
   }
 
   /**

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

@@ -33,7 +33,6 @@ import org.apache.spark.sql.connector.expressions.NamedReference
 case class CreateIndexExec(
     table: SupportsIndex,
     indexName: String,
-    indexType: String,
     ignoreIfExists: Boolean,
     columns: Seq[(NamedReference, Map[String, String])],
     properties: Map[String, String])
@@ -45,7 +44,7 @@ case class CreateIndexExec(
     }
     try {
       table.createIndex(
-        indexName, indexType, columns.unzip._1.toArray, colProperties, properties.asJava)
+        indexName, columns.unzip._1.toArray, colProperties, properties.asJava)
     } catch {
       case _: IndexAlreadyExistsException if ignoreIfExists =>
         logWarning(s"Index $indexName already exists in table ${table.name}. Ignoring.")

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

@@ -431,14 +431,13 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
       val table = a.table.asInstanceOf[ResolvedTable]
       AlterTableExec(table.catalog, table.identifier, a.changes) :: Nil
 
-    case CreateIndex(ResolvedTable(_, _, table, _),
-        indexName, indexType, ifNotExists, columns, properties) =>
+    case CreateIndex(ResolvedTable(_, _, table, _), indexName, ifNotExists, columns, properties) =>
       table match {
         case s: SupportsIndex =>
           val namedRefs = columns.map { case (field, prop) =>
             FieldReference(field.name) -> prop
           }
-          CreateIndexExec(s, indexName, indexType, ifNotExists, namedRefs, properties) :: Nil
+          CreateIndexExec(s, indexName, ifNotExists, namedRefs, properties) :: Nil
         case _ => throw QueryCompilationErrors.tableIndexNotSupportedError(
           s"CreateIndex is not supported in this table ${table.name}.")
       }

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

@@ -54,15 +54,14 @@ case class JDBCTable(ident: Identifier, schema: StructType, jdbcOptions: JDBCOpt
 
   override def createIndex(
       indexName: String,
-      indexType: String,
       columns: Array[NamedReference],
       columnsProperties: util.Map[NamedReference, util.Map[String, String]],
       properties: util.Map[String, String]): Unit = {
     JdbcUtils.withConnection(jdbcOptions) { conn =>
       JdbcUtils.classifyException(s"Failed to create index $indexName in $name",
         JdbcDialects.get(jdbcOptions.url)) {
         JdbcUtils.createIndex(
-          conn, indexName, indexType, name, columns, columnsProperties, properties, jdbcOptions)
+          conn, indexName, name, columns, columnsProperties, properties, jdbcOptions)
       }
     }
   }

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

@@ -304,7 +304,6 @@ abstract class JdbcDialect extends Serializable with Logging{
    */
   def createIndex(
       indexName: String,
-      indexType: String,
       tableName: String,
       columns: Array[NamedReference],
       columnsProperties: util.Map[NamedReference, util.Map[String, String]],

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

@@ -115,30 +115,29 @@ private case object MySQLDialect extends JdbcDialect with SQLConfHelper {
   // https://dev.mysql.com/doc/refman/8.0/en/create-index.html
   override def createIndex(
       indexName: String,
-      indexType: String,
       tableName: String,
       columns: Array[NamedReference],
       columnsProperties: util.Map[NamedReference, util.Map[String, String]],
       properties: util.Map[String, String]): String = {
     val columnList = columns.map(col => quoteIdentifier(col.fieldNames.head))
     var indexProperties: String = ""
+    var indexType = ""
     if (!properties.isEmpty) {
       properties.asScala.foreach { case (k, v) =>
-        indexProperties = indexProperties + " " + s"$k $v"
-      }
-    }
-    val iType = if (indexType.isEmpty) {
-      ""
-    } else {
-      if (indexType.length > 1 && !indexType.equalsIgnoreCase("BTREE") &&
-        !indexType.equalsIgnoreCase("HASH")) {
-        throw new UnsupportedOperationException(s"Index Type $indexType is not supported." +
-          " The supported Index Types are: BTREE and HASH")
+        if (k.equalsIgnoreCase("indexType")) {
+          if (v.equalsIgnoreCase("BTREE") || v.equalsIgnoreCase("HASH")) {
+            indexType = s"USING $v"
+          } else {
+            throw new UnsupportedOperationException(s"Index Type $v is not supported." +
+              " The supported Index Types are: BTREE and HASH")
+          }
+        } else {
+          indexProperties = indexProperties + " " + s"$k $v"
+        }
       }
-      s"USING $indexType"
     }
     // columnsProperties doesn't apply to MySQL so it is ignored
-    s"CREATE INDEX ${quoteIdentifier(indexName)} $iType ON" +
+    s"CREATE INDEX ${quoteIdentifier(indexName)} $indexType ON" +
       s" ${quoteIdentifier(tableName)} (${columnList.mkString(", ")}) $indexProperties"
   }