anoopj commented on code in PR #17433:
URL: https://github.com/apache/iceberg/pull/17433#discussion_r3685574946
##########
core/src/main/java/org/apache/iceberg/V4ManifestReader.java:
##########
@@ -263,17 +297,62 @@ private Schema readSchema(boolean hasPartitionFilter) {
if (columns != null) {
Schema selected =
caseSensitive ? fullSchema.select(columns) :
fullSchema.caseInsensitiveSelect(columns);
- return addRequiredColumns(selected, hasPartitionFilter);
+ return addRequiredColumns(fullSchema, selected, hasPartitionFilter);
}
if (requestedProjection != null) {
- return addRequiredColumns(requestedProjection, hasPartitionFilter);
+ return addRequiredColumns(fullSchema, requestedProjection,
hasPartitionFilter);
}
return fullSchema;
}
- private Schema addRequiredColumns(Schema projection, boolean
hasPartitionFilter) {
+ /** Returns the schema of everything this reader may read, including
content stats. */
+ private Schema fullSchema() {
+ Types.StructType contentStatsType = contentStatsType();
+ Schema base = TrackedFile.schema(unionPartitionType, contentStatsType);
+ if (contentStatsType.fields().isEmpty()) {
+ // schema uses the unknown type for empty stats, which cannot be
paired with the stats
+ // struct in the manifest, so drop the field instead of reading it as
unknown
+ base = TypeUtil.selectNot(base,
ImmutableSet.of(TrackedFile.CONTENT_STATS_ID));
+ }
+
+ // the read schema carries row_position (via BASE_TYPE) so the reader
can fill manifestPos
+ return TypeUtil.replaceFieldTypes(
+ base, ImmutableMap.of(TrackedFile.TRACKING.fieldId(),
TrackingStruct.BASE_TYPE));
+ }
+
+ /** Returns the stats type to read, which is empty when no stats are
needed. */
+ private Types.StructType contentStatsType() {
+ Set<Integer> fieldIds = requiredStatsFieldIds();
Review Comment:
We are running requiredStatsFieldIds() twice per build() on the
select/project paths: here, and again in addRequiredColumns() (line ~399). Each
call re-runs Binder.boundReferences(...), so the filter expression gets bound
twice, and statsReadSchema(...) walks the schema twice.
Would it make sense to compute this once, e.g. inside readSchema() ?
##########
core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java:
##########
@@ -749,6 +775,203 @@ public void invalidBuilderArguments() {
.select((Collection<String>) null))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid columns: null");
+
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .projectStats(TABLE_SCHEMA, (Collection<Integer>) null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid stats field IDs: null");
+
+ assertThatThrownBy(
+ () -> V4ManifestReader.builder(manifest,
UNPARTITIONED_SPECS).projectStats(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid table schema: null");
+
+ // empty field IDs would silently read no stats; projectStats(Schema) is
the way to read all
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .projectStats(TABLE_SCHEMA, new int[0]))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid stats field IDs: empty, use projectStats(Schema)
instead");
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void contentStatsRoundTrip(FileFormat format) throws IOException {
Review Comment:
Minor: i got feedback on my reader PR that "round trip" usually means a
serialization test. Consider renaming to something like
`readsContentStatsForAllFields`.
##########
core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java:
##########
@@ -734,21 +772,208 @@ public void unknownManifestFormatThrows() throws
IOException {
public void invalidBuilderArguments() {
InputFile manifest =
fileIO.newInputFile(tempDir.resolve("manifest.avro").toString());
- assertThatThrownBy(() -> V4ManifestReader.builder(manifest,
UNPARTITIONED_SPECS).filter(null))
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA,
UNPARTITIONED_SPECS).filter(null))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid filter: null");
assertThatThrownBy(
- () -> V4ManifestReader.builder(manifest,
UNPARTITIONED_SPECS).scanMetrics(null))
+ () ->
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .scanMetrics(null))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid scan metrics: null");
assertThatThrownBy(
() ->
- V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
.select((Collection<String>) null))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid columns: null");
+
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .projectStats((Collection<Integer>) null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid stats field IDs: null");
+
+ assertThatThrownBy(() -> V4ManifestReader.builder(manifest, null,
UNPARTITIONED_SPECS))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid table schema: null");
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void contentStatsRoundTrip(FileFormat format) throws IOException {
+ TrackedFile file = fileWithStats("s3://bucket/file.parquet",
contentStats());
+
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION, CONTENT_STATS_TYPE,
ImmutableList.of(file));
+
+ // the default read carries stats for every field so that entries can be
copied to a new
+ // manifest
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA,
UNPARTITIONED_SPECS).build()) {
+ ContentStats stats = Iterables.getOnlyElement(reader).contentStats();
+ assertThat(stats).isNotNull();
+ assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS);
+ assertFieldStats(stats.statsFor(DATA_FIELD_ID), DATA_STATS);
+ assertFieldStats(stats.statsFor(MEASURE_FIELD_ID), MEASURE_STATS);
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void projectStatsReadsOnlyRequestedColumns(FileFormat format) throws
IOException {
+ TrackedFile file = fileWithStats("s3://bucket/file.parquet",
contentStats());
+
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION, CONTENT_STATS_TYPE,
ImmutableList.of(file));
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS)
+ .projectStats(ID_FIELD_ID)
+ .build()) {
+ ContentStats stats = Iterables.getOnlyElement(reader).contentStats();
+ assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS);
+ assertThat(stats.statsFor(DATA_FIELD_ID)).isNull();
+ assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull();
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void projectStatsWithoutFieldIdsOmitsStats(FileFormat format) throws
IOException {
+ TrackedFile file = fileWithStats("s3://bucket/file.parquet",
contentStats());
+
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION, CONTENT_STATS_TYPE,
ImmutableList.of(file));
+
+ // requesting no field IDs opts out of the default projection of every
field
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS)
+ .projectStats(ImmutableList.of())
+ .build()) {
+ assertThat(Iterables.getOnlyElement(reader).contentStats()).isNull();
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void filterStatsAreProjectedWhenOmittedByCaller(FileFormat format)
throws IOException {
+ TrackedFile file = fileWithStats("s3://bucket/file.parquet",
contentStats());
+
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION, CONTENT_STATS_TYPE,
ImmutableList.of(file));
+
+ // the filter references data, so its stats are read even though the
projection omits them
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS)
+ .project(new Schema(TrackedFile.LOCATION))
+ .filter(Expressions.equal("data", "m"))
+ .build()) {
+ ContentStats stats = Iterables.getOnlyElement(reader).contentStats();
+ assertFieldStats(stats.statsFor(DATA_FIELD_ID), DATA_STATS);
+ assertThat(stats.statsFor(ID_FIELD_ID)).isNull();
+ assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull();
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void filterOnMissingColumnFails(FileFormat format) throws IOException
{
+ InputFile manifest =
+ writeManifest(
+ format,
+ EMPTY_PARTITION,
+ ImmutableList.of(dataFile("data-a.parquet",
EMPTY_PARTITION_DATA)));
+
+ // stats for the filter's columns are resolved against the table schema
when the reader is built
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .filter(Expressions.equal("missing", 34))
+ .build())
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Cannot find field 'missing' in struct: %s",
TABLE_SCHEMA.asStruct());
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void forScanPlanningReadsOnlyFilterStats(FileFormat format) throws
IOException {
+ TrackedFile file = fileWithStats("s3://bucket/file.parquet",
contentStats());
+
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION, CONTENT_STATS_TYPE,
ImmutableList.of(file));
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS)
+ .forScanPlanning()
+ .filter(Expressions.equal("id", 1))
+ .build()) {
+ ContentStats stats = Iterables.getOnlyElement(reader).contentStats();
+ assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS);
+ assertThat(stats.statsFor(DATA_FIELD_ID)).isNull();
+ assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull();
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void forScanPlanningOmitsStatsWithoutFilter(FileFormat format) throws
IOException {
+ TrackedFile file = fileWithStats("s3://bucket/file.parquet",
contentStats());
+
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION, CONTENT_STATS_TYPE,
ImmutableList.of(file));
+
+ // scan planning without a filter has no stats to evaluate, so none are
read
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS)
+ .forScanPlanning()
+ .build()) {
+ assertThat(Iterables.getOnlyElement(reader).contentStats()).isNull();
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void selectWithoutStatsOmitsStats(FileFormat format) throws
IOException {
+ TrackedFile file = fileWithStats("s3://bucket/file.parquet",
contentStats());
+
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION, CONTENT_STATS_TYPE,
ImmutableList.of(file));
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS)
+ .select("location")
+ .build()) {
+ assertThat(Iterables.getOnlyElement(reader).contentStats()).isNull();
+ }
+ }
+
+ private static void assertFieldStats(FieldStats<?> actual, FieldStats<?>
expected) {
+ assertThat(actual).isNotNull();
+ assertThat(actual.fieldId()).isEqualTo(expected.fieldId());
+ assertThat(actual.type()).isEqualTo(expected.type());
+ assertThat(actual.lowerBound()).isEqualTo(expected.lowerBound());
+ assertThat(actual.upperBound()).isEqualTo(expected.upperBound());
+ assertThat(actual.tightBounds()).isEqualTo(expected.tightBounds());
+
assertThat(actual.avgValueSizeInBytes()).isEqualTo(expected.avgValueSizeInBytes());
+
+ // the count accessors unbox a nullable Long, so they throw for stats the
field does not track
+ Types.StructType statsType = expected.type();
+ assertThat(statsType.field("value_count")).isNotNull();
+ assertThat(actual.valueCount()).isEqualTo(expected.valueCount());
+
+ if (statsType.field("null_value_count") != null) {
+ assertThat(actual.nullValueCount()).isEqualTo(expected.nullValueCount());
+ }
+
+ if (statsType.field("nan_value_count") != null) {
+ assertThat(actual.nanValueCount()).isEqualTo(expected.nanValueCount());
+ }
Review Comment:
Can we also add a multi-file test e.g. we have two files (`fileA` and
`fileB`) with differing content stats, read them back and see if the stats are
correct? This is a protection against bugs in the container reuse.
##########
core/src/main/java/org/apache/iceberg/V4ManifestReader.java:
##########
@@ -228,6 +245,22 @@ Builder project(Schema newProjection) {
return this;
}
+ /** Reads content stats for the given table field IDs instead of for every
field. */
Review Comment:
You have a good comment in the overload "Stats for fields referenced by the
{@link #filter(Expression) filter} are always read".. Replicate it here also?
##########
core/src/main/java/org/apache/iceberg/V4ManifestReader.java:
##########
@@ -122,16 +128,27 @@ private CloseableIterable<TrackedFile> open() {
Preconditions.checkArgument(
format != null, "Cannot determine format of manifest: %s",
file.location());
- CloseableIterable<TrackedFile> reader =
+ InternalData.ReadBuilder readBuilder =
InternalData.read(format, file)
.project(readSchema)
.setRootType(TrackedFileStruct.class)
.setCustomType(TrackedFile.TRACKING.fieldId(),
TrackingStruct.class)
.setCustomType(TrackedFile.DELETION_VECTOR.fieldId(),
DeletionVectorStruct.class)
.setCustomType(TrackedFile.MANIFEST_INFO.fieldId(),
ManifestInfoStruct.class)
.setCustomType(TrackedFile.PARTITION_ID, PartitionData.class)
- .reuseContainers()
- .build();
+ .reuseContainers();
+
+ // content_stats is missing when it is not projected and unknown when no
stats are read
+ Types.NestedField statsField =
readSchema.findField(TrackedFile.CONTENT_STATS_ID);
+ if (statsField != null && statsField.type().isStructType()) {
+ readBuilder.setCustomType(TrackedFile.CONTENT_STATS_ID,
ContentStatsStruct.class);
+ // content_stats holds one stats struct per projected column
+ for (Types.NestedField fieldStats :
statsField.type().asStructType().fields()) {
+ readBuilder.setCustomType(fieldStats.fieldId(),
FieldStatsStruct.class);
Review Comment:
Does geo and variant need any special handling here? Can we add a test to
make sure those types work?
--
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]
