API, Core: Support onKeep/onSkip callback for FilterIterator

api/src/main/java/org/apache/iceberg/io/CloseableIterable.java

@@ -23,6 +23,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import org.apache.iceberg.exceptions.RuntimeIOException;
@@ -41,7 +42,7 @@
   static <E> CloseableIterable<E> of(Iterable<E> iterable) {
     if (iterable instanceof CloseableIterable) {
       return (CloseableIterable<E>) iterable;
     } else if (iterable instanceof Closeable) {
       Closeable asCloseable = (Closeable) iterable;
       return combine(iterable, asCloseable);
     } else {
@@ -146,18 +147,42 @@
   static <E> CloseableIterable<E> filter(
       Counter skipCounter, CloseableIterable<E> iterable, Predicate<E> pred) {
     Preconditions.checkArgument(null != skipCounter, "Invalid counter: null");
+    return filter(iterable, pred, item -> {}, item -> skipCounter.increment());
+  }
+
+  /**
+   * Filters the given {@link CloseableIterable} with the given predicate, invoking {@code onKeep}
+   * on each accepted element and {@code onSkip} on each rejected element exactly once.
+   *
+   * @param iterable The underlying {@link CloseableIterable} to filter.
+   * @param <E> The underlying type to be iterated.
+   * @param pred The {@link Predicate} that selects elements to keep.
+   * @param onKeep The {@link Consumer} invoked once per element that satisfies {@code pred}.
+   * @param onSkip The {@link Consumer} invoked once per element that does not satisfy {@code pred}.
+   * @return A filtered {@link CloseableIterable} that fires the observers as elements are consumed.
+   */
+  static <E> CloseableIterable<E> filter(
+      CloseableIterable<E> iterable, Predicate<E> pred, Consumer<E> onKeep, Consumer<E> onSkip) {
     Preconditions.checkArgument(null != iterable, "Invalid iterable: null");
     Preconditions.checkArgument(null != pred, "Invalid predicate: null");
+    Preconditions.checkArgument(null != onKeep, "Invalid onKeep consumer: null");
+    Preconditions.checkArgument(null != onSkip, "Invalid onSkip consumer: null");
     return combine(
         () ->
             new FilterIterator<E>(iterable.iterator()) {
               @Override
               protected boolean shouldKeep(E item) {
-                boolean matches = pred.test(item);
-                if (!matches) {
-                  skipCounter.increment();
-                }
-                return matches;
+                return pred.test(item);
+              }
+
+              @Override
+              protected void onKeep(E item) {
+                onKeep.accept(item);
+              }
+
+              @Override
+              protected void onSkip(E item) {
+                onSkip.accept(item);
               }
             },
         iterable);
@@ -175,18 +200,7 @@
    */
   static <T> CloseableIterable<T> count(Counter counter, CloseableIterable<T> iterable) {
     Preconditions.checkArgument(null != counter, "Invalid counter: null");
-    Preconditions.checkArgument(null != iterable, "Invalid iterable: null");
-    return new CloseableIterable<T>() {
-      @Override
-      public CloseableIterator<T> iterator() {
-        return CloseableIterator.count(counter, iterable.iterator());
-      }
-
-      @Override
-      public void close() throws IOException {
-        iterable.close();
-      }
-    };
+    return filter(iterable, item -> true, item -> counter.increment(), item -> {});
   }
 
   static <I, O> CloseableIterable<O> transform(

api/src/main/java/org/apache/iceberg/io/FilterIterator.java

@@ -44,6 +44,12 @@ protected FilterIterator(Iterator<T> items) {
 
   protected abstract boolean shouldKeep(T item);
 
+  /** Invoked exactly once for each element accepted by {@link #shouldKeep(Object)}. */
+  protected void onKeep(T item) {}
+
+  /** Invoked exactly once for each element rejected by {@link #shouldKeep(Object)}. */
+  protected void onSkip(T item) {}
+
   @Override
   public boolean hasNext() {
     return nextReady || advance();
@@ -56,6 +62,7 @@ public T next() {
     }
 
     this.nextReady = false;
+    onKeep(next);
 
     return next;
   }
@@ -67,6 +74,7 @@ private boolean advance() {
         this.nextReady = true;
         return true;
       }
+      onSkip(next);
     }
 
     close();

api/src/test/java/org/apache/iceberg/io/TestCloseableIterable.java

@@ -264,6 +264,85 @@ public void countSkippedNullCheck() {
         .hasMessage("Invalid predicate: null");
   }
 
+  @Test
+  public void filterInvokesObserversOnMatchingItems() {
+    Counter kept = new DefaultMetricsContext().counter("kept");
+    Counter skipped = new DefaultMetricsContext().counter("skipped");
+    List<Integer> keptItems = Lists.newArrayList();
+    List<Integer> skippedItems = Lists.newArrayList();
+    CloseableIterable<Integer> items =
+        CloseableIterable.filter(
+            CloseableIterable.withNoopClose(Arrays.asList(1, 2, 3, 4, 5)),
+            x -> x % 2 == 0,
+            x -> {
+              kept.increment();
+              keptItems.add(x);
+            },
+            x -> {
+              skipped.increment();
+              skippedItems.add(x);
+            });
+
+    assertThat(items).containsExactly(2, 4);
+
+    assertThat(kept.value()).isEqualTo(2);
+    assertThat(keptItems).containsExactly(2, 4);
+
+    assertThat(skipped.value()).isEqualTo(3);
+    assertThat(skippedItems).containsExactly(1, 3, 5);
+  }
+
+  @Test
+  public void filterObserversFireExactlyOncePerElement() throws IOException {
+    Counter kept = new DefaultMetricsContext().counter("kept");
+    Counter skipped = new DefaultMetricsContext().counter("skipped");
+    CloseableIterable<Integer> items =
+        CloseableIterable.filter(
+            CloseableIterable.withNoopClose(Arrays.asList(1, 2, 3, 4)),
+            x -> x % 2 == 0,
+            x -> kept.increment(),
+            x -> skipped.increment());
+
+    try (CloseableIterator<Integer> iter = items.iterator()) {
+      // hasNext advances past the odd 1 (skip counted) and prefetches the even 2 (not yet counted)
+      assertThat(iter.hasNext()).isTrue();
+      assertThat(skipped.value()).isEqualTo(1);
+      assertThat(kept.value()).isZero();
+
+      // next delivers the prefetched 2 and increments kept
+      assertThat(iter.next()).isEqualTo(2);
+      assertThat(kept.value()).isEqualTo(1);
+
+      // second hasNext skips past 3 and prefetches 4; kept still 1 until next() is called
+      assertThat(iter.hasNext()).isTrue();
+      assertThat(skipped.value()).isEqualTo(2);
+      assertThat(kept.value()).isEqualTo(1);
+    }
+  }
+
+  @Test
+  public void filterObserversNullCheck() {
+    Predicate<Integer> pred = x -> true;
+    assertThatThrownBy(() -> CloseableIterable.filter(null, pred, x -> {}, x -> {}))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Invalid iterable: null");
+
+    assertThatThrownBy(
+            () -> CloseableIterable.filter(CloseableIterable.empty(), null, x -> {}, x -> {}))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Invalid predicate: null");
+
+    assertThatThrownBy(
+            () -> CloseableIterable.filter(CloseableIterable.empty(), pred, null, x -> {}))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Invalid onKeep consumer: null");
+
+    assertThatThrownBy(
+            () -> CloseableIterable.filter(CloseableIterable.empty(), pred, x -> {}, null))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Invalid onSkip consumer: null");
+  }
+
   @Test
   public void transformNullCheck() {
     assertThatThrownBy(() -> CloseableIterable.transform(CloseableIterable.empty(), null))

core/src/main/java/org/apache/iceberg/DeleteFileIndex.java

@@ -33,6 +33,7 @@
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.apache.iceberg.exceptions.RuntimeIOException;
@@ -601,19 +602,19 @@ private Iterable<CloseableIterable<ManifestEntry<DeleteFile>>> deleteManifestRea
 
       CloseableIterable<ManifestFile> closeableDeleteManifests =
           CloseableIterable.withNoopClose(deleteManifests);
-      CloseableIterable<ManifestFile> matchingManifests =
+      Predicate<ManifestFile> manifestPredicate =
           evalCache == null
-              ? closeableDeleteManifests
-              : CloseableIterable.filter(
-                  scanMetrics.skippedDeleteManifests(),
-                  closeableDeleteManifests,
-                  manifest ->
-                      manifest.content() == ManifestContent.DELETES
-                          && (manifest.hasAddedFiles() || manifest.hasExistingFiles())
-                          && evalCache.get(manifest.partitionSpecId()).eval(manifest));
-
-      matchingManifests =
-          CloseableIterable.count(scanMetrics.scannedDeleteManifests(), matchingManifests);
+              ? manifest -> true
+              : manifest ->
+                  manifest.content() == ManifestContent.DELETES
+                      && (manifest.hasAddedFiles() || manifest.hasExistingFiles())
+                      && evalCache.get(manifest.partitionSpecId()).eval(manifest);
+      CloseableIterable<ManifestFile> matchingManifests =
+          CloseableIterable.filter(
+              closeableDeleteManifests,
+              manifestPredicate,
+              manifest -> scanMetrics.scannedDeleteManifests().increment(),
+              manifest -> scanMetrics.skippedDeleteManifests().increment());
       return Iterables.transform(
           matchingManifests,
           manifest ->

core/src/main/java/org/apache/iceberg/ManifestGroup.java

@@ -300,38 +300,31 @@ private <T> Iterable<CloseableIterable<T>> entries(
 
     CloseableIterable<ManifestFile> closeableDataManifests =
         CloseableIterable.withNoopClose(dataManifests);
+    Predicate<ManifestFile> manifestPredicate =
+        manifest -> {
+          if (evalCache != null && !evalCache.get(manifest.partitionSpecId()).eval(manifest)) {
+            return false;
+          }
+          // 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.
+          if (ignoreDeleted && !(manifest.hasAddedFiles() || manifest.hasExistingFiles())) {
+            return false;
+          }
+          // 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.
+          if (ignoreExisting && !(manifest.hasAddedFiles() || manifest.hasDeletedFiles())) {
+            return false;
+          }
+          return true;
+        };
     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.count(scanMetrics.scannedDataManifests(), matchingManifests);
+        CloseableIterable.filter(
+            closeableDataManifests,
+            manifestPredicate,
+            manifest -> scanMetrics.scannedDataManifests().increment(),
+            manifest -> scanMetrics.skippedDataManifests().increment());
 
     return Iterables.transform(
         matchingManifests,

core/src/main/java/org/apache/iceberg/PositionDeletesTable.java

@@ -290,13 +290,12 @@ protected CloseableIterable<ScanTask> doPlanFiles() {
 
       CloseableIterable<ManifestFile> matchingManifests =
           CloseableIterable.filter(
-              scanMetrics().skippedDeleteManifests(),
               CloseableIterable.withNoopClose(manifests),
               manifest ->
                   baseTableEvalCache.get(manifest.partitionSpecId()).eval(manifest)
-                      && deletesTableEvalCache.get(manifest.partitionSpecId()).eval(manifest));
-      matchingManifests =
-          CloseableIterable.count(scanMetrics().scannedDeleteManifests(), matchingManifests);
+                      && deletesTableEvalCache.get(manifest.partitionSpecId()).eval(manifest),
+              manifest -> scanMetrics().scannedDeleteManifests().increment(),
+              manifest -> scanMetrics().skippedDeleteManifests().increment());
 
       Iterable<CloseableIterable<ScanTask>> tasks =
           CloseableIterable.transform(