andygrove opened a new issue, #6865: URL: https://github.com/apache/datafusion-comet/issues/6865
### Describe the bug On the shuffle direct-read path, `CometBlockStoreShuffleReader.readAsRawStream()` joins the fetched block streams with a `java.io.SequenceInputStream`. Its `close()` asks the enumeration for every remaining element and closes each one before it returns, as its Javadoc says. Here the enumeration is Spark's `ShuffleBlockFetcherIterator`, so each remaining element is a `next()` call. That waits in `results.take()` for the block, then sends more remote requests through `fetchUpToMaxBytes()`. `CometExecIterator.close()` closes the stream through `CometShuffleBlockIterator.close()`, either when the plan runs out of output or from its task completion listener. That listener is registered after the fetcher's own cleanup listener in `CometExecRDD.compute`, and Spark runs completion listeners in reverse order. So the fetcher is still live whenever the stream is closed, and the drain really fetches. Whenever a native plan stops reading a shuffle input before its end, closing it fetches the rest of that reduce partition only to close it. That happens when: - a limit is reached before the input runs out, - the task fails, which delays the failure report, - the task is killed without a thread interrupt. With #6805, a killed task on this path stops at its next read, but it then fetches the remaining blocks during cleanup. An interrupting kill is not affected, because the first `results.take()` throws on the interrupt flag. In the limit case, a fetch failure during the drain fails a task that had already finished its work. ### Steps to reproduce I found this by reading the code and haven't reproduced it. In local mode every block is local, and draining a local block only opens and closes a file, so the cost only shows with remote blocks. A deterministic test could read one block from the stream that `readAsRawStream()` returns for a reduce partition with several blocks, close it, and count how many times the fetch iterator's `next()` was called. ### Expected behavior Closing the direct-read stream closes the current block's stream and asks the fetch iterator for nothing more. The fetcher's own completion listener then releases whatever is still buffered without fetching more. That is what the JVM decode path in `CometBlockStoreShuffleReader.read()` does, and what Spark's shuffle reader does. One way is to have the enumeration report no more elements once `close()` has been called. ### Additional context - The Celeborn raw stream's `close()` already closes only its current reducer. - `branch-1.1` has the same `readAsRawStream()`, so 1.1.0 is affected too. - I found this while reviewing #6805. That PR doesn't change it. -- 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]
