[spark] branch branch-2.4 updated: [SPARK-27873][SQL][BRANCH-2.4] columnNameOfCorruptRecord should not be checked with column names in CSV header when disabling enforceSchema

[gu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

gurwls223 pushed a commit to branch branch-2.4
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-2.4 by this push:
     new 880cb7b  [SPARK-27873][SQL][BRANCH-2.4] columnNameOfCorruptRecord should not be checked with column names in CSV header when disabling enforceSchema
880cb7b is described below

commit 880cb7bee08172e2b82ac6fe57e4b75f8f495c1b
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Tue Jun 4 16:27:44 2019 +0900

    [SPARK-27873][SQL][BRANCH-2.4] columnNameOfCorruptRecord should not be checked with column names in CSV header when disabling enforceSchema
    
    ## What changes were proposed in this pull request?
    
    If we want to keep corrupt record when reading CSV, we provide a new column into the schema, that is `columnNameOfCorruptRecord`. But this new column isn't actually a column in CSV header. So if `enforceSchema` is disabled, `CSVHeaderChecker` throws a exception complaining that number of column in CSV header isn't equal to that in the schema.
    
    This backports the fix into branch-2.4.
    
    ## How was this patch tested?
    
    Added test.
    
    Closes #24771 from viirya/SPARK-27873-2.4.
    
    Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
    Signed-off-by: HyukjinKwon <gu...@apache.org>
---
 .../execution/datasources/csv/CSVDataSource.scala  | 12 +++++++++--
 .../sql/execution/datasources/csv/CSVSuite.scala   | 25 ++++++++++++++++++++++
 2 files changed, 35 insertions(+), 2 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
index e840ff1..62c3c16 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
@@ -210,7 +210,11 @@ object TextInputCSVDataSource extends CSVDataSource {
       // Note: if there are only comments in the first block, the header would probably
       // be not extracted.
       CSVUtils.extractHeader(lines, parser.options).foreach { header =>
-        val schema = if (columnPruning) requiredSchema else dataSchema
+        val actualRequiredSchema =
+          StructType(requiredSchema.filterNot(_.name == parser.options.columnNameOfCorruptRecord))
+        val actualDataSchema =
+          StructType(dataSchema.filterNot(_.name == parser.options.columnNameOfCorruptRecord))
+        val schema = if (columnPruning) actualRequiredSchema else actualDataSchema
         val columnNames = parser.tokenizer.parseLine(header)
         CSVDataSource.checkHeaderColumnNames(
           schema,
@@ -297,7 +301,11 @@ object MultiLineCSVDataSource extends CSVDataSource {
       caseSensitive: Boolean,
       columnPruning: Boolean): Iterator[InternalRow] = {
     def checkHeader(header: Array[String]): Unit = {
-      val schema = if (columnPruning) requiredSchema else dataSchema
+      val actualRequiredSchema =
+        StructType(requiredSchema.filterNot(_.name == parser.options.columnNameOfCorruptRecord))
+      val actualDataSchema =
+        StructType(dataSchema.filterNot(_.name == parser.options.columnNameOfCorruptRecord))
+      val schema = if (columnPruning) actualRequiredSchema else actualDataSchema
       CSVDataSource.checkHeaderColumnNames(
         schema,
         header,
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
index d59035b..df9d154 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
@@ -1836,4 +1836,29 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te
     val schema = new StructType().add("a", StringType).add("b", IntegerType)
     checkAnswer(spark.read.schema(schema).option("delimiter", delimiter).csv(input), Row("abc", 1))
   }
+
+  test("SPARK-27873: disabling enforceSchema should not fail columnNameOfCorruptRecord") {
+    Seq(false, true).foreach { multiLine =>
+      withTempPath { path =>
+        val df = Seq(("0", "2013-abc-11")).toDF("a", "b")
+        df.write
+          .option("header", "true")
+          .csv(path.getAbsolutePath)
+
+        val schema = StructType.fromDDL("a int, b date")
+        val columnNameOfCorruptRecord = "_unparsed"
+        val schemaWithCorrField = schema.add(columnNameOfCorruptRecord, StringType)
+        val readDF = spark
+          .read
+          .option("mode", "Permissive")
+          .option("header", "true")
+          .option("enforceSchema", false)
+          .option("multiLine", multiLine)
+          .option("columnNameOfCorruptRecord", columnNameOfCorruptRecord)
+          .schema(schemaWithCorrField)
+          .csv(path.getAbsoluteFile.toString)
+        checkAnswer(readDF, Row(null, null, "0,2013-abc-11") :: Nil)
+      }
+    }
+  }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org