address comments

Author: cloud-fan

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

@@ -19,6 +19,7 @@
 
 import java.util.Arrays;
 import java.util.Objects;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import com.google.common.base.Preconditions;
@@ -53,8 +54,9 @@ public String name() {
 
   @Override
   public String toString() {
-    return CatalogV2Implicits.quoteNameParts(Stream.concat(
-      Stream.of(namespace), Stream.of(name)).toArray(String[]::new));
+    return Stream.concat(Stream.of(namespace), Stream.of(name))
+      .map(CatalogV2Implicits::quote)
+      .collect(Collectors.joining("."));
   }
 
   @Override

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

@@ -299,9 +299,12 @@ public int hashCode() {
   }
 
   interface ColumnPosition {
-    First FIRST = new First();
 
-    static ColumnPosition createAfter(String column) {
+    static ColumnPosition first() {
+      return First.singleton;
+    }
+
+    static ColumnPosition after(String column) {
       return new After(column);
     }
   }
@@ -312,6 +315,8 @@ static ColumnPosition createAfter(String column) {
    * be the first one within the struct.
    */
   final class First implements ColumnPosition {
+    private static First singleton = new First();
+
     private First() {}
 
     @Override
@@ -333,7 +338,7 @@ private After(String column) {
       this.column = column;
     }
 
-    public String getColumn() {
+    public String column() {
       return column;
     }
 

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

@@ -2806,8 +2806,8 @@ 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.createAfter(ctx.afterCol.getText)
+      case SqlBaseParser.FIRST => ColumnPosition.first()
+      case SqlBaseParser.AFTER => ColumnPosition.after(ctx.afterCol.getText)
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Implicits.scala

@@ -89,7 +89,13 @@ private[sql] object CatalogV2Implicits {
   }
 
   implicit class IdentifierHelper(ident: Identifier) {
-    def quoted: String = ident.toString
+    def quoted: String = {
+      if (ident.namespace.nonEmpty) {
+        ident.namespace.map(quote).mkString(".") + "." + quote(ident.name)
+      } else {
+        quote(ident.name)
+      }
+    }
 
     def asMultipartIdentifier: Seq[String] = ident.namespace :+ ident.name
   }
@@ -109,14 +115,10 @@ private[sql] object CatalogV2Implicits {
           s"$quoted is not a valid TableIdentifier as it has more than 2 name parts.")
     }
 
-    def quoted: String = quoteNameParts(parts.toArray)
-  }
-
-  def quoteNameParts(nameParts: Array[String]): String = {
-    nameParts.map(quote).mkString(".")
+    def quoted: String = parts.map(quote).mkString(".")
   }
 
-  private def quote(part: String): String = {
+  def quote(part: String): String = {
     if (part.contains(".") || part.contains("`")) {
       s"`${part.replace("`", "``")}`"
     } else {

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

@@ -24,7 +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.{createAfter, FIRST}
+import org.apache.spark.sql.connector.catalog.TableChange.ColumnPosition.{after, 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}
@@ -542,13 +542,13 @@ class DDLParserSuite extends AnalysisTest {
     comparePlans(
       parsePlan("ALTER TABLE table_name ADD COLUMN x int FIRST"),
       AlterTableAddColumnsStatement(Seq("table_name"), Seq(
-        QualifiedColType(Seq("x"), IntegerType, None, Some(FIRST))
+        QualifiedColType(Seq("x"), IntegerType, None, Some(first()))
       )))
 
     comparePlans(
       parsePlan("ALTER TABLE table_name ADD COLUMN x int AFTER y"),
       AlterTableAddColumnsStatement(Seq("table_name"), Seq(
-        QualifiedColType(Seq("x"), IntegerType, None, Some(createAfter("y")))
+        QualifiedColType(Seq("x"), IntegerType, None, Some(after("y")))
       )))
   }
 
@@ -565,7 +565,7 @@ class DDLParserSuite extends AnalysisTest {
       parsePlan("ALTER TABLE table_name ADD COLUMN x.y.z int COMMENT 'doc', a.b string FIRST"),
       AlterTableAddColumnsStatement(Seq("table_name"), Seq(
         QualifiedColType(Seq("x", "y", "z"), IntegerType, Some("doc"), None),
-        QualifiedColType(Seq("a", "b"), StringType, None, Some(FIRST))
+        QualifiedColType(Seq("a", "b"), StringType, None, Some(first()))
       )))
   }
 
@@ -632,7 +632,7 @@ class DDLParserSuite extends AnalysisTest {
         Seq("a", "b", "c"),
         None,
         None,
-        Some(FIRST)))
+        Some(first())))
   }
 
   test("alter table: update column type, comment and position") {
@@ -644,7 +644,7 @@ class DDLParserSuite extends AnalysisTest {
         Seq("a", "b", "c"),
         Some(LongType),
         Some("new comment"),
-        Some(createAfter("d"))))
+        Some(after("d"))))
   }
 
   test("alter table: drop column") {