924060929 commented on code in PR #70410: URL: https://github.com/apache/doris/pull/70410#discussion_r4547201209
########## 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: Fixed in `290bd57bacfb`. Delete files now pass the same per-file metrics and partition predicate helper as data files, after request-local partition type adaptation and before DeleteFileIndex construction. Shared cached contents are never filtered or modified. Six parameterized cases compare SDK and streaming output for unpartitioned, bucket-source and identity-partition predicates with caching on/off. They exercise matching, query-disjoint and unknown-statistics equality deletes, alternate predicates 1/500/1 through the same cache, verify the exact delete paths sent to BE, and verify the complete cached delete list remains available. Validation: final full Iceberg connector suite, 1826 tests / 0 failures / 0 errors / 5 existing skips; standard ARM64 FE build and Checkstyle passed with two build CPUs. ########## 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: Fixed in `290bd57bacfb`. The uncached data loader is now selected after the filtered delete index is built. It retains the native requested projection for unfiltered position-delete-only scans and adds statistics when equality-delete matching or predicates need them; shared cache payloads and explicitly requested statistics remain complete. Ten parameterized cases use a 64-column table with actual manifest statistics, covering position/equality deletes, filtered/unfiltered reads, both cache policies and explicit includeColumnStats. The uncached position-only case retained all 64-column bounds before the fix and now exposes no bounds maps; the required-statistics cases retain their maps and delete associations. Validation: final full Iceberg connector suite, 1826 tests / 0 failures / 0 errors / 5 existing skips; standard ARM64 FE build and Checkstyle passed with two build CPUs. -- 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]
