github-actions[bot] commented on code in PR #70410: URL: https://github.com/apache/doris/pull/70410#discussion_r4548366265
########## 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: [P2] Drop unused statistics from surviving uncached data files With parsed caching disabled, consider a wide table with many small data manifests and an identity-partition predicate `WHERE p = 1`, where all selected files survive and there are no deletes. `filter.isPresent()` makes the provider pass `columns=null`, so this copy retains every statistics map for each survivor across the whole manifest window, although uncached consumers have finished pruning and need none of those maps. The [native SDK planner](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/ManifestGroup.java) reads statistics for filtering but drops unrequested maps when copying retained tasks. Thus an eligible window of tens of thousands of wide files adds substantial avoidable FE allocation and retained heap. Separate evaluation statistics from retained statistics: drop them after filtering when no equality-delete matching needs them, while preserving explicitly requested statistics and complete shared cache value s. This differs from the earlier rejected-file and unfiltered position-only findings: these files survive a predicate, and the scan can have zero deletes. -- 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]
