diff --git a/core/src/main/java/org/apache/iceberg/DeletionVectorStruct.java b/core/src/main/java/org/apache/iceberg/DeletionVectorStruct.java index 3f5be0756fad..efdb89578da0 100644 --- a/core/src/main/java/org/apache/iceberg/DeletionVectorStruct.java +++ b/core/src/main/java/org/apache/iceberg/DeletionVectorStruct.java @@ -64,6 +64,12 @@ public String location() { return location; } + // Package-private only so the manifest reader can store the location resolved against the + // table location; other callers must go through construction. + void setLocation(String newLocation) { + this.location = newLocation; + } + @Override public long offset() { return offset; diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java index f5ce03c7eb62..73437275b975 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java @@ -189,6 +189,12 @@ public String location() { return location; } + // Package-private only so the manifest reader can store the location resolved against the + // table location; other callers must go through construction. + void setLocation(String newLocation) { + this.location = newLocation; + } + @Override public FileFormat fileFormat() { return fileFormat; diff --git a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java index a823454845dd..dca4d67fa3e7 100644 --- a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java +++ b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java @@ -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> partitionFilters; @@ -55,16 +57,19 @@ private V4ManifestReader( Schema readSchema, Map> 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 specsById) { - return new Builder(file, specsById); + static Builder builder( + InputFile file, Map specsById, String tableLocation) { + 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 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())); + } + + 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 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 specsById) { + private Builder(InputFile file, Map 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) { diff --git a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java index c8a5cfd61a31..2b86c6bd1d39 100644 --- a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java +++ b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java @@ -48,6 +48,7 @@ import org.apache.iceberg.types.Comparators; import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.LocationUtil; import org.junit.jupiter.api.Named; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -61,13 +62,8 @@ class TestV4ManifestReader { private static final int FORMAT_VERSION_V4 = 4; private static final long RECORD_COUNT = 100L; private static final long FILE_SIZE_IN_BYTES = 1024L; - private static final DeletionVector DV = - DeletionVectorStruct.builder() - .location("s3://bucket/dv.puffin") - .offset(100L) - .sizeInBytes(50L) - .cardinality(5L) - .build(); + private static final String TABLE_LOCATION = "s3://bucket/db/table"; + private static final DeletionVector DV = dv("s3://bucket/dv.puffin"); private static final Schema TABLE_SCHEMA = new Schema( @@ -113,7 +109,8 @@ class TestV4ManifestReader { null, null); - // shared data files: FILE_A is in partition id=1, FILE_B in partition id=2 + // shared data files: FILE_A is in partition id=1, FILE_B in partition id=2. Locations are stored + // relative to the table location (the default), so the reader resolves them against the table private static final TrackedFile FILE_A = dataFile("data-a.parquet", partition(1)); private static final TrackedFile FILE_B = dataFile("data-b.parquet", partition(2)); private static final TrackedFile EQ_DELETES_A = deleteFile("eq-deletes-a.parquet", partition(1)); @@ -209,14 +206,16 @@ public void statusFiltering(FileFormat format) throws IOException { InputFile manifest = writeManifest(format, EMPTY_PARTITION, files); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).build()) { + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { assertThat(reader) .extracting(file -> file.tracking().status()) .containsExactly(EntryStatus.ADDED, EntryStatus.EXISTING, EntryStatus.MODIFIED); } try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).includeAll().build()) { + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + .includeAll() + .build()) { assertThat(reader) .extracting(file -> file.tracking().status()) .containsExactly( @@ -258,7 +257,8 @@ public void selectiveReadReturnsOnlyRequestedFields( InputFile manifest = writeManifest(format, EMPTY_PARTITION, files); - V4ManifestReader.Builder builder = V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS); + V4ManifestReader.Builder builder = + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION); configureRead.accept(builder); try (V4ManifestReader reader = builder.build()) { TrackedFile actual = Iterables.getOnlyElement(reader); @@ -316,7 +316,7 @@ public void rowFilterForcesRecordCount(FileFormat format) throws IOException { // even though the caller selected only location Schema projection = new Schema(TrackedFile.LOCATION); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) .project(projection) .filter(Expressions.equal("id", 1)) .build()) { @@ -332,7 +332,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) .select("location") .project(new Schema(TrackedFile.LOCATION))) .isInstanceOf(IllegalStateException.class) @@ -340,7 +340,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) .project(new Schema(TrackedFile.LOCATION)) .select("location")) .isInstanceOf(IllegalStateException.class) @@ -348,7 +348,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) .forScanPlanning() .select("location")) .isInstanceOf(IllegalStateException.class) @@ -356,7 +356,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) .select("location") .forScanPlanning()) .isInstanceOf(IllegalStateException.class) @@ -365,7 +365,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) .forScanPlanning() .project(new Schema(TrackedFile.LOCATION))) .isInstanceOf(IllegalStateException.class) @@ -373,7 +373,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) .project(new Schema(TrackedFile.LOCATION)) .forScanPlanning()) .isInstanceOf(IllegalStateException.class) @@ -388,7 +388,9 @@ public void projectionPreservesNarrowTrackingProjection(FileFormat format) throw writeManifest(format, EMPTY_PARTITION, ImmutableList.of(FILE_WITH_FULL_TRACKING)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).select("tracking.status").build()) { + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + .select("tracking.status") + .build()) { Tracking actual = Iterables.getOnlyElement(reader).tracking(); assertThat(actual.status()).isEqualTo(EntryStatus.ADDED); // the narrow tracking projection is not widened to the full tracking type @@ -406,7 +408,9 @@ public void forScanPlanningOmitsChangeTrackingFields(FileFormat format) throws I writeManifest(format, EMPTY_PARTITION, ImmutableList.of(FILE_WITH_FULL_TRACKING)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).forScanPlanning().build()) { + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + .forScanPlanning() + .build()) { Tracking actual = Iterables.getOnlyElement(reader).tracking(); // scan-relevant tracking fields are projected assertThat(actual.status()).isEqualTo(EntryStatus.ADDED); @@ -430,7 +434,7 @@ public void defaultReadsFullTracking(FileFormat format) throws IOException { // without scanPlanning, select, or project, the reader returns the full schema for copying to // other manifests, including the change-tracking fields try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).build()) { + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { Tracking actual = Iterables.getOnlyElement(reader).tracking(); assertThat(actual.status()).isEqualTo(EntryStatus.ADDED); assertThat(actual.snapshotId()).isEqualTo(SNAPSHOT_ID); @@ -451,7 +455,9 @@ public void projectNullReadsFullSchema(FileFormat format) throws IOException { // project(null) clears the projection and reads the full schema, like no projection at all try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).project(null).build()) { + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + .project(null) + .build()) { TrackedFile actual = Iterables.getOnlyElement(reader); assertThat(actual.location()).isEqualTo("s3://bucket/file.parquet"); assertThat(actual.fileFormat()).isEqualTo(FileFormat.PARQUET); @@ -469,12 +475,12 @@ public void partitionFilterForceProjectsFilterFields(FileFormat format) throws I // filter reads (spec_id, partition) or every row would be pruned Schema projection = new Schema(TrackedFile.LOCATION); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS, TABLE_LOCATION) .project(projection) .filter(Expressions.equal("id", 1)) .build()) { TrackedFile actual = Iterables.getOnlyElement(reader); - assertThat(actual.location()).isEqualTo(FILE_A.location()); + assertThat(actual.location()).isEqualTo(resolved(FILE_A)); assertThat(actual.specId()).isEqualTo(ID_PARTITIONING.specId()); assertThat(actual.partition().get(0, Integer.class)).isEqualTo(1); } @@ -488,12 +494,12 @@ public void selectWithPartitionFilterProjectsFilterFields(FileFormat format) thr // the caller selects only location; the reader must still project spec_id and partition // for the partition filter or every row would be pruned try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS, TABLE_LOCATION) .select("location") .filter(Expressions.equal("id", 1)) .build()) { TrackedFile actual = Iterables.getOnlyElement(reader); - assertThat(actual.location()).isEqualTo(FILE_A.location()); + assertThat(actual.location()).isEqualTo(resolved(FILE_A)); assertThat(actual.specId()).isEqualTo(ID_PARTITIONING.specId()); assertThat(actual.partition().get(0, Integer.class)).isEqualTo(1); } @@ -530,17 +536,17 @@ public void partitionFilterPrunesFilesAndCountsSkips(FileFormat format) throws I ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext()); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.equal("id", 1)) .scanMetrics(metrics) .build()) { assertThat(reader) .extracting(TrackedFile::location) .containsExactlyInAnyOrder( - FILE_A.location(), - EQ_DELETES_A.location(), - DATA_MANIFEST_REF.location(), - DELETE_MANIFEST_REF.location()); + resolved(FILE_A), + resolved(EQ_DELETES_A), + resolved(DATA_MANIFEST_REF), + resolved(DELETE_MANIFEST_REF)); } assertThat(metrics.skippedDataFiles().value()) @@ -568,7 +574,7 @@ public void rowFilterKeepsFilesWithoutStats(FileFormat format) throws IOExceptio ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext()); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) .filter(Expressions.equal("id", 1)) .scanMetrics(metrics) .build()) { @@ -588,17 +594,17 @@ public void caseInsensitivePartitionFilter(FileFormat format) throws IOException // a case-insensitive filter binds the mismatched-case "ID" reference and prunes FILE_B try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.equal("ID", 1)) .caseSensitive(false) .build()) { - assertThat(reader).extracting(TrackedFile::location).containsExactly(FILE_A.location()); + assertThat(reader).extracting(TrackedFile::location).containsExactly(resolved(FILE_A)); } // the same filter is case-sensitive by default, so "ID" fails to bind to the "id" field assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.equal("ID", 1)) .build()) .isInstanceOf(ValidationException.class) @@ -633,11 +639,13 @@ public void filterMatchesFilesAcrossDisjointSpecs(FileFormat format) throws IOEx writeManifest(format, unionType, ImmutableList.of(keepById, prunedById, keptOtherSpec)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, specsById).filter(Expressions.equal("id", 1)).build()) { + V4ManifestReader.builder(manifest, specsById, TABLE_LOCATION) + .filter(Expressions.equal("id", 1)) + .build()) { // spec0 entries are pruned by id; the spec1 entry is not partitioned by id so it survives assertThat(reader) .extracting(TrackedFile::location) - .containsExactlyInAnyOrder(keepById.location(), keptOtherSpec.location()); + .containsExactlyInAnyOrder(resolved(keepById), resolved(keptOtherSpec)); } } @@ -652,13 +660,13 @@ public void partialFilterStillPrunesOnCompatibleField(FileFormat format) throws InputFile manifest = writeManifest(format, ID_PARTITION_TYPE, ImmutableList.of(keep, prune)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.and(Expressions.equal("id", 1), Expressions.equal("data", "z"))) .build()) { assertThat(reader) .extracting(TrackedFile::location) .as("the id predicate prunes even though data is not a partition field") - .containsExactly(keep.location()); + .containsExactly(resolved(keep)); } } @@ -672,25 +680,25 @@ public void partitionFilterKeepsFileWithUnknownSpec(FileFormat format) throws IO // the filter would prune partition id=1 under spec 0, but cannot be applied to spec 5 try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.equal("id", 2)) .build()) { - assertThat(reader).extracting(TrackedFile::location).containsExactly(file.location()); + assertThat(reader).extracting(TrackedFile::location).containsExactly(resolved(file)); } } @ParameterizedTest @FieldSource("MANIFEST_FORMATS") public void partitionFilterKeepsFileWithNullSpecId(FileFormat format) throws IOException { - TrackedFile file = dataFile("no-spec.parquet", null, null); + TrackedFile file = dataFile("no-spec.parquet", (Integer) null, null); InputFile manifest = writeManifest(format, ID_PARTITION_TYPE, ImmutableList.of(file)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.equal("id", 2)) .build()) { - assertThat(reader).extracting(TrackedFile::location).containsExactly(file.location()); + assertThat(reader).extracting(TrackedFile::location).containsExactly(resolved(file)); } } @@ -706,7 +714,7 @@ public void iteratorReturnsLiveCopies(FileFormat format) throws IOException { InputFile manifest = writeManifest(format, EMPTY_PARTITION, files); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).build()) { + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { List read = Lists.newArrayList(reader); assertThat(read) .hasSize(2) @@ -723,38 +731,203 @@ public void unknownManifestFormatThrows() throws IOException { InputFile badFile = fileIO.newInputFile(tempDir.resolve("manifest-" + System.nanoTime() + ".txt").toString()); - try (V4ManifestReader reader = V4ManifestReader.builder(badFile, UNPARTITIONED_SPECS).build()) { + try (V4ManifestReader reader = + V4ManifestReader.builder(badFile, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { assertThatThrownBy(reader::iterator) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("Cannot determine format of manifest"); } } + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + public void resolvesRelativeDataFileLocation(FileFormat format) throws IOException { + TrackedFile file = dataFile("data/00000-0.parquet", EMPTY_PARTITION_DATA); + verifyLocation(format, file, TABLE_LOCATION + "/data/00000-0.parquet"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + public void absoluteDataFileLocationIsUnchanged(FileFormat format) throws IOException { + TrackedFile file = dataFile("hdfs://wh/db/table/data/00000-0.parquet", EMPTY_PARTITION_DATA); + verifyLocation(format, file, "hdfs://wh/db/table/data/00000-0.parquet"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + public void preservesNonStandardDataFileLocation(FileFormat format) throws IOException { + // a leading / or // has no URI scheme, so it is treated as relative and joined to the table + // location; the reader does not special-case authority-style or root-absolute paths + verifyLocation( + format, + dataFile("/data/00000-0.parquet", EMPTY_PARTITION_DATA), + TABLE_LOCATION + "//data/00000-0.parquet"); + verifyLocation( + format, + dataFile("//data/00000-0.parquet", EMPTY_PARTITION_DATA), + TABLE_LOCATION + "///data/00000-0.parquet"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + public void resolvesRelativeDeletionVectorLocation(FileFormat format) throws IOException { + TrackedFile file = dataFile("data/00000-0.parquet", EMPTY_PARTITION_DATA, dv("data/dv.puffin")); + + InputFile manifest = writeManifest(format, EMPTY_PARTITION, ImmutableList.of(file)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { + TrackedFile actual = Iterables.getOnlyElement(reader); + assertThat(actual.location()).isEqualTo(TABLE_LOCATION + "/data/00000-0.parquet"); + assertThat(actual.deletionVector().location()).isEqualTo(TABLE_LOCATION + "/data/dv.puffin"); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + public void resolvesLeafManifestLocation(FileFormat format) throws IOException { + TrackedFile leaf = manifestRef(FileContent.DATA_MANIFEST, "metadata/leaf.avro"); + verifyLocation(format, leaf, TABLE_LOCATION + "/metadata/leaf.avro"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + public void resolvesDataFileAndDvSchemesIndependently(FileFormat format) throws IOException { + // absolute data file paired with a relative DV, and relative data file paired with an absolute + // DV: each location's scheme is evaluated on its own + TrackedFile absoluteFileRelativeDv = + dataFile("s3://other/abs.parquet", EMPTY_PARTITION_DATA, dv("data/dv.puffin")); + TrackedFile relativeFileAbsoluteDv = + dataFile("data/rel.parquet", EMPTY_PARTITION_DATA, dv("s3://other/abs-dv.puffin")); + + InputFile manifest = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(absoluteFileRelativeDv, relativeFileAbsoluteDv)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { + List actual = Lists.newArrayList(reader); + assertThat(actual.get(0).location()).isEqualTo("s3://other/abs.parquet"); + assertThat(actual.get(0).deletionVector().location()) + .isEqualTo(TABLE_LOCATION + "/data/dv.puffin"); + assertThat(actual.get(1).location()).isEqualTo(TABLE_LOCATION + "/data/rel.parquet"); + assertThat(actual.get(1).deletionVector().location()).isEqualTo("s3://other/abs-dv.puffin"); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + public void stripsTrailingSlashFromTableLocation(FileFormat format) throws IOException { + TrackedFile file = dataFile("data/00000-0.parquet", EMPTY_PARTITION_DATA); + + InputFile manifest = writeManifest(format, EMPTY_PARTITION, ImmutableList.of(file)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION + "/").build()) { + TrackedFile actual = Iterables.getOnlyElement(reader); + assertThat(actual.location()).isEqualTo(TABLE_LOCATION + "/data/00000-0.parquet"); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + public void resolutionSkippedWhenLocationNotProjected(FileFormat format) throws IOException { + TrackedFile file = dataFile("data/00000-0.parquet", EMPTY_PARTITION_DATA); + + InputFile manifest = writeManifest(format, EMPTY_PARTITION, ImmutableList.of(file)); + + // location is not projected, so there is nothing to resolve even though it is relative + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + .select("tracking.status") + .build()) { + TrackedFile actual = Iterables.getOnlyElement(reader); + assertThat(actual.location()).isNull(); + } + } + @Test public void invalidBuilderArguments() { InputFile manifest = fileIO.newInputFile(tempDir.resolve("manifest.avro").toString()); - assertThatThrownBy(() -> V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).filter(null)) + assertThatThrownBy( + () -> + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + .filter(null)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Invalid filter: null"); assertThatThrownBy( - () -> V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).scanMetrics(null)) + () -> + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + .scanMetrics(null)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Invalid scan metrics: null"); assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) .select((Collection) null)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Invalid columns: null"); + + assertThatThrownBy(() -> V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, null)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid table location: null"); + } + + // the location a relative fixture resolves to once read against TABLE_LOCATION + private static String resolved(TrackedFile file) { + return LocationUtil.resolveLocation(TABLE_LOCATION, file.location()); + } + + // writes a single tracked file, reads it back against TABLE_LOCATION, and checks its location + private void verifyLocation(FileFormat format, TrackedFile file, String expectedLocation) + throws IOException { + InputFile manifest = writeManifest(format, EMPTY_PARTITION, ImmutableList.of(file)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { + TrackedFile actual = Iterables.getOnlyElement(reader); + assertThat(actual.location()).isEqualTo(expectedLocation); + } + } + + private static DeletionVector dv(String location) { + return DeletionVectorStruct.builder() + .location(location) + .offset(100L) + .sizeInBytes(50L) + .cardinality(5L) + .build(); } private static TrackedFile dataFile(String location, PartitionData partition) { return dataFile(location, 0, partition); } + private static TrackedFile dataFile(String location, PartitionData partition, DeletionVector dv) { + return new TrackedFileStruct( + addedTracking(), + FileContent.DATA, + FORMAT_VERSION_V4, + location, + FileFormat.PARQUET, + RECORD_COUNT, + FILE_SIZE_IN_BYTES, + 0, // spec_id + partition, + null, // content_stats + null, // sort_order_id + dv, + null, // manifest_info + null, // key_metadata + null, // split_offsets + null); // equality_ids + } + private static TrackedFile dataFile(String location, Integer specId, PartitionData partition) { return new TrackedFileStruct( addedTracking(), @@ -766,13 +939,13 @@ private static TrackedFile dataFile(String location, Integer specId, PartitionDa FILE_SIZE_IN_BYTES, specId, partition, - null, - null, - null, - null, - null, - null, - null); + null, // content_stats + null, // sort_order_id + null, // deletion_vector + null, // manifest_info + null, // key_metadata + null, // split_offsets + null); // equality_ids } private static TrackedFile deleteFile(String location, PartitionData partition) { @@ -784,15 +957,15 @@ private static TrackedFile deleteFile(String location, PartitionData partition) FileFormat.PARQUET, RECORD_COUNT, FILE_SIZE_IN_BYTES, - 0, + 0, // spec_id partition, - null, - null, - null, - null, - null, - null, - ImmutableList.of(1)); + null, // content_stats + null, // sort_order_id + null, // deletion_vector + null, // manifest_info + null, // key_metadata + null, // split_offsets + ImmutableList.of(1)); // equality_ids } private static TrackedFile manifestRef(FileContent content, String location) { @@ -880,7 +1053,8 @@ private InputFile writeManifest( private List read(InputFile manifest, Map specsById) throws IOException { - try (V4ManifestReader reader = V4ManifestReader.builder(manifest, specsById).build()) { + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, specsById, TABLE_LOCATION).build()) { return Lists.newArrayList(reader); } }