Address comments.

Author: dongjoon-hyun

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

@@ -23,6 +23,8 @@ import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.execution.datasources.orc.OrcFileFormat
+import org.apache.spark.sql.execution.datasources.parquet.ParquetSchemaConverter
 import org.apache.spark.sql.sources.BaseRelation
 
 /**
@@ -85,6 +87,13 @@ case class CreateDataSourceTableCommand(table: CatalogTable, ignoreIfExists: Boo
       }
     }
 
+    table.provider.get.toLowerCase match {
+      case "parquet" =>
+        dataSource.schema.map(_.name).foreach(ParquetSchemaConverter.checkFieldName)
+      case "orc" =>
+        dataSource.schema.map(_.name).foreach(OrcFileFormat.checkFieldName)
+    }
+
     val newTable = table.copy(
       schema = dataSource.schema,
       partitionColumnNames = partitionColumnNames,

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

@@ -0,0 +1,36 @@
+/*
+ * 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.execution.datasources.orc
+
+import org.apache.orc.TypeDescription
+
+import org.apache.spark.sql.AnalysisException
+
+private[sql] object OrcFileFormat {
+  def checkFieldName(name: String): Unit = {
+    try {
+      TypeDescription.fromString(s"struct<$name:int>")
+    } catch {
+      case _: IllegalArgumentException =>
+        throw new AnalysisException(
+          s"""Attribute name "$name" contains invalid character(s).
+             |Please use alias to rename it.
+           """.stripMargin.split("\n").mkString(" ").trim)
+    }
+  }
+}

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala

@@ -556,7 +556,7 @@ private[parquet] class ParquetSchemaConverter(
   }
 }
 
-private[parquet] object ParquetSchemaConverter {
+private[sql] object ParquetSchemaConverter {
   val SPARK_PARQUET_SCHEMA_NAME = "spark_schema"
 
   val EMPTY_MESSAGE: MessageType =

sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala

@@ -32,13 +32,13 @@ import org.apache.hadoop.io.{NullWritable, Writable}
 import org.apache.hadoop.mapred.{JobConf, OutputFormat => MapRedOutputFormat, RecordWriter, Reporter}
 import org.apache.hadoop.mapreduce._
 import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}
-import org.apache.orc.TypeDescription
 
 import org.apache.spark.TaskContext
-import org.apache.spark.sql._
+import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.execution.datasources.orc.OrcFileFormat
 import org.apache.spark.sql.hive.{HiveInspectors, HiveShim}
 import org.apache.spark.sql.sources._
 import org.apache.spark.sql.types.StructType
@@ -84,7 +84,7 @@ class OrcFileFormat extends FileFormat with DataSourceRegister with Serializable
           classOf[MapRedOutputFormat[_, _]])
     }
 
-    dataSchema.map(_.name).foreach(checkFieldName)
+    dataSchema.map(_.name).foreach(OrcFileFormat.checkFieldName)
 
     new OutputWriterFactory {
       override def newInstance(
@@ -172,18 +172,6 @@ class OrcFileFormat extends FileFormat with DataSourceRegister with Serializable
       }
     }
   }
-
-  private def checkFieldName(name: String): Unit = {
-    try {
-      TypeDescription.fromString(s"struct<$name:int>")
-    } catch {
-      case _: IllegalArgumentException =>
-        throw new AnalysisException(
-          s"""Attribute name "$name" contains invalid character(s).
-             |Please use alias to rename it.
-           """.stripMargin.split("\n").mkString(" ").trim)
-    }
-  }
 }
 
 private[orc] class OrcSerializer(dataSchema: StructType, conf: Configuration)

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

@@ -2001,13 +2001,20 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
     }
   }
 
-  test("SPARK-21912 Creating ORC datasource table should check invalid column names") {
+  test("SPARK-21912 Creating ORC/Parquet datasource table should check invalid column names") {
     withTable("orc1") {
-      Seq(" ", "?", ",", ";", "{", "}", "(", ")", "\n", "\t", "=").foreach { name =>
-        val m = intercept[AnalysisException] {
-          sql(s"CREATE TABLE orc1 USING ORC AS SELECT 1 `column$name`")
-        }.getMessage
-        assert(m.contains(s"contains invalid character(s)"))
+      Seq(" ", ",", ";", "{", "}", "(", ")", "\n", "\t", "=").foreach { name =>
+        Seq("ORC", "PARQUET").foreach { dataSource =>
+          val m = intercept[AnalysisException] {
+            sql(s"CREATE TABLE orc1(`column$name` INT) USING $dataSource")
+          }.getMessage
+          assert(m.contains(s"contains invalid character(s)"))
+
+          val m2 = intercept[AnalysisException] {
+            sql(s"CREATE TABLE orc1 USING $dataSource AS SELECT 1 `column$name`")
+          }.getMessage
+          assert(m2.contains(s"contains invalid character(s)"))
+        }
       }
     }
   }