Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/17012#discussion_r103059294
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
---
@@ -274,6 +274,11 @@ private[state] class HDFSBackedStateStoreProvider(
private def commitUpdates(newVersion: Long, map: MapType, tempDeltaFile:
Path): Path = {
synchronized {
val finalDeltaFile = deltaFile(newVersion)
+ // Renaming a file atop an existing one fails on HDFS, see
+ //
hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/filesystem/filesystem.html
+ if (fs.exists(finalDeltaFile)) {
+ fs.delete(finalDeltaFile, false)
--- End diff --
Actually, I think we should not delete the file. Otherwise, it will break
speculation. E.g., there is also a speculation ask running. It wrote
successfully and just reported to the driver that the task was successful. At
the same time, another task just deleted the final file wrongly because the
file exists.
The correct way is just skipping `rename`. We can depend on the basic
assumption on Structured Streaming that the same batch should contain the same
data.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
