address comments

Author: cloud-fan

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -855,7 +855,7 @@ intervalUnit
     ;
 
 colPosition
-    : position=FIRST | position=AFTER multipartIdentifier
+    : position=FIRST | position=AFTER afterCol=errorCapturingIdentifier
     ;
 
 dataType

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

@@ -301,13 +301,15 @@ public int hashCode() {
   interface ColumnPosition {
     First FIRST = new First();
 
-    static ColumnPosition After(String[] column) {
+    static ColumnPosition createAfter(String column) {
       return new After(column);
     }
   }
 
   /**
    * Column position FIRST means the specified column should be the first column.
+   * Note that, the specified column may be a nested field, and then FIRST means this field should
+   * be the first one within the struct.
    */
   final class First implements ColumnPosition {
     private First() {}
@@ -320,35 +322,37 @@ public String toString() {
 
   /**
    * Column position AFTER means the specified column should be put after the given `column`.
+   * Note that, the specified column may be a nested field, and then the given `column` refers to
+   * a nested field in the same struct.
    */
   final class After implements ColumnPosition {
-    private final String[] column;
+    private final String column;
 
-    private After(String[] column) {
+    private After(String column) {
       assert column != null;
       this.column = column;
     }
 
-    public String[] getColumn() {
+    public String getColumn() {
       return column;
     }
 
     @Override
     public String toString() {
-      return "AFTER " + CatalogV2Implicits.quoteNameParts(column);
+      return "AFTER " + column;
     }
 
     @Override
     public boolean equals(Object o) {
       if (this == o) return true;
       if (o == null || getClass() != o.getClass()) return false;
       After after = (After) o;
-      return Arrays.equals(column, after.column);
+      return column.equals(after.column);
     }
 
     @Override
     public int hashCode() {
-      return Arrays.hashCode(column);
+      return Objects.hash(column);
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala

@@ -45,7 +45,7 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
       createAlterTable(nameParts, catalog, tbl, changes)
 
     case AlterTableAlterColumnStatement(
-        nameParts @ NonSessionCatalogAndTable(catalog, tbl), colName, dataType, comment, pos) =>
+         nameParts @ NonSessionCatalogAndTable(catalog, tbl), colName, dataType, comment, pos) =>
       val colNameArray = colName.toArray
       val typeChange = dataType.map { newDataType =>
         TableChange.updateColumnType(colNameArray, newDataType, true)

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

@@ -2807,8 +2807,7 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
   override def visitColPosition(ctx: ColPositionContext): ColumnPosition = {
     ctx.position.getType match {
       case SqlBaseParser.FIRST => ColumnPosition.FIRST
-      case SqlBaseParser.AFTER =>
-        ColumnPosition.After(typedVisit[Seq[String]](ctx.multipartIdentifier).toArray)
+      case SqlBaseParser.AFTER => ColumnPosition.createAfter(ctx.afterCol.getText)
     }
   }
 

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

@@ -24,8 +24,7 @@ import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, GlobalTempView, Loc
 import org.apache.spark.sql.catalyst.catalog.BucketSpec
 import org.apache.spark.sql.catalyst.expressions.{EqualTo, Literal}
 import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.connector.catalog.TableChange.ColumnPosition.After
-import org.apache.spark.sql.connector.catalog.TableChange.ColumnPosition.FIRST
+import org.apache.spark.sql.connector.catalog.TableChange.ColumnPosition.{createAfter, FIRST}
 import org.apache.spark.sql.connector.expressions.{ApplyTransform, BucketTransform, DaysTransform, FieldReference, HoursTransform, IdentityTransform, LiteralValue, MonthsTransform, Transform, YearsTransform}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType, TimestampType}
@@ -549,7 +548,7 @@ class DDLParserSuite extends AnalysisTest {
     comparePlans(
       parsePlan("ALTER TABLE table_name ADD COLUMN x int AFTER y"),
       AlterTableAddColumnsStatement(Seq("table_name"), Seq(
-        QualifiedColType(Seq("x"), IntegerType, None, Some(After(Array("y"))))
+        QualifiedColType(Seq("x"), IntegerType, None, Some(createAfter("y")))
       )))
   }
 
@@ -639,13 +638,13 @@ class DDLParserSuite extends AnalysisTest {
   test("alter table: update column type, comment and position") {
     comparePlans(
       parsePlan("ALTER TABLE table_name CHANGE COLUMN a.b.c " +
-        "TYPE bigint COMMENT 'new comment' AFTER x.y"),
+        "TYPE bigint COMMENT 'new comment' AFTER d"),
       AlterTableAlterColumnStatement(
         Seq("table_name"),
         Seq("a", "b", "c"),
         Some(LongType),
         Some("new comment"),
-        Some(After(Array("x", "y")))))
+        Some(createAfter("d"))))
   }
 
   test("alter table: drop column") {