924060929 commented on code in PR #70410: URL: https://github.com/apache/doris/pull/70410#discussion_r4548085225
########## 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()); Review Comment: Fixed in `95b02f4ce592`. Uncached readers now inspect borrowed live entries, adapt partitions with the existing adapter, and apply query predicates before copying survivors into the manifest result. Discarded entries are not copied or retained. The SDK-package bridge exposes liveEntries without changing decoding or stream ownership; the reader still closes its registered streams. Each load owns its evaluator/adapter state. Cached contents remain complete. The same early boundary applies to uncached data and delete files. Their consumer does not reapply predicates or duplicate skipped counters after filtering/compaction. Eight parameterized cases inspect the actual production-selected loader result for cached/uncached, empty/selective 64-column manifests and INT-to-LONG identity partition evolution. The old uncached loader retained 16 files instead of 0/1. An additional entry-reuse test verifies only explicit retained copies survive advancement. Validation: complete Iceberg suite: 1843 tests / 0 failures / 0 errors / 5 existing skips, including 17 new cases; standard ARM64 FE build, Checkstyle and isolated plugin packaging passed with 2 build CPUs. No new SQL timing claim. ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java: ########## @@ -3234,54 +3287,62 @@ private CloseableIterable<FileScanTask> cacheBackedFileScanTasks(TableScan scan, Semaphore readPermits = IcebergStatementScope.manifestReadBudget(session, concurrency); Supplier<AutoCloseable> retainLoadOwner = IcebergStatementScope.readTableLeaseFactory(session, readHandle.getDbName(), readHandle.getTableName()); - IcebergManifestReadContext deleteReadContext = new IcebergManifestReadContext(context, readPermits, - () -> scanContext.isCancelled() || scheduler != null && scheduler.isClosed(), retainLoadOwner); - - // Phase 1 (eager): partition-prune + cache-load delete manifests into the delete-file index. + // Build the complete index before delivering data. Both manifest kinds use the same inclusive + // row-to-partition projection and bounded loader; delete loading does not require BE demand. List<DeleteFile> deleteFiles = new ArrayList<>(); - for (ManifestFile manifest : snapshot.deleteManifests(table.io())) { - scanContext.checkCancelled(); - if (manifest.content() != ManifestContent.DELETES) { - continue; - } - PartitionSpec spec = specsById.get(manifest.partitionSpecId()); - if (spec == null) { - continue; - } - if (!ManifestEvaluator.forPartitionFilter(filterExpr, spec, caseSensitive).eval(manifest)) { - continue; - } - ManifestCacheValue value = manifestCache.getCachedManifestValue(manifest, statsQueryId); - if (value == null) { - try { - value = deleteReadContext.read(manifest, m -> manifestCacheGet(m, table, statsQueryId)); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - CancellationException cancelled = new CancellationException( - "Interrupted while reading Iceberg delete manifest " + manifest.path()); - cancelled.initCause(e); - throw cancelled; + if (!deletes.isEmpty()) { + CloseableIterable<ManifestFile> matchingDeletes = getMatchingManifest(deletes, specsById, filterExpr, + scanContext::checkCancelled, scanMetrics, true); + try (IcebergManifestPrefetcher prefetcher = new IcebergManifestPrefetcher( + matchingDeletes.iterator(), concurrency, cached, deleteLoader, context, readPermits, () -> true, + scheduler, sharedDeleteLoader, retainLoadOwner, false)) { + scanContext.whenCancelled(prefetcher::cancel); + while (prefetcher.hasNext()) { + IcebergManifestPrefetcher.LoadedManifest loaded = prefetcher.next(); + scanContext.checkCancelled(); + int specId = loaded.manifest.partitionSpecId(); + ResidualEvaluator residual = residualEvaluators.get(specId); + // Manifest summaries only reject whole manifests. Match the SDK's per-file row and + // partition pruning before indexing, after type adaptation, without changing cached values. + for (DeleteFile file : loaded.files.forScan(specsById.get(specId)).getDeleteFiles()) { + scanContext.checkCancelled(); + if (matchesFileFilter(file, metricsEvaluator, residual)) { + deleteFiles.add(file); Review Comment: Fixed in `95b02f4ce592`. Retained deletes now use the native ContentFileUtil.copy policy: equalityFieldIds statistics for equality deletes, and DELETE_FILE_PATH statistics for position deletes. All predicate evaluation precedes compaction so queries on additional deleted-row payload columns still prune correctly. Uncached readers filter/compact before retaining results; cached consumers make query-local compact copies and leave shared full-statistics values unchanged. Four 64-column position/equality/cache-policy cases verify the actual retained index bounds contain only the required key/path. Four additional cache/payload-predicate cases guard the ordering of filtering and trimming. The regression also checks file identity/record counts and complete shared cached statistics. All 4 original compact-index cases failed before this fix. Validation: complete Iceberg suite: 1843 tests / 0 failures / 0 errors / 5 existing skips, including 17 new cases; standard ARM64 FE build, Checkstyle and isolated plugin packaging passed with 2 build CPUs. No new SQL timing claim. -- 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]
