From d5460115670170454e5009e4ce53ffc33b442a4d Mon Sep 17 00:00:00 2001 From: sandeep katta Date: Tue, 17 Sep 2019 23:22:29 +0530 Subject: [PATCH 1/6] fix malformed record issue --- .../spark/sql/catalyst/csv/UnivocityParser.scala | 10 +--------- .../src/test/resources/test-data/malformedRow.csv | 5 +++++ .../sql/execution/datasources/csv/CSVSuite.scala | 12 ++++++++++++ 3 files changed, 18 insertions(+), 9 deletions(-) create mode 100644 sql/core/src/test/resources/test-data/malformedRow.csv diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala index 8456b7d218ead..2dfd9db3652f1 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala @@ -188,19 +188,11 @@ class UnivocityParser( } } - private val doParse = if (requiredSchema.nonEmpty) { - (input: String) => convert(tokenizer.parseLine(input)) - } else { - // If `columnPruning` enabled and partition attributes scanned only, - // `schema` gets empty. - (_: String) => InternalRow.empty - } - /** * Parses a single CSV string and turns it into either one resulting row or no row (if the * the record is malformed). */ - def parse(input: String): InternalRow = doParse(input) + def parse(input: String): InternalRow = convert(tokenizer.parseLine(input)) private val getToken = if (options.columnPruning) { (tokens: Array[String], index: Int) => tokens(index) diff --git a/sql/core/src/test/resources/test-data/malformedRow.csv b/sql/core/src/test/resources/test-data/malformedRow.csv new file mode 100644 index 0000000000000..8cfb3eefb982c --- /dev/null +++ b/sql/core/src/test/resources/test-data/malformedRow.csv @@ -0,0 +1,5 @@ +fruit,color,price,quantity +apple,red,1,3 +banana,yellow,2,4 +orange,orange,3,5 +malformedrow 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 5afd019c11a16..6059eba7413b5 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 @@ -66,6 +66,7 @@ class CSVSuite extends QueryTest with SharedSparkSession with TestCsvData { private val unescapedQuotesFile = "test-data/unescaped-quotes.csv" private val valueMalformedFile = "test-data/value-malformed.csv" private val badAfterGoodFile = "test-data/bad_after_good.csv" + private val malformedRowFile = "test-data/malformedRow.csv" /** Verifies data and schema. */ private def verifyCars( @@ -2109,4 +2110,15 @@ class CSVSuite extends QueryTest with SharedSparkSession with TestCsvData { "expect the TextParsingException truncate the error content to be 1000 length.") } } + + test("SPARK-29101 test count with DROPMALFORMED mode") { + withSQLConf(SQLConf.CSV_PARSER_COLUMN_PRUNING.key -> "false") { + val count = spark.read + .option("header", "true") + .option("mode", "DROPMALFORMED") + .csv(testFile(malformedRowFile)) + .count() + assert(3 == count) + } + } } From ab0c5a9b1292b9a49fa029cf6d2e84aedb3d171f Mon Sep 17 00:00:00 2001 From: sandeep katta Date: Wed, 18 Sep 2019 11:03:00 +0530 Subject: [PATCH 2/6] parse only if pruning is disabled or requiredSchema is nonEmpty --- .../spark/sql/catalyst/csv/UnivocityParser.scala | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala index 2dfd9db3652f1..e494a95d41c47 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala @@ -188,11 +188,20 @@ class UnivocityParser( } } + // parse if the columnPruning is disabled or requiredSchema is nonEmpty + private val doParse = if (!options.columnPruning || requiredSchema.nonEmpty) { + (input: String) => convert(tokenizer.parseLine(input)) + } else { + // If `columnPruning` enabled and partition attributes scanned only, + // `schema` gets empty./ + (_: String) => InternalRow.empty + } + /** * Parses a single CSV string and turns it into either one resulting row or no row (if the * the record is malformed). */ - def parse(input: String): InternalRow = convert(tokenizer.parseLine(input)) + def parse(input: String): InternalRow = doParse(input) private val getToken = if (options.columnPruning) { (tokens: Array[String], index: Int) => tokens(index) From 05786cb06ca350ce1842194e941542c4eca7d122 Mon Sep 17 00:00:00 2001 From: sandeep katta Date: Wed, 18 Sep 2019 11:37:45 +0530 Subject: [PATCH 3/6] Add testcase for column pruning enable case --- .../sql/execution/datasources/csv/CSVSuite.scala | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) 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 6059eba7413b5..fd41ff4f3cfb1 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 @@ -2112,13 +2112,15 @@ class CSVSuite extends QueryTest with SharedSparkSession with TestCsvData { } test("SPARK-29101 test count with DROPMALFORMED mode") { - withSQLConf(SQLConf.CSV_PARSER_COLUMN_PRUNING.key -> "false") { - val count = spark.read - .option("header", "true") - .option("mode", "DROPMALFORMED") - .csv(testFile(malformedRowFile)) - .count() - assert(3 == count) + Seq((true, 4), (false, 3)).foreach { record => + withSQLConf(SQLConf.CSV_PARSER_COLUMN_PRUNING.key -> record._1.toString) { + val count = spark.read + .option("header", "true") + .option("mode", "DROPMALFORMED") + .csv(testFile(malformedRowFile)) + .count() + assert(record._2 == count) + } } } } From 1078e8af01b2467196ca19eff1b67517b425bcc4 Mon Sep 17 00:00:00 2001 From: sandeep katta Date: Wed, 18 Sep 2019 11:46:44 +0530 Subject: [PATCH 4/6] Add testcase for column pruning enable case --- .../org/apache/spark/sql/catalyst/csv/UnivocityParser.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala index e494a95d41c47..067f7e35233c0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala @@ -193,7 +193,7 @@ class UnivocityParser( (input: String) => convert(tokenizer.parseLine(input)) } else { // If `columnPruning` enabled and partition attributes scanned only, - // `schema` gets empty./ + // `schema` gets empty. (_: String) => InternalRow.empty } From f6a01ffb6ae6dc4c99301214cfa697d19caf2559 Mon Sep 17 00:00:00 2001 From: sandeep katta Date: Wed, 18 Sep 2019 13:35:44 +0530 Subject: [PATCH 5/6] Add testcase for column pruning enable case --- .../apache/spark/sql/catalyst/csv/UnivocityParser.scala | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala index 067f7e35233c0..39a08ec06c6a0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala @@ -188,13 +188,13 @@ class UnivocityParser( } } - // parse if the columnPruning is disabled or requiredSchema is nonEmpty - private val doParse = if (!options.columnPruning || requiredSchema.nonEmpty) { - (input: String) => convert(tokenizer.parseLine(input)) - } else { + private val doParse = if (options.columnPruning && requiredSchema.isEmpty) { // If `columnPruning` enabled and partition attributes scanned only, // `schema` gets empty. (_: String) => InternalRow.empty + } else { + // parse if the columnPruning is disabled or requiredSchema is nonEmpty + (input: String) => convert(tokenizer.parseLine(input)) } /** From f2c25f068b29e2a71a7e4eacaa075e67001a2652 Mon Sep 17 00:00:00 2001 From: sandeep katta Date: Wed, 18 Sep 2019 14:19:01 +0530 Subject: [PATCH 6/6] fixed review comment --- .../spark/sql/execution/datasources/csv/CSVSuite.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 fd41ff4f3cfb1..835fa4ef166b2 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 @@ -2112,14 +2112,14 @@ class CSVSuite extends QueryTest with SharedSparkSession with TestCsvData { } test("SPARK-29101 test count with DROPMALFORMED mode") { - Seq((true, 4), (false, 3)).foreach { record => - withSQLConf(SQLConf.CSV_PARSER_COLUMN_PRUNING.key -> record._1.toString) { + Seq((true, 4), (false, 3)).foreach { case (csvColumnPruning, expectedCount) => + withSQLConf(SQLConf.CSV_PARSER_COLUMN_PRUNING.key -> csvColumnPruning.toString) { val count = spark.read .option("header", "true") .option("mode", "DROPMALFORMED") .csv(testFile(malformedRowFile)) .count() - assert(record._2 == count) + assert(expectedCount == count) } } }