From 1f82ca8993ae655a6742263f5f06880d1a1c3027 Mon Sep 17 00:00:00 2001 From: Jorge Molina <70665012+Joorgem@users.noreply.github.com> Date: Fri, 31 Jul 2026 08:59:44 -0300 Subject: [PATCH 1/2] [SPARK-58458][SQL][TESTS] Add coverage for the record split when multiLine is disabled --- .../execution/datasources/csv/CSVSuite.scala | 57 +++++++++++++++++++ 1 file changed, 57 insertions(+) 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 71eb34134920b..bc7a75917b12c 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 @@ -4072,6 +4072,63 @@ 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. It becomes a NULL-tailed head plus a fragment, which changes the row + // count as well as the values. None of that is currently covered by a test. + val schema = new StructType() + .add("id", StringType) + .add("nome", StringType) + .add("uf", StringType) + + 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, its trailing + // field is nulled, and the remainder of the quoted value starts a new record. + checkAnswer( + spark.read.schema(schema).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 line break survives inside the value. + checkAnswer( + spark.read.schema(schema).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 carry unbalanced quotes, so both are malformed and DROPMALFORMED + // discards the pair -- the whole source record is lost, not just the damaged half. + checkAnswer( + spark.read.schema(schema).option("mode", "DROPMALFORMED").csv(path.getAbsolutePath), + Row("1", "ACME LTDA", "SP") :: + Row("3", "OUTRA EMPRESA", "MG") :: Nil) + + // With columnNameOfCorruptRecord declared, both halves are captured with their raw + // text, which is the only configuration in which the split leaves a trace. + val withCorrupt = schema.add("_corrupt", StringType) + checkAnswer( + spark.read + .schema(withCorrupt) + .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) + } + } } class CSVv1Suite extends CSVSuite { From 3a9ef31af6ff298591694b9b7c06c4cab9062227 Mon Sep 17 00:00:00 2001 From: Jorge Molina <70665012+Joorgem@users.noreply.github.com> Date: Sat, 1 Aug 2026 14:25:23 -0300 Subject: [PATCH 2/2] Add FAILFAST coverage, and pin the case where a half is not malformed Answers review on this PR and on #57608. @uros-b: FAILFAST was missing. Added, asserting FAILED_READ_FILE.NO_HINT the way the suite existing FAILFAST test does. @szehon-ho pointed out on #57608 that both halves are not always malformed: each physical line is parsed alone, univocity STOP_AT_DELIMITER does not fail on an unclosed quote, and token count is what decides it. That condemns this test comment, which blamed unbalanced quotes -- wrong, even though the assertions passed. Corrected, and a second case added: a 2-column schema where the leading half token count matches, so it comes back clean and unflagged with a silently truncated value, and DROPMALFORMED keeps it. The case where both halves die is the benign one; that is the one worth locking down. --- .../execution/datasources/csv/CSVSuite.scala | 76 +++++++++++++++---- 1 file changed, 61 insertions(+), 15 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 bc7a75917b12c..27c2c872cd2ea 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 @@ -4078,13 +4078,18 @@ abstract class CSVSuite // 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. It becomes a NULL-tailed head plus a fragment, which changes the row - // count as well as the values. None of that is currently covered by a test. - val schema = new StructType() + // 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, @@ -4092,35 +4097,45 @@ abstract class CSVSuite "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, its trailing - // field is nulled, and the remainder of the quoted value starts a new record. + // Three records in, four rows out: record 2 is cut at the line break. checkAnswer( - spark.read.schema(schema).csv(path.getAbsolutePath), + 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 line break survives inside the value. + // multiLine reassembles the record, so the break survives inside the value. checkAnswer( - spark.read.schema(schema).option("multiLine", true).csv(path.getAbsolutePath), + 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 carry unbalanced quotes, so both are malformed and DROPMALFORMED - // discards the pair -- the whole source record is lost, not just the damaged half. + // Both halves being malformed, DROPMALFORMED discards the pair and the whole source + // record is lost -- not just the damaged half. checkAnswer( - spark.read.schema(schema).option("mode", "DROPMALFORMED").csv(path.getAbsolutePath), + spark.read.schema(threeCols).option("mode", "DROPMALFORMED").csv(path.getAbsolutePath), Row("1", "ACME LTDA", "SP") :: Row("3", "OUTRA EMPRESA", "MG") :: Nil) - // With columnNameOfCorruptRecord declared, both halves are captured with their raw - // text, which is the only configuration in which the split leaves a trace. - val withCorrupt = schema.add("_corrupt", StringType) + // 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(withCorrupt) + .schema(threeCols.add("_corrupt", StringType)) .option("columnNameOfCorruptRecord", "_corrupt") .csv(path.getAbsolutePath), Row("1", "ACME LTDA", "SP", null) :: @@ -4128,6 +4143,37 @@ abstract class CSVSuite 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) + + // And nothing flags it: the corrupt-record column stays null on the truncated row. + checkAnswer( + spark.read + .schema(twoCols.add("_corrupt", StringType)) + .option("columnNameOfCorruptRecord", "_corrupt") + .csv(path.getAbsolutePath), + Row("1", "hello", null) :: + Row("world\"", null, "world\"") :: Nil) + } } }