[SPARK-14460] [SQL] properly handling of column name contains space

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala

@@ -31,6 +31,7 @@ import org.apache.spark.sql.execution.datasources.{BucketSpec, CreateTableUsingA
 import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils
 import org.apache.spark.sql.execution.streaming.{MemoryPlan, MemorySink, StreamExecution}
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.jdbc.JdbcDialects
 import org.apache.spark.sql.sources.HadoopFsRelation
 import org.apache.spark.util.Utils
 
@@ -496,7 +497,7 @@ final class DataFrameWriter private[sql](df: DataFrame) {
     // connectionProperties should override settings in extraOptions
     props.putAll(connectionProperties)
     val conn = JdbcUtils.createConnectionFactory(url, props)()
-
+    val dialect = JdbcDialects.get(url)
     try {
       var tableExists = JdbcUtils.tableExists(conn, url, table)
 
@@ -515,7 +516,7 @@ final class DataFrameWriter private[sql](df: DataFrame) {
 
       // Create the table if the table didn't exist.
       if (!tableExists) {
-        val schema = JdbcUtils.schemaString(df, url)
+        val schema = JdbcUtils.schemaString(dialect, df, url)
         val sql = s"CREATE TABLE $table ($schema)"
         val statement = conn.createStatement
         try {

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

@@ -96,8 +96,9 @@ object JdbcUtils extends Logging {
   /**
    * Returns a PreparedStatement that inserts a row into table via conn.
    */
-  def insertStatement(conn: Connection, table: String, rddSchema: StructType): PreparedStatement = {
-    val columns = rddSchema.fields.map(_.name).mkString(",")
+  def insertStatement(dialect: JdbcDialect, conn: Connection, table: String, rddSchema: StructType)
+      : PreparedStatement = {
+    val columns = rddSchema.fields.map(f => quoteColumnName(dialect, f.name)).mkString(",")
     val placeholders = rddSchema.fields.map(_ => "?").mkString(",")
     val sql = s"INSERT INTO $table ($columns) VALUES ($placeholders)"
     conn.prepareStatement(sql)
@@ -169,7 +170,7 @@ object JdbcUtils extends Logging {
       if (supportsTransactions) {
         conn.setAutoCommit(false) // Everything in the same db transaction.
       }
-      val stmt = insertStatement(conn, table, rddSchema)
+      val stmt = insertStatement(dialect, conn, table, rddSchema)
       try {
         var rowCount = 0
         while (iterator.hasNext) {
@@ -245,14 +246,24 @@ object JdbcUtils extends Logging {
     Array[Byte]().iterator
   }
 
+  /**
+   * The utility to add quote to the column name based on its dialect
+   * @param dialect the JDBC dialect
+   * @param columnName the input column name
+   * @return the quoted column name
+   */
+  private def quoteColumnName(dialect: JdbcDialect, columnName: String): String = {
+    dialect.quoteIdentifier(columnName)
+  }
+
   /**
    * Compute the schema string for this RDD.
    */
-  def schemaString(df: DataFrame, url: String): String = {
+  def schemaString(dialect: JdbcDialect, df: DataFrame, url: String): String = {
     val sb = new StringBuilder()
     val dialect = JdbcDialects.get(url)
     df.schema.fields foreach { field =>
-      val name = field.name
+      val name = quoteColumnName(dialect, field.name)
       val typ: String = getJdbcType(field.dataType, dialect).databaseTypeDefinition
       val nullable = if (field.nullable) "" else "NOT NULL"
       sb.append(s", $name $typ $nullable")

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala

@@ -52,7 +52,8 @@ class JDBCWriteSuite extends SharedSQLContext with BeforeAndAfter {
     conn1.prepareStatement("insert into test.people values ('mary', 2)").executeUpdate()
     conn1.prepareStatement("drop table if exists test.people1").executeUpdate()
     conn1.prepareStatement(
-      "create table test.people1 (name TEXT(32) NOT NULL, theid INTEGER NOT NULL)").executeUpdate()
+      "create table test.people1 (name TEXT(32) NOT NULL, `the id` INTEGER NOT NULL)")
+      .executeUpdate()
     conn1.commit()
 
     sql(
@@ -151,4 +152,10 @@ class JDBCWriteSuite extends SharedSQLContext with BeforeAndAfter {
     assert(2 === sqlContext.read.jdbc(url1, "TEST.PEOPLE1", properties).count)
     assert(2 === sqlContext.read.jdbc(url1, "TEST.PEOPLE1", properties).collect()(0).length)
   }
+
+  test("SPARK-14460: Insert into table with column containing space") {
+    val df = sqlContext.createDataFrame(sparkContext.parallelize(arr2x2), schema2)
+    df.write.insertInto("PEOPLE1")
+    assert(2 === sqlContext.read.jdbc(url1, "TEST.PEOPLE1", properties).count)
+  }
 }