You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "clintropolis (via GitHub)" <gi...@apache.org> on 2026/05/01 22:24:44 UTC
[PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
clintropolis opened a new pull request, #19397:
URL: https://github.com/apache/druid/pull/19397
### Description
This PR adds an async variant of `CursorFactory.makeCursorHolder` and migrates MSQ frame processors to use it, so that future cursor factories backed by partial / lazy-loaded storage can perform I/O without blocking the processing
thread. This PR introduces no partial-loading behavior on its own, instead it just establishes the integration shape for more upcoming partial-segment work that will follow this PR. Every existing cursor factory remains synchronous via the default implementation which just calls `Futures.immediateFuture(makeCursorHolder(spec))`.
Worth noting, I am planning some changes in a separate PR for V10 segments so that a partial segment can provide a `TimeBoundaryInspector` for usage by `GroupByPreShuffleFrameProcessor` without needing to download any column data so we can avoid making an async variant of it.
changes:
* add `CursorFactory.makeCursorHolderAsync(CursorBuildSpec)` for cursor factories backed by partial downloads can do I/O without blocking processing threads, with a default implementation returning `Futures.immediateFuture(makeCursorHolder(spec))` so existing implementations remain async-correct without changes
* add `GroupingEngine.processAsync` returning `ListenableFuture<Sequence<ResultRow>>` that uses `makeCursorHolderAsync`, extracting shared `processWithCursorHolder` helper from `GroupingEngine.process()`
* migrate `ScanQueryFrameProcessor.runWithSegment` to call `makeCursorHolderAsync` and yield via `ReturnOrAwait.awaitAllFutures` while the future is pending
* migrate `GroupByPreShuffleFrameProcessor.runWithSegment` cursor path to call `GroupingEngine.processAsync` and yield via `ReturnOrAwait.awaitAllFutures`
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #19397:
URL: https://github.com/apache/druid/pull/19397#discussion_r3205078651
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.druid.segment;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.guava.FutureUtils;
+
+import java.io.Closeable;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where cancelling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * Typical usage from a non-blocking caller (e.g. an MSQ frame processor):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point — close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ * The wrapper has no blocking accessor by design: callers must wait for {@link #isReady()} via
+ * {@link #addReadyCallback} (so the wait can be expressed as a yield) and then call {@link #release()}.
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Already-ready {@link AsyncCursorHolder} backed by an immediately-available {@link CursorHolder}. The default
+ * synchronous {@link CursorFactory#makeCursorHolderAsync} implementation uses this.
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ return new AsyncCursorHolder(Futures.immediateFuture(holder));
+ }
+
+ /**
+ * Wraps a {@link ListenableFuture}{@code <CursorHolder>}. Implementations doing real I/O should produce the future
+ * (e.g. by submitting to an executor) and pass it here.
+ */
+ public static AsyncCursorHolder fromFuture(ListenableFuture<CursorHolder> future)
+ {
+ return new AsyncCursorHolder(future);
+ }
+
+ private final ListenableFuture<CursorHolder> future;
+ private boolean closed = false;
+ private boolean disposed = false;
+
+ private AsyncCursorHolder(ListenableFuture<CursorHolder> future)
+ {
+ this.future = future;
+ // If close() lands before the future completes we still need to dispose the eventually-produced holder. Register
+ // a callback up-front so the holder is closed whenever it arrives if we've been closed in the meantime.
+ Futures.addCallback(
+ future,
+ new FutureCallback<>()
+ {
+ @Override
+ public void onSuccess(CursorHolder holder)
+ {
+ synchronized (AsyncCursorHolder.this) {
+ if (closed && !disposed) {
+ disposed = true;
+ try {
+ holder.close();
+ }
+ catch (Throwable ignored) {
+ // Best-effort — we're already in cleanup; nothing to do if close itself fails.
+ }
+ }
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t)
+ {
+ // Nothing to dispose: no holder was produced.
+ }
+ },
+ MoreExecutors.directExecutor()
+ );
+ }
+
+ /**
+ * Whether {@link #release()} can return without blocking. Once true, stays true. Callers that need to wait for
+ * readiness without blocking the current thread should register a {@link #addReadyCallback} and yield.
+ */
+ public boolean isReady()
+ {
+ return future.isDone();
+ }
+
+ /**
+ * Take ownership of the underlying {@link CursorHolder}. After this returns, {@link #close()} on this
+ * {@code AsyncCursorHolder} is a no-op; the caller is responsible for closing the returned holder. Useful when
+ * passing the holder to another component (e.g. a cursor-lifecycle manager) that takes ownership of it.
+ * <p>
+ * Throws {@link IllegalStateException} if the holder is not yet ready, has already been released, or this wrapper
+ * has been closed. Throws if the underlying load failed. Does not block; callers must check {@link #isReady()}
+ * first (typically by yielding via a {@link #addReadyCallback}-driven wait pattern).
+ */
+ public synchronized CursorHolder release()
+ {
+ if (closed) {
+ throw new IllegalStateException("AsyncCursorHolder is already closed");
+ }
+ if (disposed) {
+ throw new IllegalStateException("AsyncCursorHolder has already been released");
+ }
+ if (!future.isDone()) {
+ throw new IllegalStateException("AsyncCursorHolder is not ready yet");
+ }
+ final CursorHolder holder = FutureUtils.getUncheckedImmediately(future);
+ disposed = true;
+ return holder;
+ }
+
+ /**
+ * Register a callback to fire when {@link #isReady()} becomes true (whether the load succeeded or failed). If the
+ * holder is already ready, the callback fires synchronously. Multiple callbacks may be registered; each fires once.
+ */
+ public void addReadyCallback(Runnable callback)
+ {
+ future.addListener(callback, MoreExecutors.directExecutor());
+ }
+
+ /**
+ * Close the holder. Safe at any lifecycle point and idempotent:
+ * <ul>
+ * <li>Already-loaded: closes the underlying {@link CursorHolder} immediately.</li>
+ * <li>Loading in progress: marks for disposal; the produced holder is closed when the load completes.</li>
+ * <li>Load failed: no-op (nothing to dispose).</li>
+ * <li>Already closed: no-op.</li>
+ * </ul>
+ * Note that closing does NOT cancel an in-flight load — cancelling a future-of-Closeable is the exact lifecycle
+ * hazard this class exists to prevent. The load completes normally and the resulting holder is closed promptly.
Review Comment:
This is nicer, though I ended up not using `Either` for the field since it didn't save me much and was kind of awkward since internally we are always dealing with one or the other (i did use it to make it easier to share a set method internally though)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #19397:
URL: https://github.com/apache/druid/pull/19397#discussion_r3206215427
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.druid.segment;
+
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.frame.processor.ReturnOrAwait;
+import org.apache.druid.java.util.common.Either;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where canceling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * <h3>Producer protocol</h3>
+ * Producers feed results in via {@link #set(CursorHolder)} or {@link #setException(Throwable)}, both of which return
+ * a boolean. If they return {@code false}, this wrapper has already been closed and the producer is responsible for
+ * closing whatever holder it just produced.
+ * Producers may pass a {@link Runnable} canceler at construction time which runs on {@link #close()} when the wrapper
+ * is closed before the {@link #set} has been called, giving the producer an opportunity to abort its work. The canceler
+ * is best-effort: a producer may have already produced the holder by the time it observes cancellation, in which case
+ * its {@link #set} call will return false and it must close the holder it tried to set.
+ * <p>
+ * <h3>Consumer protocol</h3>
+ * Consumers wait for {@link #isReady()} via {@link #addReadyCallback}, and {@link #release()} to transfer ownership of
+ * the {@link CursorHolder} (or throw the producer exception). Calling {@link #release()} before {@link #isReady()}
+ * returns {@code true}, multiple times, or after this holder has been closed will throw a {@link DruidException}.
+ * <p>
+ * For example (using {@link ReturnOrAwait} to show intended yield-then-resume usage pattern):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point, close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Completed {@link AsyncCursorHolder} backed by an already available {@link CursorHolder}
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ final AsyncCursorHolder result = new AsyncCursorHolder(null);
+ result.set(holder);
+ return result;
+ }
+
+ @Nullable
+ private final Runnable canceler;
+
+ @GuardedBy("this")
+ @Nullable
+ private CursorHolder result = null;
+ @GuardedBy("this")
+ @Nullable
+ private Throwable error = null;
+ @GuardedBy("this")
+ private boolean closed = false;
+ @GuardedBy("this")
+ private boolean disposed = false;
+ @GuardedBy("this")
+ private final List<Runnable> readyCallbacks = new ArrayList<>();
+
+ /**
+ * @param canceler optional callback invoked from {@link #close()} when the wrapper is closed before the load has
+ * completed ({@link #set} or {@link #setException}). Producers that support cancellation should
+ * provide one; producers that don't can pass {@code null}, in which case {@link #close()} just stops
+ * observing the result.
+ */
+ public AsyncCursorHolder(@Nullable Runnable canceler)
+ {
+ this.canceler = canceler;
+ }
+
+ /**
+ * Allows producer to mark the load successful with the given holder. Returns {@code true} if accepted, {@code false}
+ * if this wrapper has already been closed, in which case the producer is responsible for closing {@link CursorHolder}
+ * itself. Throws {@link DruidException} if the load was already completed (from prior calls to this method or
+ * {@link #setException}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean set(CursorHolder holder)
+ {
+ return setInternal(Either.value(holder));
+ }
+
+ /**
+ * Allows producer to mark the load as failed. Returns {@code true} if accepted, {@code false} if this wrapper has
+ * already been closed (no holder was produced, so there's nothing for the producer to clean up). Throws
+ * {@link DruidException} if the load was already completed (from prior calls to this method or {@link #set}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean setException(Throwable t)
+ {
+ return setInternal(Either.error(t));
+ }
+
+ private boolean setInternal(Either<Throwable, CursorHolder> value)
+ {
+ final List<Runnable> callbacksToFire;
+ synchronized (this) {
+ if (closed) {
+ return false;
+ }
+ if (result != null || error != null) {
+ throw DruidException.defensive("AsyncCursorHolder is already completed");
+ }
+ if (value.isError()) {
+ error = value.error();
+ } else {
+ result = value.valueOrThrow();
+ }
+ callbacksToFire = drainCallbacks();
+ }
+ fireCallbacks(callbacksToFire);
+ return true;
+ }
+
+ /**
+ * Whether the load has completed (successfully or with failure). Once true, stays true. Callers that need to wait
+ * for readiness without blocking the current thread should register a {@link #addReadyCallback} and yield.
+ */
+ public synchronized boolean isReady()
+ {
+ return result != null || error != null;
+ }
+
+ /**
+ * Take ownership of the underlying {@link CursorHolder}. After this returns, {@link #close()} on this
+ * {@code AsyncCursorHolder} is a no-op; the caller is responsible for closing the returned holder. Useful when
+ * passing the holder to another component (e.g. a cursor-lifecycle manager) that takes ownership of it.
+ * <p>
+ * Throws {@link DruidException} if the holder is not yet ready, has already been released, or this wrapper
+ * has been closed. Wraps and rethrows the failure if the underlying load failed. Does not block; callers must
+ * check {@link #isReady()} first (typically by yielding via a {@link #addReadyCallback}-driven wait pattern).
+ */
+ public synchronized CursorHolder release()
+ {
+ if (closed) {
+ throw DruidException.defensive("AsyncCursorHolder is already closed");
+ }
+ if (disposed) {
+ throw DruidException.defensive("AsyncCursorHolder has already been released");
+ }
+ if (error != null) {
+ // pass through as is
+ if (error instanceof RuntimeException runtime) {
+ throw runtime;
+ }
+ throw DruidException.forPersona(DruidException.Persona.DEVELOPER)
+ .ofCategory(DruidException.Category.UNCATEGORIZED)
+ .build(error, error.getMessage());
+ }
+ if (result == null) {
+ throw DruidException.defensive("AsyncCursorHolder is not ready yet");
+ }
+ disposed = true;
+ return result;
+ }
+
+ /**
+ * Register a callback to fire when {@link #isReady()} becomes true (whether the load succeeded or failed). If the
+ * holder is already ready, the callback fires synchronously on the calling thread. Otherwise it fires on whatever
+ * thread invokes {@link #set} or {@link #setException}, outside the wrapper's lock so the callback may safely
+ * re-enter the wrapper. Multiple callbacks may be registered; each fires once.
+ */
+ public void addReadyCallback(Runnable callback)
+ {
+ final boolean fireImmediately;
+ synchronized (this) {
+ if (result != null || error != null) {
+ fireImmediately = true;
+ } else {
+ readyCallbacks.add(callback);
+ fireImmediately = false;
+ }
+ }
+ if (fireImmediately) {
+ callback.run();
+ }
+ }
+
+ /**
+ * Close the wrapper. Safe at any lifecycle point and idempotent:
+ * <ul>
+ * <li>Already-loaded: closes the underlying {@link CursorHolder} immediately.</li>
+ * <li>Loading in progress: invokes the canceler (if one was supplied at construction). The producer may still
+ * call {@link #set} / {@link #setException} after this; if the producer wins the race and calls {@code set}
+ * with a holder, {@code set} returns false and the producer is responsible for closing it.</li>
+ * <li>Load failed: no-op (no holder was produced).</li>
+ * <li>Already released or already closed: no-op.</li>
+ * </ul>
+ */
+ @Override
+ public void close()
+ {
+ final CursorHolder holderToClose;
+ final Runnable cancelerToRun;
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ if (disposed) {
+ // Ownership was already transferred via release(); the caller manages the holder lifecycle.
+ return;
+ }
+ if (result != null) {
+ // Result is here and no one has released it; we close it.
+ disposed = true;
+ holderToClose = result;
+ cancelerToRun = null;
+ } else if (error != null) {
+ // Load already failed; nothing to dispose.
+ holderToClose = null;
+ cancelerToRun = null;
+ } else {
+ // Load not yet completed; signal cancellation to the producer (if any).
+ holderToClose = null;
+ cancelerToRun = canceler;
+ }
+ }
+ if (holderToClose != null) {
+ try {
+ holderToClose.close();
+ }
+ catch (Throwable ignored) {
+ // Best-effort cleanup
+ }
+ }
+ if (cancelerToRun != null) {
+ try {
+ cancelerToRun.run();
+ }
+ catch (Throwable ignored) {
+ // Best-effort cancel
+ }
+ }
+ }
+
+ @GuardedBy("this")
+ private List<Runnable> drainCallbacks()
+ {
+ final List<Runnable> snapshot = List.copyOf(readyCallbacks);
+ readyCallbacks.clear();
+ return snapshot;
+ }
+
+ private static void fireCallbacks(List<Runnable> callbacks)
+ {
+ for (Runnable cb : callbacks) {
+ try {
+ cb.run();
+ }
+ catch (Throwable ignored) {
+ // Best-effort; one bad callback shouldn't break others.
Review Comment:
`log.warn` something, otherwise a callback failure will be completely silent, and this could make debugging something difficult down the road. Especially since the likely outcome of an readiness callback failing would be that things just get stalled out.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.druid.segment;
+
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.frame.processor.ReturnOrAwait;
+import org.apache.druid.java.util.common.Either;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where canceling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * <h3>Producer protocol</h3>
+ * Producers feed results in via {@link #set(CursorHolder)} or {@link #setException(Throwable)}, both of which return
+ * a boolean. If they return {@code false}, this wrapper has already been closed and the producer is responsible for
+ * closing whatever holder it just produced.
+ * Producers may pass a {@link Runnable} canceler at construction time which runs on {@link #close()} when the wrapper
+ * is closed before the {@link #set} has been called, giving the producer an opportunity to abort its work. The canceler
+ * is best-effort: a producer may have already produced the holder by the time it observes cancellation, in which case
+ * its {@link #set} call will return false and it must close the holder it tried to set.
+ * <p>
+ * <h3>Consumer protocol</h3>
+ * Consumers wait for {@link #isReady()} via {@link #addReadyCallback}, and {@link #release()} to transfer ownership of
+ * the {@link CursorHolder} (or throw the producer exception). Calling {@link #release()} before {@link #isReady()}
+ * returns {@code true}, multiple times, or after this holder has been closed will throw a {@link DruidException}.
+ * <p>
+ * For example (using {@link ReturnOrAwait} to show intended yield-then-resume usage pattern):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point, close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Completed {@link AsyncCursorHolder} backed by an already available {@link CursorHolder}
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ final AsyncCursorHolder result = new AsyncCursorHolder(null);
+ result.set(holder);
+ return result;
+ }
+
+ @Nullable
+ private final Runnable canceler;
+
+ @GuardedBy("this")
+ @Nullable
+ private CursorHolder result = null;
+ @GuardedBy("this")
+ @Nullable
+ private Throwable error = null;
+ @GuardedBy("this")
+ private boolean closed = false;
+ @GuardedBy("this")
+ private boolean disposed = false;
+ @GuardedBy("this")
+ private final List<Runnable> readyCallbacks = new ArrayList<>();
+
+ /**
+ * @param canceler optional callback invoked from {@link #close()} when the wrapper is closed before the load has
+ * completed ({@link #set} or {@link #setException}). Producers that support cancellation should
+ * provide one; producers that don't can pass {@code null}, in which case {@link #close()} just stops
+ * observing the result.
+ */
+ public AsyncCursorHolder(@Nullable Runnable canceler)
+ {
+ this.canceler = canceler;
+ }
+
+ /**
+ * Allows producer to mark the load successful with the given holder. Returns {@code true} if accepted, {@code false}
+ * if this wrapper has already been closed, in which case the producer is responsible for closing {@link CursorHolder}
+ * itself. Throws {@link DruidException} if the load was already completed (from prior calls to this method or
+ * {@link #setException}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean set(CursorHolder holder)
+ {
+ return setInternal(Either.value(holder));
+ }
+
+ /**
+ * Allows producer to mark the load as failed. Returns {@code true} if accepted, {@code false} if this wrapper has
+ * already been closed (no holder was produced, so there's nothing for the producer to clean up). Throws
+ * {@link DruidException} if the load was already completed (from prior calls to this method or {@link #set}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean setException(Throwable t)
+ {
+ return setInternal(Either.error(t));
+ }
+
+ private boolean setInternal(Either<Throwable, CursorHolder> value)
+ {
+ final List<Runnable> callbacksToFire;
+ synchronized (this) {
+ if (closed) {
+ return false;
+ }
+ if (result != null || error != null) {
+ throw DruidException.defensive("AsyncCursorHolder is already completed");
+ }
+ if (value.isError()) {
+ error = value.error();
+ } else {
+ result = value.valueOrThrow();
+ }
+ callbacksToFire = drainCallbacks();
+ }
+ fireCallbacks(callbacksToFire);
+ return true;
+ }
+
+ /**
+ * Whether the load has completed (successfully or with failure). Once true, stays true. Callers that need to wait
+ * for readiness without blocking the current thread should register a {@link #addReadyCallback} and yield.
+ */
+ public synchronized boolean isReady()
+ {
+ return result != null || error != null;
+ }
+
+ /**
+ * Take ownership of the underlying {@link CursorHolder}. After this returns, {@link #close()} on this
+ * {@code AsyncCursorHolder} is a no-op; the caller is responsible for closing the returned holder. Useful when
+ * passing the holder to another component (e.g. a cursor-lifecycle manager) that takes ownership of it.
+ * <p>
+ * Throws {@link DruidException} if the holder is not yet ready, has already been released, or this wrapper
+ * has been closed. Wraps and rethrows the failure if the underlying load failed. Does not block; callers must
+ * check {@link #isReady()} first (typically by yielding via a {@link #addReadyCallback}-driven wait pattern).
+ */
+ public synchronized CursorHolder release()
+ {
+ if (closed) {
+ throw DruidException.defensive("AsyncCursorHolder is already closed");
+ }
+ if (disposed) {
+ throw DruidException.defensive("AsyncCursorHolder has already been released");
+ }
+ if (error != null) {
+ // pass through as is
+ if (error instanceof RuntimeException runtime) {
Review Comment:
or `Error`?
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.druid.segment;
+
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.frame.processor.ReturnOrAwait;
+import org.apache.druid.java.util.common.Either;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where canceling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * <h3>Producer protocol</h3>
+ * Producers feed results in via {@link #set(CursorHolder)} or {@link #setException(Throwable)}, both of which return
+ * a boolean. If they return {@code false}, this wrapper has already been closed and the producer is responsible for
+ * closing whatever holder it just produced.
+ * Producers may pass a {@link Runnable} canceler at construction time which runs on {@link #close()} when the wrapper
+ * is closed before the {@link #set} has been called, giving the producer an opportunity to abort its work. The canceler
+ * is best-effort: a producer may have already produced the holder by the time it observes cancellation, in which case
+ * its {@link #set} call will return false and it must close the holder it tried to set.
+ * <p>
+ * <h3>Consumer protocol</h3>
+ * Consumers wait for {@link #isReady()} via {@link #addReadyCallback}, and {@link #release()} to transfer ownership of
+ * the {@link CursorHolder} (or throw the producer exception). Calling {@link #release()} before {@link #isReady()}
+ * returns {@code true}, multiple times, or after this holder has been closed will throw a {@link DruidException}.
+ * <p>
+ * For example (using {@link ReturnOrAwait} to show intended yield-then-resume usage pattern):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point, close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Completed {@link AsyncCursorHolder} backed by an already available {@link CursorHolder}
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ final AsyncCursorHolder result = new AsyncCursorHolder(null);
+ result.set(holder);
+ return result;
+ }
+
+ @Nullable
+ private final Runnable canceler;
+
+ @GuardedBy("this")
+ @Nullable
+ private CursorHolder result = null;
+ @GuardedBy("this")
+ @Nullable
+ private Throwable error = null;
+ @GuardedBy("this")
+ private boolean closed = false;
+ @GuardedBy("this")
+ private boolean disposed = false;
+ @GuardedBy("this")
+ private final List<Runnable> readyCallbacks = new ArrayList<>();
+
+ /**
+ * @param canceler optional callback invoked from {@link #close()} when the wrapper is closed before the load has
+ * completed ({@link #set} or {@link #setException}). Producers that support cancellation should
+ * provide one; producers that don't can pass {@code null}, in which case {@link #close()} just stops
+ * observing the result.
+ */
+ public AsyncCursorHolder(@Nullable Runnable canceler)
+ {
+ this.canceler = canceler;
+ }
+
+ /**
+ * Allows producer to mark the load successful with the given holder. Returns {@code true} if accepted, {@code false}
+ * if this wrapper has already been closed, in which case the producer is responsible for closing {@link CursorHolder}
+ * itself. Throws {@link DruidException} if the load was already completed (from prior calls to this method or
+ * {@link #setException}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean set(CursorHolder holder)
+ {
+ return setInternal(Either.value(holder));
+ }
+
+ /**
+ * Allows producer to mark the load as failed. Returns {@code true} if accepted, {@code false} if this wrapper has
+ * already been closed (no holder was produced, so there's nothing for the producer to clean up). Throws
+ * {@link DruidException} if the load was already completed (from prior calls to this method or {@link #set}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean setException(Throwable t)
+ {
+ return setInternal(Either.error(t));
+ }
+
+ private boolean setInternal(Either<Throwable, CursorHolder> value)
+ {
+ final List<Runnable> callbacksToFire;
+ synchronized (this) {
+ if (closed) {
+ return false;
+ }
+ if (result != null || error != null) {
+ throw DruidException.defensive("AsyncCursorHolder is already completed");
+ }
+ if (value.isError()) {
+ error = value.error();
+ } else {
+ result = value.valueOrThrow();
+ }
+ callbacksToFire = drainCallbacks();
+ }
+ fireCallbacks(callbacksToFire);
+ return true;
+ }
+
+ /**
+ * Whether the load has completed (successfully or with failure). Once true, stays true. Callers that need to wait
+ * for readiness without blocking the current thread should register a {@link #addReadyCallback} and yield.
+ */
+ public synchronized boolean isReady()
+ {
+ return result != null || error != null;
+ }
+
+ /**
+ * Take ownership of the underlying {@link CursorHolder}. After this returns, {@link #close()} on this
+ * {@code AsyncCursorHolder} is a no-op; the caller is responsible for closing the returned holder. Useful when
+ * passing the holder to another component (e.g. a cursor-lifecycle manager) that takes ownership of it.
+ * <p>
+ * Throws {@link DruidException} if the holder is not yet ready, has already been released, or this wrapper
+ * has been closed. Wraps and rethrows the failure if the underlying load failed. Does not block; callers must
+ * check {@link #isReady()} first (typically by yielding via a {@link #addReadyCallback}-driven wait pattern).
+ */
+ public synchronized CursorHolder release()
+ {
+ if (closed) {
+ throw DruidException.defensive("AsyncCursorHolder is already closed");
+ }
+ if (disposed) {
+ throw DruidException.defensive("AsyncCursorHolder has already been released");
+ }
+ if (error != null) {
+ // pass through as is
+ if (error instanceof RuntimeException runtime) {
+ throw runtime;
+ }
+ throw DruidException.forPersona(DruidException.Persona.DEVELOPER)
+ .ofCategory(DruidException.Category.UNCATEGORIZED)
+ .build(error, error.getMessage());
+ }
+ if (result == null) {
+ throw DruidException.defensive("AsyncCursorHolder is not ready yet");
+ }
+ disposed = true;
+ return result;
+ }
+
+ /**
+ * Register a callback to fire when {@link #isReady()} becomes true (whether the load succeeded or failed). If the
+ * holder is already ready, the callback fires synchronously on the calling thread. Otherwise it fires on whatever
+ * thread invokes {@link #set} or {@link #setException}, outside the wrapper's lock so the callback may safely
+ * re-enter the wrapper. Multiple callbacks may be registered; each fires once.
+ */
+ public void addReadyCallback(Runnable callback)
+ {
+ final boolean fireImmediately;
+ synchronized (this) {
+ if (result != null || error != null) {
+ fireImmediately = true;
+ } else {
+ readyCallbacks.add(callback);
+ fireImmediately = false;
+ }
+ }
+ if (fireImmediately) {
+ callback.run();
+ }
+ }
+
+ /**
+ * Close the wrapper. Safe at any lifecycle point and idempotent:
+ * <ul>
+ * <li>Already-loaded: closes the underlying {@link CursorHolder} immediately.</li>
+ * <li>Loading in progress: invokes the canceler (if one was supplied at construction). The producer may still
+ * call {@link #set} / {@link #setException} after this; if the producer wins the race and calls {@code set}
+ * with a holder, {@code set} returns false and the producer is responsible for closing it.</li>
+ * <li>Load failed: no-op (no holder was produced).</li>
+ * <li>Already released or already closed: no-op.</li>
+ * </ul>
+ */
+ @Override
+ public void close()
+ {
+ final CursorHolder holderToClose;
+ final Runnable cancelerToRun;
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ if (disposed) {
+ // Ownership was already transferred via release(); the caller manages the holder lifecycle.
+ return;
+ }
+ if (result != null) {
+ // Result is here and no one has released it; we close it.
Review Comment:
I think this logic would end up being simpler if you had nulled out `result` in `release`. It would then just be: always close `result` if it is nonnull, because if it's nonnull, we own it.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.druid.segment;
+
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.frame.processor.ReturnOrAwait;
+import org.apache.druid.java.util.common.Either;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where canceling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * <h3>Producer protocol</h3>
+ * Producers feed results in via {@link #set(CursorHolder)} or {@link #setException(Throwable)}, both of which return
+ * a boolean. If they return {@code false}, this wrapper has already been closed and the producer is responsible for
+ * closing whatever holder it just produced.
+ * Producers may pass a {@link Runnable} canceler at construction time which runs on {@link #close()} when the wrapper
+ * is closed before the {@link #set} has been called, giving the producer an opportunity to abort its work. The canceler
+ * is best-effort: a producer may have already produced the holder by the time it observes cancellation, in which case
+ * its {@link #set} call will return false and it must close the holder it tried to set.
+ * <p>
+ * <h3>Consumer protocol</h3>
+ * Consumers wait for {@link #isReady()} via {@link #addReadyCallback}, and {@link #release()} to transfer ownership of
+ * the {@link CursorHolder} (or throw the producer exception). Calling {@link #release()} before {@link #isReady()}
+ * returns {@code true}, multiple times, or after this holder has been closed will throw a {@link DruidException}.
+ * <p>
+ * For example (using {@link ReturnOrAwait} to show intended yield-then-resume usage pattern):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point, close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Completed {@link AsyncCursorHolder} backed by an already available {@link CursorHolder}
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ final AsyncCursorHolder result = new AsyncCursorHolder(null);
+ result.set(holder);
+ return result;
+ }
+
+ @Nullable
+ private final Runnable canceler;
+
+ @GuardedBy("this")
+ @Nullable
+ private CursorHolder result = null;
+ @GuardedBy("this")
+ @Nullable
+ private Throwable error = null;
+ @GuardedBy("this")
+ private boolean closed = false;
+ @GuardedBy("this")
+ private boolean disposed = false;
+ @GuardedBy("this")
+ private final List<Runnable> readyCallbacks = new ArrayList<>();
+
+ /**
+ * @param canceler optional callback invoked from {@link #close()} when the wrapper is closed before the load has
+ * completed ({@link #set} or {@link #setException}). Producers that support cancellation should
+ * provide one; producers that don't can pass {@code null}, in which case {@link #close()} just stops
+ * observing the result.
+ */
+ public AsyncCursorHolder(@Nullable Runnable canceler)
+ {
+ this.canceler = canceler;
+ }
+
+ /**
+ * Allows producer to mark the load successful with the given holder. Returns {@code true} if accepted, {@code false}
+ * if this wrapper has already been closed, in which case the producer is responsible for closing {@link CursorHolder}
+ * itself. Throws {@link DruidException} if the load was already completed (from prior calls to this method or
+ * {@link #setException}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean set(CursorHolder holder)
+ {
+ return setInternal(Either.value(holder));
+ }
+
+ /**
+ * Allows producer to mark the load as failed. Returns {@code true} if accepted, {@code false} if this wrapper has
+ * already been closed (no holder was produced, so there's nothing for the producer to clean up). Throws
+ * {@link DruidException} if the load was already completed (from prior calls to this method or {@link #set}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean setException(Throwable t)
+ {
+ return setInternal(Either.error(t));
+ }
+
+ private boolean setInternal(Either<Throwable, CursorHolder> value)
+ {
+ final List<Runnable> callbacksToFire;
+ synchronized (this) {
+ if (closed) {
+ return false;
+ }
+ if (result != null || error != null) {
+ throw DruidException.defensive("AsyncCursorHolder is already completed");
+ }
+ if (value.isError()) {
+ error = value.error();
+ } else {
+ result = value.valueOrThrow();
+ }
+ callbacksToFire = drainCallbacks();
+ }
+ fireCallbacks(callbacksToFire);
+ return true;
+ }
+
+ /**
+ * Whether the load has completed (successfully or with failure). Once true, stays true. Callers that need to wait
+ * for readiness without blocking the current thread should register a {@link #addReadyCallback} and yield.
+ */
+ public synchronized boolean isReady()
+ {
+ return result != null || error != null;
+ }
+
+ /**
+ * Take ownership of the underlying {@link CursorHolder}. After this returns, {@link #close()} on this
+ * {@code AsyncCursorHolder} is a no-op; the caller is responsible for closing the returned holder. Useful when
+ * passing the holder to another component (e.g. a cursor-lifecycle manager) that takes ownership of it.
+ * <p>
+ * Throws {@link DruidException} if the holder is not yet ready, has already been released, or this wrapper
+ * has been closed. Wraps and rethrows the failure if the underlying load failed. Does not block; callers must
+ * check {@link #isReady()} first (typically by yielding via a {@link #addReadyCallback}-driven wait pattern).
+ */
+ public synchronized CursorHolder release()
+ {
+ if (closed) {
+ throw DruidException.defensive("AsyncCursorHolder is already closed");
+ }
+ if (disposed) {
+ throw DruidException.defensive("AsyncCursorHolder has already been released");
+ }
+ if (error != null) {
+ // pass through as is
+ if (error instanceof RuntimeException runtime) {
+ throw runtime;
+ }
+ throw DruidException.forPersona(DruidException.Persona.DEVELOPER)
+ .ofCategory(DruidException.Category.UNCATEGORIZED)
+ .build(error, error.getMessage());
+ }
+ if (result == null) {
+ throw DruidException.defensive("AsyncCursorHolder is not ready yet");
+ }
+ disposed = true;
Review Comment:
set `result` to `null` here to allow GC? The AsyncCursorHolder can live longer than the reference from `release()` if the AsyncCursorHolder is added to a high-level `Closer` or `withBaggage` or something like that.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.druid.segment;
+
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.frame.processor.ReturnOrAwait;
+import org.apache.druid.java.util.common.Either;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where canceling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * <h3>Producer protocol</h3>
+ * Producers feed results in via {@link #set(CursorHolder)} or {@link #setException(Throwable)}, both of which return
+ * a boolean. If they return {@code false}, this wrapper has already been closed and the producer is responsible for
+ * closing whatever holder it just produced.
+ * Producers may pass a {@link Runnable} canceler at construction time which runs on {@link #close()} when the wrapper
+ * is closed before the {@link #set} has been called, giving the producer an opportunity to abort its work. The canceler
+ * is best-effort: a producer may have already produced the holder by the time it observes cancellation, in which case
+ * its {@link #set} call will return false and it must close the holder it tried to set.
+ * <p>
+ * <h3>Consumer protocol</h3>
+ * Consumers wait for {@link #isReady()} via {@link #addReadyCallback}, and {@link #release()} to transfer ownership of
+ * the {@link CursorHolder} (or throw the producer exception). Calling {@link #release()} before {@link #isReady()}
+ * returns {@code true}, multiple times, or after this holder has been closed will throw a {@link DruidException}.
+ * <p>
+ * For example (using {@link ReturnOrAwait} to show intended yield-then-resume usage pattern):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point, close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Completed {@link AsyncCursorHolder} backed by an already available {@link CursorHolder}
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ final AsyncCursorHolder result = new AsyncCursorHolder(null);
+ result.set(holder);
+ return result;
+ }
+
+ @Nullable
+ private final Runnable canceler;
+
+ @GuardedBy("this")
+ @Nullable
+ private CursorHolder result = null;
+ @GuardedBy("this")
+ @Nullable
+ private Throwable error = null;
+ @GuardedBy("this")
+ private boolean closed = false;
+ @GuardedBy("this")
+ private boolean disposed = false;
+ @GuardedBy("this")
+ private final List<Runnable> readyCallbacks = new ArrayList<>();
+
+ /**
+ * @param canceler optional callback invoked from {@link #close()} when the wrapper is closed before the load has
+ * completed ({@link #set} or {@link #setException}). Producers that support cancellation should
+ * provide one; producers that don't can pass {@code null}, in which case {@link #close()} just stops
+ * observing the result.
+ */
+ public AsyncCursorHolder(@Nullable Runnable canceler)
+ {
+ this.canceler = canceler;
+ }
+
+ /**
+ * Allows producer to mark the load successful with the given holder. Returns {@code true} if accepted, {@code false}
+ * if this wrapper has already been closed, in which case the producer is responsible for closing {@link CursorHolder}
+ * itself. Throws {@link DruidException} if the load was already completed (from prior calls to this method or
+ * {@link #setException}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
Review Comment:
What does this comment mean? Seems like there's only one lock (`this`) and it should be fine to re-enter it.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.druid.segment;
+
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.frame.processor.ReturnOrAwait;
+import org.apache.druid.java.util.common.Either;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where canceling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * <h3>Producer protocol</h3>
+ * Producers feed results in via {@link #set(CursorHolder)} or {@link #setException(Throwable)}, both of which return
+ * a boolean. If they return {@code false}, this wrapper has already been closed and the producer is responsible for
+ * closing whatever holder it just produced.
+ * Producers may pass a {@link Runnable} canceler at construction time which runs on {@link #close()} when the wrapper
+ * is closed before the {@link #set} has been called, giving the producer an opportunity to abort its work. The canceler
+ * is best-effort: a producer may have already produced the holder by the time it observes cancellation, in which case
+ * its {@link #set} call will return false and it must close the holder it tried to set.
+ * <p>
+ * <h3>Consumer protocol</h3>
+ * Consumers wait for {@link #isReady()} via {@link #addReadyCallback}, and {@link #release()} to transfer ownership of
+ * the {@link CursorHolder} (or throw the producer exception). Calling {@link #release()} before {@link #isReady()}
+ * returns {@code true}, multiple times, or after this holder has been closed will throw a {@link DruidException}.
+ * <p>
+ * For example (using {@link ReturnOrAwait} to show intended yield-then-resume usage pattern):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point, close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Completed {@link AsyncCursorHolder} backed by an already available {@link CursorHolder}
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ final AsyncCursorHolder result = new AsyncCursorHolder(null);
+ result.set(holder);
+ return result;
+ }
+
+ @Nullable
+ private final Runnable canceler;
+
+ @GuardedBy("this")
+ @Nullable
+ private CursorHolder result = null;
+ @GuardedBy("this")
+ @Nullable
+ private Throwable error = null;
+ @GuardedBy("this")
+ private boolean closed = false;
+ @GuardedBy("this")
+ private boolean disposed = false;
+ @GuardedBy("this")
+ private final List<Runnable> readyCallbacks = new ArrayList<>();
+
+ /**
+ * @param canceler optional callback invoked from {@link #close()} when the wrapper is closed before the load has
+ * completed ({@link #set} or {@link #setException}). Producers that support cancellation should
+ * provide one; producers that don't can pass {@code null}, in which case {@link #close()} just stops
+ * observing the result.
+ */
+ public AsyncCursorHolder(@Nullable Runnable canceler)
+ {
+ this.canceler = canceler;
+ }
+
+ /**
+ * Allows producer to mark the load successful with the given holder. Returns {@code true} if accepted, {@code false}
+ * if this wrapper has already been closed, in which case the producer is responsible for closing {@link CursorHolder}
+ * itself. Throws {@link DruidException} if the load was already completed (from prior calls to this method or
+ * {@link #setException}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean set(CursorHolder holder)
+ {
+ return setInternal(Either.value(holder));
+ }
+
+ /**
+ * Allows producer to mark the load as failed. Returns {@code true} if accepted, {@code false} if this wrapper has
+ * already been closed (no holder was produced, so there's nothing for the producer to clean up). Throws
+ * {@link DruidException} if the load was already completed (from prior calls to this method or {@link #set}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean setException(Throwable t)
+ {
+ return setInternal(Either.error(t));
+ }
+
+ private boolean setInternal(Either<Throwable, CursorHolder> value)
+ {
+ final List<Runnable> callbacksToFire;
+ synchronized (this) {
+ if (closed) {
+ return false;
+ }
+ if (result != null || error != null) {
+ throw DruidException.defensive("AsyncCursorHolder is already completed");
+ }
+ if (value.isError()) {
+ error = value.error();
+ } else {
+ result = value.valueOrThrow();
+ }
+ callbacksToFire = drainCallbacks();
+ }
+ fireCallbacks(callbacksToFire);
+ return true;
+ }
+
+ /**
+ * Whether the load has completed (successfully or with failure). Once true, stays true. Callers that need to wait
+ * for readiness without blocking the current thread should register a {@link #addReadyCallback} and yield.
+ */
+ public synchronized boolean isReady()
+ {
+ return result != null || error != null;
+ }
+
+ /**
+ * Take ownership of the underlying {@link CursorHolder}. After this returns, {@link #close()} on this
+ * {@code AsyncCursorHolder} is a no-op; the caller is responsible for closing the returned holder. Useful when
+ * passing the holder to another component (e.g. a cursor-lifecycle manager) that takes ownership of it.
+ * <p>
+ * Throws {@link DruidException} if the holder is not yet ready, has already been released, or this wrapper
+ * has been closed. Wraps and rethrows the failure if the underlying load failed. Does not block; callers must
+ * check {@link #isReady()} first (typically by yielding via a {@link #addReadyCallback}-driven wait pattern).
+ */
+ public synchronized CursorHolder release()
+ {
+ if (closed) {
+ throw DruidException.defensive("AsyncCursorHolder is already closed");
+ }
+ if (disposed) {
+ throw DruidException.defensive("AsyncCursorHolder has already been released");
+ }
+ if (error != null) {
+ // pass through as is
+ if (error instanceof RuntimeException runtime) {
+ throw runtime;
+ }
+ throw DruidException.forPersona(DruidException.Persona.DEVELOPER)
+ .ofCategory(DruidException.Category.UNCATEGORIZED)
+ .build(error, error.getMessage());
+ }
+ if (result == null) {
+ throw DruidException.defensive("AsyncCursorHolder is not ready yet");
+ }
+ disposed = true;
+ return result;
+ }
+
+ /**
+ * Register a callback to fire when {@link #isReady()} becomes true (whether the load succeeded or failed). If the
+ * holder is already ready, the callback fires synchronously on the calling thread. Otherwise it fires on whatever
+ * thread invokes {@link #set} or {@link #setException}, outside the wrapper's lock so the callback may safely
+ * re-enter the wrapper. Multiple callbacks may be registered; each fires once.
+ */
+ public void addReadyCallback(Runnable callback)
+ {
+ final boolean fireImmediately;
+ synchronized (this) {
+ if (result != null || error != null) {
+ fireImmediately = true;
+ } else {
+ readyCallbacks.add(callback);
+ fireImmediately = false;
+ }
+ }
+ if (fireImmediately) {
+ callback.run();
+ }
+ }
+
+ /**
+ * Close the wrapper. Safe at any lifecycle point and idempotent:
+ * <ul>
+ * <li>Already-loaded: closes the underlying {@link CursorHolder} immediately.</li>
+ * <li>Loading in progress: invokes the canceler (if one was supplied at construction). The producer may still
+ * call {@link #set} / {@link #setException} after this; if the producer wins the race and calls {@code set}
+ * with a holder, {@code set} returns false and the producer is responsible for closing it.</li>
+ * <li>Load failed: no-op (no holder was produced).</li>
+ * <li>Already released or already closed: no-op.</li>
+ * </ul>
+ */
+ @Override
+ public void close()
+ {
+ final CursorHolder holderToClose;
+ final Runnable cancelerToRun;
+ synchronized (this) {
+ if (closed) {
+ return;
Review Comment:
Instead of idempotent close, how about throwing a defensive error here? Double-close can be indicative of resource management problems, and an exception would help us identify those potential issues.
If you do this then the javadoc should be updated too.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.druid.segment;
+
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.frame.processor.ReturnOrAwait;
+import org.apache.druid.java.util.common.Either;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where canceling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * <h3>Producer protocol</h3>
+ * Producers feed results in via {@link #set(CursorHolder)} or {@link #setException(Throwable)}, both of which return
+ * a boolean. If they return {@code false}, this wrapper has already been closed and the producer is responsible for
+ * closing whatever holder it just produced.
+ * Producers may pass a {@link Runnable} canceler at construction time which runs on {@link #close()} when the wrapper
+ * is closed before the {@link #set} has been called, giving the producer an opportunity to abort its work. The canceler
+ * is best-effort: a producer may have already produced the holder by the time it observes cancellation, in which case
+ * its {@link #set} call will return false and it must close the holder it tried to set.
+ * <p>
+ * <h3>Consumer protocol</h3>
+ * Consumers wait for {@link #isReady()} via {@link #addReadyCallback}, and {@link #release()} to transfer ownership of
+ * the {@link CursorHolder} (or throw the producer exception). Calling {@link #release()} before {@link #isReady()}
+ * returns {@code true}, multiple times, or after this holder has been closed will throw a {@link DruidException}.
+ * <p>
+ * For example (using {@link ReturnOrAwait} to show intended yield-then-resume usage pattern):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point, close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Completed {@link AsyncCursorHolder} backed by an already available {@link CursorHolder}
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ final AsyncCursorHolder result = new AsyncCursorHolder(null);
+ result.set(holder);
+ return result;
+ }
+
+ @Nullable
+ private final Runnable canceler;
+
+ @GuardedBy("this")
+ @Nullable
+ private CursorHolder result = null;
+ @GuardedBy("this")
+ @Nullable
+ private Throwable error = null;
+ @GuardedBy("this")
+ private boolean closed = false;
+ @GuardedBy("this")
+ private boolean disposed = false;
+ @GuardedBy("this")
+ private final List<Runnable> readyCallbacks = new ArrayList<>();
+
+ /**
+ * @param canceler optional callback invoked from {@link #close()} when the wrapper is closed before the load has
+ * completed ({@link #set} or {@link #setException}). Producers that support cancellation should
+ * provide one; producers that don't can pass {@code null}, in which case {@link #close()} just stops
+ * observing the result.
+ */
+ public AsyncCursorHolder(@Nullable Runnable canceler)
+ {
+ this.canceler = canceler;
+ }
+
+ /**
+ * Allows producer to mark the load successful with the given holder. Returns {@code true} if accepted, {@code false}
+ * if this wrapper has already been closed, in which case the producer is responsible for closing {@link CursorHolder}
+ * itself. Throws {@link DruidException} if the load was already completed (from prior calls to this method or
+ * {@link #setException}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean set(CursorHolder holder)
Review Comment:
This method should reject null `holder` with a defensive exception.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #19397:
URL: https://github.com/apache/druid/pull/19397#discussion_r3176204717
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java:
##########
@@ -297,21 +304,29 @@ protected ReturnOrAwait<SegmentsInputSlice> runWithDataServerQuery(final DataSer
protected ReturnOrAwait<Unit> runWithSegment(final SegmentReferenceHolder segmentHolder) throws IOException
{
if (cursor == null) {
- final Segment segment = mapSegment(segmentHolder, closer);
- final CursorFactory cursorFactory = segment.as(CursorFactory.class);
- if (cursorFactory == null) {
- throw DruidException.defensive(
- "Null cursor factory found. Probably trying to issue a query against a segment being memory unmapped."
+ if (cursorHolderFuture == null) {
+ final Segment segment = mapSegment(segmentHolder, closer);
+ final CursorFactory cursorFactory = segment.as(CursorFactory.class);
+ if (cursorFactory == null) {
+ throw DruidException.defensive(
+ "Null cursor factory found. Probably trying to issue a query against a segment being memory unmapped."
+ );
+ }
+
+ cursorHolderFuture = cursorFactory.makeCursorHolderAsync(
Review Comment:
Handling futures that return closeable things is tricky. Maybe we can improve it by changing the return of `makeCursorHolderAsync` from `ListenableFuture<CursorHolder>` to `AsyncCursorHolder` that is closeable and has methods `get()` (blocks if not ready), `close()` (closes the resource no matter where it is in its lifecycle), and `addReadyCallback(Runnable)` (used by nonblocking callers to learn when `get` is ready).
The problem with the future approach is that once this call site gets the `cursorHolderFuture`, it's responsible for monitoring the future and closing `cursorHolder` if the future resolves successfully. This has to be done even if the processor is canceled before it has a chance to run through normally. It requires extra carefulness and is easy to mess up.
One way it can be handled is by attaching a callback in `cleanup` that closes the holder in `onSuccess`, like:
```
if (cursorHolderFuture != null) {
Futures.addCallback(
cursorHolderFuture,
new FutureCallback<>() {
void onSuccess(CursorHolder holder) { holder.close(); }
void onFailure(Throwable t) { /* nothing */ }
}
);
}
```
But even with this structure, it's important watch out for pitfalls. A big one is that you can never cancel a future that returns a closeable thing like CursorHolder. Cancellation of the future can cause the object to be created, then orphaned and eventually GCed without being closed.
If we can avoid these problems by returning something *directly* closeable (like this `AsyncCursorHolder` idea) rather than future-of-closeable, then the caller code becomes simpler.
##########
processing/src/main/java/org/apache/druid/query/groupby/GroupingEngine.java:
##########
@@ -494,21 +534,38 @@ public Sequence<ResultRow> process(
if (intervals.size() != 1) {
throw new IAE("Should only have one interval, got[%s]", intervals);
}
+ }
- final ResourceHolder<ByteBuffer> bufferHolder = bufferPool.take();
+ private Sequence<ResultRow> processWithCursorHolder(
+ GroupByQuery query,
+ CursorFactory cursorFactory,
+ CursorHolder cursorHolder,
+ @Nullable TimeBoundaryInspector timeBoundaryInspector,
+ NonBlockingPool<ByteBuffer> bufferPool,
+ CursorBuildSpec buildSpec
+ )
+ {
+ final GroupByQueryConfig querySpecificConfig = configSupplier.get().withOverrides(query);
+
+ final ResourceHolder<ByteBuffer> bufferHolder;
+ try {
+ bufferHolder = bufferPool.take();
+ }
+ catch (Throwable e) {
+ CloseableUtils.closeAndWrapExceptions(cursorHolder);
Review Comment:
`throw CloseableUtils.closeAndWrapInCatch(e, cursorHolder)` will properly retain exceptions from closing `cursorHolder` as suppressed exceptions on `e`.
Although, there's probably some way of structuring this code to use the `Closer` to handle this better. Like, create the `Closer` first, register `cursorHolder` right after it's created, start the main `try`, then register `bufferHolder` after it's acquired. If the buffer fails to be acquired then the `catch` will close the `Closer` and release the `cursorHolder`.
##########
processing/src/main/java/org/apache/druid/query/groupby/GroupingEngine.java:
##########
@@ -482,8 +484,46 @@ public Sequence<ResultRow> process(
@Nullable GroupByQueryMetrics groupByQueryMetrics
)
{
- final GroupByQueryConfig querySpecificConfig = configSupplier.get().withOverrides(query);
+ validateForProcess(query, cursorFactory);
+ final CursorBuildSpec buildSpec = makeCursorBuildSpec(query, groupByQueryMetrics);
+ final CursorHolder cursorHolder = cursorFactory.makeCursorHolder(buildSpec);
+ return processWithCursorHolder(query, cursorFactory, cursorHolder, timeBoundaryInspector, bufferPool, buildSpec);
+ }
+
+ /**
+ * Asynchronous variant of {@link #process} that obtains the {@link CursorHolder} from
+ * {@link CursorFactory#makeCursorHolderAsync} so callers running on threads that must not block on I/O
+ * (e.g. MSQ frame processors) can yield via {@link org.apache.druid.frame.processor.ReturnOrAwait#awaitAllFutures}
+ * until the cursor holder is ready.
+ * <p>
+ * The processing-buffer reservation from {@code bufferPool} is deferred until the cursor holder is available, so
+ * the buffer is not held during cursor-holder I/O.
+ */
+ public ListenableFuture<Sequence<ResultRow>> processAsync(
+ GroupByQuery query,
+ CursorFactory cursorFactory,
+ @Nullable TimeBoundaryInspector timeBoundaryInspector,
+ NonBlockingPool<ByteBuffer> bufferPool,
+ @Nullable GroupByQueryMetrics groupByQueryMetrics
+ )
+ {
+ validateForProcess(query, cursorFactory);
+ final CursorBuildSpec buildSpec = makeCursorBuildSpec(query, groupByQueryMetrics);
+ return FutureUtils.transform(
+ cursorFactory.makeCursorHolderAsync(buildSpec),
+ cursorHolder -> processWithCursorHolder(
Review Comment:
Won't this end up doing the main processing in whatever thread happened to resolve the `makeCursorHolderAsync` future? (Because `FutureUtils.transform` uses a direct executor.) Possibly that'd be in a virtual storage loader thread.
I think we'll need to either adjust this method to accept a `ListenableExecutorService` that will be used to run `processWithCursorHolder`, or break it up so callers first call `groupingEngine.makeCursorHolderAsync` and then call `groupingEngine.processCursorHolder`.
##########
processing/src/main/java/org/apache/druid/query/groupby/GroupingEngine.java:
##########
@@ -482,8 +484,46 @@ public Sequence<ResultRow> process(
@Nullable GroupByQueryMetrics groupByQueryMetrics
)
{
- final GroupByQueryConfig querySpecificConfig = configSupplier.get().withOverrides(query);
+ validateForProcess(query, cursorFactory);
+ final CursorBuildSpec buildSpec = makeCursorBuildSpec(query, groupByQueryMetrics);
+ final CursorHolder cursorHolder = cursorFactory.makeCursorHolder(buildSpec);
+ return processWithCursorHolder(query, cursorFactory, cursorHolder, timeBoundaryInspector, bufferPool, buildSpec);
+ }
+
+ /**
+ * Asynchronous variant of {@link #process} that obtains the {@link CursorHolder} from
+ * {@link CursorFactory#makeCursorHolderAsync} so callers running on threads that must not block on I/O
+ * (e.g. MSQ frame processors) can yield via {@link org.apache.druid.frame.processor.ReturnOrAwait#awaitAllFutures}
+ * until the cursor holder is ready.
Review Comment:
This javadoc feels a bit too specific to me. It's enough to say that this is an asynchronous variant of `process` that uses `CursorFactory#makeCursorHolderAsync` to avoid blocking on acquisition of `CursorHolder`.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #19397:
URL: https://github.com/apache/druid/pull/19397#discussion_r3176204717
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java:
##########
@@ -297,21 +304,29 @@ protected ReturnOrAwait<SegmentsInputSlice> runWithDataServerQuery(final DataSer
protected ReturnOrAwait<Unit> runWithSegment(final SegmentReferenceHolder segmentHolder) throws IOException
{
if (cursor == null) {
- final Segment segment = mapSegment(segmentHolder, closer);
- final CursorFactory cursorFactory = segment.as(CursorFactory.class);
- if (cursorFactory == null) {
- throw DruidException.defensive(
- "Null cursor factory found. Probably trying to issue a query against a segment being memory unmapped."
+ if (cursorHolderFuture == null) {
+ final Segment segment = mapSegment(segmentHolder, closer);
+ final CursorFactory cursorFactory = segment.as(CursorFactory.class);
+ if (cursorFactory == null) {
+ throw DruidException.defensive(
+ "Null cursor factory found. Probably trying to issue a query against a segment being memory unmapped."
+ );
+ }
+
+ cursorHolderFuture = cursorFactory.makeCursorHolderAsync(
Review Comment:
Handling futures that return closeable things is tricky. Maybe we can improve it by changing the return of `makeCursorHolderAsync` from `ListenableFuture<CursorHolder>` to `AsyncCursorHolder` that is closeable and has methods `get()` (blocks if not ready), `close()` (closes the resource no matter where it is in its lifecycle), and `addReadyCallback(Runnable)` (used by nonblocking callers to learn when `get` is ready).
The problem with the future approach is that once this call site gets the `cursorHolderFuture`, it's responsible for monitoring the future and closing `cursorHolder` if the future resolves successfully. This has to be done even if the processor is canceled before it has a chance to run through normally. It requires extra carefulness and is easy to mess up.
One way it can be handled is by attaching a callback in `cleanup` that closes the holder in `onSuccess`, like:
```
if (cursorHolderFuture != null) {
Futures.addCallback(
cursorHolderFuture,
new FutureCallback<>() {
void onSuccess(CursorHolder holder) { holder.close(); }
void onFailure(Throwable t) { /* nothing */ }
}
);
}
```
But even with this structure, it's important watch out for pitfalls. A big one is that you can never cancel a future that returns a closeable thing. Cancellation of the future can cause the object to be created, then orphaned and eventually GCed without being closed.
If we can avoid these problems by returning something *directly* closeable (like this `AsyncCursorHolder` idea) rather than future-of-closeable, then the caller code becomes simpler.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #19397:
URL: https://github.com/apache/druid/pull/19397#discussion_r3206343297
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.druid.segment;
+
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.frame.processor.ReturnOrAwait;
+import org.apache.druid.java.util.common.Either;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where canceling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * <h3>Producer protocol</h3>
+ * Producers feed results in via {@link #set(CursorHolder)} or {@link #setException(Throwable)}, both of which return
+ * a boolean. If they return {@code false}, this wrapper has already been closed and the producer is responsible for
+ * closing whatever holder it just produced.
+ * Producers may pass a {@link Runnable} canceler at construction time which runs on {@link #close()} when the wrapper
+ * is closed before the {@link #set} has been called, giving the producer an opportunity to abort its work. The canceler
+ * is best-effort: a producer may have already produced the holder by the time it observes cancellation, in which case
+ * its {@link #set} call will return false and it must close the holder it tried to set.
+ * <p>
+ * <h3>Consumer protocol</h3>
+ * Consumers wait for {@link #isReady()} via {@link #addReadyCallback}, and {@link #release()} to transfer ownership of
+ * the {@link CursorHolder} (or throw the producer exception). Calling {@link #release()} before {@link #isReady()}
+ * returns {@code true}, multiple times, or after this holder has been closed will throw a {@link DruidException}.
+ * <p>
+ * For example (using {@link ReturnOrAwait} to show intended yield-then-resume usage pattern):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point, close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Completed {@link AsyncCursorHolder} backed by an already available {@link CursorHolder}
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ final AsyncCursorHolder result = new AsyncCursorHolder(null);
+ result.set(holder);
+ return result;
+ }
+
+ @Nullable
+ private final Runnable canceler;
+
+ @GuardedBy("this")
+ @Nullable
+ private CursorHolder result = null;
+ @GuardedBy("this")
+ @Nullable
+ private Throwable error = null;
+ @GuardedBy("this")
+ private boolean closed = false;
+ @GuardedBy("this")
+ private boolean disposed = false;
+ @GuardedBy("this")
+ private final List<Runnable> readyCallbacks = new ArrayList<>();
+
+ /**
+ * @param canceler optional callback invoked from {@link #close()} when the wrapper is closed before the load has
+ * completed ({@link #set} or {@link #setException}). Producers that support cancellation should
+ * provide one; producers that don't can pass {@code null}, in which case {@link #close()} just stops
+ * observing the result.
+ */
+ public AsyncCursorHolder(@Nullable Runnable canceler)
+ {
+ this.canceler = canceler;
+ }
+
+ /**
+ * Allows producer to mark the load successful with the given holder. Returns {@code true} if accepted, {@code false}
+ * if this wrapper has already been closed, in which case the producer is responsible for closing {@link CursorHolder}
+ * itself. Throws {@link DruidException} if the load was already completed (from prior calls to this method or
+ * {@link #setException}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
Review Comment:
oh, i guess bad wording; like its more like avoiding problems like if the callback needs to hold some other lock we don't have any weird ordering issues like if that lock is some lock shared between callback thread and producing thread, will try to clarify
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19397:
URL: https://github.com/apache/druid/pull/19397#discussion_r3181976806
##########
processing/src/main/java/org/apache/druid/query/groupby/GroupingEngine.java:
##########
@@ -494,21 +534,38 @@ public Sequence<ResultRow> process(
if (intervals.size() != 1) {
throw new IAE("Should only have one interval, got[%s]", intervals);
}
+ }
- final ResourceHolder<ByteBuffer> bufferHolder = bufferPool.take();
+ private Sequence<ResultRow> processWithCursorHolder(
+ GroupByQuery query,
+ CursorFactory cursorFactory,
+ CursorHolder cursorHolder,
+ @Nullable TimeBoundaryInspector timeBoundaryInspector,
+ NonBlockingPool<ByteBuffer> bufferPool,
+ CursorBuildSpec buildSpec
+ )
+ {
+ final GroupByQueryConfig querySpecificConfig = configSupplier.get().withOverrides(query);
Review Comment:
P2 Close cursorHolder if query config override throws
processWithCursorHolder now receives an already-created CursorHolder, but computes querySpecificConfig before entering any cleanup-protected block. If withOverrides throws, for example due to an invalid groupBy query context value, the CursorHolder returned by makeCursorHolderAsync/makeCursorHolder is never closed. Move this config resolution before acquiring the holder, or wrap it so cursorHolder is closed on every pre-sequence failure.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis merged PR #19397:
URL: https://github.com/apache/druid/pull/19397
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #19397:
URL: https://github.com/apache/druid/pull/19397#discussion_r3205086129
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.druid.segment;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.guava.FutureUtils;
+
+import java.io.Closeable;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where cancelling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * Typical usage from a non-blocking caller (e.g. an MSQ frame processor):
Review Comment:
The example code got a bit weird trying to get rid of `ReturnOrAwait`, so instead of tried to frame it that the example is using 'ReturnOrAwait' to show how you're supposed to use this thing, since the important part is the yield-then-resume when ready pattern, which i think makes it feel less like this is a thing for MSQ than the last iteration did?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #19397:
URL: https://github.com/apache/druid/pull/19397#discussion_r3206538286
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.druid.segment;
+
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.frame.processor.ReturnOrAwait;
+import org.apache.druid.java.util.common.Either;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where canceling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * <h3>Producer protocol</h3>
+ * Producers feed results in via {@link #set(CursorHolder)} or {@link #setException(Throwable)}, both of which return
+ * a boolean. If they return {@code false}, this wrapper has already been closed and the producer is responsible for
+ * closing whatever holder it just produced.
+ * Producers may pass a {@link Runnable} canceler at construction time which runs on {@link #close()} when the wrapper
+ * is closed before the {@link #set} has been called, giving the producer an opportunity to abort its work. The canceler
+ * is best-effort: a producer may have already produced the holder by the time it observes cancellation, in which case
+ * its {@link #set} call will return false and it must close the holder it tried to set.
+ * <p>
+ * <h3>Consumer protocol</h3>
+ * Consumers wait for {@link #isReady()} via {@link #addReadyCallback}, and {@link #release()} to transfer ownership of
+ * the {@link CursorHolder} (or throw the producer exception). Calling {@link #release()} before {@link #isReady()}
+ * returns {@code true}, multiple times, or after this holder has been closed will throw a {@link DruidException}.
+ * <p>
+ * For example (using {@link ReturnOrAwait} to show intended yield-then-resume usage pattern):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point, close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Completed {@link AsyncCursorHolder} backed by an already available {@link CursorHolder}
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ final AsyncCursorHolder result = new AsyncCursorHolder(null);
+ result.set(holder);
+ return result;
+ }
+
+ @Nullable
+ private final Runnable canceler;
+
+ @GuardedBy("this")
+ @Nullable
+ private CursorHolder result = null;
+ @GuardedBy("this")
+ @Nullable
+ private Throwable error = null;
+ @GuardedBy("this")
+ private boolean closed = false;
+ @GuardedBy("this")
+ private boolean disposed = false;
+ @GuardedBy("this")
+ private final List<Runnable> readyCallbacks = new ArrayList<>();
+
+ /**
+ * @param canceler optional callback invoked from {@link #close()} when the wrapper is closed before the load has
+ * completed ({@link #set} or {@link #setException}). Producers that support cancellation should
+ * provide one; producers that don't can pass {@code null}, in which case {@link #close()} just stops
+ * observing the result.
+ */
+ public AsyncCursorHolder(@Nullable Runnable canceler)
+ {
+ this.canceler = canceler;
+ }
+
+ /**
+ * Allows producer to mark the load successful with the given holder. Returns {@code true} if accepted, {@code false}
+ * if this wrapper has already been closed, in which case the producer is responsible for closing {@link CursorHolder}
+ * itself. Throws {@link DruidException} if the load was already completed (from prior calls to this method or
+ * {@link #setException}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean set(CursorHolder holder)
+ {
+ return setInternal(Either.value(holder));
+ }
+
+ /**
+ * Allows producer to mark the load as failed. Returns {@code true} if accepted, {@code false} if this wrapper has
+ * already been closed (no holder was produced, so there's nothing for the producer to clean up). Throws
+ * {@link DruidException} if the load was already completed (from prior calls to this method or {@link #set}).
+ * <p>
+ * Callbacks registered via {@link #addReadyCallback} fire outside the lock to avoid re-entrancy deadlocks.
+ */
+ public boolean setException(Throwable t)
+ {
+ return setInternal(Either.error(t));
+ }
+
+ private boolean setInternal(Either<Throwable, CursorHolder> value)
+ {
+ final List<Runnable> callbacksToFire;
+ synchronized (this) {
+ if (closed) {
+ return false;
+ }
+ if (result != null || error != null) {
+ throw DruidException.defensive("AsyncCursorHolder is already completed");
+ }
+ if (value.isError()) {
+ error = value.error();
+ } else {
+ result = value.valueOrThrow();
+ }
+ callbacksToFire = drainCallbacks();
+ }
+ fireCallbacks(callbacksToFire);
+ return true;
+ }
+
+ /**
+ * Whether the load has completed (successfully or with failure). Once true, stays true. Callers that need to wait
+ * for readiness without blocking the current thread should register a {@link #addReadyCallback} and yield.
+ */
+ public synchronized boolean isReady()
+ {
+ return result != null || error != null;
+ }
+
+ /**
+ * Take ownership of the underlying {@link CursorHolder}. After this returns, {@link #close()} on this
+ * {@code AsyncCursorHolder} is a no-op; the caller is responsible for closing the returned holder. Useful when
+ * passing the holder to another component (e.g. a cursor-lifecycle manager) that takes ownership of it.
+ * <p>
+ * Throws {@link DruidException} if the holder is not yet ready, has already been released, or this wrapper
+ * has been closed. Wraps and rethrows the failure if the underlying load failed. Does not block; callers must
+ * check {@link #isReady()} first (typically by yielding via a {@link #addReadyCallback}-driven wait pattern).
+ */
+ public synchronized CursorHolder release()
+ {
+ if (closed) {
+ throw DruidException.defensive("AsyncCursorHolder is already closed");
+ }
+ if (disposed) {
+ throw DruidException.defensive("AsyncCursorHolder has already been released");
+ }
+ if (error != null) {
+ // pass through as is
+ if (error instanceof RuntimeException runtime) {
+ throw runtime;
+ }
+ throw DruidException.forPersona(DruidException.Persona.DEVELOPER)
+ .ofCategory(DruidException.Category.UNCATEGORIZED)
+ .build(error, error.getMessage());
+ }
+ if (result == null) {
+ throw DruidException.defensive("AsyncCursorHolder is not ready yet");
+ }
+ disposed = true;
+ return result;
+ }
+
+ /**
+ * Register a callback to fire when {@link #isReady()} becomes true (whether the load succeeded or failed). If the
+ * holder is already ready, the callback fires synchronously on the calling thread. Otherwise it fires on whatever
+ * thread invokes {@link #set} or {@link #setException}, outside the wrapper's lock so the callback may safely
+ * re-enter the wrapper. Multiple callbacks may be registered; each fires once.
+ */
+ public void addReadyCallback(Runnable callback)
+ {
+ final boolean fireImmediately;
+ synchronized (this) {
+ if (result != null || error != null) {
+ fireImmediately = true;
+ } else {
+ readyCallbacks.add(callback);
+ fireImmediately = false;
+ }
+ }
+ if (fireImmediately) {
+ callback.run();
+ }
+ }
+
+ /**
+ * Close the wrapper. Safe at any lifecycle point and idempotent:
+ * <ul>
+ * <li>Already-loaded: closes the underlying {@link CursorHolder} immediately.</li>
+ * <li>Loading in progress: invokes the canceler (if one was supplied at construction). The producer may still
+ * call {@link #set} / {@link #setException} after this; if the producer wins the race and calls {@code set}
+ * with a holder, {@code set} returns false and the producer is responsible for closing it.</li>
+ * <li>Load failed: no-op (no holder was produced).</li>
+ * <li>Already released or already closed: no-op.</li>
+ * </ul>
+ */
+ @Override
+ public void close()
+ {
+ final CursorHolder holderToClose;
+ final Runnable cancelerToRun;
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ if (disposed) {
+ // Ownership was already transferred via release(); the caller manages the holder lifecycle.
+ return;
+ }
+ if (result != null) {
+ // Result is here and no one has released it; we close it.
Review Comment:
I added the nulling out of `result` in `release()`, however i still have the `disposed` flag because otherwise calling release would allow something to be able to call `set`/`setException` again which seems unchill, so I've kept this short-circuit on `disposed` here as is and haven't really changed this part.
We do essentially always call close on the result if we get to this part where we check not null, stuff maybe looks a bit more funny because we actually call close and the canceler outside of the synchronized block?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #19397:
URL: https://github.com/apache/druid/pull/19397#discussion_r3203838552
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.druid.segment;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.guava.FutureUtils;
+
+import java.io.Closeable;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where cancelling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * Typical usage from a non-blocking caller (e.g. an MSQ frame processor):
Review Comment:
No reason to include `(e.g. an MSQ frame processor)` here. I would also edit the code sample to be less MSQ specific.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.druid.segment;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.guava.FutureUtils;
+
+import java.io.Closeable;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where cancelling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * Typical usage from a non-blocking caller (e.g. an MSQ frame processor):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point — close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ * The wrapper has no blocking accessor by design: callers must wait for {@link #isReady()} via
+ * {@link #addReadyCallback} (so the wait can be expressed as a yield) and then call {@link #release()}.
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Already-ready {@link AsyncCursorHolder} backed by an immediately-available {@link CursorHolder}. The default
+ * synchronous {@link CursorFactory#makeCursorHolderAsync} implementation uses this.
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ return new AsyncCursorHolder(Futures.immediateFuture(holder));
+ }
+
+ /**
+ * Wraps a {@link ListenableFuture}{@code <CursorHolder>}. Implementations doing real I/O should produce the future
+ * (e.g. by submitting to an executor) and pass it here.
+ */
+ public static AsyncCursorHolder fromFuture(ListenableFuture<CursorHolder> future)
+ {
+ return new AsyncCursorHolder(future);
+ }
+
+ private final ListenableFuture<CursorHolder> future;
+ private boolean closed = false;
Review Comment:
use `@GuardedBy` on `closed` and `disposed`, since synchronization is crucial given how they're used.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.druid.segment;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.guava.FutureUtils;
+
+import java.io.Closeable;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where cancelling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * Typical usage from a non-blocking caller (e.g. an MSQ frame processor):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point — close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ * The wrapper has no blocking accessor by design: callers must wait for {@link #isReady()} via
+ * {@link #addReadyCallback} (so the wait can be expressed as a yield) and then call {@link #release()}.
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Already-ready {@link AsyncCursorHolder} backed by an immediately-available {@link CursorHolder}. The default
+ * synchronous {@link CursorFactory#makeCursorHolderAsync} implementation uses this.
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ return new AsyncCursorHolder(Futures.immediateFuture(holder));
+ }
+
+ /**
+ * Wraps a {@link ListenableFuture}{@code <CursorHolder>}. Implementations doing real I/O should produce the future
Review Comment:
The `@link` and `@code` seem weird here. Also the "Implementations doing real I/O…" bit doesn't seem necessary.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.druid.segment;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.guava.FutureUtils;
+
+import java.io.Closeable;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where cancelling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * Typical usage from a non-blocking caller (e.g. an MSQ frame processor):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point — close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ * The wrapper has no blocking accessor by design: callers must wait for {@link #isReady()} via
+ * {@link #addReadyCallback} (so the wait can be expressed as a yield) and then call {@link #release()}.
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Already-ready {@link AsyncCursorHolder} backed by an immediately-available {@link CursorHolder}. The default
+ * synchronous {@link CursorFactory#makeCursorHolderAsync} implementation uses this.
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ return new AsyncCursorHolder(Futures.immediateFuture(holder));
+ }
+
+ /**
+ * Wraps a {@link ListenableFuture}{@code <CursorHolder>}. Implementations doing real I/O should produce the future
+ * (e.g. by submitting to an executor) and pass it here.
+ */
+ public static AsyncCursorHolder fromFuture(ListenableFuture<CursorHolder> future)
+ {
+ return new AsyncCursorHolder(future);
+ }
+
+ private final ListenableFuture<CursorHolder> future;
+ private boolean closed = false;
+ private boolean disposed = false;
+
+ private AsyncCursorHolder(ListenableFuture<CursorHolder> future)
+ {
+ this.future = future;
+ // If close() lands before the future completes we still need to dispose the eventually-produced holder. Register
+ // a callback up-front so the holder is closed whenever it arrives if we've been closed in the meantime.
+ Futures.addCallback(
+ future,
+ new FutureCallback<>()
+ {
+ @Override
+ public void onSuccess(CursorHolder holder)
+ {
+ synchronized (AsyncCursorHolder.this) {
+ if (closed && !disposed) {
+ disposed = true;
+ try {
+ holder.close();
+ }
+ catch (Throwable ignored) {
+ // Best-effort — we're already in cleanup; nothing to do if close itself fails.
+ }
+ }
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t)
+ {
+ // Nothing to dispose: no holder was produced.
+ }
+ },
+ MoreExecutors.directExecutor()
+ );
+ }
+
+ /**
+ * Whether {@link #release()} can return without blocking. Once true, stays true. Callers that need to wait for
+ * readiness without blocking the current thread should register a {@link #addReadyCallback} and yield.
+ */
+ public boolean isReady()
+ {
+ return future.isDone();
+ }
+
+ /**
+ * Take ownership of the underlying {@link CursorHolder}. After this returns, {@link #close()} on this
+ * {@code AsyncCursorHolder} is a no-op; the caller is responsible for closing the returned holder. Useful when
+ * passing the holder to another component (e.g. a cursor-lifecycle manager) that takes ownership of it.
+ * <p>
+ * Throws {@link IllegalStateException} if the holder is not yet ready, has already been released, or this wrapper
Review Comment:
`DruidException.defensive` seems more standard.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.druid.segment;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.guava.FutureUtils;
+
+import java.io.Closeable;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where cancelling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * Typical usage from a non-blocking caller (e.g. an MSQ frame processor):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point — close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ * The wrapper has no blocking accessor by design: callers must wait for {@link #isReady()} via
+ * {@link #addReadyCallback} (so the wait can be expressed as a yield) and then call {@link #release()}.
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Already-ready {@link AsyncCursorHolder} backed by an immediately-available {@link CursorHolder}. The default
+ * synchronous {@link CursorFactory#makeCursorHolderAsync} implementation uses this.
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ return new AsyncCursorHolder(Futures.immediateFuture(holder));
+ }
+
+ /**
+ * Wraps a {@link ListenableFuture}{@code <CursorHolder>}. Implementations doing real I/O should produce the future
+ * (e.g. by submitting to an executor) and pass it here.
+ */
+ public static AsyncCursorHolder fromFuture(ListenableFuture<CursorHolder> future)
+ {
+ return new AsyncCursorHolder(future);
+ }
+
+ private final ListenableFuture<CursorHolder> future;
+ private boolean closed = false;
+ private boolean disposed = false;
+
+ private AsyncCursorHolder(ListenableFuture<CursorHolder> future)
+ {
+ this.future = future;
+ // If close() lands before the future completes we still need to dispose the eventually-produced holder. Register
+ // a callback up-front so the holder is closed whenever it arrives if we've been closed in the meantime.
+ Futures.addCallback(
+ future,
+ new FutureCallback<>()
+ {
+ @Override
+ public void onSuccess(CursorHolder holder)
+ {
+ synchronized (AsyncCursorHolder.this) {
+ if (closed && !disposed) {
+ disposed = true;
+ try {
+ holder.close();
+ }
+ catch (Throwable ignored) {
+ // Best-effort — we're already in cleanup; nothing to do if close itself fails.
+ }
+ }
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t)
+ {
+ // Nothing to dispose: no holder was produced.
+ }
+ },
+ MoreExecutors.directExecutor()
+ );
+ }
+
+ /**
+ * Whether {@link #release()} can return without blocking. Once true, stays true. Callers that need to wait for
+ * readiness without blocking the current thread should register a {@link #addReadyCallback} and yield.
+ */
+ public boolean isReady()
+ {
+ return future.isDone();
+ }
+
+ /**
+ * Take ownership of the underlying {@link CursorHolder}. After this returns, {@link #close()} on this
+ * {@code AsyncCursorHolder} is a no-op; the caller is responsible for closing the returned holder. Useful when
+ * passing the holder to another component (e.g. a cursor-lifecycle manager) that takes ownership of it.
+ * <p>
+ * Throws {@link IllegalStateException} if the holder is not yet ready, has already been released, or this wrapper
+ * has been closed. Throws if the underlying load failed. Does not block; callers must check {@link #isReady()}
+ * first (typically by yielding via a {@link #addReadyCallback}-driven wait pattern).
+ */
+ public synchronized CursorHolder release()
+ {
+ if (closed) {
+ throw new IllegalStateException("AsyncCursorHolder is already closed");
+ }
+ if (disposed) {
+ throw new IllegalStateException("AsyncCursorHolder has already been released");
+ }
+ if (!future.isDone()) {
+ throw new IllegalStateException("AsyncCursorHolder is not ready yet");
+ }
+ final CursorHolder holder = FutureUtils.getUncheckedImmediately(future);
+ disposed = true;
+ return holder;
+ }
+
+ /**
+ * Register a callback to fire when {@link #isReady()} becomes true (whether the load succeeded or failed). If the
+ * holder is already ready, the callback fires synchronously. Multiple callbacks may be registered; each fires once.
+ */
+ public void addReadyCallback(Runnable callback)
+ {
+ future.addListener(callback, MoreExecutors.directExecutor());
+ }
+
+ /**
+ * Close the holder. Safe at any lifecycle point and idempotent:
+ * <ul>
+ * <li>Already-loaded: closes the underlying {@link CursorHolder} immediately.</li>
+ * <li>Loading in progress: marks for disposal; the produced holder is closed when the load completes.</li>
+ * <li>Load failed: no-op (nothing to dispose).</li>
+ * <li>Already closed: no-op.</li>
+ * </ul>
+ * Note that closing does NOT cancel an in-flight load — cancelling a future-of-Closeable is the exact lifecycle
+ * hazard this class exists to prevent. The load completes normally and the resulting holder is closed promptly.
+ */
+ @Override
+ public synchronized void close()
+ {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ if (disposed) {
+ // Ownership of the holder was already transferred (via release) or already disposed; nothing to do here.
+ return;
+ }
+ if (future.isDone() && !future.isCancelled()) {
+ try {
+ final CursorHolder holder = FutureUtils.getUncheckedImmediately(future);
+ disposed = true;
+ holder.close();
+ }
+ catch (Throwable t) {
+ // Future completed with a failure, no holder to close.
Review Comment:
Comment isn't entirely right. The `holder.close()` might also have failed. I don't think we want to suppress that.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.druid.segment;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.guava.FutureUtils;
+
+import java.io.Closeable;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where cancelling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * Typical usage from a non-blocking caller (e.g. an MSQ frame processor):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point — close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ * The wrapper has no blocking accessor by design: callers must wait for {@link #isReady()} via
+ * {@link #addReadyCallback} (so the wait can be expressed as a yield) and then call {@link #release()}.
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Already-ready {@link AsyncCursorHolder} backed by an immediately-available {@link CursorHolder}. The default
+ * synchronous {@link CursorFactory#makeCursorHolderAsync} implementation uses this.
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ return new AsyncCursorHolder(Futures.immediateFuture(holder));
+ }
+
+ /**
+ * Wraps a {@link ListenableFuture}{@code <CursorHolder>}. Implementations doing real I/O should produce the future
+ * (e.g. by submitting to an executor) and pass it here.
+ */
+ public static AsyncCursorHolder fromFuture(ListenableFuture<CursorHolder> future)
+ {
+ return new AsyncCursorHolder(future);
+ }
+
+ private final ListenableFuture<CursorHolder> future;
+ private boolean closed = false;
+ private boolean disposed = false;
+
+ private AsyncCursorHolder(ListenableFuture<CursorHolder> future)
+ {
+ this.future = future;
+ // If close() lands before the future completes we still need to dispose the eventually-produced holder. Register
+ // a callback up-front so the holder is closed whenever it arrives if we've been closed in the meantime.
+ Futures.addCallback(
+ future,
+ new FutureCallback<>()
+ {
+ @Override
+ public void onSuccess(CursorHolder holder)
+ {
+ synchronized (AsyncCursorHolder.this) {
Review Comment:
We can avoid having two different `holder.close()` places by attaching the callback in `close` itself. Although, if we avoid futures entirely as suggested in another comment, this comment will become irrelevant.
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.druid.segment;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.guava.FutureUtils;
+
+import java.io.Closeable;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where cancelling the future or letting a caller fail before consuming the future
Review Comment:
canceling (spelling)
##########
processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.druid.segment;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.guava.FutureUtils;
+
+import java.io.Closeable;
+
+/**
+ * Closeable wrapper around an asynchronously-loaded {@link CursorHolder}, returned by
+ * {@link CursorFactory#makeCursorHolderAsync}. Designed to make lifecycle management safe even when the holder is
+ * still loading: callers receive a single Closeable handle and can register it once with their cleanup machinery,
+ * regardless of where the underlying load is in its lifecycle.
+ * <p>
+ * The hazard this exists to avoid: returning a {@code ListenableFuture<CursorHolder>} (or similar future-of-Closeable)
+ * makes correct cleanup error-prone, where cancelling the future or letting a caller fail before consuming the future
+ * can orphan the produced holder, leaking the underlying resources. By exposing a Closeable that internally tracks the
+ * load and disposes whatever has materialized, callers don't have to write that bookkeeping themselves.
+ * <p>
+ * Typical usage from a non-blocking caller (e.g. an MSQ frame processor):
+ * <pre>{@code
+ * if (asyncHolder == null) {
+ * asyncHolder = cursorFactory.makeCursorHolderAsync(spec);
+ * closer.register(asyncHolder); // safe at any lifecycle point — close() handles in-flight loads
+ * }
+ * if (!asyncHolder.isReady()) {
+ * SettableFuture<?> awaitFuture = SettableFuture.create();
+ * asyncHolder.addReadyCallback(() -> awaitFuture.set(null));
+ * return ReturnOrAwait.awaitAllFutures(List.of(awaitFuture));
+ * }
+ * final CursorHolder holder = asyncHolder.release(); // ownership transfers to the caller
+ * // ... use holder; close it when done (or hand it to a component that owns its lifecycle) ...
+ * }</pre>
+ * The wrapper has no blocking accessor by design: callers must wait for {@link #isReady()} via
+ * {@link #addReadyCallback} (so the wait can be expressed as a yield) and then call {@link #release()}.
+ */
+public class AsyncCursorHolder implements Closeable
+{
+ /**
+ * Already-ready {@link AsyncCursorHolder} backed by an immediately-available {@link CursorHolder}. The default
+ * synchronous {@link CursorFactory#makeCursorHolderAsync} implementation uses this.
+ */
+ public static AsyncCursorHolder completed(CursorHolder holder)
+ {
+ return new AsyncCursorHolder(Futures.immediateFuture(holder));
+ }
+
+ /**
+ * Wraps a {@link ListenableFuture}{@code <CursorHolder>}. Implementations doing real I/O should produce the future
+ * (e.g. by submitting to an executor) and pass it here.
+ */
+ public static AsyncCursorHolder fromFuture(ListenableFuture<CursorHolder> future)
+ {
+ return new AsyncCursorHolder(future);
+ }
+
+ private final ListenableFuture<CursorHolder> future;
+ private boolean closed = false;
+ private boolean disposed = false;
+
+ private AsyncCursorHolder(ListenableFuture<CursorHolder> future)
+ {
+ this.future = future;
+ // If close() lands before the future completes we still need to dispose the eventually-produced holder. Register
+ // a callback up-front so the holder is closed whenever it arrives if we've been closed in the meantime.
+ Futures.addCallback(
+ future,
+ new FutureCallback<>()
+ {
+ @Override
+ public void onSuccess(CursorHolder holder)
+ {
+ synchronized (AsyncCursorHolder.this) {
+ if (closed && !disposed) {
+ disposed = true;
+ try {
+ holder.close();
+ }
+ catch (Throwable ignored) {
+ // Best-effort — we're already in cleanup; nothing to do if close itself fails.
+ }
+ }
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t)
+ {
+ // Nothing to dispose: no holder was produced.
+ }
+ },
+ MoreExecutors.directExecutor()
+ );
+ }
+
+ /**
+ * Whether {@link #release()} can return without blocking. Once true, stays true. Callers that need to wait for
+ * readiness without blocking the current thread should register a {@link #addReadyCallback} and yield.
+ */
+ public boolean isReady()
+ {
+ return future.isDone();
+ }
+
+ /**
+ * Take ownership of the underlying {@link CursorHolder}. After this returns, {@link #close()} on this
+ * {@code AsyncCursorHolder} is a no-op; the caller is responsible for closing the returned holder. Useful when
+ * passing the holder to another component (e.g. a cursor-lifecycle manager) that takes ownership of it.
+ * <p>
+ * Throws {@link IllegalStateException} if the holder is not yet ready, has already been released, or this wrapper
+ * has been closed. Throws if the underlying load failed. Does not block; callers must check {@link #isReady()}
+ * first (typically by yielding via a {@link #addReadyCallback}-driven wait pattern).
+ */
+ public synchronized CursorHolder release()
+ {
+ if (closed) {
+ throw new IllegalStateException("AsyncCursorHolder is already closed");
+ }
+ if (disposed) {
+ throw new IllegalStateException("AsyncCursorHolder has already been released");
+ }
+ if (!future.isDone()) {
+ throw new IllegalStateException("AsyncCursorHolder is not ready yet");
+ }
+ final CursorHolder holder = FutureUtils.getUncheckedImmediately(future);
+ disposed = true;
+ return holder;
+ }
+
+ /**
+ * Register a callback to fire when {@link #isReady()} becomes true (whether the load succeeded or failed). If the
+ * holder is already ready, the callback fires synchronously. Multiple callbacks may be registered; each fires once.
+ */
+ public void addReadyCallback(Runnable callback)
+ {
+ future.addListener(callback, MoreExecutors.directExecutor());
+ }
+
+ /**
+ * Close the holder. Safe at any lifecycle point and idempotent:
+ * <ul>
+ * <li>Already-loaded: closes the underlying {@link CursorHolder} immediately.</li>
+ * <li>Loading in progress: marks for disposal; the produced holder is closed when the load completes.</li>
+ * <li>Load failed: no-op (nothing to dispose).</li>
+ * <li>Already closed: no-op.</li>
+ * </ul>
+ * Note that closing does NOT cancel an in-flight load — cancelling a future-of-Closeable is the exact lifecycle
+ * hazard this class exists to prevent. The load completes normally and the resulting holder is closed promptly.
Review Comment:
Really we should structure this to be capable of canceling the in-flight load. I think to make it work we'll want to not base this class on futures at all. Something like:
- Accept a `Runnable canceler` in the constructor
- Replace the `future` field with `Either<Throwable, T>`
- Expose `set` and `setException` methods that return `boolean` (true for set accepted, false for set not accepted)
- In `close`, call `canceler` if the set methods haven't been called yet
- Whatever provides the actual `CursorHolder` should close it if `set` returns false
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: add async CursorFactory API and migrate MSQ frame processors to use it (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #19397:
URL: https://github.com/apache/druid/pull/19397#discussion_r3176204717
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java:
##########
@@ -297,21 +304,29 @@ protected ReturnOrAwait<SegmentsInputSlice> runWithDataServerQuery(final DataSer
protected ReturnOrAwait<Unit> runWithSegment(final SegmentReferenceHolder segmentHolder) throws IOException
{
if (cursor == null) {
- final Segment segment = mapSegment(segmentHolder, closer);
- final CursorFactory cursorFactory = segment.as(CursorFactory.class);
- if (cursorFactory == null) {
- throw DruidException.defensive(
- "Null cursor factory found. Probably trying to issue a query against a segment being memory unmapped."
+ if (cursorHolderFuture == null) {
+ final Segment segment = mapSegment(segmentHolder, closer);
+ final CursorFactory cursorFactory = segment.as(CursorFactory.class);
+ if (cursorFactory == null) {
+ throw DruidException.defensive(
+ "Null cursor factory found. Probably trying to issue a query against a segment being memory unmapped."
+ );
+ }
+
+ cursorHolderFuture = cursorFactory.makeCursorHolderAsync(
Review Comment:
Handling futures that return closeable things is tricky. Maybe we can improve it by changing the return of `makeCursorHolderAsync` from `ListenableFuture<CursorHolder>` to `AsyncCursorHolder` that is closeable and has methods `get()` (blocks if not ready), `close()` (closes the resource no matter where it is in its lifecycle), and `addReadyCallback(Runnable)` (used by nonblocking callers to learn when `get` is ready).
The problem with the future approach is that once this call site gets the `cursorHolderFuture`, it's responsible for monitoring the future and closing `cursorHolder` if the future resolves successfully. This has to be done even if the processor is canceled before it has a chance to run through normally. It requires extra carefulness and is easy to mess up.
One way it can be handled is by attaching a callback in `cleanup` that closes the holder in `onSuccess`, like:
```
if (cursorHolderFuture != null) {
Futures.addCallback(
cursorHolderFuture,
new FutureCallback<>() {
void onSuccess(CursorHolder holder) { holder.close(); }
void onFailure(Throwable t) { /* nothing */ }
}
);
}
```
But even with this structure, it's important watch out for pitfalls. A big one is that you can never cancel a future that returns a closeable thing. Cancellation of the future can cause the object to be orphaned and eventually GCed without being closed (if the object is created before cancellation has a chance to interrupt whatever was creating it).
If we can avoid these problems by returning something *directly* closeable (like this `AsyncCursorHolder` idea) rather than future-of-closeable, then the caller code becomes simpler.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org