From c8f1a5283669dc9b019c3e5c77b2ed5d1fb14959 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Fri, 19 Jul 2019 09:57:04 +0800 Subject: [PATCH 1/8] init pr --- .../org/apache/spark/sql/catalyst/csv/CSVOptions.scala | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala index 1268fcffcfcd0..293951b2d2dde 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala @@ -189,6 +189,11 @@ class CSVOptions( */ val emptyValueInWrite = emptyValue.getOrElse("\"\"") + /** + * The max error content in CSV parser/writer exception message. + */ + val maxErrorContentLength = 1024 + /** * A string between two consecutive JSON records. */ @@ -220,6 +225,7 @@ class CSVOptions( writerSettings.setSkipEmptyLines(true) writerSettings.setQuoteAllFields(quoteAll) writerSettings.setQuoteEscapingEnabled(escapeQuotes) + writerSettings.setErrorContentLength(maxErrorContentLength) writerSettings } @@ -246,6 +252,7 @@ class CSVOptions( lineSeparatorInRead.foreach { _ => settings.setNormalizeLineEndingsWithinQuotes(!multiLine) } + settings.setErrorContentLength(maxErrorContentLength) settings } From eb027acf6d6ac59aea0a94b8527032381d4f232a Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Fri, 19 Jul 2019 10:59:14 +0800 Subject: [PATCH 2/8] update --- .../scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala index 293951b2d2dde..f5d7caa811096 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala @@ -190,7 +190,7 @@ class CSVOptions( val emptyValueInWrite = emptyValue.getOrElse("\"\"") /** - * The max error content in CSV parser/writer exception message. + * The max error content length in CSV parser/writer exception message. */ val maxErrorContentLength = 1024 From c40c3d280389a13668c43b55fea7bbb625ae61af Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Fri, 19 Jul 2019 21:23:01 +0800 Subject: [PATCH 3/8] add test --- .../spark/sql/catalyst/csv/CSVOptions.scala | 17 ++++++----- .../execution/datasources/csv/CSVSuite.scala | 29 ++++++++++++++++++- 2 files changed, 38 insertions(+), 8 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala index f5d7caa811096..0fd624f7b9364 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala @@ -189,11 +189,6 @@ class CSVOptions( */ val emptyValueInWrite = emptyValue.getOrElse("\"\"") - /** - * The max error content length in CSV parser/writer exception message. - */ - val maxErrorContentLength = 1024 - /** * A string between two consecutive JSON records. */ @@ -225,7 +220,7 @@ class CSVOptions( writerSettings.setSkipEmptyLines(true) writerSettings.setQuoteAllFields(quoteAll) writerSettings.setQuoteEscapingEnabled(escapeQuotes) - writerSettings.setErrorContentLength(maxErrorContentLength) + writerSettings.setErrorContentLength(CSVOptions.maxErrorContentLength) writerSettings } @@ -252,8 +247,16 @@ class CSVOptions( lineSeparatorInRead.foreach { _ => settings.setNormalizeLineEndingsWithinQuotes(!multiLine) } - settings.setErrorContentLength(maxErrorContentLength) + settings.setErrorContentLength(CSVOptions.maxErrorContentLength) settings } } + +object CSVOptions { + /** + * The max error content length in CSV parser/writer exception message. + */ + val maxErrorContentLength = 1024 + +} 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 2e7d682a3bbca..ee6cf6e87bb03 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 @@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.datasources.csv import java.io.{ByteArrayOutputStream, EOFException, File, FileOutputStream} import java.nio.charset.{Charset, StandardCharsets, UnsupportedCharsetException} -import java.nio.file.Files +import java.nio.file.{Files, StandardOpenOption} import java.sql.{Date, Timestamp} import java.text.SimpleDateFormat import java.util.Locale @@ -28,6 +28,7 @@ import java.util.zip.GZIPOutputStream import scala.collection.JavaConverters._ import scala.util.Properties +import com.univocity.parsers.common.TextParsingException import org.apache.commons.lang3.time.FastDateFormat import org.apache.hadoop.io.SequenceFile.CompressionType import org.apache.hadoop.io.compress.GzipCodec @@ -36,6 +37,7 @@ import org.apache.log4j.spi.LoggingEvent import org.apache.spark.{SparkException, TestUtils} import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row} +import org.apache.spark.sql.catalyst.csv.CSVOptions import org.apache.spark.sql.catalyst.util.DateTimeUtils import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.{SharedSQLContext, SQLTestUtils} @@ -2085,4 +2087,29 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te } } } + + test("SPARK-28431: prevent CSV datasource throw TextParsingException with large size message") { + withTempDir { dir => + val maxCharsPerCol = 10000 + val str = "a" * (maxCharsPerCol + 1) + + val csvFile = new File(dir, "data.csv") + Files.write( + csvFile.toPath, + Seq(str).asJava, + StandardOpenOption.CREATE, StandardOpenOption.WRITE + ) + + val errMsg = intercept[TextParsingException] { + spark.read + .option("maxCharsPerColumn", maxCharsPerCol) + .csv(csvFile.getAbsolutePath) + .count() + }.getMessage + + assert(errMsg.contains("..." + "a" * CSVOptions.maxErrorContentLength), + s"expect the TextParsingException truncate the error content to be " + + s"${CSVOptions.maxErrorContentLength} length.") + } + } } From 6f28b8160f96f8a2b30ffdcace11b505012b9bf4 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Fri, 19 Jul 2019 22:05:28 +0800 Subject: [PATCH 4/8] update test --- .../apache/spark/sql/execution/datasources/csv/CSVSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 ee6cf6e87bb03..f6152ba7fb2af 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 @@ -2107,7 +2107,7 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te .count() }.getMessage - assert(errMsg.contains("..." + "a" * CSVOptions.maxErrorContentLength), + assert(errMsg.endsWith("=..." + "a" * CSVOptions.maxErrorContentLength), s"expect the TextParsingException truncate the error content to be " + s"${CSVOptions.maxErrorContentLength} length.") } From 4a11dd304b485ab8566c287504061a9151912905 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Sat, 20 Jul 2019 10:21:06 +0800 Subject: [PATCH 5/8] update --- .../scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala index 0fd624f7b9364..090e8e2fa0fb6 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala @@ -253,10 +253,10 @@ class CSVOptions( } } -object CSVOptions { +private[spark] object CSVOptions { /** * The max error content length in CSV parser/writer exception message. */ - val maxErrorContentLength = 1024 + val maxErrorContentLength = 1000 } From 2c354a09d6baa9dbbda10050d75716adc7493472 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Sun, 21 Jul 2019 11:35:58 +0800 Subject: [PATCH 6/8] update --- .../spark/sql/catalyst/csv/CSVOptions.scala | 17 +++++++---------- .../execution/datasources/csv/CSVSuite.scala | 5 ++--- 2 files changed, 9 insertions(+), 13 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala index 090e8e2fa0fb6..7d52847216cc9 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala @@ -163,6 +163,11 @@ class CSVOptions( val inputBufferSize = 128 + /** + * The max error content length in CSV parser/writer exception message. + */ + val maxErrorContentLength = 1000 + val isCommentSet = this.comment != '\u0000' val samplingRatio = @@ -220,7 +225,7 @@ class CSVOptions( writerSettings.setSkipEmptyLines(true) writerSettings.setQuoteAllFields(quoteAll) writerSettings.setQuoteEscapingEnabled(escapeQuotes) - writerSettings.setErrorContentLength(CSVOptions.maxErrorContentLength) + writerSettings.setErrorContentLength(maxErrorContentLength) writerSettings } @@ -247,16 +252,8 @@ class CSVOptions( lineSeparatorInRead.foreach { _ => settings.setNormalizeLineEndingsWithinQuotes(!multiLine) } - settings.setErrorContentLength(CSVOptions.maxErrorContentLength) + settings.setErrorContentLength(maxErrorContentLength) settings } } - -private[spark] object CSVOptions { - /** - * The max error content length in CSV parser/writer exception message. - */ - val maxErrorContentLength = 1000 - -} 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 f6152ba7fb2af..692e8702813da 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 @@ -2107,9 +2107,8 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te .count() }.getMessage - assert(errMsg.endsWith("=..." + "a" * CSVOptions.maxErrorContentLength), - s"expect the TextParsingException truncate the error content to be " + - s"${CSVOptions.maxErrorContentLength} length.") + assert(errMsg.contains("..."), + s"expect the TextParsingException truncate the error content to be 1000 length.") } } } From f8c3f7fa5f2bb64ec57fc2d33aae77fd0165d26b Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Mon, 22 Jul 2019 12:03:51 +0800 Subject: [PATCH 7/8] update --- .../apache/spark/sql/execution/datasources/csv/CSVSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 692e8702813da..f3ab1b812e4c2 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 @@ -2108,7 +2108,7 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te }.getMessage assert(errMsg.contains("..."), - s"expect the TextParsingException truncate the error content to be 1000 length.") + "expect the TextParsingException truncate the error content to be 1000 length.") } } } From 9e8fccabf1effa76b58527f6f061a2c8e525ab07 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Mon, 22 Jul 2019 14:56:44 +0800 Subject: [PATCH 8/8] update --- .../spark/sql/execution/datasources/csv/CSVSuite.scala | 10 ++++------ 1 file changed, 4 insertions(+), 6 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 f3ab1b812e4c2..c772b621939aa 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 @@ -37,7 +37,6 @@ import org.apache.log4j.spi.LoggingEvent import org.apache.spark.{SparkException, TestUtils} import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row} -import org.apache.spark.sql.catalyst.csv.CSVOptions import org.apache.spark.sql.catalyst.util.DateTimeUtils import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.{SharedSQLContext, SQLTestUtils} @@ -2089,21 +2088,20 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te } test("SPARK-28431: prevent CSV datasource throw TextParsingException with large size message") { - withTempDir { dir => + withTempPath { path => val maxCharsPerCol = 10000 val str = "a" * (maxCharsPerCol + 1) - val csvFile = new File(dir, "data.csv") Files.write( - csvFile.toPath, - Seq(str).asJava, + path.toPath, + str.getBytes(StandardCharsets.UTF_8), StandardOpenOption.CREATE, StandardOpenOption.WRITE ) val errMsg = intercept[TextParsingException] { spark.read .option("maxCharsPerColumn", maxCharsPerCol) - .csv(csvFile.getAbsolutePath) + .csv(path.getAbsolutePath) .count() }.getMessage