Tweaks data schema passed to FSBasedRelations

- Only data columns should be passed - Data schema should be converted to nullable version (see SPARK-5950 and PR apache#4826)
jeanlyn · May 12, 2015 · 1f9b1a5 · 1f9b1a5
1 parent 43ba50e
commit 1f9b1a5
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 13 deletions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/sources/commands.scala b/sql/core/src/main/scala/org/apache/spark/sql/sources/commands.scala
@@ -197,10 +197,11 @@ private[sql] case class InsertIntoFSBasedRelation(
       if (needsConversion) {
         val converter = CatalystTypeConverters.createToScalaConverter(dataSchema)
         while (iterator.hasNext) {
-          val row = converter(iterator.next()).asInstanceOf[Row]
+          val row = iterator.next()
           val partitionPart = partitionProj(row)
           val dataPart = dataProj(row)
-          writerContainer.outputWriterForRow(partitionPart).write(dataPart)
+          val convertedDataPart = converter(dataPart).asInstanceOf[Row]
+          writerContainer.outputWriterForRow(partitionPart).write(convertedDataPart)
         }
       } else {
         while (iterator.hasNext) {

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/sources/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/sources/ddl.scala
@@ -239,6 +239,8 @@ private[sql] object ResolvedDataSource {
             SparkHadoopUtil.get.globPath(patternPath).map(_.toString).toArray
           }
 
+          val dataSchema = StructType(schema.filterNot(f => partitionColumns.contains(f.name)))
+
           dataSource.createRelation(
             sqlContext,
             paths,
@@ -279,7 +281,7 @@ private[sql] object ResolvedDataSource {
       schema.find(_.name == col).getOrElse {
         throw new RuntimeException(s"Partition column $col not found in schema $schema")
       }
-    })
+    }).asNullable
   }
 
   /** Create a [[ResolvedDataSource]] for saving the content of the given [[DataFrame]]. */
@@ -305,10 +307,11 @@ private[sql] object ResolvedDataSource {
           val fs = path.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
           path.makeQualified(fs.getUri, fs.getWorkingDirectory)
         }
+        val dataSchema = StructType(data.schema.filterNot(f => partitionColumns.contains(f.name)))
         val r = dataSource.createRelation(
           sqlContext,
           Array(outputPath.toString),
-          Some(data.schema),
+          Some(dataSchema.asNullable),
           Some(partitionColumnsSchema(data.schema, partitionColumns)),
           caseInsensitiveOptions)
         sqlContext.executePlan(

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala
@@ -191,7 +191,7 @@ case class CreateMetastoreDataSourceAsSelect(
         case SaveMode.Append =>
           // Check if the specified data source match the data source of the existing table.
           val resolved = ResolvedDataSource(
-            sqlContext, Some(query.schema), partitionColumns, provider, optionsWithPath)
+            sqlContext, Some(query.schema.asNullable), partitionColumns, provider, optionsWithPath)
           val createdRelation = LogicalRelation(resolved.relation)
           EliminateSubQueries(sqlContext.table(tableName).logicalPlan) match {
             case l @ LogicalRelation(_: InsertableRelation | _: FSBasedRelation) =>
@@ -203,14 +203,13 @@ case class CreateMetastoreDataSourceAsSelect(
                   s"table $tableName and using its data source and options."
                 val errorMessage =
                   s"""
-                |$errorDescription
-                |== Relations ==
-                |${sideBySide(
-                s"== Expected Relation ==" ::
-                  l.toString :: Nil,
-                s"== Actual Relation ==" ::
-                  createdRelation.toString :: Nil).mkString("\n")}
-              """.stripMargin
+                     |$errorDescription
+                     |== Relations ==
+                     |${sideBySide(
+                        s"== Expected Relation ==" :: l.toString :: Nil,
+                        s"== Actual Relation ==" :: createdRelation.toString :: Nil
+                      ).mkString("\n")}
+                   """.stripMargin
                 throw new AnalysisException(errorMessage)
               }
               existingSchema = Some(l.schema)