private void flushDeletes() {
    if (posDeletes.isEmpty()) {
      return;
    }

    posDeletes.forEach(
        (filePath, posDeletes) -> {
          if (posDeletes.size() <= 0) {
            return;
          }
          posDeletes.sort(Comparator.comparingLong(PosRow::pos));
          String fileName = TableFileUtil.getFileNameWithoutExt(filePath.get().toString());
          String fileDir = TableFileUtil.getFileDir(filePath.get().toString());
          String deleteFilePath =
              format.addExtension(
                  String.format(
                      "%s/%s",
                      fileDir,
                      TableFileUtil.optimizingPosDeleteFileName(fileName, fileNameSuffix)));
          EncryptedOutputFile outputFile =
              encryptionManager.encrypt(fileIO.newOutputFile(deleteFilePath));

          PositionDeleteWriter<T> writer =
              appenderFactory.newPosDeleteWriter(outputFile, format, partition);
          PositionDelete<T> posDelete = PositionDelete.create();
          try (PositionDeleteWriter<T> closeableWriter = writer) {
            posDeletes.forEach(
                posRow ->
                    closeableWriter.write(
                        posDelete.set(filePath.get(), posRow.pos(), posRow.row())));
          } catch (IOException e) {
            setFailure(e);
            throw new UncheckedIOException(
                "Failed to write the sorted path/pos pairs to pos-delete file: "
                    + outputFile.encryptingOutputFile().location(),
                e);
          }
          // Add the referenced data files.
          referencedDataFiles.addAll(writer.referencedDataFiles());

          // Add the completed delete files.
          completedFiles.add(writer.toDeleteFile());
        });

    // Clear the buffered pos-deletions.
    posDeletes.clear();
  }
Positional delete file writer for iceberg tables. Write to different delete file for every data
file. The output delete files are named with pattern:
{data_file_name}-delete-{delete_file_suffix}.

@param <T> to indicate the record data type.
amoro-format-iceberg/src/main/java/org/apache/amoro/io/writer/IcebergFanoutPosDeleteWriter.java