-
Notifications
You must be signed in to change notification settings - Fork 3.4k
Core: Resolve relative paths in V4 manifest reader #17434
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -37,6 +37,7 @@ | |
| 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.LocationUtil; | ||
| import org.apache.iceberg.util.Pair; | ||
| import org.apache.iceberg.util.StructProjection; | ||
|
|
||
|
|
@@ -46,6 +47,7 @@ class V4ManifestReader extends CloseableGroup implements CloseableIterable<Track | |
| private final Schema readSchema; | ||
| private final boolean includeAll; | ||
| private final ScanMetrics scanMetrics; | ||
| private final String tableLocation; | ||
|
|
||
| // partition filters keyed by spec ID; empty when no partition filter applies | ||
| private final Map<Integer, Pair<Evaluator, StructProjection>> partitionFilters; | ||
|
|
@@ -55,16 +57,19 @@ private V4ManifestReader( | |
| Schema readSchema, | ||
| Map<Integer, Pair<Evaluator, StructProjection>> partitionFilters, | ||
| boolean includeAll, | ||
| ScanMetrics scanMetrics) { | ||
| ScanMetrics scanMetrics, | ||
| String tableLocation) { | ||
| this.file = file; | ||
| this.readSchema = readSchema; | ||
| this.partitionFilters = partitionFilters; | ||
| this.includeAll = includeAll; | ||
| this.scanMetrics = scanMetrics; | ||
| this.tableLocation = tableLocation; | ||
| } | ||
|
|
||
| static Builder builder(InputFile file, Map<Integer, PartitionSpec> specsById) { | ||
| return new Builder(file, specsById); | ||
| static Builder builder( | ||
| InputFile file, Map<Integer, PartitionSpec> specsById, String tableLocation) { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think making this required is correct since we either have it from the existing metadata or it needs to be provided (e.g. by catalog). I don't see a way of knowing wheater paths are absolute, so this seems reasonable.
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks for confirming. |
||
| return new Builder(file, specsById, tableLocation); | ||
| } | ||
|
|
||
| /** Returns copies of the tracked files that match this reader's configured filters. */ | ||
|
|
@@ -81,7 +86,7 @@ public CloseableIterator<TrackedFile> iterator() { | |
| entries = CloseableIterable.filter(entries, entry -> entry.tracking().isLive()); | ||
| } | ||
|
|
||
| return CloseableIterable.transform(entries, TrackedFile::copy).iterator(); | ||
| return CloseableIterable.transform(entries, this::copyResolved).iterator(); | ||
| } | ||
|
|
||
| private boolean matchesPartition(TrackedFile trackedFile) { | ||
|
|
@@ -146,6 +151,22 @@ private TrackedFile prepare(TrackedFile trackedFile) { | |
| return trackedFile; | ||
| } | ||
|
|
||
| // resolves stored locations against the table location | ||
| private TrackedFile copyResolved(TrackedFile trackedFile) { | ||
| TrackedFileStruct copy = (TrackedFileStruct) trackedFile.copy(); | ||
| if (copy.location() != null) { | ||
| copy.setLocation(LocationUtil.resolveLocation(tableLocation, copy.location())); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Seems like we could make this lazy, but I assume that if we're projecting location we're going to pay the cost at some point. If we want to switch to a reuse based approach, that would be more efficient (e.g. push the table location down and only resolve with
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We are already doing a copy, and this PR doesn't make anything worse. Yes, we can always change this in future to make this lazy/zero copy. |
||
| } | ||
|
|
||
| DeletionVector dv = copy.deletionVector(); | ||
| if (dv != null && dv.location() != null) { | ||
| ((DeletionVectorStruct) dv) | ||
| .setLocation(LocationUtil.resolveLocation(tableLocation, dv.location())); | ||
| } | ||
|
|
||
| return copy; | ||
| } | ||
|
|
||
| private static boolean isManifest(TrackedFile trackedFile) { | ||
| FileContent content = trackedFile.contentType(); | ||
| return content == FileContent.DATA_MANIFEST || content == FileContent.DELETE_MANIFEST; | ||
|
|
@@ -156,6 +177,7 @@ static class Builder { | |
| private final Types.StructType unionPartitionType; | ||
| private final Map<Integer, PartitionSpec> specsById; | ||
| private final Schema fullSchema; | ||
| private final String tableLocation; | ||
| private Expression rowFilter = Expressions.alwaysTrue(); | ||
| private boolean caseSensitive = true; | ||
| private boolean includeAll = false; | ||
|
|
@@ -164,9 +186,11 @@ static class Builder { | |
| private Schema requestedProjection = null; | ||
| private ScanMetrics scanMetrics = ScanMetrics.noop(); | ||
|
|
||
| private Builder(InputFile file, Map<Integer, PartitionSpec> specsById) { | ||
| private Builder(InputFile file, Map<Integer, PartitionSpec> specsById, String tableLocation) { | ||
| Preconditions.checkArgument(tableLocation != null, "Invalid table location: null"); | ||
| this.file = file; | ||
| this.specsById = specsById; | ||
| this.tableLocation = LocationUtil.stripTrailingSlash(tableLocation); | ||
| this.unionPartitionType = Partitioning.unionPartitionTypes(specsById.values()); | ||
| Schema base = TrackedFile.schema(unionPartitionType, Types.StructType.of()); | ||
| // the read schema carries row_position (via BASE_TYPE) so the reader can fill manifestPos | ||
|
|
@@ -250,7 +274,12 @@ V4ManifestReader build() { | |
|
|
||
| boolean hasPartitionFilter = !partitionFilters.isEmpty(); | ||
| return new V4ManifestReader( | ||
| file, readSchema(hasPartitionFilter), partitionFilters, includeAll, scanMetrics); | ||
| file, | ||
| readSchema(hasPartitionFilter), | ||
| partitionFilters, | ||
| includeAll, | ||
| scanMetrics, | ||
| tableLocation); | ||
| } | ||
|
|
||
| private Schema readSchema(boolean hasPartitionFilter) { | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
An alternative to exposing the package private
setLocationmethod is doing it throughsetwith ordinal. But that would mean computing location's projected index from the read schema at resolution time and handling the not-projected case, for both the entry and its nested DV.The package-private
setLocationwrites the field directly and bypasses the projection layer, so it's correct and fast regardless of what's projected.