Add ALTER TABLE and address comments.

Author: dongjoon-hyun

sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala

@@ -853,22 +853,17 @@ object DDLUtils {
   }
 
   private[sql] def checkFieldNames(table: CatalogTable): Unit = {
-    table.provider.get.toLowerCase(Locale.ROOT) match {
-      case "hive" =>
-        val serde = table.storage.serde
-        if (serde == HiveSerDe.sourceToSerDe("orc").get.serde) {
-          OrcFileFormat.checkFieldNames(table.dataSchema)
-        } else if (serde == HiveSerDe.sourceToSerDe("parquet").get.serde) {
-          ParquetSchemaConverter.checkFieldNames(table.dataSchema)
-        }
-
-      case "parquet" =>
-        ParquetSchemaConverter.checkFieldNames(table.dataSchema)
-
-      case "orc" =>
-        OrcFileFormat.checkFieldNames(table.dataSchema)
-
-      case _ =>
+    val serde = table.storage.serde
+    if (serde == HiveSerDe.sourceToSerDe("orc").get.serde) {
+      OrcFileFormat.checkFieldNames(table.dataSchema)
+    } else if (serde == HiveSerDe.sourceToSerDe("parquet").get.serde) {
+      ParquetSchemaConverter.checkFieldNames(table.dataSchema)
+    } else {
+      table.provider.get.toLowerCase(Locale.ROOT) match {
+        case "parquet" => ParquetSchemaConverter.checkFieldNames(table.dataSchema)
+        case "orc" => OrcFileFormat.checkFieldNames(table.dataSchema)
+        case _ =>
+      }
     }
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

@@ -206,6 +206,9 @@ case class AlterTableAddColumnsCommand(
       reorderedSchema.map(_.name), "in the table definition of " + table.identifier,
       conf.caseSensitiveAnalysis)
 
+    val newDataSchema = StructType(catalogTable.dataSchema ++ columns)
+    DDLUtils.checkFieldNames(catalogTable.copy(schema = newDataSchema))
+
     catalog.alterTableSchema(
       table, catalogTable.schema.copy(fields = reorderedSchema.toArray))
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala

@@ -29,7 +29,7 @@ private[sql] object OrcFileFormat {
     } catch {
       case _: IllegalArgumentException =>
         throw new AnalysisException(
-          s"""Attribute name "$name" contains invalid character(s).
+          s"""Column name "$name" contains invalid character(s).
              |Please use alias to rename it.
            """.stripMargin.split("\n").mkString(" ").trim)
     }

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala

@@ -2001,9 +2001,9 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
     }
   }
 
-  test("SPARK-21912 Creating ORC/Parquet datasource table should check invalid column names") {
-    withTable("t21912") {
-      Seq(" ", ",", ";", "{", "}", "(", ")", "\n", "\t", "=").foreach { name =>
+  test("SPARK-21912 ORC/Parquet table should not create invalid column names") {
+    Seq(" ", ",", ";", "{", "}", "(", ")", "\n", "\t", "=").foreach { name =>
+      withTable("t21912") {
         Seq("ORC", "PARQUET").foreach { source =>
           val m = intercept[AnalysisException] {
             sql(s"CREATE TABLE t21912(`col$name` INT) USING $source")
@@ -2022,6 +2022,15 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
             assert(m3.contains(s"contains invalid character(s)"))
           }
         }
+
+        // TODO: After SPARK-21929, we need to check ORC, too.
+        Seq("PARQUET").foreach { source =>
+          sql(s"CREATE TABLE t21912(`col` INT) USING $source")
+          val m = intercept[AnalysisException] {
+            sql(s"ALTER TABLE t21912 ADD COLUMNS(`col$name` INT)")
+          }.getMessage
+          assert(m.contains(s"contains invalid character(s)"))
+        }
       }
     }
   }