diff --git a/api/src/main/java/org/apache/iceberg/io/CloseableIterable.java b/api/src/main/java/org/apache/iceberg/io/CloseableIterable.java index 34c561bc373d..294146e39dec 100644 --- a/api/src/main/java/org/apache/iceberg/io/CloseableIterable.java +++ b/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; @@ -146,18 +147,42 @@ protected boolean shouldKeep(E item) { static CloseableIterable filter( Counter skipCounter, CloseableIterable iterable, Predicate 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 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 CloseableIterable filter( + CloseableIterable iterable, Predicate pred, Consumer onKeep, Consumer 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(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 @@ protected boolean shouldKeep(E item) { */ static CloseableIterable count(Counter counter, CloseableIterable iterable) { Preconditions.checkArgument(null != counter, "Invalid counter: null"); - Preconditions.checkArgument(null != iterable, "Invalid iterable: null"); - return new CloseableIterable() { - @Override - public CloseableIterator 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 CloseableIterable transform( diff --git a/api/src/main/java/org/apache/iceberg/io/FilterIterator.java b/api/src/main/java/org/apache/iceberg/io/FilterIterator.java index f721e19dd106..686d05fcdaf7 100644 --- a/api/src/main/java/org/apache/iceberg/io/FilterIterator.java +++ b/api/src/main/java/org/apache/iceberg/io/FilterIterator.java @@ -44,6 +44,12 @@ protected FilterIterator(Iterator 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(); diff --git a/api/src/test/java/org/apache/iceberg/io/TestCloseableIterable.java b/api/src/test/java/org/apache/iceberg/io/TestCloseableIterable.java index 9b6ff64afd84..adc42ecf7495 100644 --- a/api/src/test/java/org/apache/iceberg/io/TestCloseableIterable.java +++ b/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 keptItems = Lists.newArrayList(); + List skippedItems = Lists.newArrayList(); + CloseableIterable 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 items = + CloseableIterable.filter( + CloseableIterable.withNoopClose(Arrays.asList(1, 2, 3, 4)), + x -> x % 2 == 0, + x -> kept.increment(), + x -> skipped.increment()); + + try (CloseableIterator 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 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)) diff --git a/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java b/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java index 872fcd212b8a..1b216e802a53 100644 --- a/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java +++ b/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>> deleteManifestRea CloseableIterable closeableDeleteManifests = CloseableIterable.withNoopClose(deleteManifests); - CloseableIterable matchingManifests = + Predicate 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 matchingManifests = + CloseableIterable.filter( + closeableDeleteManifests, + manifestPredicate, + manifest -> scanMetrics.scannedDeleteManifests().increment(), + manifest -> scanMetrics.skippedDeleteManifests().increment()); return Iterables.transform( matchingManifests, manifest -> diff --git a/core/src/main/java/org/apache/iceberg/ManifestGroup.java b/core/src/main/java/org/apache/iceberg/ManifestGroup.java index 39b66d558b34..4c6bba065b5a 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestGroup.java +++ b/core/src/main/java/org/apache/iceberg/ManifestGroup.java @@ -300,38 +300,31 @@ private Iterable> entries( CloseableIterable closeableDataManifests = CloseableIterable.withNoopClose(dataManifests); + Predicate 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 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, diff --git a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java index e205e6a9426e..f33458a70531 100644 --- a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java +++ b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java @@ -290,13 +290,12 @@ protected CloseableIterable doPlanFiles() { CloseableIterable 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> tasks = CloseableIterable.transform(