You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ma...@apache.org on 2022/09/17 08:00:15 UTC
[spark] branch branch-3.3 updated: [SPARK-40468][SQL] Fix column pruning in CSV when _corrupt_record is selected
This is an automated email from the ASF dual-hosted git repository.
maxgekk pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new d616da74a4e [SPARK-40468][SQL] Fix column pruning in CSV when _corrupt_record is selected
d616da74a4e is described below
commit d616da74a4ed7202b3480ebc234eb109dcc86fb9
Author: Ivan Sadikov <iv...@databricks.com>
AuthorDate: Sat Sep 17 10:59:32 2022 +0300
[SPARK-40468][SQL] Fix column pruning in CSV when _corrupt_record is selected
### What changes were proposed in this pull request?
The PR fixes an issue when depending on the name of the `_corrupt_record` field, column pruning would behave differently for a record that has no parsing errors.
For example, with a CSV file like this (c1 and c2 columns):
```
1,a
```
Before the patch, the following query would return:
```scala
val df = spark.read
.schema("c1 int, c2 string, x string, _corrupt_record string")
.csv("file:/tmp/file.csv")
.withColumn("x", lit("A"))
Result:
+---+---+---+---------------+
|c1 |c2 |x |_corrupt_record|
+---+---+---+---------------+
|1 |a |A |1,a |
+---+---+---+---------------+
```
However, if you rename the corrupt record column, the result is different (the original, arguably correct, behaviour before https://github.com/apache/spark/commit/959694271e30879c944d7fd5de2740571012460a):
```scala
val df = spark.read
.option("columnNameCorruptRecord", "corrupt_record")
.schema("c1 int, c2 string, x string, corrupt_record string")
.csv("file:/tmp/file.csv") .withColumn("x", lit("A"))
+---+---+---+--------------+
|c1 |c2 |x |corrupt_record|
+---+---+---+--------------+
|1 |a |A |null |
+---+---+---+--------------+
```
This patch fixes the former so both results would return `null` for corrupt record as there are no parsing issues. Note that https://issues.apache.org/jira/browse/SPARK-38523 is still fixed and works correctly.
### Why are the changes needed?
Fixes a bug where corrupt record would be non-null even though the record has no parsing errors.
### Does this PR introduce _any_ user-facing change?
Yes, fixes the output of corrupt record with additional columns provided by user. Everything should be unchanged outside of that scenario.
### How was this patch tested?
I added a unit test that reproduces the issue.
Closes #37909 from sadikovi/SPARK-40468.
Authored-by: Ivan Sadikov <iv...@databricks.com>
Signed-off-by: Max Gekk <ma...@gmail.com>
(cherry picked from commit 0776f9e7bcb10612eb977ed4884e9848aea86c33)
Signed-off-by: Max Gekk <ma...@gmail.com>
---
.../execution/datasources/csv/CSVFileFormat.scala | 4 +---
.../sql/execution/datasources/v2/csv/CSVScan.scala | 3 +--
.../sql/execution/datasources/csv/CSVSuite.scala | 26 ++++++++++++++++++++++
3 files changed, 28 insertions(+), 5 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
index 8d952507840..93679516a8c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
@@ -100,8 +100,7 @@ class CSVFileFormat extends TextBasedFileFormat with DataSourceRegister {
hadoopConf: Configuration): (PartitionedFile) => Iterator[InternalRow] = {
val broadcastedHadoopConf =
sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
- val columnPruning = sparkSession.sessionState.conf.csvColumnPruning &&
- !requiredSchema.exists(_.name == sparkSession.sessionState.conf.columnNameOfCorruptRecord)
+ val columnPruning = sparkSession.sessionState.conf.csvColumnPruning
val parsedOptions = new CSVOptions(
options,
columnPruning,
@@ -154,4 +153,3 @@ class CSVFileFormat extends TextBasedFileFormat with DataSourceRegister {
}
}
-
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVScan.scala
index 5c33a1047a1..d81223b48a5 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVScan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVScan.scala
@@ -44,8 +44,7 @@ case class CSVScan(
dataFilters: Seq[Expression] = Seq.empty)
extends TextBasedFileScan(sparkSession, options) {
- val columnPruning = sparkSession.sessionState.conf.csvColumnPruning &&
- !readDataSchema.exists(_.name == sparkSession.sessionState.conf.columnNameOfCorruptRecord)
+ val columnPruning = sparkSession.sessionState.conf.csvColumnPruning
private lazy val parsedOptions: CSVOptions = new CSVOptions(
options.asScala.toMap,
columnPruning = columnPruning,
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 dd42f48d716..5f4f8f84cf8 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
@@ -1655,6 +1655,32 @@ abstract class CSVSuite
Row(1, Date.valueOf("1983-08-04"), null) :: Nil)
}
+ test("SPARK-40468: column pruning with the corrupt record column") {
+ withTempPath { path =>
+ Seq("1,a").toDF()
+ .repartition(1)
+ .write.text(path.getAbsolutePath)
+
+ // Corrupt record column with the default name should return null instead of "1,a"
+ val corruptRecordCol = spark.sessionState.conf.columnNameOfCorruptRecord
+ var df = spark.read
+ .schema(s"c1 int, c2 string, x string, ${corruptRecordCol} string")
+ .csv(path.getAbsolutePath)
+ .selectExpr("c1", "c2", "'A' as x", corruptRecordCol)
+
+ checkAnswer(df, Seq(Row(1, "a", "A", null)))
+
+ // Corrupt record column with the user-provided name should return null instead of "1,a"
+ df = spark.read
+ .schema(s"c1 int, c2 string, x string, _invalid string")
+ .option("columnNameCorruptRecord", "_invalid")
+ .csv(path.getAbsolutePath)
+ .selectExpr("c1", "c2", "'A' as x", "_invalid")
+
+ checkAnswer(df, Seq(Row(1, "a", "A", null)))
+ }
+ }
+
test("SPARK-23846: schema inferring touches less data if samplingRatio < 1.0") {
// Set default values for the DataSource parameters to make sure
// that whole test file is mapped to only one partition. This will guarantee
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org