private Predicate<StructForDelete<T>> applyEqDeletesForSchema(
      Map.Entry<Set<Integer>, Schema> deleteSchemaEntry,
      BloomFilter<StructLike> bloomFilter,
      List<Closeable> structMapCloseable) {
    Set<Integer> ids = deleteSchemaEntry.getKey();
    Schema deleteSchema = deleteSchemaEntry.getValue();
    Iterable<DeleteFile> eqDeletes = eqDeleteFilesByDeleteIds.get(ids);

    InternalRecordWrapper internalRecordWrapper =
        new InternalRecordWrapper(deleteSchema.asStruct());

    CloseableIterable<RecordWithLsn> deleteRecords =
        CloseableIterable.transform(
            CloseableIterable.concat(
                Iterables.transform(
                    eqDeletes,
                    s ->
                        CloseableIterable.transform(
                            openFile(s, deleteSchema),
                            r -> new RecordWithLsn(s.dataSequenceNumber(), r)))),
            RecordWithLsn::recordCopy);

    StructLikeBaseMap<Long> structLikeMap =
        structLikeCollections.createStructLikeMap(deleteSchema.asStruct());

    // init map
    try (CloseableIterable<RecordWithLsn> deletes = deleteRecords) {
      Iterator<RecordWithLsn> it =
          getFileIO() == null ? deletes.iterator() : getFileIO().doAs(deletes::iterator);
      while (it.hasNext()) {
        RecordWithLsn recordWithLsn = it.next();
        StructLike deletePK = internalRecordWrapper.copyFor(recordWithLsn.getRecord());
        if (filterEqDelete && !bloomFilter.mightContain(deletePK)) {
          continue;
        }
        Long lsn = recordWithLsn.getLsn();
        Long old = structLikeMap.get(deletePK);
        if (old == null || old.compareTo(lsn) <= 0) {
          structLikeMap.put(deletePK, lsn);
        }
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
    structMapCloseable.add(structLikeMap);

    return structForDelete -> {
      StructProjection deleteProjection =
          StructProjection.create(requiredSchema, deleteSchema).wrap(structForDelete.getPk());
      StructLike dataPk = internalRecordWrapper.copyFor(deleteProjection);
      Long dataLSN = structForDelete.getLsn();
      Long deleteLsn = structLikeMap.get(dataPk);
      if (deleteLsn == null) {
        return false;
      }

      return deleteLsn.compareTo(dataLSN) > 0;
    };
  }
Get all delete ids of equality delete files

<p>For example, if there are two equality delete fields, one is [1, 2] and another is [1], the
delete ids will be [1, 2].

@return delete ids
amoro-format-iceberg/src/main/java/org/apache/amoro/io/reader/CombinedDeleteFilter.java