private Predicate<T> applyEqDeletes() {
    if (eqPredicate != null) {
      return eqPredicate;
    }

    if (eqDeletes.isEmpty()) {
      return record -> false;
    }

    Schema pkSchema = TypeUtil.select(requiredSchema, primaryKeyId);
    // a projection to select and reorder fields of the file schema to match the delete rows
    StructProjection deletePKProjectRow = StructProjection.create(deleteSchema, pkSchema);
    StructProjection dataPKProjectRow = StructProjection.create(requiredSchema, pkSchema);

    Iterable<CloseableIterable<Record>> deleteRecords =
        Iterables.transform(eqDeletes, this::openDeletes);

    // copy the delete records because they will be held in a map
    CloseableIterable<Record> records =
        CloseableIterable.transform(CloseableIterable.concat(deleteRecords), Record::copy);
    if (deleteNodeFilter != null) {
      records = deleteNodeFilter.filter(records);
    }

    InternalRecordWrapper internalRecordWrapper =
        new InternalRecordWrapper(deleteSchema.asStruct());
    CloseableIterable<StructLike> structLikeIterable =
        CloseableIterable.transform(records, internalRecordWrapper::copyFor);

    StructLikeBaseMap<ChangedLsn> structLikeMap =
        structLikeCollections.createStructLikeMap(pkSchema.asStruct());
    // init map
    try (CloseableIterable<StructLike> deletes = structLikeIterable) {
      Iterator<StructLike> it =
          getFileIO() == null ? deletes.iterator() : getFileIO().doAs(deletes::iterator);
      while (it.hasNext()) {
        StructLike structLike = it.next();
        StructLike deletePK = deletePKProjectRow.copyFor(structLike);
        ChangedLsn deleteLsn = deleteLSN(structLike);

        ChangedLsn old = structLikeMap.get(deletePK);
        if (old == null || old.compareTo(deleteLsn) <= 0) {
          structLikeMap.put(deletePK, deleteLsn);
        }
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }

    Predicate<T> isInDeleteSet =
        record -> {
          StructLike data = asStructLike(record);
          StructLike dataPk = dataPKProjectRow.copyFor(data);
          ChangedLsn dataLSN = dataLSN(data);
          ChangedLsn deleteLsn = structLikeMap.get(dataPk);
          if (deleteLsn == null) {
            return false;
          }

          return deleteLsn.compareTo(dataLSN) > 0;
        };

    this.eqPredicate = new CloseablePredicate<>(isInDeleteSet, structLikeMap);
    return isInDeleteSet;
  }
@return The data in equity delete file
amoro-format-iceberg/src/main/java/org/apache/amoro/io/reader/MixedDeleteFilter.java