rdblue commented on code in PR #16958: URL: https://github.com/apache/iceberg/pull/16958#discussion_r3606804878
########## 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: I don't think that `project` is the right method for configuration. In the library, we typically use `project` to pass a specific schema to read. That schema is not modified if it is used because we assume that the caller knows what they are doing and used `project` on purpose. We also use `select` to select columns by name. This has the same behavior as SQL: `select("a", "b")` projects columns `a` and `b`. In `ManifestReader`, we had configuration methods for both (it was an early API) but the contract of `project` doesn't allow you to mess with that schema. The behavior was to use the selected columns (and potentially add more stats columns) if `select` was called, even if `project` was also called. If `select` was not called then `project` would be used. `select` is used a lot more than `project` is used because that's the easy way to get the columns we want out of the file with readable code (`select(file_path, record_count, ...)`). Project is used for cases like metadata tables where we are showing users the data but not actually using it directly from the library. I think that we should fix the API of the builder. For internal uses, `select` is the right method to implement. We can also add `project`, but if we do it should be mutually exclusive with `select`. If you call `select` then you can't use `project` and vice versa. Stats are another dimension to the read schema. We want to take the filter that is passed in and project the stats structs required by that filter (the read schema method in `StatsUtil`). We also need a way for callers to request all stats or stats for a specific set of fields. The `copy` methods accept `Set<Integer>` to pass field IDs, but I think we probably want to configure this using field names. (Side note: generating these schemas requires knowing the table schema in this builder.) -- 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]
