Comet writer doesn't respect `OVERWRITE` mode

[comphead]

Describe the bug

Comet fails if the folder exists and cannot overwrite it

Steps to reproduce

Folder output2 exists

spark.read.parquet("/tmp/writertest/input").repartition(1).write.mode("overwrite").parquet("/tmp/writertest/output2")

25/12/23 08:31:40 ERROR CometExecIterator: Native execution for task 17 failed
org.apache.comet.CometNativeException: Failed to create output file '/tmp/writertest/output2/part-00000-00017.parquet': Is a directory (os error 21)

Expected behavior

Should overwrite the folder with new content

Additional context

No response

[coderfender]

Taking a look at this

[comphead]

some more details on implementation

The flow is like

    ↓
DataFrameWriter.saveInternal() 
    ↓
InsertIntoHadoopFsRelationCommand.run()
    ↓
[Line 131] deleteMatchingPartitions(fs, qualifiedOutputPath, customPartitionLocations, committer)
    ↓
[Line 238] committer.deleteWithJob(fs, staticPrefixPath, true)
    ↓
[Line 183] fs.delete(path, recursive=true)  ← ACTUAL DELETION HAPPENS HERE

This Spark implementation now

    val doInsertion = if (mode == SaveMode.Append) {
      true
    } else {
      val pathExists = fs.exists(qualifiedOutputPath)
      (mode, pathExists) match {
        case (SaveMode.ErrorIfExists, true) =>
          throw QueryCompilationErrors.outputPathAlreadyExistsError(qualifiedOutputPath)
        case (SaveMode.Overwrite, true) =>
          if (ifPartitionNotExists && matchingPartitions.nonEmpty) {
            false
          } else if (dynamicPartitionOverwrite) {
            // For dynamic partition overwrite, do not delete partition directories ahead.
            true
          } else {
            deleteMatchingPartitions(fs, qualifiedOutputPath, customPartitionLocations, committer)
            true
          }
        case (SaveMode.Overwrite, _) | (SaveMode.ErrorIfExists, false) =>
          true
        case (SaveMode.Ignore, exists) =>
          !exists
        case (s, exists) =>
          throw QueryExecutionErrors.saveModeUnsupportedError(s, exists)
      }
    }

The keypoint here, we need to support dynamicPartitionOverwrite or fallback to Spark in this case

[coderfender]

take