github-actions[bot] commented on code in PR #70410:
URL: https://github.com/apache/doris/pull/70410#discussion_r4547307099
##########
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:
[P2] Compact delete statistics before retaining the query-local index
With caching disabled, two small data manifests enable this route for an
unfiltered v2 table with many equality-delete files. Equality deletes may
legally contain additional deleted-row columns, so a delete keyed only on `id`
can carry statistics for a wide row payload. This line retains every map from
each matching delete in the complete index for the query's lifetime;
`builderFor(deleteFiles)` does not trim them. The native
`DeleteFileIndex.loadDeleteFiles` instead copies only `equalityFieldIds()`
statistics (or `DELETE_FILE_PATH` for position deletes). Thus retained metadata
grows with all payload columns rather than the delete keys, independently of
the bounded manifest window. Compact a request-local copy after predicate
evaluation, preserving the shared cache. This is distinct from the earlier
delete-pruning finding: every delete in this unfiltered example is relevant and
must remain indexed.
##########
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:
[P2] Prune uncached data files before materializing the manifest
An uncached scan of an unpartitioned table with multiple small, known-count
data manifests reaches this reader even when `WHERE id = 1` rejects every file
by its bounds. Manifest partition summaries cannot eliminate those manifests.
Here every file and its full statistics are copied into the manifest result;
`ManifestFileScanTaskIterator` applies the predicate only afterward. On wide
tables, the prefetch window can therefore retain maps for up to 30,000 rejected
files, adding substantial allocation and heap pressure to a scan producing no
splits. The native SDK applies `filterRows` before copying/buffering surviving
entries. Apply the same query-local pruning during uncached materialization,
after the required partition adaptation, while keeping shared cached contents
complete. This is distinct from the existing unfiltered position-delete
projection thread: this trigger has no deletes and concerns retaining files
that the predicate already excludes.
--
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]
