@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();
              }
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