stevenzwu commented on code in PR #18147:
URL: https://github.com/apache/iceberg/pull/18147#discussion_r4031441653
##########
core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java:
##########
@@ -732,6 +739,274 @@ public void statsFilterMissingColumnFailure() {
.hasMessageContaining("Cannot find field 'missing' in struct: %s",
TABLE_SCHEMA.asStruct());
}
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void statsFilterRecordCountFiltering(FileFormat format) throws
IOException {
+ TrackedFile emptyTrackedFile =
+ new TrackedFileStruct(
+ ADDED_TRACKING,
+ FileContent.DATA,
+ FORMAT_VERSION_V4,
+ "s3://bucket/table/empty-file.parquet",
+ FileFormat.PARQUET,
+ 0, // file contains no records
+ 100L,
+ null,
+ null,
+ null,
+ SortOrder.unsorted().orderId(),
+ null,
+ null,
+ null,
+ List.of(4L),
+ null);
+
+ ManifestFile manifest =
+ writeManifest(format, UNPARTITIONED_TYPE,
ImmutableList.of(emptyTrackedFile, FILE_D));
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+
+ List<TrackedFile> actualFiles = read(builder);
+
+ assertThat(actualFiles)
+ .usingComparatorForType(FILE_COMPARATOR, TrackedFile.class)
+ .containsExactly(FILE_D);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void statsFilterInvalidRecordCountNotFiltered(FileFormat format)
throws IOException {
+ // A bug in old writers produced Avro files with record_count=-1 (unknown)
+ TrackedFile invalidRecordCountFile =
+ new TrackedFileStruct(
+ ADDED_TRACKING,
+ FileContent.DATA,
+ FORMAT_VERSION_V4,
+ "s3://bucket/table/very-old.avro",
+ FileFormat.AVRO,
+ -1, // mimic invalid record count in old Avro metadata
+ 100L,
+ null,
+ null,
+ null,
+ SortOrder.unsorted().orderId(),
+ null,
+ null,
+ null,
+ List.of(4L),
+ null);
+
+ ManifestFile manifest =
+ writeManifest(format, UNPARTITIONED_TYPE,
ImmutableList.of(invalidRecordCountFile, FILE_D));
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+
+ List<TrackedFile> actualFiles = read(builder);
+
+ assertThat(actualFiles)
+ .usingComparatorForType(FILE_COMPARATOR, TrackedFile.class)
+ .containsExactly(invalidRecordCountFile, FILE_D);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void statsFilterDataFileBoundsFiltering(FileFormat format) throws
IOException {
+ // FILE_C has stats {id in [0, 99], data in [a, z]}, FILE_D has no stats
+ ManifestFile manifest =
+ writeManifest(format, UNPARTITIONED_TYPE, ImmutableList.of(FILE_C,
FILE_D));
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .filter(Expressions.equal("id", 105)) // eliminates FILE_C
+ .metricsConfig(METRICS_CONFIG);
+
+ List<TrackedFile> actualFiles = read(builder);
+
+ assertThat(actualFiles)
+ .usingComparatorForType(FILE_COMPARATOR, TrackedFile.class)
+ .containsExactly(FILE_D);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void statsFilterManifestBoundsFiltering(FileFormat format) throws
IOException {
+ // DATA_MANIFEST_WITH_STATS_REF has stats {id in [0, 99], data in [a, z]}
+ ManifestFile manifest =
+ writeManifest(
+ format,
+ UNPARTITIONED_TYPE,
+ ImmutableList.of(DATA_MANIFEST_REF, DATA_MANIFEST_WITH_STATS_REF));
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .filter(Expressions.equal("id", 105)) // eliminates the manifest
with stats
+ .metricsConfig(METRICS_CONFIG);
+
+ List<TrackedFile> actualFiles = read(builder);
+
+ assertThat(actualFiles)
+ .usingComparatorForType(FILE_COMPARATOR, TrackedFile.class)
+ .containsExactly(DATA_MANIFEST_REF);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void statsFilterBoundsFilteringWithForScanPlanning(FileFormat format)
throws IOException {
+ // FILE_C has stats {id in [0, 99], data in [a, z]}, FILE_D has no stats
+ ManifestFile manifest =
+ writeManifest(format, UNPARTITIONED_TYPE, ImmutableList.of(FILE_C,
FILE_D));
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .forScanPlanning() // does not project unused stats
+ .filter(Expressions.equal("id", 105)) // eliminates FILE_C
+ .metricsConfig(METRICS_CONFIG);
+
+ List<TrackedFile> actualFiles = read(builder);
+
+ assertThat(actualFiles)
+ .usingComparatorForType(FILE_COMPARATOR, TrackedFile.class)
+ .containsExactly(FILE_D);
+ }
+
+ @ParameterizedTest
+ @FieldSource("PROJECTION_CASES")
+ public void
statsFilterBoundsFilteringWithProjection(Consumer<V4ManifestReader.Builder>
config)
+ throws IOException {
+ // FILE_C has stats {id in [0, 99], data in [a, z]}, FILE_D has no stats
+ // config projects just the file location, but stats are automatically
projected for the filter
+ ManifestFile manifest =
+ writeManifest(FileFormat.PARQUET, UNPARTITIONED_TYPE,
ImmutableList.of(FILE_C, FILE_D));
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .filter(Expressions.equal("id", 105)) // eliminates FILE_C
+ .metricsConfig(METRICS_CONFIG);
+
+ config.accept(builder);
+
+ List<TrackedFile> actualFiles = read(builder);
+
+
assertThat(actualFiles).extracting(TrackedFile::location).containsExactly(FILE_D.location());
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void statsFilterCaseSensitivity(FileFormat format) throws IOException
{
Review Comment:
nit: should we call this `statsFilterCaseInsensitive`?
##########
core/src/main/java/org/apache/iceberg/V4ManifestReader.java:
##########
@@ -97,18 +101,30 @@ Schema readSchema() {
/** Returns copies of the tracked files that match this reader's configured
filters. */
@Override
public CloseableIterator<TrackedFile> iterator() {
- CloseableIterable<TrackedFile> entries =
CloseableIterable.transform(open(), this::prepare);
- if (!partitionFilters.isEmpty()) {
- // manifests have no partition, so the partition filter cannot apply to
them
- entries =
- CloseableIterable.filter(entries, entry -> isManifest(entry) ||
matchesPartition(entry));
- }
+ CloseableIterable<TrackedFile> files = CloseableIterable.transform(open(),
this::prepare);
if (!includeAll) {
- entries = CloseableIterable.filter(entries, entry ->
entry.tracking().isLive());
+ files = CloseableIterable.filter(files, file ->
file.tracking().isLive());
Review Comment:
Status-first makes sense. Stats before partition is the opposite of
`ManifestReader`, which short-circuits `evaluator.eval(partition) &&
metricsEvaluator.eval(file)`. Partition eval is typically cheaper than bounds,
so files that fail the partition residual still pay for
`InclusiveStatsEvaluator` here. When partition filtering is rewritten, worth
combining into one predicate with partition first (or keeping partition ahead
of stats).
##########
api/src/main/java/org/apache/iceberg/io/CloseableIterable.java:
##########
@@ -132,6 +133,38 @@ protected boolean shouldKeep(E item) {
iterable);
}
+ /**
+ * Filters the given {@link CloseableIterable} and passes each skipped item
to a {@link Consumer}.
+ *
+ * @param skipCallback A consumer used to handle skipped items
+ * @param iterable The underlying {@link CloseableIterable} to filter
+ * @param <E> The underlying type to be iterated
+ * @return A filtered {@link CloseableIterable} that skips items the
predicate does not match
+ */
+ static <E> CloseableIterable<E> filter(
+ Consumer<E> skipCallback, CloseableIterable<E> iterable, Predicate<E>
pred) {
+ Preconditions.checkArgument(null != iterable, "Invalid iterable: null");
+ Preconditions.checkArgument(null != pred, "Invalid predicate: null");
+
+ if (skipCallback != null) {
Review Comment:
I'd require a non-null `skipCallback` like the `Counter` overload, and drop
the null branch. The 2-arg `filter` already covers "no callback", and
`filter(null, iterable, pred)` is ambiguous between `Counter` and `Consumer`.
##########
core/src/main/java/org/apache/iceberg/V4ManifestReader.java:
##########
@@ -97,18 +101,30 @@ Schema readSchema() {
/** Returns copies of the tracked files that match this reader's configured
filters. */
@Override
public CloseableIterator<TrackedFile> iterator() {
- CloseableIterable<TrackedFile> entries =
CloseableIterable.transform(open(), this::prepare);
- if (!partitionFilters.isEmpty()) {
- // manifests have no partition, so the partition filter cannot apply to
them
- entries =
- CloseableIterable.filter(entries, entry -> isManifest(entry) ||
matchesPartition(entry));
- }
+ CloseableIterable<TrackedFile> files = CloseableIterable.transform(open(),
this::prepare);
if (!includeAll) {
- entries = CloseableIterable.filter(entries, entry ->
entry.tracking().isLive());
+ files = CloseableIterable.filter(files, file ->
file.tracking().isLive());
+ }
+
+ if (statsFilter != null) {
+ files =
+ CloseableIterable.filter(
+ this::incrementSkipCount,
+ files,
+ file -> statsFilter.eval(file.contentStats(),
file.recordCount()));
+ } else {
+ files =
+ CloseableIterable.filter(
+ this::incrementSkipCount, files, file -> file.recordCount() !=
0L);
Review Comment:
This always-on `recordCount != 0` path diverges from `ManifestReader`, which
only runs `InclusiveMetricsEvaluator` when there is a row/partition filter.
Empty files are still table entries: once this reader is used for rewrite,
expire, or `includeAll()` metadata listing, dropping them would lose files.
`InclusiveStatsEvaluator` already rejects `recordCount == 0` when `statsFilter
!= null`. I would gate this skip on scan planning (or drop the else branch) so
unfiltered reads still return empty files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
