You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/08/12 23:57:19 UTC
(spark) branch branch-4.0 updated: Revert "[SPARK-52482][SQL][CORE] Improve exception handling for reading certain corrupt zstd files"
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new f86b60d79a71 Revert "[SPARK-52482][SQL][CORE] Improve exception handling for reading certain corrupt zstd files"
f86b60d79a71 is described below
commit f86b60d79a712aeac61ef2f64c5bf41be0a39f63
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Aug 12 16:57:08 2025 -0700
Revert "[SPARK-52482][SQL][CORE] Improve exception handling for reading certain corrupt zstd files"
This reverts commit ec36b8b395f2be90cc759d1e314239c7c1f4afbd.
---
.../execution/datasources/DataSourceUtils.scala | 6 -----
.../sql/execution/datasources/FileScanRDD.scala | 5 ++--
.../datasources/v2/FilePartitionReader.scala | 8 +++---
.../sql/execution/datasources/csv/CSVSuite.scala | 29 +---------------------
4 files changed, 7 insertions(+), 41 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala
index 7b267e8b2545..81eadcc263c6 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala
@@ -17,7 +17,6 @@
package org.apache.spark.sql.execution.datasources
-import java.io.IOException
import java.util.Locale
import scala.jdk.CollectionConverters._
@@ -193,11 +192,6 @@ object DataSourceUtils extends PredicateHelper {
QueryExecutionErrors.sparkUpgradeInWritingDatesError(format, config)
}
- def shouldIgnoreCorruptFileException(e: Throwable): Boolean = e match {
- case _: RuntimeException | _: IOException | _: InternalError => true
- case _ => false
- }
-
def createDateRebaseFuncInRead(
rebaseMode: LegacyBehaviorPolicy.Value,
format: String): Int => Int = rebaseMode match {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala
index 9a28d2e4f664..e468807f4ffd 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.execution.datasources
-import java.io.{Closeable, FileNotFoundException}
+import java.io.{Closeable, FileNotFoundException, IOException}
import java.net.URI
import org.apache.hadoop.fs.Path
@@ -269,8 +269,7 @@ class FileScanRDD(
// Throw FileNotFoundException even if `ignoreCorruptFiles` is true
case e: FileNotFoundException if !ignoreMissingFiles => throw e
case e @ (_ : AccessControlException | _ : BlockMissingException) => throw e
- case e if ignoreCorruptFiles &&
- DataSourceUtils.shouldIgnoreCorruptFileException(e) =>
+ case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles =>
logWarning(log"Skipped the rest of the content in the corrupted file: " +
log"${MDC(PATH, currentFile)}", e)
finished = true
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala
index 2911e17591a3..be4f5dcb65aa 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala
@@ -16,7 +16,7 @@
*/
package org.apache.spark.sql.execution.datasources.v2
-import java.io.FileNotFoundException
+import java.io.{FileNotFoundException, IOException}
import org.apache.hadoop.hdfs.BlockMissingException
import org.apache.hadoop.security.AccessControlException
@@ -26,7 +26,7 @@ import org.apache.spark.internal.LogKeys.{CURRENT_FILE, PARTITIONED_FILE_READER}
import org.apache.spark.rdd.InputFileBlockHolder
import org.apache.spark.sql.catalyst.FileSourceOptions
import org.apache.spark.sql.connector.read.PartitionReader
-import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedFile}
+import org.apache.spark.sql.execution.datasources.PartitionedFile
class FilePartitionReader[T](
files: Iterator[PartitionedFile],
@@ -53,7 +53,7 @@ class FilePartitionReader[T](
currentReader = null
case e @ (_ : AccessControlException | _ : BlockMissingException) =>
throw FileDataSourceV2.attachFilePath(file.urlEncodedPath, e)
- case e if ignoreCorruptFiles && DataSourceUtils.shouldIgnoreCorruptFileException(e) =>
+ case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles =>
logWarning(
s"Skipped the rest of the content in the corrupted file.", e)
currentReader = null
@@ -71,7 +71,7 @@ class FilePartitionReader[T](
} catch {
case e @ (_ : AccessControlException | _ : BlockMissingException) =>
throw FileDataSourceV2.attachFilePath(currentReader.file.urlEncodedPath, e)
- case e if ignoreCorruptFiles && DataSourceUtils.shouldIgnoreCorruptFileException(e) =>
+ case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles =>
logWarning(log"Skipped the rest of the content in the corrupted file: " +
log"${MDC(PARTITIONED_FILE_READER, currentReader)}", e)
false
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 296a04df03b5..850e887ac8e7 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
@@ -18,9 +18,8 @@
package org.apache.spark.sql.execution.datasources.csv
import java.io.{EOFException, File}
-import java.net.URI
import java.nio.charset.{Charset, StandardCharsets}
-import java.nio.file.{Files, Paths, StandardOpenOption}
+import java.nio.file.{Files, StandardOpenOption}
import java.sql.{Date, Timestamp}
import java.text.SimpleDateFormat
import java.time._
@@ -3493,32 +3492,6 @@ abstract class CSVSuite
val textParsingException = malformedCSVException.getCause.asInstanceOf[TextParsingException]
assert(textParsingException.getCause.isInstanceOf[ArrayIndexOutOfBoundsException])
}
-
- test("corrupted ZSTD compressed csv respects ignoreCorruptFiles") {
- withTempDir { dir =>
- val originalFile = new File(dir, "original.csv.zst")
- val corruptedHeadFile = new File(dir, "corrupted_head.csv.zst")
- val corruptedTailFile = new File(dir, "corrupted_tail.csv.zst")
- val bytes = Files.readAllBytes(Paths.get(new URI(testFile(zstCompressedCarsFile))))
- Files.write(originalFile.toPath(), bytes)
- Files.write(corruptedHeadFile.toPath(), bytes.drop(10))
- Files.write(corruptedTailFile.toPath(), bytes.dropRight(10))
-
- withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> "true") {
- val df = spark.read.format("csv").option("header", "true").load(dir.getAbsolutePath)
- // check that the entries from originalFile are still read
- assert(df.count() == 3)
- }
-
- withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> "false") {
- val ex = intercept[SparkException] {
- spark.read.format("csv").option("header", "true").load(dir.getAbsolutePath).collect()
- }
- checkErrorMatchPVals(ex, "FAILED_READ_FILE.NO_HINT",
- Map("path" -> ".*corrupted.*\\.csv\\.zst"))
- }
- }
- }
}
class CSVv1Suite extends CSVSuite {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org