Core: Read content stats from v4 Manifest #17433
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,6 +22,9 @@ | |
| import java.util.Collection; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.stream.Collectors; | ||
| import java.util.stream.Stream; | ||
| import org.apache.iceberg.expressions.Binder; | ||
| import org.apache.iceberg.expressions.Evaluator; | ||
| import org.apache.iceberg.expressions.Expression; | ||
| import org.apache.iceberg.expressions.Expressions; | ||
|
|
@@ -32,11 +35,14 @@ | |
| import org.apache.iceberg.io.InputFile; | ||
| import org.apache.iceberg.metrics.ScanMetrics; | ||
| import org.apache.iceberg.relocated.com.google.common.base.Preconditions; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Maps; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Sets; | ||
| import org.apache.iceberg.types.TypeUtil; | ||
| import org.apache.iceberg.types.Types; | ||
| import org.apache.iceberg.util.ArrayUtil; | ||
| import org.apache.iceberg.util.LocationUtil; | ||
| import org.apache.iceberg.util.Pair; | ||
| import org.apache.iceberg.util.StructProjection; | ||
|
|
@@ -68,8 +74,11 @@ private V4ManifestReader( | |
| } | ||
|
|
||
| static Builder builder( | ||
| InputFile file, | ||
| Schema tableSchema, | ||
| Map<Integer, PartitionSpec> specsById, | ||
| String tableLocation) { | ||
| return new Builder(file, tableSchema, specsById, tableLocation); | ||
| } | ||
|
|
||
| /** Returns copies of the tracked files that match this reader's configured filters. */ | ||
|
|
@@ -127,16 +136,26 @@ private CloseableIterable<TrackedFile> open() { | |
| Preconditions.checkArgument( | ||
| format != null, "Cannot determine format of manifest: %s", file.location()); | ||
|
|
||
| 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(); | ||
|
|
||
| // content_stats is missing from the read schema when no stats are read | ||
| Types.NestedField statsField = readSchema.findField(TrackedFile.CONTENT_STATS_ID); | ||
| if (statsField != null) { | ||
| readBuilder.setCustomType(TrackedFile.CONTENT_STATS_ID, ContentStatsStruct.class); | ||
| for (Types.NestedField fieldStats : statsField.type().asStructType().fields()) { | ||
| readBuilder.setCustomType(fieldStats.fieldId(), FieldStatsStruct.class); | ||
| } | ||
| } | ||
|
|
||
| CloseableIterable<TrackedFile> reader = readBuilder.build(); | ||
| addCloseable(reader); | ||
| return reader; | ||
| } | ||
|
|
@@ -174,29 +193,32 @@ private static boolean isManifest(TrackedFile trackedFile) { | |
|
|
||
| static class Builder { | ||
| private final InputFile file; | ||
| private final Schema tableSchema; | ||
| private final Types.StructType unionPartitionType; | ||
| private final Map<Integer, PartitionSpec> specsById; | ||
| private final String tableLocation; | ||
| private Expression rowFilter = Expressions.alwaysTrue(); | ||
| private boolean caseSensitive = true; | ||
| private boolean includeAll = false; | ||
| private boolean scanPlanning = false; | ||
| private Collection<String> columns = null; | ||
| private Schema requestedProjection = null; | ||
| private Set<Integer> fieldIdsWithRequestedStats = null; | ||
| private MetricsConfig metricsConfig = null; | ||
| private ScanMetrics scanMetrics = ScanMetrics.noop(); | ||
|
|
||
| private Builder( | ||
| InputFile file, | ||
| Schema tableSchema, | ||
| Map<Integer, PartitionSpec> specsById, | ||
| String tableLocation) { | ||
| Preconditions.checkArgument(tableSchema != null, "Invalid table schema: null"); | ||
| Preconditions.checkArgument(tableLocation != null, "Invalid table location: null"); | ||
| this.file = file; | ||
| this.tableSchema = tableSchema; | ||
| this.specsById = specsById; | ||
| this.tableLocation = LocationUtil.stripTrailingSlash(tableLocation); | ||
| this.unionPartitionType = Partitioning.unionPartitionTypes(specsById.values()); | ||
| } | ||
|
|
||
| /** Sets a filter; files that cannot match the expression are skipped. */ | ||
|
|
@@ -252,6 +274,39 @@ Builder project(Schema newProjection) { | |
| return this; | ||
| } | ||
|
|
||
| /** | ||
| * Reads content stats for the given table field IDs instead of for every field. Stats for | ||
| * fields referenced by the {@link #filter(Expression) filter} are always read. | ||
| * | ||
| * <p>Passing no field IDs reads only the stats that the filter needs. | ||
| */ | ||
|
nastra marked this conversation as resolved.
|
||
| Builder projectStats(int... fieldIds) { | ||
| Preconditions.checkArgument(fieldIds != null, "Invalid field IDs: null"); | ||
| return projectStats(ArrayUtil.toIntList(fieldIds)); | ||
| } | ||
|
|
||
| /** | ||
| * Reads content stats for the given table field IDs instead of for every field. Stats for | ||
| * fields referenced by the {@link #filter(Expression) filter} are always read. | ||
| * | ||
| * <p>Passing an empty iterable reads only the stats that the filter needs. | ||
| */ | ||
| Builder projectStats(Iterable<Integer> fieldIds) { | ||
| Preconditions.checkArgument(fieldIds != null, "Invalid field IDs: null"); | ||
| this.fieldIdsWithRequestedStats = ImmutableSet.copyOf(fieldIds); | ||
| return this; | ||
| } | ||
|
|
||
| /** | ||
| * Sets the metrics config that determines which stats the manifest holds. Defaults to the | ||
| * config produced by the table's default metrics properties. | ||
| */ | ||
| Builder metricsConfig(MetricsConfig newMetricsConfig) { | ||
| Preconditions.checkArgument(newMetricsConfig != null, "Invalid metrics config: null"); | ||
| this.metricsConfig = newMetricsConfig; | ||
| return this; | ||
| } | ||
|
|
||
| Builder scanMetrics(ScanMetrics newScanMetrics) { | ||
| Preconditions.checkArgument(newScanMetrics != null, "Invalid scan metrics: null"); | ||
| this.scanMetrics = newScanMetrics; | ||
|
|
@@ -283,26 +338,128 @@ V4ManifestReader build() { | |
| } | ||
|
|
||
| private Schema readSchema(boolean hasPartitionFilter) { | ||
| Types.StructType requiredStatsType = | ||
| StatsUtil.statsReadSchema( | ||
| tableSchema, | ||
| fieldIdsWithRequiredStats( | ||
| tableSchema, fieldIdsWithRequestedStats, rowFilter, caseSensitive)); | ||
| Schema fullSchema = fullSchema(contentStatsType(requiredStatsType)); | ||
| if (scanPlanning) { | ||
| // scan planning does not read the change-tracking fields omitted by SCAN_TYPE | ||
| return TypeUtil.replaceFieldTypes( | ||
| fullSchema, ImmutableMap.of(TrackedFile.TRACKING.fieldId(), TrackingStruct.SCAN_TYPE)); | ||
| } | ||
|
|
||
| Types.StructType partitionType = hasPartitionFilter ? unionPartitionType : null; | ||
| if (columns != null) { | ||
| Schema selected = | ||
| caseSensitive ? fullSchema.select(columns) : fullSchema.caseInsensitiveSelect(columns); | ||
| return addRequiredColumns( | ||
| fullSchema, selected, requiredStatsType, partitionType, rowFilter); | ||
| } | ||
|
|
||
| if (requestedProjection != null) { | ||
| return addRequiredColumns( | ||
| fullSchema, requestedProjection, requiredStatsType, partitionType, rowFilter); | ||
| } | ||
|
|
||
| return fullSchema; | ||
| } | ||
|
|
||
| /** Returns the schema of everything this reader may read, including content stats. */ | ||
| private Schema fullSchema(Types.StructType contentStatsType) { | ||
| Schema base = TrackedFile.schema(unionPartitionType, contentStatsType); | ||
| if (contentStatsType.fields().isEmpty()) { | ||
| // the schema uses the unknown type for empty stats, but readers fail to pair the stats | ||
| // struct stored in the manifest with unknown, so drop the field instead of projecting it | ||
| 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. | ||
| * | ||
| * <p>Stats for every field the manifest holds are read unless the caller narrows them with | ||
| * {@link #forScanPlanning()}, {@link #projectStats(Iterable)}, or a {@link #project(Schema) | ||
| * projection} that carries its own stats, because copying entries into a new manifest needs all | ||
| * of them. A {@link #filter(Expression) filter} therefore never narrows the stats that are | ||
| * read; it only widens a set the caller has already narrowed. | ||
| */ | ||
| private Types.StructType contentStatsType(Types.StructType requiredStatsType) { | ||
|
Contributor
There was a problem hiding this comment. I don't think the stats struct handling in this class is correct. There are a few uses of this class:
The content stats projection depends on those cases. For metadata tables, the schema passed in will be based on the current table manifest schema, so we don't have to worry about adding content stats. It's similar for Scan planning should use Reads for table operations should produce all stats fields by using the table's current manifest schema. That is produced using
Contributor
Author
There was a problem hiding this comment. 2 + 4 should have been covered already and I just added reads for table operations that use a metrics config. @stevenzwu this also addresses the issue you brought up in #17433 (comment). I've also added handling for 1 |
||
| if (scanPlanning || fieldIdsWithRequestedStats != null) { | ||
| return requiredStatsType; | ||
| } | ||
|
|
||
| if (requestedProjection != null) { | ||
| return union(requiredStatsType, projectedStatsType(requestedProjection)); | ||
| } | ||
|
|
||
| return StatsUtil.statsWriteSchema(tableSchema, metricsConfig()); | ||
| } | ||
|
|
||
| private MetricsConfig metricsConfig() { | ||
| if (metricsConfig == null) { | ||
| this.metricsConfig = MetricsConfig.from(ImmutableMap.of(), tableSchema, null); | ||
| } | ||
|
|
||
| return metricsConfig; | ||
| } | ||
|
|
||
| /** Returns the requested field IDs, plus the field IDs referenced by the row filter. */ | ||
| private static Set<Integer> fieldIdsWithRequiredStats( | ||
| Schema tableSchema, | ||
| Set<Integer> fieldIdsWithRequestedStats, | ||
| Expression rowFilter, | ||
| boolean caseSensitive) { | ||
| Set<Integer> fieldIds = Sets.newHashSet(); | ||
| if (fieldIdsWithRequestedStats != null) { | ||
| fieldIds.addAll(fieldIdsWithRequestedStats); | ||
| } | ||
|
|
||
| if (rowFilter != Expressions.alwaysTrue()) { | ||
| fieldIds.addAll( | ||
| Binder.boundReferences( | ||
| tableSchema.asStruct(), ImmutableList.of(rowFilter), caseSensitive)); | ||
| } | ||
|
|
||
| return fieldIds; | ||
| } | ||
|
|
||
| /** Returns the fields of both stats structs */ | ||
| private static Types.StructType union( | ||
| Types.StructType statsType, Types.StructType otherStatsType) { | ||
| if (otherStatsType.fields().isEmpty()) { | ||
| return statsType; | ||
| } | ||
|
|
||
| return Types.StructType.of( | ||
| Stream.concat( | ||
| statsType.fields().stream(), | ||
| otherStatsType.fields().stream() | ||
| .filter(field -> statsType.field(field.fieldId()) == null)) | ||
| .collect(Collectors.toList())); | ||
| } | ||
|
|
||
| /** Returns the content stats struct carried by the projection, which may be empty. */ | ||
| private static Types.StructType projectedStatsType(Schema projection) { | ||
| Types.NestedField statsField = projection.findField(TrackedFile.CONTENT_STATS_ID); | ||
| if (statsField == null || !statsField.type().isStructType()) { | ||
| return Types.StructType.of(); | ||
| } | ||
|
|
||
| return statsField.type().asStructType(); | ||
| } | ||
|
|
||
| private static Schema addRequiredColumns( | ||
| Schema fullSchema, | ||
|
nastra marked this conversation as resolved.
|
||
| Schema projection, | ||
| Types.StructType requiredStatsType, | ||
| Types.StructType partitionType, | ||
| Expression rowFilter) { | ||
| Set<Integer> projectedIds = Sets.newHashSet(TypeUtil.getProjectedIds(projection)); | ||
|
|
||
| // fields the reader consumes internally: status for liveness filtering, row_position for | ||
|
|
@@ -315,13 +472,14 @@ private Schema addRequiredColumns(Schema projection, boolean hasPartitionFilter) | |
| projectedIds.add(TrackedFile.RECORD_COUNT.fieldId()); | ||
| } | ||
|
|
||
| if (partitionType != null) { | ||
| projectedIds.add(TrackedFile.SPEC_ID.fieldId()); | ||
| projectedIds.add(TrackedFile.PARTITION_ID); | ||
| projectedIds.addAll(TypeUtil.getProjectedIds(partitionType)); | ||
| } | ||
|
|
||
| projectedIds.addAll(TypeUtil.getProjectedIds(requiredStatsType)); | ||
|
|
||
| // project instead of select to preserve narrow struct projections from the caller | ||
| return TypeUtil.project(fullSchema, projectedIds); | ||
| } | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.