gaborkaszab commented on code in PR #16958: URL: https://github.com/apache/iceberg/pull/16958#discussion_r3518776622
########## core/src/main/java/org/apache/iceberg/V4ManifestReader.java: ########## @@ -0,0 +1,276 @@ +/* + * 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.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> { + private final InputFile file; + private final Types.StructType partitionType; + private final Schema fileProjection; + private final ScanMetrics scanMetrics; + + // partition pruning state, keyed by spec ID; empty when no filtering is required + private final Map<Integer, Evaluator> partitionEvaluators; + private final Map<Integer, StructProjection> partitionProjections; + + private V4ManifestReader( + InputFile file, + Types.StructType partitionType, + Map<Integer, Evaluator> partitionEvaluators, + Map<Integer, StructProjection> partitionProjections, + Schema fileProjection, + ScanMetrics scanMetrics) { + this.file = file; + this.partitionType = partitionType; + this.partitionEvaluators = partitionEvaluators; + this.partitionProjections = partitionProjections; + this.fileProjection = fileProjection; + this.scanMetrics = scanMetrics; + } + + static Builder builder( + InputFile file, Schema tableSchema, Map<Integer, PartitionSpec> specsById) { + return new Builder(file, tableSchema, specsById); + } + + /** Returns all tracked files in this manifest, regardless of status. */ + CloseableIterable<TrackedFile> allFiles() { + return files(false /* all files */); + } + + /** Returns tracked files whose tracking {@link Tracking#isLive() is live}. */ + CloseableIterable<TrackedFile> liveFiles() { + return files(true /* only live files */); + } + + /** Returns live tracked files, each as an independent copy. */ + @Override + public CloseableIterator<TrackedFile> iterator() { + return CloseableIterable.transform(liveFiles(), TrackedFile::copy).iterator(); + } + + private CloseableIterable<TrackedFile> files(boolean onlyLive) { + CloseableIterable<TrackedFile> entries = CloseableIterable.transform(open(), this::prepare); + if (!partitionEvaluators.isEmpty()) { + entries = CloseableIterable.filter(entries, this::matchesPartition); + } + + if (onlyLive) { + entries = CloseableIterable.filter(entries, entry -> entry.tracking().isLive()); + } + + return entries; + } + + private boolean matchesPartition(TrackedFile trackedFile) { + FileContent content = trackedFile.contentType(); + if (content == FileContent.DATA_MANIFEST || content == FileContent.DELETE_MANIFEST) { + // manifest references are expanded later and are not pruned by the partition filter + return true; + } + + Integer specId = trackedFile.specId(); + Evaluator evaluator = specId != null ? partitionEvaluators.get(specId) : null; + StructProjection projection = specId != null ? partitionProjections.get(specId) : null; + Preconditions.checkState( Review Comment: `specId` is an optional field in the V4 schema. I double checked the proposal doc and the spec PR and none of them gives extra info on the behavior difference between leaf manifest and data/delete file entries. I get the point, we return early for leaf manifests so here we only have data/delete entries where we expect `specId` not to be null. I think in the spec we should articulate better that it is mandatory for those entries. Will leave a comment. Since the field is optional, shouldn't we have an extra Preconditions check to verify that? -- 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]
