924060929 commented on code in PR #69298: URL: https://github.com/apache/doris/pull/69298#discussion_r4321335873
########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestPrefetcher.java: ########## @@ -0,0 +1,356 @@ +// 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.doris.connector.spi.ConnectorContext; +import org.apache.doris.connector.spi.ConnectorTaskScheduler; + +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.io.CloseableIterator; + +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.FutureTask; +import java.util.concurrent.RunnableFuture; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; +import java.util.function.Function; + +/** + * Completion-ordered, single-consumer manifest prefetch. close() may run on the query-cancellation thread. + * Only the immutable manifest-list iterator and cached-only probes run under the short registry lock; + * remote loading, future waits and cancellation all run outside it. At most window results are retained. + */ +final class IcebergManifestPrefetcher implements CloseableIterator<IcebergManifestPrefetcher.LoadedManifest> { + static final class LoadedManifest { + final ManifestFile manifest; + final ManifestCacheValue files; + + LoadedManifest(ManifestFile manifest, ManifestCacheValue files) { + this.manifest = manifest; + this.files = files; + } + } + + private static final class Pending { + final ManifestFile manifest; + final ManifestCacheValue cached; + RunnableFuture<ManifestCacheValue> future; + + Pending(ManifestFile manifest, ManifestCacheValue cached, RunnableFuture<ManifestCacheValue> future) { + this.manifest = manifest; + this.cached = cached; + this.future = future; + } + } + + private static final Pending CLOSED = new Pending(null, null, null); + + private final CloseableIterator<ManifestFile> manifests; + private final int window; + private final Function<ManifestFile, ManifestCacheValue> cached; + private final Function<ManifestFile, ManifestCacheValue> loader; + private final ConnectorContext context; + private final ConnectorTaskScheduler scheduler; + private final Semaphore readPermits; + private final BooleanSupplier allowPrefetch; + private final Deque<Pending> pending = new ArrayDeque<>(); + // Entries stay in pending until consumed, so completed results also count against the window. + // Completion callbacks never acquire the registry lock. Reserve one slot for the close signal. + private final BlockingQueue<Pending> ready; + private volatile boolean closed; + private boolean coldManifestConsumed; + + IcebergManifestPrefetcher(CloseableIterator<ManifestFile> manifests, int window, + Function<ManifestFile, ManifestCacheValue> cached, + Function<ManifestFile, ManifestCacheValue> loader, ConnectorContext context, Semaphore readPermits) { + this(manifests, window, cached, loader, context, readPermits, () -> true); + } + + IcebergManifestPrefetcher(CloseableIterator<ManifestFile> manifests, int window, + Function<ManifestFile, ManifestCacheValue> cached, + Function<ManifestFile, ManifestCacheValue> loader, ConnectorContext context, Semaphore readPermits, + BooleanSupplier allowPrefetch) { + this(manifests, window, cached, loader, context, readPermits, allowPrefetch, null); + } + + IcebergManifestPrefetcher(CloseableIterator<ManifestFile> manifests, int window, + Function<ManifestFile, ManifestCacheValue> cached, + Function<ManifestFile, ManifestCacheValue> loader, ConnectorContext context, Semaphore readPermits, + BooleanSupplier allowPrefetch, ConnectorTaskScheduler scheduler) { + this.scheduler = scheduler; + this.manifests = manifests; + this.window = window; + this.ready = new ArrayBlockingQueue<>(window + 1); + this.cached = cached; + this.loader = loader; + this.context = context; + this.readPermits = readPermits; + this.allowPrefetch = allowPrefetch; + } + + @Override + public boolean hasNext() { + synchronized (pending) { + return !closed && (!pending.isEmpty() || manifests.hasNext()); + } + } + + @Override + public LoadedManifest next() { + Pending head = null; + List<ScheduledLoad> submissions = null; + try { + synchronized (pending) { + if (closed) { + throw new CancellationException("Iceberg manifest source is closed"); + } + if (pending.isEmpty()) { + if (!manifests.hasNext()) { + throw new NoSuchElementException(); + } + ManifestFile manifest = manifests.next(); + ManifestCacheValue value = cached.apply(manifest); + if (value != null) { + // Warm cache: no futures, submissions, or speculative iterator traversal. + return new LoadedManifest(manifest, value); + } + if (window == 1 || !coldManifestConsumed || !allowPrefetch.getAsBoolean()) { + // A required serial read has no executor handoff. Expand only when useful work + // needs more metadata, without making progress depend on backend requests. + head = new Pending(manifest, null, null); + } else { + submissions = new ArrayList<>(window); + pending.addLast(submit(manifest, null, submissions)); + } + } + if (head == null) { + // If a matching file was found, pause further speculative submissions until the + // backend asks for another batch. Already-started shared reads may still complete. + while (allowPrefetch.getAsBoolean() && pending.size() < window && manifests.hasNext()) { + ManifestFile manifest = manifests.next(); + if (submissions == null) { + submissions = new ArrayList<>(window); + } + pending.addLast(submit(manifest, cached.apply(manifest), submissions)); + } + } + } + // The offline scheduler may execute inline. Even that path must never do IO under + // the pending lock: close() must be able to cancel and wake the consumer at any time. + if (submissions != null) { + submissions.forEach(ScheduledLoad::admit); + } + if (head == null) { + head = awaitReady(); + } + ManifestCacheValue value; + if (head.future == null && head.cached == null) { + value = load(head.manifest); + } else if (head.cached != null) { + value = head.cached; + } else { + // The callable publishes readiness just before FutureTask records its outcome. + // get() supplies the completion barrier and propagates failures, including Errors. + value = head.future.get(); + } + synchronized (pending) { + if (closed) { + throw new CancellationException("Iceberg manifest source is closed"); + } + if (head.future != null || head.cached != null) { + pending.remove(head); + } + coldManifestConsumed = true; + } + return new LoadedManifest(head.manifest, value); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw fail(new RuntimeException("Interrupted while loading Iceberg manifest", e)); + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + throw fail(new RuntimeException("Failed to load Iceberg manifest " + head.manifest.path(), cause)); + } catch (RuntimeException e) { + throw fail(e); + } + } + + private Pending awaitReady() throws InterruptedException { + Pending help; + synchronized (pending) { + if (closed) { + throw new CancellationException("Iceberg manifest source is closed"); + } + Pending completed = ready.poll(); + if (completed != null) { + return completed; + } + help = pending.getFirst(); + } + // If every worker is occupied by a parent, run our own queued work to guarantee progress. + // If a worker already owns it, run() is a no-op: wait for ANY completion, not this future. + if (scheduler == null) { + help.future.run(); + } + Pending completed = ready.take(); + if (closed) { + throw new CancellationException("Iceberg manifest source is closed"); + } + return completed; + } + + private Pending submit(ManifestFile manifest, ManifestCacheValue value, List<ScheduledLoad> submissions) { + Pending task = new Pending(manifest, value, null); + if (value != null) { + ready.add(task); + return task; + } + if (scheduler != null) { + ScheduledLoad load = new ScheduledLoad(task); + task.future = load; + submissions.add(load); + return task; + } + Callable<ManifestCacheValue> load = () -> { + try { + return load(manifest); + } finally { + if (!closed) { + // A close racing this publication can leave at most window completions plus + // its wakeup signal. add() must never block a metadata worker on consumption. + ready.add(task); + } + } + }; + task.future = context == null ? new FutureTask<>(load) : context.submitMetadataTask(load); + return task; + } + + /** A rejected admission keeps only this window entry; it never blocks a scheduler worker. */ + private final class ScheduledLoad extends FutureTask<ManifestCacheValue> { + private final Pending pendingLoad; + private final Runnable onSpace = this::admit; + + private ScheduledLoad(Pending pendingLoad) { + super(() -> load(pendingLoad.manifest)); + this.pendingLoad = pendingLoad; + } + + private void admit() { + if (isDone()) { + return; + } + try { + scheduler.tryExecute(this, this::setException, onSpace); + } catch (RuntimeException e) { + setException(e); + } + // Cancellation can race callback registration in tryExecute. + if (isDone()) { + scheduler.removeSpaceListener(onSpace); + } + } + + @Override + protected void done() { + scheduler.removeSpaceListener(onSpace); + if (!closed) { + ready.add(pendingLoad); + } + } + } + + private ManifestCacheValue load(ManifestFile manifest) throws InterruptedException { + // Permits are acquired ONLY by a running task, never by queued futures. Consequently a waiter + // cannot depend on an unstarted permit holder when all shared-pool workers are occupied. + // Another scan of the same query can hold every permit inside remote IO. Query cleanup + // closes its group before this producer can close the source, so observe that lifetime + // while waiting as well as after admission, without interrupting a shared cache loader. + checkReadOpen(); + while (!readPermits.tryAcquire(100, TimeUnit.MILLISECONDS)) { + checkReadOpen(); + } + ClassLoader previous = Thread.currentThread().getContextClassLoader(); + try { + checkReadOpen(); + Thread.currentThread().setContextClassLoader(IcebergManifestPrefetcher.class.getClassLoader()); + if (context == null) { + return loader.apply(manifest); + } + try { + return context.executeAuthenticated(() -> loader.apply(manifest)); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException("Failed to load Iceberg manifest " + manifest.path(), e); + } + } finally { + Thread.currentThread().setContextClassLoader(previous); + readPermits.release(); + } + } + + private void checkReadOpen() { + if (closed || (scheduler != null && scheduler.isClosed())) { + throw new CancellationException("Iceberg manifest source is closed"); + } + } + + private RuntimeException fail(RuntimeException failure) { + try { + close(); + } catch (IOException closeFailure) { + failure.addSuppressed(closeFailure); + } + return failure; + } + + @Override + public void close() throws IOException { + List<Pending> unused; + synchronized (pending) { + if (closed) { + return; + } + closed = true; + unused = new ArrayList<>(pending); + pending.clear(); + ready.clear(); + ready.add(CLOSED); + } + for (Pending task : unused) { + if (task.future != null) { + // MetaCache single-flights this immutable load across queries. Interrupting its elected + // loader would also fail unrelated waiters. Cancel queued work, but let an ongoing shared + // read finish; closed is checked before any subsequent manifest read starts. + task.future.cancel(false); Review Comment: Fixed in ff44a870a718eebf4b954177892218501b134dad. Each actual Manifest load retains a lease on the exact resolved table/FileIO owner for the entire callable body. Cancellation of its Future does not release that lease while I/O is still running; queued cancelled work acquires none. Shared-load followers now detach and resume by completion callback without occupying metadata workers. Tests cover blocked running reads, cancellation, owner closure, shared completion, and exception propagation. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/split/FileScanRangeCompactor.java: ########## @@ -0,0 +1,136 @@ +// 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.datasource.split; + +import org.apache.doris.thrift.TFileScanRange; +import org.apache.doris.thrift.TFileScanRangeParams; +import org.apache.doris.thrift.TQueryOptions; +import org.apache.doris.thrift.TScanRangeLocations; +import org.apache.doris.thrift.TScanRangeParams; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** Reduces repeated file-scan wrappers only at the transport boundary, after scan assignment. */ +public final class FileScanRangeCompactor { + private FileScanRangeCompactor() { + } + + /** + * Compact only after fetching: maxBatchSize counts the original split descriptions, not envelopes. + * Every range comes from FileQueryScanNode.splitToScanRange, so file-specific attributes remain in + * TFileRangeDesc. Keep adjacent envelopes separate when their scan parameters or locations differ. + */ + public static List<TScanRangeLocations> compactSplitBatch(List<TScanRangeLocations> scanRanges) { + if (scanRanges.size() < 2) { + return scanRanges; + } + List<TScanRangeLocations> compacted = new ArrayList<>(); + TScanRangeLocations previous = null; + TFileScanRange previousFiles = null; + boolean copied = false; + for (TScanRangeLocations current : scanRanges) { + TFileScanRange files = current.getScanRange().getExtScanRange().getFileScanRange(); Review Comment: Fixed in ff44a870a718eebf4b954177892218501b134dad. SplitProducerTest now constructs complete TScanRange/TExternalScanRange/TFileScanRange envelopes with params and ranges, so the compactor exercises valid production input instead of racing into a null fixture. ########## fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionBuilder.java: ########## @@ -174,9 +175,20 @@ public ConnectorSession build() { } } } + ConnectorStatementScope scope = captureStatementScope(); + if (scope != ConnectorStatementScope.NONE && connectContext != null) { + StmtExecutor owner = connectContext.getExecutor(); + if (owner != null) { + scope.computeIfAbsent(QueryTaskGroups.SCOPE_KEY, () -> { Review Comment: Fixed in ff44a870a718eebf4b954177892218501b134dad. The outer E230 replan loop now replaces the closed connector statement scope after failed-attempt cleanup and before planning again, following the existing inner retry lifecycle. The regression verifies the new attempt can use both task groups and that split/metadata pool group counts return to their original baseline. -- 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]
