924060929 commented on code in PR #70410: URL: https://github.com/apache/doris/pull/70410#discussion_r4548760089
########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestReader.java: ########## @@ -0,0 +1,213 @@ +// 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.doris.connector.iceberg; + +import org.apache.iceberg.ContentFile; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.ManifestContent; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; +import org.apache.iceberg.ManifestReaderAccess; +import org.apache.iceberg.MetadataColumns; +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.PartitionField; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.ContentFileUtil; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; +import java.util.function.Predicate; + +/** Reads immutable manifest contents independently of whether the result is retained in a cache. */ +final class IcebergManifestReader { + /** One manifest's request-local predicate and partition adapter, confined to its loading worker. */ + static final class ReadFilter { + private final Types.StructType partitionType; + private final Predicate<ContentFile<?>> predicate; + private final Runnable skipped; + private boolean partitionTypeChecked; + private IcebergPartitionValueAdapter adapter; + + ReadFilter(PartitionSpec spec, Predicate<ContentFile<?>> predicate, Runnable skipped) { + this.partitionType = spec.rawPartitionType(); + this.predicate = predicate; + this.skipped = skipped; + } + + private void checkPartitionType(ContentFile<?> file) { + if (!partitionTypeChecked) { + Types.StructType physical = ((PartitionData) file.partition()).getPartitionType(); + if (!physical.equals(partitionType)) { + adapter = new IcebergPartitionValueAdapter(physical, partitionType); + } + partitionTypeChecked = true; + } + } + + DataFile adapt(DataFile file) { + checkPartitionType(file); + return adapter == null ? file : adapter.adapt(file); + } + + DeleteFile adapt(DeleteFile file) { + checkPartitionType(file); + return adapter == null ? file : adapter.adapt(file); + } + + boolean matches(ContentFile<?> file) { + if (predicate.test(file)) { + return true; + } + skipped.run(); + return false; + } + } + + private IcebergManifestReader() { + } + + static ManifestCacheValue read(ManifestFile manifest, Table table, + ManifestContent content, boolean estimateWeight) { + return read(manifest, table, table.io(), content, estimateWeight, null, null, () -> { }); + } + + static Function<ManifestFile, ManifestCacheValue> uncachedLoader(Table table, List<String> columns, + Runnable checkCancelled) { + return uncachedLoader(table, columns, null, checkCancelled); + } + + static Function<ManifestFile, ManifestCacheValue> uncachedLoader(Table table, List<String> columns, + Function<ManifestFile, ReadFilter> filters, Runnable checkCancelled) { + // One request-local IO view, shared by this scan's loads, never by the catalog/cache. + FileIO io = new IcebergScanFileIO(table.io(), checkCancelled); + return manifest -> read(manifest, table, io, manifest.content(), false, columns, filters, checkCancelled); + } + + private static ManifestCacheValue read(ManifestFile manifest, Table table, FileIO io, + ManifestContent content, boolean estimateWeight, List<String> columns, + Function<ManifestFile, ReadFilter> filters, Runnable checkCancelled) { + checkCancelled.run(); + ReadFilter filter = filters == null ? null : filters.apply(manifest); + try { + if (content == ManifestContent.DELETES) { + return ManifestCacheValue.forDeleteFiles( + loadDeleteFiles(manifest, table, io, filter, checkCancelled), estimateWeight); + } + return ManifestCacheValue.forDataFiles( + loadDataFiles(manifest, table, io, columns, filter, checkCancelled), estimateWeight); + } catch (IOException e) { + throw new RuntimeException("Failed to read iceberg manifest " + manifest.path(), e); + } + } + + private static List<DataFile> loadDataFiles(ManifestFile manifest, Table table, FileIO io, + List<String> columns, ReadFilter filter, Runnable checkCancelled) throws IOException { + List<DataFile> dataFiles = new ArrayList<>(); + // .copy() is mandatory: the ManifestReader iterator reuses the same object across iterations. + try (ManifestReader<DataFile> reader = ManifestFiles.read(manifest, io, + manifestReadSpecs(manifest, table))) { + if (columns != null) { + reader.select(columns); + } + for (DataFile borrowed : ManifestReaderAccess.liveFiles(reader)) { + checkCancelled.run(); + DataFile file = filter == null ? borrowed : filter.adapt(borrowed); + if (filter == null || filter.matches(file)) { + dataFiles.add(file.copy()); Review Comment: Fixed in 41183a309266. The uncached reader now separates read-time statistics from retained metadata using an immutable projection derived from the native scan context. Predicates run with the required statistics, then window copies keep only explicitly requested fields plus equality-delete keys. Delete association happens before the matching-only keys are stripped from final tasks; shared cached values remain complete and unchanged. The new six-case regression failed in all three cache-disabled variants before the fix. It now covers 72 combinations of cache policy, partition/ordinary-column filters, delete type and none/partial/key/all/empty statistics requests, comparing task bounds and delete association with the native SDK. All 1849 Iceberg tests pass (five skips), and the standard ARM64 FE build/Checkstyle passes with two build CPUs. End-to-end SQL timing was not rerun for this repair. -- 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]
