You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ka...@apache.org on 2024/07/08 07:24:16 UTC
(spark) branch master updated: [SPARK-48802][SS][FOLLOWUP] FileStreamSource maxCachedFiles set to 0 causes batch with no files to be processed
This is an automated email from the ASF dual-hosted git repository.
kabhwan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 872420dbc7ae [SPARK-48802][SS][FOLLOWUP] FileStreamSource maxCachedFiles set to 0 causes batch with no files to be processed
872420dbc7ae is described below
commit 872420dbc7ae7867e449aac6e6a7f6ef13929dff
Author: ragnarok56 <ke...@gmail.com>
AuthorDate: Mon Jul 8 16:24:02 2024 +0900
[SPARK-48802][SS][FOLLOWUP] FileStreamSource maxCachedFiles set to 0 causes batch with no files to be processed
### What changes were proposed in this pull request?
This is a followup to a bug identified from https://github.com/apache/spark/pull/45362. When setting `maxCachedFiles` to 0 (to force a full relisting of files for each batch, see https://issues.apache.org/jira/browse/SPARK-44924) subsequent batches of files would be skipped due to a logic error that carried forward an empty array of `unreadFiles` which was only being null checked. This update includes additional checks to verify that `unreadFiles` is also non-empty as a guard condit [...]
### Why are the changes needed?
Setting the `maxCachedFiles` configuration to 0 would inadvertently cause every other batch to contain 0 files, which is an unexpected behavior for users.
### Does this PR introduce _any_ user-facing change?
Fixes the case where users may want to always perform a full listing of files each batch by setting `maxCachedFiles` to 0
### How was this patch tested?
New test added to verify `maxCachedFiles` set to 0 would perform a file listing each batch
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #47195 from ragnarok56/filestreamsource-maxcachedfiles-edgecase.
Lead-authored-by: ragnarok56 <ke...@gmail.com>
Co-authored-by: Kevin Nacios <ke...@gmail.com>
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
.../sql/execution/streaming/FileStreamSource.scala | 9 ++-
.../sql/streaming/FileStreamSourceSuite.scala | 65 ++++++++++++++++++++++
2 files changed, 71 insertions(+), 3 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
index 4a9b2d11b7e0..25c6d454dfd5 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
@@ -168,7 +168,7 @@ class FileStreamSource(
* there is no race here, so the cost of `synchronized` should be rare.
*/
private def fetchMaxOffset(limit: ReadLimit): FileStreamSourceOffset = synchronized {
- val newFiles = if (unreadFiles != null) {
+ val newFiles = if (unreadFiles != null && unreadFiles.nonEmpty) {
logDebug(s"Reading from unread files - ${unreadFiles.size} files are available.")
unreadFiles
} else {
@@ -227,13 +227,16 @@ class FileStreamSource(
case _: ReadAllAvailable => (newFiles, null)
}
- if (unselectedFiles != null && unselectedFiles.nonEmpty) {
+ // need to ensure that if maxCachedFiles is set to 0 that the next batch will be forced to
+ // list files again
+ if (unselectedFiles != null && unselectedFiles.nonEmpty && maxCachedFiles > 0) {
logTrace(s"Taking first $maxCachedFiles unread files.")
unreadFiles = unselectedFiles.take(maxCachedFiles)
logTrace(s"${unreadFiles.size} unread files are available for further batches.")
} else {
unreadFiles = null
- logTrace(s"No unread file is available for further batches.")
+ logTrace(s"No unread file is available for further batches or maxCachedFiles has been set " +
+ s" to 0 to disable caching.")
}
batchFiles.foreach { case NewFileEntry(p, _, timestamp) =>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala
index ca4f2a7f26ce..634311b669a8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala
@@ -2448,6 +2448,71 @@ class FileStreamSourceSuite extends FileStreamSourceTest {
}
}
+ test("SPARK-48802: Ensure maxCachedFiles set to 0 forces each batch to list files") {
+ withCountListingLocalFileSystemAsLocalFileSystem {
+ withThreeTempDirs { case (src, meta, tmp) =>
+ val options = Map("latestFirst" -> "false", "maxFilesPerTrigger" -> "10",
+ "maxCachedFiles" -> "0")
+ val scheme = CountListingLocalFileSystem.scheme
+ val source = new FileStreamSource(spark, s"$scheme:///${src.getCanonicalPath}/*/*", "text",
+ StructType(Nil), Seq.empty, meta.getCanonicalPath, options)
+ val _metadataLog = PrivateMethod[FileStreamSourceLog](Symbol("metadataLog"))
+ val metadataLog = source invokePrivate _metadataLog()
+
+ def verifyBatch(
+ offset: FileStreamSourceOffset,
+ expectedBatchId: Long,
+ inputFiles: Seq[File],
+ expectedFileOffset: Int,
+ expectedFilesInBatch: Int,
+ expectedListingCount: Int): Unit = {
+ val batchId = offset.logOffset
+ assert(batchId === expectedBatchId)
+
+ val files = metadataLog.get(batchId).getOrElse(Array.empty[FileEntry])
+ assert(files.forall(_.batchId == batchId))
+
+ val actualInputFiles = files.map { p => p.sparkPath.toUri.getPath }
+ val expectedInputFiles = inputFiles.slice(
+ expectedFileOffset,
+ expectedFileOffset + expectedFilesInBatch
+ )
+ .map(_.getCanonicalPath)
+ assert(actualInputFiles === expectedInputFiles)
+
+ assert(expectedListingCount === CountListingLocalFileSystem.pathToNumListStatusCalled
+ .get(src.getCanonicalPath).map(_.get()).getOrElse(0))
+ }
+
+ CountListingLocalFileSystem.resetCount()
+
+ // provide 20 files in src, with sequential "last modified" to guarantee ordering
+ val inputFiles = (0 to 19).map { idx =>
+ val f = createFile(idx.toString, new File(src, idx.toString), tmp)
+ f.setLastModified(idx * 10000)
+ f
+ }
+
+ // each batch should perform a listing since maxCachedFiles is set to 0
+ val offsetBatch0 = source.latestOffset(FileStreamSourceOffset(-1L), ReadLimit.maxFiles(10))
+ .asInstanceOf[FileStreamSourceOffset]
+ verifyBatch(offsetBatch0, expectedBatchId = 0, inputFiles,
+ expectedFileOffset = 0, expectedFilesInBatch = 10, expectedListingCount = 1)
+ val offsetBatch1 = source.latestOffset(FileStreamSourceOffset(-1L), ReadLimit.maxFiles(10))
+ .asInstanceOf[FileStreamSourceOffset]
+ verifyBatch(offsetBatch1, expectedBatchId = 1, inputFiles,
+ expectedFileOffset = 10, expectedFilesInBatch = 10, expectedListingCount = 2)
+
+ // validate no remaining files and another listing is performed
+ val offsetBatch = source.latestOffset(FileStreamSourceOffset(-1L), ReadLimit.maxFiles(10))
+ .asInstanceOf[FileStreamSourceOffset]
+ assert(1 === offsetBatch.logOffset)
+ assert(3 === CountListingLocalFileSystem.pathToNumListStatusCalled
+ .get(src.getCanonicalPath).map(_.get()).getOrElse(0))
+ }
+ }
+ }
+
test("SPARK-48314: Don't cache unread files when using Trigger.AvailableNow") {
withCountListingLocalFileSystemAsLocalFileSystem {
withThreeTempDirs { case (src, meta, tmp) =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org