private <T> Iterable<CloseableIterable<T>> entries(
      BiFunction<ManifestFile, CloseableIterable<ManifestEntry<?>>, CloseableIterable<T>> entryFn) {

    Snapshot snapshot =
        snapshotId != null ? ops.current().snapshot(snapshotId) : ops.current().currentSnapshot();

    // snapshot could be null when the table just gets created
    if (snapshot == null) {
      return CloseableIterable.empty();
    }

    List<ManifestFile> manifests;
    if (manifestContent == null) {
      manifests = snapshot.allManifests(io);
    } else if (manifestContent == ManifestContent.DATA) {
      manifests = snapshot.dataManifests(io);
    } else {
      manifests = snapshot.deleteManifests(io);
    }

    LoadingCache<Integer, ManifestEvaluator> evalCache =
        specsById == null
            ? null
            : Caffeine.newBuilder()
                .build(
                    specId -> {
                      PartitionSpec spec = specsById.get(specId);
                      Expression partitionFilter = partitionFilter(specId);
                      return ManifestEvaluator.forPartitionFilter(
                          Expressions.and(
                              partitionFilter,
                              Projections.inclusive(spec, caseSensitive).project(dataFilter)),
                          spec,
                          caseSensitive);
                    });

    Evaluator evaluator;
    if (fileFilter != null && fileFilter != Expressions.alwaysTrue()) {
      evaluator = new Evaluator(DataFile.getType(EMPTY_STRUCT), fileFilter, caseSensitive);
    } else {
      evaluator = null;
    }

    CloseableIterable<ManifestFile> closeableDataManifests =
        CloseableIterable.withNoopClose(manifests);
    CloseableIterable<ManifestFile> matchingManifests =
        evalCache == null
            ? closeableDataManifests
            : CloseableIterable.filter(
                scanMetrics.skippedDataManifests(),
                closeableDataManifests,
                manifest -> evalCache.get(manifest.partitionSpecId()).eval(manifest));

    if (ignoreDeleted) {
      // only scan manifests that have entries other than deletes
      // remove any manifests that don't have any existing or added files. if either the added or
      // existing files count is missing, the manifest must be scanned.
      matchingManifests =
          CloseableIterable.filter(
              scanMetrics.skippedDataManifests(),
              matchingManifests,
              manifest -> manifest.hasAddedFiles() || manifest.hasExistingFiles());
    }

    if (ignoreExisting) {
      // only scan manifests that have entries other than existing
      // remove any manifests that don't have any deleted or added files. if either the added or
      // deleted files count is missing, the manifest must be scanned.
      matchingManifests =
          CloseableIterable.filter(
              scanMetrics.skippedDataManifests(),
              matchingManifests,
              manifest -> manifest.hasAddedFiles() || manifest.hasDeletedFiles());
    }

    matchingManifests =
        CloseableIterable.filter(
            scanMetrics.skippedDataManifests(), matchingManifests, manifestPredicate);
    matchingManifests =
        CloseableIterable.count(scanMetrics.scannedDataManifests(), matchingManifests);

    return Iterables.transform(
        matchingManifests,
        manifest ->
            new CloseableIterable<T>() {
              private CloseableIterable<T> iterable;

              @Override
              public CloseableIterator<T> iterator() {
                ManifestReader<?> reader =
                    ManifestFiles.open(manifest, io, specsById)
                        .filterRows(dataFilter)
                        .filterPartitions(partitionSet)
                        .caseSensitive(caseSensitive)
                        .select(columns)
                        .scanMetrics(scanMetrics);

                CloseableIterable<? extends ManifestEntry<?>> tmpEntries;
                if (ignoreDeleted) {
                  tmpEntries = reader.liveEntries();
                } else {
                  tmpEntries = reader.entries();
                }

                CloseableIterable<ManifestEntry<?>> entries =
                    CloseableIterable.transform(tmpEntries, ManifestEntry.class::cast);

                if (ignoreExisting) {
                  entries =
                      CloseableIterable.filter(
                          scanMetrics.skippedDataFiles(),
                          entries,
                          entry -> entry.status() != ManifestEntry.Status.EXISTING);
                }

                if (evaluator != null) {
                  entries =
                      CloseableIterable.filter(
                          scanMetrics.skippedDataFiles(),
                          entries,
                          entry -> evaluator.eval((GenericDataFile) entry.file()));
                }

                iterable = entryFn.apply(manifest, entries);

                return iterable.iterator();
              }

              @Override
              public void close() throws IOException {
                if (iterable != null) {
                  iterable.close();
                }
              }
            });
  }
Base results on files in the snapshot that was current as of a timestamp.

@param timestampMillis a timestamp in milliseconds
@return this for method chaining
amoro-format-iceberg/src/main/java/org/apache/iceberg/IcebergFindFiles.java