This is an automated email from the ASF dual-hosted git repository.
stevenzwu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 6938d2191f Core: Resolve relative paths in V4 manifest reader (#17434)
6938d2191f is described below
commit 6938d2191f1a7fb93173b53d783144e2f7a0013a
Author: Anoop Johnson <[email protected]>
AuthorDate: Wed Aug 5 11:53:43 2026 -0700
Core: Resolve relative paths in V4 manifest reader (#17434)
---
.../org/apache/iceberg/DeletionVectorStruct.java | 6 +
.../java/org/apache/iceberg/TrackedFileStruct.java | 6 +
.../java/org/apache/iceberg/V4ManifestReader.java | 41 +++-
.../org/apache/iceberg/TestV4ManifestReader.java | 272 ++++++++++++++++-----
4 files changed, 254 insertions(+), 71 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/DeletionVectorStruct.java
b/core/src/main/java/org/apache/iceberg/DeletionVectorStruct.java
index 461e472e59..2695c59a60 100644
--- a/core/src/main/java/org/apache/iceberg/DeletionVectorStruct.java
+++ b/core/src/main/java/org/apache/iceberg/DeletionVectorStruct.java
@@ -74,6 +74,12 @@ class DeletionVectorStruct extends SupportsIndexProjection
implements DeletionVe
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 f5ce03c7eb..73437275b9 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 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
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 a823454845..dca4d67fa3 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.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.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 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
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) {
+ return new Builder(file, specsById, tableLocation);
}
/** Returns copies of the tracked files that match this reader's configured
filters. */
@@ -81,7 +86,7 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
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 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
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 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
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 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
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 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
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 c8a5cfd61a..3f2cbdd4e1 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.transforms.Transforms;
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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
// 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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
assertThatThrownBy(
() ->
- V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS,
TABLE_LOCATION)
.forScanPlanning()
.select("location"))
.isInstanceOf(IllegalStateException.class)
@@ -356,7 +356,7 @@ class TestV4ManifestReader {
assertThatThrownBy(
() ->
- V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS,
TABLE_LOCATION)
.select("location")
.forScanPlanning())
.isInstanceOf(IllegalStateException.class)
@@ -365,7 +365,7 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
// 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 @@ class TestV4ManifestReader {
// 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 @@ class TestV4ManifestReader {
// 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 @@ class TestV4ManifestReader {
// 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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
// 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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
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 @@ class TestV4ManifestReader {
// 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 @@ class TestV4ManifestReader {
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<TrackedFile> read = Lists.newArrayList(reader);
assertThat(read)
.hasSize(2)
@@ -723,38 +731,171 @@ class TestV4ManifestReader {
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(format.addExtension("data/00000-0"),
EMPTY_PARTITION_DATA);
+ verifyLocationAfterWriteReadRoundTrip(format, file, resolved(file));
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void resolvesRelativeDeletionVectorLocation(FileFormat format) throws
IOException {
+ TrackedFile file =
+ dataFile(format.addExtension("data/00000-0"), 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(resolved(file));
+ assertThat(actual.deletionVector().location())
+ .isEqualTo(LocationUtil.resolveLocation(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");
+ verifyLocationAfterWriteReadRoundTrip(format, leaf, resolved(leaf));
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void mixedAbsoluteAndRelativePathWithDataFileAndDV(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<TrackedFile> actual = Lists.newArrayList(reader);
+ // absolute locations pass through unchanged; relative ones resolve
against the table location
+ assertThat(actual.get(0).location()).isEqualTo("s3://other/abs.parquet");
+ assertThat(actual.get(0).deletionVector().location())
+ .isEqualTo(LocationUtil.resolveLocation(TABLE_LOCATION,
"data/dv.puffin"));
+ assertThat(actual.get(1).location())
+ .isEqualTo(LocationUtil.resolveLocation(TABLE_LOCATION,
"data/rel.parquet"));
+
assertThat(actual.get(1).deletionVector().location()).isEqualTo("s3://other/abs-dv.puffin");
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void resolutionSkippedWhenLocationNotProjected(FileFormat format)
throws IOException {
+ TrackedFile file = dataFile(format.addExtension("data/00000-0"),
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<String>) 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 verifyLocationAfterWriteReadRoundTrip(
+ 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 +907,13 @@ class TestV4ManifestReader {
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 +925,15 @@ class TestV4ManifestReader {
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 +1021,8 @@ class TestV4ManifestReader {
private List<TrackedFile> read(InputFile manifest, Map<Integer,
PartitionSpec> 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);
}
}
