public static int deleteFiles(
      AuthenticatedFileIO io, Set<String> files, ExecutorService workPool) {
    if (files == null || files.isEmpty()) {
      return 0;
    }

    AtomicInteger failedFileCnt = new AtomicInteger(0);
    if (io.supportBulkOperations()) {
      try {
        io.asBulkFileIO().deleteFiles(files);
      } catch (BulkDeletionFailureException e) {
        failedFileCnt.set(e.numberFailedObjects());
        LOG.warn("Failed to bulk delete {} files", e.numberFailedObjects(), e);
      } catch (RuntimeException e) {
        failedFileCnt.set(files.size());
        LOG.warn("Failed to bulk delete files", e);
      }
    } else {
      if (workPool != null) {
        Tasks.foreach(files)
            .executeWith(workPool)
            .noRetry()
            .suppressFailureWhenFinished()
            .onFailure(
                (file, exc) -> {
                  failedFileCnt.addAndGet(1);
                  LOG.warn("Failed to delete file {}", file, exc);
                })
            .run(io::deleteFile);
      } else {
        files.forEach(
            f -> {
              try {
                io.deleteFile(f);
              } catch (RuntimeException e) {
                failedFileCnt.addAndGet(1);
                LOG.warn("Failed to delete file {}", f, e);
              }
            });
      }
    }

    return files.size() - failedFileCnt.get();
  }
Helper to delete files. Bulk deletion is used if possible.

@param io mixed-format file io
@param files files to delete
@param workPool executor pool. Only applicable for non-bulk FileIO
@return deleted file count
amoro-format-iceberg/src/main/java/org/apache/amoro/utils/TableFileUtil.java