szehon-ho commented on code in PR #57671:
URL: https://github.com/apache/spark/pull/57671#discussion_r3788006531
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala:
##########
@@ -4072,6 +4072,109 @@ abstract class CSVSuite
}
}
}
+
+ test("SPARK-58458: a quoted line break splits the record when multiLine is
disabled") {
Review Comment:
[P3 / question] The two cases here are independent scenarios, and the second
is the more valuable of the two: it is the one where a silently truncated row
survives `DROPMALFORMED`. Right now it sits at the bottom of a test whose name
mentions only the split, so nothing in the name carries that point.
Would you consider two tests, along the lines of `... a quoted line break
splits the record when multiLine is disabled` and `... a split half whose token
count matches the schema is not malformed`? That would also make each runnable
on its own with `-z`. Spark has precedent both ways, so treat this as a
preference rather than a request.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala:
##########
@@ -4072,6 +4072,109 @@ abstract class CSVSuite
}
}
}
+
+ test("SPARK-58458: a quoted line break splits the record when multiLine is
disabled") {
+ // A line break inside a quoted value is valid CSV (RFC 4180 section 2.6)
and occurs in
+ // published datasets. In the default non-multiLine mode the input is
split on the line
+ // separator before the CSV tokenizer runs -- HadoopFileLinesReader feeds
+ // UnivocityParser.parseLine one physical line at a time -- so the record
cannot be
+ // reassembled: the leading value is truncated and the remainder starts a
new record.
+ //
+ // Whether a resulting half is *malformed* is a separate question from the
split, and is
+ // decided mainly by token count: univocity's default STOP_AT_DELIMITER
does not fail on
+ // an unclosed quote. The two cases below differ in exactly that, they
behave differently
+ // under DROPMALFORMED, and the difference is the part worth pinning.
+ val threeCols = new StructType()
+ .add("id", StringType)
+ .add("nome", StringType)
+ .add("uf", StringType)
+
+ // Case 1 -- both halves carry 2 tokens against a 3-column schema, so both
are malformed.
+ withTempPath { path =>
+ Files.write(
+ path.toPath,
+ ("1,ACME LTDA,SP\n" +
+ "2,\"EMPRESA COM\nQUEBRA DE LINHA\",RJ\n" +
+ "3,OUTRA EMPRESA,MG\n").getBytes(StandardCharsets.UTF_8))
+
+ // Three records in, four rows out: record 2 is cut at the line break.
+ checkAnswer(
+ spark.read.schema(threeCols).csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("2", "EMPRESA COM", null) ::
+ Row("QUEBRA DE LINHA\"", "RJ", null) ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // multiLine reassembles the record, so the break survives inside the
value.
+ checkAnswer(
+ spark.read.schema(threeCols).option("multiLine",
true).csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("2", "EMPRESA COM\nQUEBRA DE LINHA", "RJ") ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // Both halves being malformed, DROPMALFORMED discards the pair and the
whole source
+ // record is lost -- not just the damaged half.
+ checkAnswer(
+ spark.read.schema(threeCols).option("mode",
"DROPMALFORMED").csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // FAILFAST refuses the file rather than returning a split record.
+ val e = intercept[SparkException] {
+ spark.read
+ .schema(threeCols)
+ .option("mode", "FAILFAST")
+ .csv(path.getAbsolutePath)
+ .collect()
+ }
+ checkErrorMatchPVals(
+ exception = e,
+ condition = "FAILED_READ_FILE.NO_HINT",
+ parameters = Map("path" -> s".*${path.getName}.*"))
Review Comment:
[P2] This assertion has almost no pinning power. `FAILED_READ_FILE.NO_HINT`
plus the path is what *any* read failure on this file produces, so it certifies
little more than "the read raised". It would still pass if the split stopped
happening the way the rest of this test documents, for instance if the leading
half were padded to the schema length so that only the trailing fragment failed.
The suite already establishes the stronger convention in `test("test for
FAILFAST parsing mode")`, walking the chain down to the offending record. That
works here too, and the innermost cause carries the raw line:
`UnivocityParser.convert` builds it from `currentInput` on a token-count
mismatch, and `FailureSafeParser` unwraps the `LazyBadRecordCauseWrapper` under
`FailFastMode`. So the cause can assert that the record was cut exactly at the
quoted line break, which is the fact this test exists to capture:
```suggestion
checkErrorMatchPVals(
exception = e,
condition = "FAILED_READ_FILE.NO_HINT",
parameters = Map("path" -> s".*${path.getName}.*"))
val cause = e.getCause.asInstanceOf[SparkException]
assert(cause.getCondition ==
"MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION")
checkError(
exception = cause.getCause.asInstanceOf[SparkRuntimeException],
condition = "MALFORMED_CSV_RECORD",
parameters = Map("badRecord" -> "2,\"EMPRESA COM"))
```
I derived that chain by reading the source rather than by running it, so
please confirm the exact `badRecord` text. The leading half should fail first,
being earlier in the file.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala:
##########
@@ -4072,6 +4072,109 @@ abstract class CSVSuite
}
}
}
+
+ test("SPARK-58458: a quoted line break splits the record when multiLine is
disabled") {
+ // A line break inside a quoted value is valid CSV (RFC 4180 section 2.6)
and occurs in
+ // published datasets. In the default non-multiLine mode the input is
split on the line
+ // separator before the CSV tokenizer runs -- HadoopFileLinesReader feeds
+ // UnivocityParser.parseLine one physical line at a time -- so the record
cannot be
+ // reassembled: the leading value is truncated and the remainder starts a
new record.
+ //
+ // Whether a resulting half is *malformed* is a separate question from the
split, and is
+ // decided mainly by token count: univocity's default STOP_AT_DELIMITER
does not fail on
+ // an unclosed quote. The two cases below differ in exactly that, they
behave differently
+ // under DROPMALFORMED, and the difference is the part worth pinning.
+ val threeCols = new StructType()
+ .add("id", StringType)
+ .add("nome", StringType)
+ .add("uf", StringType)
+
+ // Case 1 -- both halves carry 2 tokens against a 3-column schema, so both
are malformed.
+ withTempPath { path =>
+ Files.write(
+ path.toPath,
+ ("1,ACME LTDA,SP\n" +
+ "2,\"EMPRESA COM\nQUEBRA DE LINHA\",RJ\n" +
+ "3,OUTRA EMPRESA,MG\n").getBytes(StandardCharsets.UTF_8))
+
+ // Three records in, four rows out: record 2 is cut at the line break.
+ checkAnswer(
+ spark.read.schema(threeCols).csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("2", "EMPRESA COM", null) ::
+ Row("QUEBRA DE LINHA\"", "RJ", null) ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // multiLine reassembles the record, so the break survives inside the
value.
+ checkAnswer(
+ spark.read.schema(threeCols).option("multiLine",
true).csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("2", "EMPRESA COM\nQUEBRA DE LINHA", "RJ") ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // Both halves being malformed, DROPMALFORMED discards the pair and the
whole source
+ // record is lost -- not just the damaged half.
+ checkAnswer(
+ spark.read.schema(threeCols).option("mode",
"DROPMALFORMED").csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // FAILFAST refuses the file rather than returning a split record.
+ val e = intercept[SparkException] {
+ spark.read
+ .schema(threeCols)
+ .option("mode", "FAILFAST")
+ .csv(path.getAbsolutePath)
+ .collect()
+ }
+ checkErrorMatchPVals(
+ exception = e,
+ condition = "FAILED_READ_FILE.NO_HINT",
+ parameters = Map("path" -> s".*${path.getName}.*"))
+
+ // Only with columnNameOfCorruptRecord declared does the split leave a
trace.
+ checkAnswer(
+ spark.read
+ .schema(threeCols.add("_corrupt", StringType))
+ .option("columnNameOfCorruptRecord", "_corrupt")
+ .csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP", null) ::
+ Row("2", "EMPRESA COM", null, "2,\"EMPRESA COM") ::
+ Row("QUEBRA DE LINHA\"", "RJ", null, "QUEBRA DE LINHA\",RJ") ::
+ Row("3", "OUTRA EMPRESA", "MG", null) :: Nil)
Review Comment:
[P2] The `count()` and column-pruning caveat is the one claim from #57608
that this test leaves unpinned, and it is the most fragile of them. Consider
adding it after the `FAILFAST` check above.
Now that #57608 has merged (`bf76295bba7`), the `mode` row it added ends
with:
> An action requiring no columns (a bare `count()`, for example) may surface
none of this because of column pruning.
That rests entirely on a narrow shortcut in `UnivocityParser`:
```scala
val parse: String => Option[InternalRow] = {
if (columnPruning && requiredSchema.isEmpty) {
(_: String) => Some(InternalRow.empty)
} else {
(input: String) => convert(parseLine(input))
}
}
```
Of the four documented consequences it is both the most surprising, since
even `FAILFAST` returns the inflated count rather than raising, and the one
most exposed to a silent refactor, because it depends on that shortcut, on
`isColumnPruningEnabled`, and on the
`spark.sql.csv.parser.columnPruning.enabled` default all staying as they are.
It is also the cheapest to add, and it contrasts sharply with the `FAILFAST`
assertion just above:
```scala
// Column pruning means an action requiring no columns never parses a
record, so even
// FAILFAST reports the inflated count rather than refusing the file.
assert(spark.read.schema(threeCols).option("mode", "FAILFAST")
.csv(path.getAbsolutePath).count() === 4)
```
Worth confirming it holds for v1 and v2 alike; the `filters push down` test
in this suite already relies on `count()` pushing an empty schema in both.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala:
##########
@@ -4072,6 +4072,109 @@ abstract class CSVSuite
}
}
}
+
+ test("SPARK-58458: a quoted line break splits the record when multiLine is
disabled") {
+ // A line break inside a quoted value is valid CSV (RFC 4180 section 2.6)
and occurs in
+ // published datasets. In the default non-multiLine mode the input is
split on the line
+ // separator before the CSV tokenizer runs -- HadoopFileLinesReader feeds
+ // UnivocityParser.parseLine one physical line at a time -- so the record
cannot be
+ // reassembled: the leading value is truncated and the remainder starts a
new record.
+ //
+ // Whether a resulting half is *malformed* is a separate question from the
split, and is
+ // decided mainly by token count: univocity's default STOP_AT_DELIMITER
does not fail on
+ // an unclosed quote. The two cases below differ in exactly that, they
behave differently
+ // under DROPMALFORMED, and the difference is the part worth pinning.
Review Comment:
[P3] `STOP_AT_DELIMITER` is Spark's default rather than univocity's. Spark
sets it explicitly on every read, in `CSVOptions`:
```scala
val unescapedQuoteHandling: UnescapedQuoteHandling =
UnescapedQuoteHandling.valueOf(parameters
.getOrElse(UNESCAPED_QUOTE_HANDLING,
"STOP_AT_DELIMITER").toUpperCase(Locale.ROOT))
```
together with `settings.setUnescapedQuoteHandling(unescapedQuoteHandling)`
further down, so univocity's own default never comes into play. A reader
following this comment would go looking in the wrong library. In fairness, the
phrasing came out of the #57608 thread, mine included.
It is worth fixing beyond the attribution, because the option is
load-bearing for what this test asserts: the asymmetry between the stripped
leading quote (`EMPRESA COM`) and the retained trailing one (`QUEBRA DE
LINHA"`) follows from `STOP_AT_DELIMITER`, and `STOP_AT_CLOSING_QUOTE` would
produce different halves, as the SPARK-33566 test earlier in this suite shows.
Wording it as "Spark's default `unescapedQuoteHandling` (`STOP_AT_DELIMITER`)"
would correct the attribution and bound the claim at the same time.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala:
##########
@@ -4072,6 +4072,109 @@ abstract class CSVSuite
}
}
}
+
+ test("SPARK-58458: a quoted line break splits the record when multiLine is
disabled") {
+ // A line break inside a quoted value is valid CSV (RFC 4180 section 2.6)
and occurs in
+ // published datasets. In the default non-multiLine mode the input is
split on the line
+ // separator before the CSV tokenizer runs -- HadoopFileLinesReader feeds
+ // UnivocityParser.parseLine one physical line at a time -- so the record
cannot be
+ // reassembled: the leading value is truncated and the remainder starts a
new record.
+ //
+ // Whether a resulting half is *malformed* is a separate question from the
split, and is
+ // decided mainly by token count: univocity's default STOP_AT_DELIMITER
does not fail on
+ // an unclosed quote. The two cases below differ in exactly that, they
behave differently
+ // under DROPMALFORMED, and the difference is the part worth pinning.
+ val threeCols = new StructType()
+ .add("id", StringType)
+ .add("nome", StringType)
+ .add("uf", StringType)
+
+ // Case 1 -- both halves carry 2 tokens against a 3-column schema, so both
are malformed.
+ withTempPath { path =>
+ Files.write(
+ path.toPath,
+ ("1,ACME LTDA,SP\n" +
+ "2,\"EMPRESA COM\nQUEBRA DE LINHA\",RJ\n" +
+ "3,OUTRA EMPRESA,MG\n").getBytes(StandardCharsets.UTF_8))
+
+ // Three records in, four rows out: record 2 is cut at the line break.
+ checkAnswer(
+ spark.read.schema(threeCols).csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("2", "EMPRESA COM", null) ::
+ Row("QUEBRA DE LINHA\"", "RJ", null) ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // multiLine reassembles the record, so the break survives inside the
value.
+ checkAnswer(
+ spark.read.schema(threeCols).option("multiLine",
true).csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("2", "EMPRESA COM\nQUEBRA DE LINHA", "RJ") ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // Both halves being malformed, DROPMALFORMED discards the pair and the
whole source
+ // record is lost -- not just the damaged half.
+ checkAnswer(
+ spark.read.schema(threeCols).option("mode",
"DROPMALFORMED").csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // FAILFAST refuses the file rather than returning a split record.
+ val e = intercept[SparkException] {
+ spark.read
+ .schema(threeCols)
+ .option("mode", "FAILFAST")
+ .csv(path.getAbsolutePath)
+ .collect()
+ }
+ checkErrorMatchPVals(
+ exception = e,
+ condition = "FAILED_READ_FILE.NO_HINT",
+ parameters = Map("path" -> s".*${path.getName}.*"))
+
+ // Only with columnNameOfCorruptRecord declared does the split leave a
trace.
+ checkAnswer(
+ spark.read
+ .schema(threeCols.add("_corrupt", StringType))
+ .option("columnNameOfCorruptRecord", "_corrupt")
+ .csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP", null) ::
+ Row("2", "EMPRESA COM", null, "2,\"EMPRESA COM") ::
+ Row("QUEBRA DE LINHA\"", "RJ", null, "QUEBRA DE LINHA\",RJ") ::
+ Row("3", "OUTRA EMPRESA", "MG", null) :: Nil)
+ }
+
+ // Case 2 -- the leading half's token count matches the schema, so it is
NOT malformed.
+ // It is a clean row carrying a silently truncated value, and
DROPMALFORMED keeps it,
+ // which is the opposite of what the mode's name suggests for a damaged
record.
+ val twoCols = new StructType()
+ .add("id", StringType)
+ .add("valor", StringType)
+
+ withTempPath { path =>
+ Files.write(
+ path.toPath, "1,\"hello\nworld\"\n".getBytes(StandardCharsets.UTF_8))
+
+ checkAnswer(
+ spark.read.schema(twoCols).csv(path.getAbsolutePath),
+ Row("1", "hello") ::
+ Row("world\"", null) :: Nil)
+
+ // The truncated row survives; only the trailing fragment is dropped.
+ checkAnswer(
+ spark.read.schema(twoCols).option("mode",
"DROPMALFORMED").csv(path.getAbsolutePath),
+ Row("1", "hello") :: Nil)
Review Comment:
[P3] Case 2 is described as the case worth having, yet it gets the lightest
coverage: no `FAILFAST` and no `multiLine = true` contrast, both of which Case
1 has.
`FAILFAST` here would pin a sharper asymmetry than anything currently
asserted, namely that the read still fails even though the truncated row is
individually valid, because the trailing fragment is not. Since @uros-b asked
for FAILFAST coverage, applying it only to the case where nothing silently
wrong survives leaves that request half-addressed.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala:
##########
@@ -4072,6 +4072,109 @@ abstract class CSVSuite
}
}
}
+
+ test("SPARK-58458: a quoted line break splits the record when multiLine is
disabled") {
+ // A line break inside a quoted value is valid CSV (RFC 4180 section 2.6)
and occurs in
+ // published datasets. In the default non-multiLine mode the input is
split on the line
+ // separator before the CSV tokenizer runs -- HadoopFileLinesReader feeds
+ // UnivocityParser.parseLine one physical line at a time -- so the record
cannot be
+ // reassembled: the leading value is truncated and the remainder starts a
new record.
+ //
+ // Whether a resulting half is *malformed* is a separate question from the
split, and is
+ // decided mainly by token count: univocity's default STOP_AT_DELIMITER
does not fail on
+ // an unclosed quote. The two cases below differ in exactly that, they
behave differently
+ // under DROPMALFORMED, and the difference is the part worth pinning.
+ val threeCols = new StructType()
+ .add("id", StringType)
+ .add("nome", StringType)
+ .add("uf", StringType)
+
+ // Case 1 -- both halves carry 2 tokens against a 3-column schema, so both
are malformed.
+ withTempPath { path =>
+ Files.write(
+ path.toPath,
+ ("1,ACME LTDA,SP\n" +
+ "2,\"EMPRESA COM\nQUEBRA DE LINHA\",RJ\n" +
+ "3,OUTRA EMPRESA,MG\n").getBytes(StandardCharsets.UTF_8))
+
+ // Three records in, four rows out: record 2 is cut at the line break.
+ checkAnswer(
+ spark.read.schema(threeCols).csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("2", "EMPRESA COM", null) ::
+ Row("QUEBRA DE LINHA\"", "RJ", null) ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // multiLine reassembles the record, so the break survives inside the
value.
+ checkAnswer(
+ spark.read.schema(threeCols).option("multiLine",
true).csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("2", "EMPRESA COM\nQUEBRA DE LINHA", "RJ") ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // Both halves being malformed, DROPMALFORMED discards the pair and the
whole source
+ // record is lost -- not just the damaged half.
+ checkAnswer(
+ spark.read.schema(threeCols).option("mode",
"DROPMALFORMED").csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP") ::
+ Row("3", "OUTRA EMPRESA", "MG") :: Nil)
+
+ // FAILFAST refuses the file rather than returning a split record.
+ val e = intercept[SparkException] {
+ spark.read
+ .schema(threeCols)
+ .option("mode", "FAILFAST")
+ .csv(path.getAbsolutePath)
+ .collect()
+ }
+ checkErrorMatchPVals(
+ exception = e,
+ condition = "FAILED_READ_FILE.NO_HINT",
+ parameters = Map("path" -> s".*${path.getName}.*"))
+
+ // Only with columnNameOfCorruptRecord declared does the split leave a
trace.
+ checkAnswer(
+ spark.read
+ .schema(threeCols.add("_corrupt", StringType))
+ .option("columnNameOfCorruptRecord", "_corrupt")
+ .csv(path.getAbsolutePath),
+ Row("1", "ACME LTDA", "SP", null) ::
+ Row("2", "EMPRESA COM", null, "2,\"EMPRESA COM") ::
+ Row("QUEBRA DE LINHA\"", "RJ", null, "QUEBRA DE LINHA\",RJ") ::
+ Row("3", "OUTRA EMPRESA", "MG", null) :: Nil)
+ }
+
+ // Case 2 -- the leading half's token count matches the schema, so it is
NOT malformed.
+ // It is a clean row carrying a silently truncated value, and
DROPMALFORMED keeps it,
+ // which is the opposite of what the mode's name suggests for a damaged
record.
Review Comment:
[Open question] Should the test signpost that these values are not contracts?
This comment is right that `DROPMALFORMED` keeping the truncated row is the
opposite of what the mode's name suggests. Making that row visible as corrupt
would be a defensible future improvement, and it would necessarily change these
assertions.
For a test that deliberately pins behaviour it simultaneously describes as
surprising, one line saying so would stop a future contributor from reading the
failure as a regression. Something like: "These rows pin current behaviour, not
a guarantee. A change that surfaces the truncated row as corrupt should update
this test rather than preserve it."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
