dongjoon-hyun commented on code in PR #44636:
URL: https://github.com/apache/spark/pull/44636#discussion_r1448536927
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala:
##########
@@ -166,6 +186,23 @@ class FileStreamSource(
// implies "sourceOptions.latestFirst = true" which we want to refresh
the list per batch
(newFiles.take(files.maxFiles()), null)
+ case files: ReadMaxBytes if !sourceOptions.latestFirst =>
+ // we can cache and reuse remaining fetched list of files in further
batches
+ val (bFiles, usFiles) = takeFilesUntilMax(newFiles, files.maxBytes())
+ if (usFiles.map(_.size).sum < files.maxBytes() *
DISCARD_UNSEEN_FILES_RATIO) {
Review Comment:
`takeFilesUntilMax` had better returns the total size together. Then,
- we can avoid this re-computation, `usFiles.map(_.size).sum`.
- we can avoid the chance of overflow here if `takeFilesUntilMax` guarantees
that once.
--
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]
