Fail LookupOuterOperator's future when getOuterPositionIterator() fails
Previously, when `getOuterPositionIterator()` threw exception, the
`SettableFuture` handed over to `LookupJoinOperatorFactory` would never
be satisfied. Since exception was in a future callback, it would be
generally ignored, resulting in hanged query.
@@ -19,17 +19,16 @@ import com.facebook.presto.operator.LookupSource.OuterPositionIterator;
import com.facebook.presto.spi.type.Type;
import com.facebook.presto.sql.planner.plan.PlanNodeId;
import com.google.common.collect.ImmutableList;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.SettableFuture;
+import com.google.common.util.concurrent.ListenableFuture;
import java.util.List;
import java.util.Optional;
-import java.util.function.Consumer;
import static com.facebook.presto.operator.LookupJoinOperators.JoinType.INNER;
import static com.facebook.presto.operator.LookupJoinOperators.JoinType.PROBE_OUTER;
import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.util.concurrent.Futures.transform;
+import static com.google.common.util.concurrent.Futures.transformAsync;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static java.util.Objects.requireNonNull;
@@ -81,14 +80,13 @@ public class LookupJoinOperatorFactory
this.outerOperatorFactory = Optional.empty();
}
else {
- // when all join operators finish, set the outer position future to start the outer operator
- SettableFuture<OuterPositionIterator> outerPositionsFuture = SettableFuture.create();
- probeReferenceCount.getFreeFuture().addListener(() -> {
- // lookup source may not be finished yet, so add a listener
- Futures.addCallback(
- lookupSourceFactory.createLookupSource(),
- new OnSuccessFutureCallback<>(lookupSource -> outerPositionsFuture.set(lookupSource.getOuterPositionIterator())));
- }, directExecutor());
+ // when all join operators finish (and lookup source is ready), set the outer position future to start the outer operator
+ ListenableFuture<LookupSource> lookupSourceAfterProbeFinished = transformAsync(probeReferenceCount.getFreeFuture(), ignored -> lookupSourceFactory.createLookupSource());
+ ListenableFuture<OuterPositionIterator> outerPositionsFuture = transform(lookupSourceAfterProbeFinished, lookupSource -> {
+ try (LookupSource ignore = lookupSource) {
+ return lookupSource.getOuterPositionIterator();
+ }
+ });
lookupSourceFactoryUsersCount.retain();
this.outerOperatorFactory = Optional.of(new LookupOuterOperatorFactory(operatorId, planNodeId, outerPositionsFuture, probeOutputTypes, buildOutputTypes, lookupSourceFactoryUsersCount));
@@ -167,27 +165,4 @@ public class LookupJoinOperatorFactory
{
return outerOperatorFactory;
}
-
- // We use a public class to avoid access problems with the isolated class loaders
- public static class OnSuccessFutureCallback<T>
- implements FutureCallback<T>
- {
- private final Consumer<T> onSuccess;
-
- public OnSuccessFutureCallback(Consumer<T> onSuccess)
- {
- this.onSuccess = onSuccess;
- }
-
- @Override
- public void onSuccess(T result)
- {
- onSuccess.accept(result);
- }
-
- @Override
- public void onFailure(Throwable t)
- {
- }
- }
}
