stevenzwu commented on code in PR #16958:
URL: https://github.com/apache/iceberg/pull/16958#discussion_r3618050125
##########
core/src/main/java/org/apache/iceberg/Partitioning.java:
##########
@@ -243,6 +243,17 @@ public static StructType partitionType(Table table) {
"table partition", specs, allActiveFieldIds(table.schema(), specs));
}
+ /**
+ * Builds a unified partition type containing all partition fields from the
given specs, including
+ * fields whose source columns are no longer present in the table schema.
+ *
+ * @param specs the partition specs to unify
+ * @return the constructed unified partition type
+ */
+ static StructType unionPartitionTypes(Collection<PartitionSpec> specs) {
+ return buildPartitionProjectionType("table partition", specs,
allFieldIds(specs));
Review Comment:
nit on the type name, maybe "union partition"?
It is only used in exception error msg. Here is another caller in addition
to L242 above.
```
public static StructType groupingKeyType(Schema schema,
Collection<PartitionSpec> specs) {
return buildPartitionProjectionType("grouping key", specs,
commonActiveFieldIds(schema, specs));
}
```
##########
core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java:
##########
@@ -0,0 +1,738 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.metrics.DefaultMetricsContext;
+import org.apache.iceberg.metrics.ScanMetrics;
+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.Lists;
+import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestV4ManifestReader {
+ private static final long SNAPSHOT_ID = 42L;
+ 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 int SORT_ORDER_ID = 1;
+ private static final String DV_LOCATION = "s3://bucket/dv.puffin";
+ private static final long DV_OFFSET = 100L;
+ private static final long DV_SIZE_IN_BYTES = 50L;
+ private static final long DV_CARDINALITY = 5L;
+
+ private static final Schema TABLE_SCHEMA =
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()), optional(2, "data",
Types.StringType.get()));
+ private static final PartitionSpec SPEC =
+ PartitionSpec.builderFor(TABLE_SCHEMA).identity("id").build();
+ private static final Types.StructType PARTITION_TYPE = SPEC.partitionType();
+ private static final Types.StructType EMPTY_PARTITION =
Types.StructType.of();
+ private static final PartitionData EMPTY_PARTITION_DATA = new
PartitionData(EMPTY_PARTITION);
+ private static final Map<Integer, PartitionSpec> PARTITIONED_SPECS =
+ ImmutableMap.of(SPEC.specId(), SPEC);
+ private static final Map<Integer, PartitionSpec> UNPARTITIONED_SPECS =
+ ImmutableMap.of(PartitionSpec.unpartitioned().specId(),
PartitionSpec.unpartitioned());
+
+ @Parameter private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ protected static List<FileFormat> parameters() {
+ return Arrays.asList(FileFormat.AVRO, FileFormat.PARQUET);
Review Comment:
nit: ` ImmutableList.of(...)`
##########
core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java:
##########
@@ -370,34 +446,62 @@ public void testPartitionFilterCountsSkippedDeleteFiles()
throws IOException {
public void testPartitionFilterKeepsManifestReferences() throws IOException {
TrackedFile keep = dataFile("data-1.parquet", partition(1));
TrackedFile prune = dataFile("data-2.parquet", partition(2));
- ManifestInfo info = new ManifestInfoStruct(1, 0, 0, 0, 1L, 0L, 0L, 0L, 1L,
null, null);
- TrackedFile manifestRef =
- new TrackedFileStruct(
- addedTracking(),
- FileContent.DATA_MANIFEST,
- FORMAT_VERSION_V4,
- "leaf.parquet",
- FileFormat.PARQUET,
- partition(2),
- RECORD_COUNT,
- FILE_SIZE_IN_BYTES,
- 0,
- null,
- null,
- null,
- info,
- null,
- null,
- null);
+ // a real manifest reference has a null spec_id and no partition tuple;
these refs carry a
+ // spec and a tuple that fails the filter so that pruning would be
detected if the manifest
+ // passthrough broke
+ TrackedFile dataManifestRef = manifestRef(FileContent.DATA_MANIFEST,
"data-leaf.parquet");
+ TrackedFile deleteManifestRef = manifestRef(FileContent.DELETE_MANIFEST,
"delete-leaf.parquet");
- InputFile manifest = writeManifest(PARTITION_TYPE, ImmutableList.of(keep,
prune, manifestRef));
+ InputFile manifest =
+ writeManifest(
+ PARTITION_TYPE, ImmutableList.of(keep, prune, dataManifestRef,
deleteManifestRef));
try (V4ManifestReader reader =
newReader(manifest,
PARTITIONED_SPECS).filterRows(Expressions.equal("id", 1)).build()) {
assertThat(reader)
.extracting(TrackedFile::location)
- .containsExactlyInAnyOrder(keep.location(), manifestRef.location());
+ .containsExactlyInAnyOrder(
+ keep.location(), dataManifestRef.location(),
deleteManifestRef.location());
Review Comment:
nit: should we also check the scan metrics like the one below?
```
assertThat(metrics.skippedDataFiles().value()).isEqualTo(1L);
assertThat(metrics.skippedDeleteFiles().value()).isEqualTo(0L);
```
##########
core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java:
##########
@@ -235,6 +248,69 @@ public void testProjectionRestrictsFields() throws
IOException {
}
}
+ @TestTemplate
+ public void testSelectRestrictsFields() throws IOException {
+ TrackedFile file = dataFile("s3://bucket/file.parquet",
EMPTY_PARTITION_DATA);
+
+ InputFile manifest = writeManifest(EMPTY_PARTITION,
ImmutableList.of(file));
+
+ try (V4ManifestReader reader =
+ newReader(manifest, UNPARTITIONED_SPECS)
+ .select(ImmutableList.of("location", "record_count"))
+ .build()) {
+ TrackedFile actual = Lists.newArrayList(reader).get(0);
+ assertThat(actual.location()).isEqualTo(file.location());
+ assertThat(actual.recordCount()).isEqualTo(RECORD_COUNT);
+ // tracking and content_type are always projected, even though the
caller omitted them
+ assertThat(actual.tracking()).isNotNull();
+ assertThat(actual.tracking().status()).isEqualTo(EntryStatus.ADDED);
+ assertThat(actual.contentType()).isEqualTo(FileContent.DATA);
+ // file_format and spec_id are null because they were not selected
Review Comment:
nit: can we add a coverage for partitioned table where these two fields are
always selected?
##########
core/src/main/java/org/apache/iceberg/V4ManifestReader.java:
##########
@@ -0,0 +1,272 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.expressions.Evaluator;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Projections;
+import org.apache.iceberg.io.CloseableGroup;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+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.Lists;
+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.StructProjection;
+
+/** Reader that reads a v4+ manifest file as {@link TrackedFile}s. */
+class V4ManifestReader extends CloseableGroup implements
CloseableIterable<TrackedFile> {
+ // tracking fields read on the scan path; row_position backs
Tracking.manifestPos
+ private static final Types.StructType SCAN_TRACKING =
+ Types.StructType.of(
+ Tracking.STATUS,
+ Tracking.SNAPSHOT_ID,
+ Tracking.SEQUENCE_NUMBER,
+ Tracking.FILE_SEQUENCE_NUMBER,
+ Tracking.FIRST_ROW_ID,
+ MetadataColumns.ROW_POSITION);
+
+ private final InputFile file;
+ private final Schema readSchema;
+ private final boolean onlyLive;
+ private final ScanMetrics scanMetrics;
+
+ // partition pruning state, keyed by spec ID
+ private final Map<Integer, Evaluator> partitionEvaluators;
+ private final Map<Integer, StructProjection> partitionProjections;
+
+ private V4ManifestReader(
+ InputFile file,
+ Schema readSchema,
+ Map<Integer, Evaluator> partitionEvaluators,
+ Map<Integer, StructProjection> partitionProjections,
+ boolean onlyLive,
+ ScanMetrics scanMetrics) {
+ this.file = file;
+ this.readSchema = readSchema;
+ this.partitionEvaluators = partitionEvaluators;
+ this.partitionProjections = partitionProjections;
+ this.onlyLive = onlyLive;
+ this.scanMetrics = scanMetrics;
+ }
+
+ static Builder builder(InputFile file, Map<Integer, PartitionSpec>
specsById) {
+ return new Builder(file, specsById);
+ }
+
+ /** Returns copies of the tracked files that match this reader's configured
filters. */
+ @Override
+ public CloseableIterator<TrackedFile> iterator() {
+ CloseableIterable<TrackedFile> entries =
CloseableIterable.transform(open(), this::prepare);
+ if (!partitionEvaluators.isEmpty()) {
+ // manifest references are expanded later and are not pruned by the
partition filter
+ entries =
+ CloseableIterable.filter(entries, entry -> isManifest(entry) ||
matchesPartition(entry));
+ }
+
+ if (onlyLive) {
+ entries = CloseableIterable.filter(entries, entry ->
entry.tracking().isLive());
+ }
+
+ return CloseableIterable.transform(entries, TrackedFile::copy).iterator();
+ }
+
+ private static boolean isManifest(TrackedFile trackedFile) {
+ FileContent content = trackedFile.contentType();
+ return content == FileContent.DATA_MANIFEST || content ==
FileContent.DELETE_MANIFEST;
+ }
+
+ private boolean matchesPartition(TrackedFile trackedFile) {
+ Integer specId = trackedFile.specId();
+ Evaluator evaluator = specId != null ? partitionEvaluators.get(specId) :
null;
+ StructProjection projection = specId != null ?
partitionProjections.get(specId) : null;
+ Preconditions.checkState(
+ evaluator != null && projection != null,
+ "Cannot apply partition filter: spec ID %s is not one of the known
specs %s in manifest %s",
+ specId,
+ partitionEvaluators.keySet(),
+ file.location());
+
+ boolean matches = evaluator.eval(projection.wrap(trackedFile.partition()));
+ if (!matches) {
+ incrementSkipCount(trackedFile.contentType());
+ }
+
+ return matches;
+ }
+
+ private void incrementSkipCount(FileContent content) {
+ switch (content) {
+ case DATA:
+ scanMetrics.skippedDataFiles().increment();
+ break;
+ case EQUALITY_DELETES:
+ scanMetrics.skippedDeleteFiles().increment();
+ break;
+ case DATA_MANIFEST:
+ scanMetrics.skippedDataManifests().increment();
+ break;
+ case DELETE_MANIFEST:
+ scanMetrics.skippedDeleteManifests().increment();
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported content type: " +
content);
+ }
+ }
+
+ private CloseableIterable<TrackedFile> open() {
+ FileFormat format = FileFormat.fromFileName(file.location());
+ Preconditions.checkArgument(
+ format != null, "Cannot determine format of manifest: %s",
file.location());
+
+ CloseableIterable<TrackedFile> reader =
+ 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();
+ addCloseable(reader);
+ return reader;
+ }
+
+ private TrackedFile prepare(TrackedFile trackedFile) {
+ Tracking tracking = trackedFile.tracking();
+ // manifestLocation is not stored in the manifest; the reader fills it in
+ if (tracking instanceof TrackingStruct) {
+ ((TrackingStruct) tracking).setManifestLocation(file.location());
+ }
+
+ return trackedFile;
+ }
+
+ static class Builder {
+ private final InputFile file;
+ private final Types.StructType partitionType;
+ private final Map<Integer, PartitionSpec> specsById;
+ private Expression rowFilter = Expressions.alwaysTrue();
+ private boolean caseSensitive = true;
+ private boolean onlyLive = false;
+ private Schema fileProjection = null;
+ private ScanMetrics scanMetrics = ScanMetrics.noop();
+
+ private Builder(InputFile file, Map<Integer, PartitionSpec> specsById) {
+ this.file = file;
+ this.partitionType =
Partitioning.unionPartitionTypes(specsById.values());
+ this.specsById = specsById;
+ }
+
+ /** Sets a row filter; files that cannot match the expression are skipped.
*/
+ Builder filterRows(Expression expr) {
+ Preconditions.checkArgument(expr != null, "Invalid row filter: null");
+ this.rowFilter = expr;
+ return this;
+ }
+
+ Builder caseSensitive(boolean isCaseSensitive) {
+ this.caseSensitive = isCaseSensitive;
+ return this;
+ }
+
+ /** Returns only files whose tracking {@link Tracking#isLive() is live}. */
+ Builder liveOnly() {
+ this.onlyLive = true;
+ return this;
+ }
+
+ Builder project(Schema newFileProjection) {
Review Comment:
Concretely, the current `project(Schema)` path modifies the caller's schema
in three ways at
`251:272:core/src/main/java/org/apache/iceberg/V4ManifestReader.java`:
1. A narrower `tracking` sub-projection is widened to full
`Tracking.scanSchema()` — `TypeUtil.getProjectedIds` returns both the parent
`TRACKING.fieldId` and any nested ids, and `TypeUtil.select` uses
`selectFullTypes=true`
(`119:123:api/src/main/java/org/apache/iceberg/types/TypeUtil.java`), so
parent-in-set short-circuits back to the whole struct.
2. `TRACKING` and `CONTENT_TYPE` are force-added at L264-265 even if the
caller omitted them.
3. `SPEC_ID` and `PARTITION_ID` are force-added when a row filter is active
(L267-268).
Worth noting: `TypeUtil.project`
(`75:87:api/src/main/java/org/apache/iceberg/types/TypeUtil.java`) uses
`selectFullTypes=false`, so it *does* preserve a narrower sub-projection.
`readSchema()` can use `TypeUtil.project` to avoid the projection widening.
parent struct widening problem will be especially problematic when content
stats are plumbed in, since there could be dozens or hundreds of column stats.
--
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]
