rich7420 opened a new issue, #7594:
URL: https://github.com/apache/datafusion-comet/issues/7594
### Describe the bug
`CometNativeShuffleWriter` builds its temp data file path with
`dataFile.getPath.replace(".data", ".data.tmp")`
([CometNativeShuffleWriter.scala#L121](https://github.com/apache/datafusion-comet/blob/8c145b742b1501087f5aef204bca4dd5d7f90a9e/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala#L121)),
and nothing deletes that file when the task fails.
1. `String.replace` replaces every `.data` in the path, not just the
extension. If the local dir path contains `.data`, the temp path points into a
directory that doesn't exist, and every native shuffle write fails:
```
org.apache.comet.CometNativeException: shuffle write error: Os { code: 2,
kind: NotFound, message: "No such file or directory" }
```
2. The native writer creates the temp file before it reads any input
([local_partition_writer.rs#L113-L118](https://github.com/apache/datafusion-comet/blob/8c145b742b1501087f5aef204bca4dd5d7f90a9e/native/shuffle/src/writers/local/local_partition_writer.rs#L113-L118)).
When the map task fails, `write` and `stop(false)` clean up only the remote
output, so the file stays in the block manager directory. `unregisterShuffle`
removes only the data, index and checksum files, so the temp file isn't removed
with the shuffle either. The files I saw were empty, but with a single output
partition, the writer writes to the file while it reads input
([single_partition.rs#L59-L71](https://github.com/apache/datafusion-comet/blob/8c145b742b1501087f5aef204bca4dd5d7f90a9e/native/shuffle/src/partitioners/single_partition.rs#L59-L71)),
so a failure there can leave data behind. Spark's
`BypassMergeSortShuffleWriter` and `UnsafeShuffleWriter` delete their temp file
through `ShuffleMapOutputWriter.abort`.
### Steps to reproduce
Local mode, with Comet shuffle enabled and `spark.comet.shuffle.mode=native`:
1. Set `spark.local.dir` to a directory named `local.data` and run:
```scala
spark.range(0, 1000, 1, 2).write.parquet(path)
spark.read.parquet(path).repartition(4, col("id")).collect()
```
The job fails with the error above. With a directory named `local-plain`,
it returns all 1000 rows through a `CometNativeShuffle` exchange.
2. With a plain local dir, make the map task fail:
```scala
spark.conf.set("spark.sql.ansi.enabled", "true")
spark.read.parquet(path)
.selectExpr("id", "cast(case when id = 700 then 'x' else cast(id as
string) end as int) as v")
.repartition(4, col("id"))
.collect()
```
The job fails with `CAST_INVALID_INPUT` as expected, and
`blockmgr-*/*/shuffle_0_3_0.data.tmp` is still in the local dir afterwards. A
UDF that throws leaves a temp file the same way.
Reproduced at 4de670439 on the default Spark 4.1 profile. The linked files
are unchanged on main.
### Expected behavior
The temp file is created next to the data file regardless of the local dir
path, and is deleted when the map task fails.
### Additional context
Both problems involve the same temp file, so one change should cover them:
create the temp file with `IndexShuffleBlockResolver.createTempFile`, which
Spark's `LocalDiskShuffleMapOutputWriter` uses, and delete it when `write`
fails. I can open a PR with tests for both cases.
--
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]
