github-actions[bot] commented on code in PR #70410: URL: https://github.com/apache/doris/pull/70410#discussion_r4545895515
########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestReader.java: ########## @@ -0,0 +1,142 @@ +// 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.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.ManifestContent; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; +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 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; + +/** Reads immutable manifest contents independently of whether the result is retained in a cache. */ +final class IcebergManifestReader { + private IcebergManifestReader() { + } + + static ManifestCacheValue read(ManifestFile manifest, Table table, + ManifestContent content, boolean estimateWeight) { + return read(manifest, table, table.io(), content, estimateWeight, null, () -> { }); + } + + static Function<ManifestFile, ManifestCacheValue> uncachedLoader(Table table, List<String> columns, + 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, checkCancelled); + } + + private static ManifestCacheValue read(ManifestFile manifest, Table table, FileIO io, + ManifestContent content, boolean estimateWeight, List<String> columns, Runnable checkCancelled) { + checkCancelled.run(); + try { + if (content == ManifestContent.DELETES) { + return ManifestCacheValue.forDeleteFiles( + loadDeleteFiles(manifest, table, io, checkCancelled), estimateWeight); + } + return ManifestCacheValue.forDataFiles( + loadDataFiles(manifest, table, io, columns, 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, 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 dataFile : reader) { + checkCancelled.run(); + dataFiles.add(dataFile.copy()); + } + } + return dataFiles; + } + + /** Recover the manifest writer's partition source types, or read its header when that schema expired. */ + private static Map<Integer, PartitionSpec> manifestReadSpecs(ManifestFile manifest, Table table) { + PartitionSpec spec = table.specs().get(manifest.partitionSpecId()); + // Fixed-result transforms (bucket/year/month/day/hour) remain readable without their source + // schema. Iceberg resolves their result from UNKNOWN; do not require expired source schemas. + boolean missingSource = spec.rawPartitionType().fields().stream() + .anyMatch(field -> field.type().typeId() == org.apache.iceberg.types.Type.TypeID.UNKNOWN); + if (!missingSource) { + return Collections.singletonMap(spec.specId(), spec); + } + Long writerSnapshotId = manifest.snapshotId(); + Snapshot writerSnapshot = writerSnapshotId == null ? null : table.snapshot(writerSnapshotId); + Schema writerSchema = writerSnapshot == null ? null : table.schemas().get(writerSnapshot.schemaId()); + if (writerSchema == null) { + // The manifest header retains its physical partition type after snapshot/schema expiration. + return null; + } + Map<Integer, Types.NestedField> sources = new LinkedHashMap<>(); + Set<String> reservedNames = new HashSet<>(); + spec.fields().forEach(field -> reservedNames.add(field.name())); + for (PartitionField field : spec.fields()) { + Types.NestedField source = writerSchema.findField(field.sourceId()); + if (source == null) { + return null; + } + String name = "source_" + field.sourceId(); + while (!reservedNames.add(name)) { + name += "_"; + } + sources.put(field.sourceId(), Types.NestedField.optional(field.sourceId(), name, source.type())); + } + PartitionSpec readable = spec.toUnbound().bind(new Schema( + new ArrayList<>(sources.values())), true); + return Collections.singletonMap(readable.specId(), readable); + } + + private static List<DeleteFile> loadDeleteFiles(ManifestFile manifest, Table table, FileIO io, + Runnable checkCancelled) throws IOException { + List<DeleteFile> deleteFiles = new ArrayList<>(); + try (ManifestReader<DeleteFile> reader = + ManifestFiles.readDeleteManifest(manifest, io, manifestReadSpecs(manifest, table))) { Review Comment: [P2] Preserve per-file predicate pruning for delete files In the new cache-disabled route, consider a v2 unpartitioned table with two small data manifests, a data file with required `id` bounds `[1,1000]`, and newer equality deletes bounded to `id=500`. For `WHERE id=1`, the SDK delete loader applies `filterRows` and discards those deletes; this reader copies every live delete, and `getMatchingManifest` only prunes partition summaries. `DeleteFileIndex.forDataFile` compares against the whole data-file bounds, so it attaches them all. With many such deletes, FE builds an irrelevant index and BE receives extra files to read/index. Apply SDK-equivalent per-file row/partition pruning before building the request-local index, keeping shared cached values immutable. This is distinct from the earlier fixed manifest-projection/parallelism findings. ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java: ########## @@ -3173,50 +3216,63 @@ private SplitPlan planFileScanTaskWithManifestCache(TableScan scan, } /** - * PERF-04: the lazy, cache-backed {@link FileScanTask} enumeration shared by the synchronous - * ({@link #planFileScanTaskWithManifestCache}), streaming ({@link #streamSplits}), and COUNT(*) - * ({@link #planCountPushdown}) paths whenever {@link #isManifestCacheEnabled()}. Ported faithfully from legacy - * {@code IcebergScanNode.planFileScanTaskWithManifestCache}: partition-prune manifests with a - * {@link ManifestEvaluator}, read each surviving manifest's data/delete files THROUGH THE CACHE, then per data - * file apply the {@link InclusiveMetricsEvaluator} (file-stats prune) + {@link ResidualEvaluator} (partition - * residual) and attach its deletes via a {@link DeleteFileIndex}. Produces WHOLE-FILE - * {@link BaseFileScanTask}s (callers byte-offset-split via {@link TableScanUtil#splitFiles}). - * - * <p><b>Phase 1 is eager</b> (delete-manifest reads + {@link DeleteFileIndex} build): a data file's deletes - * need the full index before any data task can be produced, and running it here (on the caller's thread) lets - * the streaming/count callers catch a cache failure and fall back to the SDK path. Bounded like the SDK - * {@code planFiles()} (which also reads all delete manifests up front); delete manifests are far fewer than - * data manifests. <b>Phase 2 is lazy</b>: the returned iterable's iterator flat-maps the matching data - * manifests, yielding one surviving task at a time WITHOUT materializing the list, so a million-file streaming - * scan keeps FE heap bounded (peak = the delete index + a bounded manifest window + the split queue). - * - * <p>{@code statsQueryId} is nullable: a non-null id tallies cache hits/misses under that query (the - * synchronous + COUNT paths); {@code null} preserves the existing no-stats streaming path. - * Cache counters support concurrent prefetch workers. + * Shared manifest-level enumeration. Build the complete delete index before delivering data tasks, then + * read matching data manifests through the bounded prefetcher and prune individual files lazily. Caching + * only changes lookup/load functions; disabled caches retain results solely within the current window. + * {@code statsQueryId} is nullable and records accesses only when caching is enabled. */ - private CloseableIterable<FileScanTask> cacheBackedFileScanTasks(TableScan scan, + private CloseableIterable<FileScanTask> manifestFileScanTasks(TableScan scan, ConnectorSession session, Table table, Optional<ConnectorExpression> filter, String statsQueryId, IcebergTableHandle readHandle) { - return cacheBackedFileScanTasks(scan, session, table, filter, statsQueryId, + return manifestFileScanTasks(scan, session, table, filter, statsQueryId, new ManifestReadControl(true), readHandle); } - private CloseableIterable<FileScanTask> cacheBackedFileScanTasks(TableScan scan, + private CloseableIterable<FileScanTask> manifestFileScanTasks(TableScan scan, ConnectorSession session, Table table, Optional<ConnectorExpression> filter, String statsQueryId, ManifestReadControl readControl, IcebergTableHandle readHandle) { - return cacheBackedFileScanTasks(scan, session, table, filter, statsQueryId, readControl, readHandle, + return manifestFileScanTasks(scan, session, table, filter, statsQueryId, readControl, readHandle, new ConnectorScanContext()); } - private CloseableIterable<FileScanTask> cacheBackedFileScanTasks(TableScan scan, + private CloseableIterable<FileScanTask> manifestFileScanTasks(TableScan scan, ConnectorSession session, Table table, Optional<ConnectorExpression> filter, String statsQueryId, ManifestReadControl readControl, IcebergTableHandle readHandle, ConnectorScanContext scanContext) { + return manifestFileScanTasks(scan, session, table, filter, statsQueryId, readControl, + readHandle, scanContext, null); + } + + private CloseableIterable<FileScanTask> manifestFileScanTasks(TableScan scan, + ConnectorSession session, Table table, Optional<ConnectorExpression> filter, String statsQueryId, + ManifestReadControl readControl, IcebergTableHandle readHandle, ConnectorScanContext scanContext, + ScanMetrics scanMetrics) { scanContext.checkCancelled(); ConnectorTaskScheduler scheduler = requireActiveManifestScheduler(session); + // Cache policy only selects how immutable contents are obtained. Admission, completion order, + // cancellation and the retained-result window are identical with and without a cache. Snapshot snapshot = scan.snapshot(); if (snapshot == null) { return CloseableIterable.withNoopClose(Collections.emptyList()); } + List<ManifestFile> data = snapshot.dataManifests(table.io()); + List<ManifestFile> deletes = snapshot.deleteManifests(table.io()); + if (scanMetrics != null) { + scanMetrics.totalDataManifests().increment(data.size()); + scanMetrics.totalDeleteManifests().increment(deletes.size()); + } + boolean cacheEnabled = isManifestCacheEnabled(); + // Cached values serve later predicates/deletes and must retain all metrics. Without either, + // uncached scans keep the SDK's projection instead of decoding/copying unused statistics maps. + List<String> columns = cacheEnabled || filter.isPresent() || !deletes.isEmpty() Review Comment: [P2] Retain the native data projection for position-only deletes With caching disabled, any delete manifest sets `columns=null`, including an unfiltered v2 scan containing only position deletes. For an eligible wide table with multiple small data manifests, `IcebergManifestReader` therefore decodes, copies and retains every data file's statistics maps across the manifest window, although position-delete matching uses only path/partition/sequence and the filter is `alwaysTrue`. The [SDK planner](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/ManifestGroup.java) adds data statistics for equality deletes; its normal projection omits them here. A 30,000-entry bound does not bound these per-column maps, so this adds substantial FE allocation and retained heap for an ordinary position-delete workload. Choose the uncached data projection from the actual delete-index requirements so position-only scans keep the native no-statistics projection. -- 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]
