API, Core: Support onKeep/onSkip callback for FilterIterator#17118
API, Core: Support onKeep/onSkip callback for FilterIterator#17118dramaticlly wants to merge 2 commits into
Conversation
Add a new filter overload that invokes a Consumer on each accepted and rejected element, so callers can attribute both survivors and skipped entries to metrics in a single pass. The counter-based filter and count helpers delegate to this new primitive. Migrate the manifest-level scan pipelines in DeleteFileIndex, PositionDeletesTable, and ManifestGroup — each previously chained a skip-counting filter with a scan-counting count call — to a single filter call with both onKeep and onSkip observers. File-level filters in ManifestReader and ManifestGroup remain on the counter-based overload since ScanMetrics only exposes a skipped counter at that layer. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
FYI @stevenzwu @anoopj @nastra if you want to take a look. |
| iterable.close(); | ||
| } | ||
| }; | ||
| return filter(iterable, item -> true, item -> counter.increment(), item -> {}); |
There was a problem hiding this comment.
count now delegates to the FilterIterator-based filter, which increments during hasNext()/advance() rather than on next() like CloseableIterator.count. The javadoc still says the counter is incremented on each Iterator#next() call, so peeking via hasNext() without a following next() over-counts. Update the javadoc, or keep count on CloseableIterator.count.
There was a problem hiding this comment.
Thanks @wombatu-kun for the review. I updated to change FilterIterator directly so that we keep exactly once semantics for skip on hasNext() and kept on next(), same as before.
Move the keep/skip observation into FilterIterator via overridable onKeep(T) and onSkip(T) hook methods with no-op defaults. onSkip fires inside advance() right after shouldKeep returns false, onKeep fires inside next() right before the accepted element is delivered — matching the historical timing of the counter-based helpers. This aligns with what Ryan Blue proposed on the original PR (apache#5268 review comment r929192438): put counter support directly in FilterIterator rather than layering it via predicates or wrapping iterators. It also lets CloseableIterable.filter drop its outer next()-intercepting wrapper.
| this.nextReady = true; | ||
| return true; | ||
| } | ||
| onSkip(next); |
There was a problem hiding this comment.
@rdblue I think this align with your previous suggestion in #5268 (comment). A custom callback enable different counter for v4 heterogenous entry of trackedFile
dramaticlly
changed the title
|
If this is mainly for v4, maybe we should wait until the basic plumbing of write and read get in first. Then we will have a better idea on how other pieces (like this) fit in. |
3 participants
What
Add a new filter overload that invokes a Consumer on each accepted and rejected element, so callers can attribute both kept and skipped entries to metrics in a single pass.
Why
This help provide a path for hetergeneous entry of
TrackedFilein v4, where a given entry can carry contentType of DATA, EQUALITY_DELETES, DATA_MANIFEST, or DELETE_MANIFEST within a single root manifest. Callback is better suited than a dedicated counter to allow more flexibility. Plus, importorg.apache.iceberg.metrics.Counterinorg.apache.iceberg.io. CloseableIterableis not ideal for a general purpose iteration utility.Existing caller
Migrate the manifest-level scan pipelines in
DeleteFileIndex,PositionDeletesTable, andManifestGroupwhere each previously chained a skip-counting filter with a scan-counting count call. Now replaced with a single filter call with both onKeep and onSkip callback. File-level filters in ManifestReader and ManifestGroup remain on the counter-based overload since ScanMetrics only exposes a skipped counter at that layer.