!1395 Spiller for join and using Spilled buildSide for RightOuter queries

Merge pull request !1395 from i-robot/pull301

hetu-docs/en/admin/properties.md

@@ -161,6 +161,30 @@ This section describes the most important config properties that may be used to
 >
 > This config property can be overridden by the `spill_window_operator` session property.
 
+
+### `experimental.spill-build-for-outer-join-enabled`
+
+> -   **Type:** `boolean`
+> -   **Default value:** `false`
+>
+> Enables spill feature for right-outer and full-outer join operations.
+>
+>
+>
+> This config property can be overridden by the `spill_build_for_outer_join_enabled` session property.
+
+### `experimental.inner-join-spill-filter-enabled`
+
+> -   **Type:** `boolean`
+> -   **Default value:** `false`
+>
+> Enables bloom filter based build-side spill matching for probe side spill decision.
+>
+>
+>
+> This config property can be overridden by the `inner_join_spill_filter_enabled` session property.
+
+
 ### `experimental.spill-reuse-tablescan`
 
 > -   **Type:** `boolean`

presto-benchmark/src/main/java/io/prestosql/benchmark/HashBuildAndJoinBenchmark.java

@@ -98,7 +98,7 @@ public class HashBuildAndJoinBenchmark
                         .collect(toImmutableList()),
                 1,
                 requireNonNull(ImmutableMap.of(), "layout is null"),
-                false));
+                false, false));
         HashBuilderOperatorFactory hashBuilder = new HashBuilderOperatorFactory(
                 2,
                 new PlanNodeId("test"),

presto-benchmark/src/main/java/io/prestosql/benchmark/HashBuildBenchmark.java

@@ -69,7 +69,7 @@ public class HashBuildBenchmark
                         .collect(toImmutableList()),
                 1,
                 requireNonNull(ImmutableMap.of(), "layout is null"),
-                false));
+                false, false));
         HashBuilderOperatorFactory hashBuilder = new HashBuilderOperatorFactory(
                 1,
                 new PlanNodeId("test"),

presto-benchmark/src/main/java/io/prestosql/benchmark/HashJoinBenchmark.java

@@ -78,7 +78,7 @@ public class HashJoinBenchmark
                             .collect(toImmutableList()),
                     1,
                     requireNonNull(ImmutableMap.of(), "layout is null"),
-                    false));
+                    false, false));
             HashBuilderOperatorFactory hashBuilder = new HashBuilderOperatorFactory(
                     1,
                     new PlanNodeId("test"),

presto-main/pom.xml

@@ -579,6 +579,12 @@
             <groupId>org.objenesis</groupId>
             <artifactId>objenesis</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>org.roaringbitmap</groupId>
+            <artifactId>RoaringBitmap</artifactId>
+            <version>0.9.0</version>
+        </dependency>
     </dependencies>
 
     <build>

presto-main/src/main/java/io/prestosql/SystemSessionProperties.java

@@ -102,6 +102,8 @@ public final class SystemSessionProperties
     public static final String SPILL_ORDER_BY = "spill_order_by";
     public static final String SPILL_NON_BLOCKING_ORDERBY = "spill_non_blocking_orderby";
     public static final String SPILL_WINDOW_OPERATOR = "spill_window_operator";
+    public static final String SPILL_OUTER_JOIN_ENABLED = "spill_build_for_outer_join_enabled";
+    public static final String INNER_JOIN_SPILL_FILTER_ENABLED = "inner_join_spill_filter_enabled";
     public static final String AGGREGATION_OPERATOR_UNSPILL_MEMORY_LIMIT = "aggregation_operator_unspill_memory_limit";
     public static final String OPTIMIZE_DISTINCT_AGGREGATIONS = "optimize_mixed_distinct_aggregations";
     public static final String ITERATIVE_OPTIMIZER = "iterative_optimizer_enabled";
@@ -503,6 +505,16 @@ public final class SystemSessionProperties
                         "Spill orderby in non blocking manner",
                         featuresConfig.isNonBlockingSpill(),
                         false),
+                booleanProperty(
+                        SPILL_OUTER_JOIN_ENABLED,
+                        "Enable build side spill for Right or Full Outer Join",
+                        featuresConfig.isSpillBuildForOuterJoinEnabled(),
+                        false),
+                booleanProperty(
+                        INNER_JOIN_SPILL_FILTER_ENABLED,
+                        "Enable build side spill matching optimization for Inner Join",
+                        featuresConfig.isInnerJoinSpillFilterEnabled(),
+                        false),
                 booleanProperty(
                         SPILL_WINDOW_OPERATOR,
                         "Spill in WindowOperator if spill_enabled is also set",
@@ -1063,6 +1075,16 @@ public final class SystemSessionProperties
         return session.getSystemProperty(SPILL_ENABLED, Boolean.class);
     }
 
+    public static boolean isSpillForOuterJoinEnabled(Session session)
+    {
+        return session.getSystemProperty(SPILL_OUTER_JOIN_ENABLED, Boolean.class);
+    }
+
+    public static boolean isInnerJoinSpillFilteringEnabled(Session session)
+    {
+        return session.getSystemProperty(INNER_JOIN_SPILL_FILTER_ENABLED, Boolean.class);
+    }
+
     public static boolean isNonBlockingSpillOrderby(Session session)
     {
         return session.getSystemProperty(SPILL_NON_BLOCKING_ORDERBY, Boolean.class);

presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java

@@ -773,7 +773,7 @@ public class SqlQueryScheduler
                     }
 
                     // perform some scheduling work
-                    /* Todo(nitin) get groupSize specification from the ResourceGroupManager */
+                    /* Get groupSize specification from the ResourceGroupManager */
                     int maxSplitGroupSize = getOptimalSmallSplitGroupSize();
                     ScheduleResult result = stageSchedulers.get(stage.getStageId())
                             .schedule(maxSplitGroupSize);

presto-main/src/main/java/io/prestosql/operator/HashBuilderOperator.java

@@ -14,7 +14,10 @@
 package io.prestosql.operator;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.AbstractIterator;
 import com.google.common.collect.ImmutableList;
+import com.google.common.hash.BloomFilter;
+import com.google.common.hash.Funnels;
 import com.google.common.io.Closer;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.prestosql.execution.Lifespan;
@@ -22,10 +25,12 @@ import io.prestosql.memory.context.LocalMemoryContext;
 import io.prestosql.snapshot.SingleInputSnapshotState;
 import io.prestosql.snapshot.Spillable;
 import io.prestosql.spi.Page;
+import io.prestosql.spi.block.Block;
 import io.prestosql.spi.plan.PlanNodeId;
 import io.prestosql.spi.snapshot.BlockEncodingSerdeProvider;
 import io.prestosql.spi.snapshot.MarkerPage;
 import io.prestosql.spi.snapshot.RestorableConfig;
+import io.prestosql.spi.type.BigintType;
 import io.prestosql.spiller.SingleStreamSpiller;
 import io.prestosql.spiller.SingleStreamSpillerFactory;
 import io.prestosql.sql.gen.JoinFilterFunctionCompiler.JoinFilterFunctionFactory;
@@ -33,11 +38,15 @@ import io.prestosql.sql.gen.JoinFilterFunctionCompiler.JoinFilterFunctionFactory
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.Serializable;
+import java.io.UncheckedIOException;
 import java.nio.file.Path;
 import java.util.ArrayDeque;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -51,6 +60,7 @@ import static com.google.common.base.Verify.verify;
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.airlift.concurrent.MoreFutures.checkSuccess;
 import static io.airlift.concurrent.MoreFutures.getDone;
+import static io.prestosql.SystemSessionProperties.isInnerJoinSpillFilteringEnabled;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 
@@ -239,7 +249,7 @@ public class HashBuilderOperator
 
     private State state = State.CONSUMING_INPUT;
     private Optional<ListenableFuture<?>> lookupSourceNotNeeded = Optional.empty();
-    private final SpilledLookupSourceHandle spilledLookupSourceHandle = new SpilledLookupSourceHandle();
+    private final SpilledLookupSourceHandle spilledLookupSourceHandle;
     private Optional<SingleStreamSpiller> spiller = Optional.empty();
     private ListenableFuture<?> spillInProgress = NOT_BLOCKED;
     private Optional<ListenableFuture<List<Page>>> unspillInProgress = Optional.empty();
@@ -248,6 +258,7 @@ public class HashBuilderOperator
     private OptionalLong lookupSourceChecksum = OptionalLong.empty();
 
     private Optional<Runnable> finishMemoryRevoke = Optional.empty();
+    private BloomFilter<Long> spillBloom;
 
     private final SingleInputSnapshotState snapshotState;
     // Snapshot: special logic for taking an extra snapshot for HashBuilder operator, for outer joins.
@@ -306,6 +317,15 @@ public class HashBuilderOperator
         this.spillEnabled = spillEnabled;
         this.singleStreamSpillerFactory = requireNonNull(singleStreamSpillerFactory, "singleStreamSpillerFactory is null");
         this.snapshotState = operatorContext.isSnapshotEnabled() ? SingleInputSnapshotState.forOperator(this, operatorContext) : null;
+
+        if (preComputedHashChannel.isPresent() && spillEnabled && isInnerJoinSpillFilteringEnabled(operatorContext.getDriverContext().getSession())) {
+            this.spillBloom = BloomFilter.create(Funnels.longFunnel(), 10_000, 0.01);
+        }
+        else {
+            this.spillBloom = null;
+        }
+
+        spilledLookupSourceHandle = new SpilledLookupSourceHandle(spillBloom);
     }
 
     @Override
@@ -408,9 +428,24 @@ public class HashBuilderOperator
     {
         checkState(spillInProgress.isDone(), "Previous spill still in progress");
         checkSuccess(spillInProgress, "spilling failed");
+        updateBloom(page);
         spillInProgress = getSpiller().spill(page);
     }
 
+    private void updateBloom(Page page)
+    {
+        if (spillBloom == null) {
+            return;
+        }
+
+        int hashChannel = preComputedHashChannel.getAsInt();
+        Block hashes = page.getBlock(hashChannel);
+        for (int i = 0; i < page.getPositionCount(); i++) {
+            //Todo make generic: spillBloom.put(type.get(hashes, i));
+            spillBloom.put(BigintType.BIGINT.getLong(hashes, i));
+        }
+    }
+
     @Override
     public ListenableFuture<?> startMemoryRevoke()
     {
@@ -463,7 +498,21 @@ public class HashBuilderOperator
                 index.getTypes(),
                 operatorContext.getSpillContext().newLocalSpillContext(),
                 operatorContext.newLocalSystemMemoryContext(HashBuilderOperator.class.getSimpleName())));
-        return getSpiller().spill(index.getPages());
+        return getSpiller().spill(new AbstractIterator<Page>()
+        {
+            private Iterator<Page> spillPages = index.getPages();
+
+            @Override
+            protected Page computeNext()
+            {
+                if (!spillPages.hasNext()) {
+                    return endOfData();
+                }
+                Page page = spillPages.next();
+                updateBloom(page);
+                return page;
+            }
+        });
     }
 
     @Override
@@ -652,6 +701,7 @@ public class HashBuilderOperator
         localUserMemoryContext.setBytes(index.getEstimatedSize().toBytes());
 
         close();
+        spilledLookupSourceHandle.setDisposeCompleted();
     }
 
     private LookupSourceSupplier buildLookupSource()
@@ -739,6 +789,17 @@ public class HashBuilderOperator
         if (spiller.isPresent()) {
             myState.spiller = spiller.get().capture(serdeProvider);
         }
+
+        if (spillBloom != null) {
+            ByteArrayOutputStream bos = new ByteArrayOutputStream();
+            try {
+                spillBloom.writeTo(bos);
+                myState.spillBloom = bos.toByteArray();
+            }
+            catch (IOException e) {
+                throw new UncheckedIOException(e);
+            }
+        }
         return myState;
     }
 
@@ -767,6 +828,15 @@ public class HashBuilderOperator
             }
             this.spiller.get().restore(myState.spiller, serdeProvider);
         }
+        if (myState.spillBloom != null) {
+            ByteArrayInputStream bis = new ByteArrayInputStream((byte[]) myState.spillBloom);
+            try {
+                spillBloom = BloomFilter.readFrom(bis, Funnels.longFunnel());
+            }
+            catch (IOException e) {
+                throw new UncheckedIOException(e);
+            }
+        }
         if (oldState == State.CONSUMING_INPUT && this.state == State.SPILLING_INPUT) {
             lookupSourceFactory.setPartitionSpilledLookupSourceHandle(partitionIndex, spilledLookupSourceHandle);
         }
@@ -791,5 +861,7 @@ public class HashBuilderOperator
         private String state;
         private boolean alreadyFinished;
         private Object spiller;
+
+        private Object spillBloom;
     }
 }

presto-main/src/main/java/io/prestosql/operator/JoinBridge.java

@@ -30,4 +30,14 @@ public interface JoinBridge
     void destroy();
 
     ListenableFuture<?> whenBuildFinishes();
+
+    default ListenableFuture<?> whenMemProbeFinishes()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    default boolean isOuterEarlyStartEnabled()
+    {
+        return false;
+    }
 }

presto-main/src/main/java/io/prestosql/operator/JoinBridgeManager.java

@@ -15,6 +15,7 @@
 package io.prestosql.operator;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.prestosql.execution.Lifespan;
@@ -30,6 +31,7 @@ import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.util.concurrent.Futures.transform;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
+import static io.airlift.concurrent.MoreFutures.whenAnyComplete;
 import static io.prestosql.operator.PipelineExecutionStrategy.UNGROUPED_EXECUTION;
 import static java.util.Objects.requireNonNull;
 
@@ -162,6 +164,12 @@ public class JoinBridgeManager<T extends JoinBridge>
         return internalJoinBridgeDataManager.getOuterPositionsFuture(lifespan);
     }
 
+    public void probeOperatorFinished(Lifespan lifespan)
+    {
+        initializeIfNecessary();
+        internalJoinBridgeDataManager.probeOperatorFinish(lifespan);
+    }
+
     private static <T extends JoinBridge> InternalJoinBridgeDataManager<T> internalJoinBridgeDataManager(
             PipelineExecutionStrategy probeExecutionStrategy,
             PipelineExecutionStrategy buildExecutionStrategy,
@@ -213,6 +221,8 @@ public class JoinBridgeManager<T extends JoinBridge>
         void outerOperatorCreated(Lifespan lifespan);
 
         void outerOperatorClosed(Lifespan lifespan);
+
+        void probeOperatorFinish(Lifespan lifespan);
     }
 
     // 1 probe, 1 lookup source
@@ -252,6 +262,7 @@ public class JoinBridgeManager<T extends JoinBridge>
         public void probeOperatorFactoryClosed(Lifespan lifespan)
         {
             checkArgument(Lifespan.taskWide().equals(lifespan));
+            joinLifecycle.releaseForProbeInMem();
             joinLifecycle.releaseForProbe();
         }
 
@@ -289,6 +300,13 @@ public class JoinBridgeManager<T extends JoinBridge>
             checkArgument(Lifespan.taskWide().equals(lifespan));
             joinLifecycle.releaseForOuter();
         }
+
+        @Override
+        public void probeOperatorFinish(Lifespan lifespan)
+        {
+            checkArgument(Lifespan.taskWide().equals(lifespan));
+            joinLifecycle.releaseForProbeInMem();
+        }
     }
 
     // N probe, N lookup source; one-to-one mapping, bijective
@@ -331,6 +349,7 @@ public class JoinBridgeManager<T extends JoinBridge>
         public void probeOperatorFactoryClosed(Lifespan lifespan)
         {
             checkArgument(!Lifespan.taskWide().equals(lifespan));
+            data(lifespan).joinLifecycle.releaseForProbeInMem();
             data(lifespan).joinLifecycle.releaseForProbe();
         }
 
@@ -369,6 +388,13 @@ public class JoinBridgeManager<T extends JoinBridge>
             data(lifespan).joinLifecycle.releaseForOuter();
         }
 
+        @Override
+        public void probeOperatorFinish(Lifespan lifespan)
+        {
+            checkArgument(!Lifespan.taskWide().equals(lifespan));
+            data(lifespan).joinLifecycle.releaseForProbeInMem();
+        }
+
         private JoinBridgeAndLifecycle<T> data(Lifespan lifespan)
         {
             checkArgument(!Lifespan.taskWide().equals(lifespan));
@@ -421,6 +447,7 @@ public class JoinBridgeManager<T extends JoinBridge>
         @Override
         public void probeOperatorFactoryClosedForAllLifespans()
         {
+            joinLifecycle.releaseForProbeInMem();
             joinLifecycle.releaseForProbe();
         }
 
@@ -464,11 +491,18 @@ public class JoinBridgeManager<T extends JoinBridge>
             checkArgument(Lifespan.taskWide().equals(lifespan), "join bridge is not partitioned");
             joinLifecycle.releaseForOuter();
         }
+
+        @Override
+        public void probeOperatorFinish(Lifespan lifespan)
+        {
+            joinLifecycle.releaseForProbeInMem();
+        }
     }
 
     private static class JoinLifecycle
     {
         private final ReferenceCount probeReferenceCount;
+        private final ReferenceCount probeInMemReferenceCount;
         private final ReferenceCount outerReferenceCount;
 
         private final ListenableFuture<?> whenBuildAndProbeFinishes;
@@ -486,9 +520,12 @@ public class JoinBridgeManager<T extends JoinBridge>
             // * Each probe operator factory count as 1
             // * Each probe operator count as 1
             probeReferenceCount = new ReferenceCount(probeFactoryCount);
+            probeInMemReferenceCount = new ReferenceCount(probeFactoryCount);
+
+            ListenableFuture<?> whenMemProbeFinish = whenAnyComplete(ImmutableList.of(probeInMemReferenceCount.getFreeFuture(), probeReferenceCount.getFreeFuture()));
+            whenBuildAndProbeFinishes = Futures.whenAllSucceed(joinBridge.whenBuildFinishes(), whenMemProbeFinish).call(() -> joinBridge.whenMemProbeFinishes(), directExecutor());
+            whenAllFinishes = Futures.whenAllSucceed(whenBuildAndProbeFinishes, probeReferenceCount.getFreeFuture(), outerReferenceCount.getFreeFuture()).call(() -> null, directExecutor());
 
-            whenBuildAndProbeFinishes = Futures.whenAllSucceed(joinBridge.whenBuildFinishes(), probeReferenceCount.getFreeFuture()).call(() -> null, directExecutor());
-            whenAllFinishes = Futures.whenAllSucceed(whenBuildAndProbeFinishes, outerReferenceCount.getFreeFuture()).call(() -> null, directExecutor());
             whenAllFinishes.addListener(joinBridge::destroy, directExecutor());
         }
 
@@ -500,6 +537,7 @@ public class JoinBridgeManager<T extends JoinBridge>
         private void retainForProbe()
         {
             probeReferenceCount.retain();
+            probeInMemReferenceCount.retain();
         }
 
         private void releaseForProbe()
@@ -516,6 +554,11 @@ public class JoinBridgeManager<T extends JoinBridge>
         {
             outerReferenceCount.release();
         }
+
+        public void releaseForProbeInMem()
+        {
+            probeInMemReferenceCount.release();
+        }
     }
 
     private static class FreezeOnReadCounter

presto-main/src/main/java/io/prestosql/operator/LookupJoinOperator.java

@@ -44,6 +44,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalInt;
+import java.util.function.BiPredicate;
 import java.util.function.IntPredicate;
 import java.util.function.Supplier;
 
@@ -53,6 +54,7 @@ import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.airlift.concurrent.MoreFutures.addSuccessCallback;
 import static io.airlift.concurrent.MoreFutures.checkSuccess;
 import static io.airlift.concurrent.MoreFutures.getDone;
+import static io.prestosql.SystemSessionProperties.isInnerJoinSpillFilteringEnabled;
 import static io.prestosql.operator.LookupJoinOperators.JoinType.FULL_OUTER;
 import static io.prestosql.operator.LookupJoinOperators.JoinType.PROBE_OUTER;
 import static java.lang.String.format;
@@ -73,7 +75,7 @@ import static java.util.Objects.requireNonNull;
 @RestorableConfig(uncapturedFields = {"probeTypes", "joinProbeFactory", "afterClose", "hashGenerator", "lookupSourceFactory",
         "partitioningSpillerFactory", "lookupSourceProviderFuture", "lookupSourceProvider", "probe", "outputPage",
         "partitionGenerator", "spillInProgress", "unspilling", "currentPartition",
-        "unspilledLookupSource", "unspilledInputPages", "snapshotState"})
+        "unspilledLookupSource", "unspilledInputPages", "snapshotState", "afterMemOpFinish"})
 public class LookupJoinOperator
         implements Operator
 {
@@ -84,6 +86,7 @@ public class LookupJoinOperator
     private final List<Type> probeTypes;
     private final JoinProbeFactory joinProbeFactory;
     private final Runnable afterClose;
+    private Runnable afterMemOpFinish;
     private final OptionalInt lookupJoinsCount;
     private final HashGenerator hashGenerator;
     private final LookupSourceFactory lookupSourceFactory;
@@ -94,6 +97,7 @@ public class LookupJoinOperator
     private final LookupJoinPageBuilder pageBuilder;
 
     private final boolean probeOnOuterSide;
+    private final boolean spillBypassEnabled;
 
     private final ListenableFuture<LookupSourceProvider> lookupSourceProviderFuture;
     private LookupSourceProvider lookupSourceProvider;
@@ -136,7 +140,8 @@ public class LookupJoinOperator
             Runnable afterClose,
             OptionalInt lookupJoinsCount,
             HashGenerator hashGenerator,
-            PartitioningSpillerFactory partitioningSpillerFactory)
+            PartitioningSpillerFactory partitioningSpillerFactory,
+            Runnable afterMemOpFinish)
     {
         this.operatorContext = requireNonNull(operatorContext, "operatorContext is null");
         this.forked = forked;
@@ -145,6 +150,7 @@ public class LookupJoinOperator
         requireNonNull(joinType, "joinType is null");
         // Cannot use switch case here, because javac will synthesize an inner class and cause IllegalAccessError
         probeOnOuterSide = joinType == PROBE_OUTER || joinType == FULL_OUTER;
+        spillBypassEnabled = probeOnOuterSide || !isInnerJoinSpillFilteringEnabled(operatorContext.getDriverContext().getSession());
 
         this.joinProbeFactory = requireNonNull(joinProbeFactory, "joinProbeFactory is null");
         this.afterClose = requireNonNull(afterClose, "afterClose is null");
@@ -159,6 +165,8 @@ public class LookupJoinOperator
 
         this.pageBuilder = new LookupJoinPageBuilder(buildOutputTypes);
         this.snapshotState = operatorContext.isSnapshotEnabled() ? SingleInputSnapshotState.forOperator(this, operatorContext) : null;
+
+        this.afterMemOpFinish = afterMemOpFinish;
     }
 
     @Override
@@ -269,7 +277,9 @@ public class LookupJoinOperator
 
         Page newPage = page;
         if (spillInfoSnapshot.hasSpilled()) {
-            newPage = spillAndMaskSpilledPositions(page, spillInfoSnapshot.getSpillMask());
+            newPage = spillAndMaskSpilledPositions(page,
+                    spillInfoSnapshot.getSpillMask(),
+                    (spillBypassEnabled) ? (i, j) -> true : spillInfoSnapshot.getSpillMatcher());
             if (newPage.getPositionCount() == 0) {
                 return;
             }
@@ -295,7 +305,12 @@ public class LookupJoinOperator
         return true;
     }
 
-    private Page spillAndMaskSpilledPositions(Page page, IntPredicate spillMask)
+    private static Long getHashValue(HashGenerator hashGenerator, Object position, Object page)
+    {
+        return hashGenerator.hashPosition((int) position, (Page) page);
+    }
+
+    private Page spillAndMaskSpilledPositions(Page page, IntPredicate spillMask, BiPredicate<Integer, Long> spillMatcher)
     {
         checkState(spillInProgress.isDone(), "Previous spill still in progress");
         checkSuccess(spillInProgress, "spilling failed");
@@ -305,10 +320,11 @@ public class LookupJoinOperator
                     probeTypes,
                     getPartitionGenerator(),
                     operatorContext.getSpillContext().newLocalSpillContext(),
-                    operatorContext.newAggregateSystemMemoryContext()));
+                    operatorContext.newAggregateSystemMemoryContext(),
+                    hashGenerator::hashPosition));
         }
 
-        PartitioningSpillResult result = spiller.get().partitionAndSpill(page, spillMask);
+        PartitioningSpillResult result = spiller.get().partitionAndSpill(page, spillMask, spillMatcher);
         spillInProgress = result.getSpillingFuture();
         return result.getRetained();
     }
@@ -363,7 +379,10 @@ public class LookupJoinOperator
              * Let LookupSourceFactory know LookupSources can be disposed as far as we're concerned.
              */
             verify(partitionedConsumption == null, "partitioned consumption already started");
+            lookupSourceProvider.close();
             partitionedConsumption = lookupSourceFactory.finishProbeOperator(lookupJoinsCount);
+            afterMemOpFinish.run();
+            afterMemOpFinish = () -> {};
             unspilling = true;
         }
 
@@ -586,6 +605,7 @@ public class LookupJoinOperator
         try (Closer closer = Closer.create()) {
             // `afterClose` must be run last.
             // Closer is documented to mimic try-with-resource, which implies close will happen in reverse order.
+            closer.register(afterMemOpFinish::run);
             closer.register(afterClose::run);
 
             closer.register(pageBuilder::reset);
@@ -667,12 +687,19 @@ public class LookupJoinOperator
         private final boolean hasSpilled;
         private final long spillEpoch;
         private final IntPredicate spillMask;
+        private final BiPredicate<Integer, Long> spillMatcher;
 
         public SpillInfoSnapshot(boolean hasSpilled, long spillEpoch, IntPredicate spillMask)
+        {
+            this(hasSpilled, spillEpoch, spillMask, (a, b) -> true);
+        }
+
+        public SpillInfoSnapshot(boolean hasSpilled, long spillEpoch, IntPredicate spillMask, BiPredicate<Integer, Long> spillMatcher)
         {
             this.hasSpilled = hasSpilled;
             this.spillEpoch = spillEpoch;
             this.spillMask = requireNonNull(spillMask, "spillMask is null");
+            this.spillMatcher = requireNonNull(spillMatcher, "spillMater is null");
         }
 
         public static SpillInfoSnapshot from(LookupSourceLease lookupSourceLease)
@@ -680,7 +707,8 @@ public class LookupJoinOperator
             return new SpillInfoSnapshot(
                     lookupSourceLease.hasSpilled(),
                     lookupSourceLease.spillEpoch(),
-                    lookupSourceLease.getSpillMask());
+                    lookupSourceLease.getSpillMask(),
+                    lookupSourceLease.getSpillMatcher());
         }
 
         public static SpillInfoSnapshot noSpill()
@@ -702,6 +730,11 @@ public class LookupJoinOperator
         {
             return spillMask;
         }
+
+        public BiPredicate<Integer, Long> getSpillMatcher()
+        {
+            return spillMatcher;
+        }
     }
 
     // This class must be public because LookupJoinOperator is isolated.
@@ -853,7 +886,10 @@ public class LookupJoinOperator
                     i -> {
                         throw new UnsupportedOperationException();
                     },
-                    i -> {}));
+                    i -> {},
+                    i -> {
+                        throw new UnsupportedOperationException();
+                    }));
         }
         else {
             this.partitionedConsumption = null;

presto-main/src/main/java/io/prestosql/operator/LookupJoinOperatorFactory.java

@@ -157,7 +157,8 @@ public class LookupJoinOperatorFactory
                 () -> joinBridgeManager.probeOperatorClosed(driverContext.getLifespan()),
                 totalOperatorsCount,
                 probeHashGenerator,
-                partitioningSpillerFactory);
+                partitioningSpillerFactory,
+                () -> joinBridgeManager.probeOperatorFinished(driverContext.getLifespan()));
     }
 
     @Override

presto-main/src/main/java/io/prestosql/operator/LookupOuterOperator.java

@@ -124,7 +124,7 @@ public class LookupOuterOperator
 
     private final OperatorContext operatorContext;
     private final LookupSourceFactory lookupSourceFactory;
-    private final ListenableFuture<OuterPositionIterator> outerPositionsFuture;
+    private ListenableFuture<OuterPositionIterator> outerPositionsFuture;
 
     private final List<Type> probeOutputTypes;
     private final Runnable onClose;
@@ -274,7 +274,12 @@ public class LookupOuterOperator
         }
 
         if (outputPositionsFinished) {
-            close();
+            outerPositionsFuture = outerPositions.getNextBatch();
+            outerPositions = null;
+            outerPositions = tryGetFutureValue(outerPositionsFuture).orElse(null);
+            if (outerPositions == null) {
+                close();
+            }
         }
         return page;
     }

presto-main/src/main/java/io/prestosql/operator/LookupSourceFactory.java

@@ -13,6 +13,7 @@
  */
 package io.prestosql.operator;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.prestosql.spi.plan.Symbol;
 import io.prestosql.spi.snapshot.MarkerPage;
@@ -45,7 +46,20 @@ public interface LookupSourceFactory
                 i -> {
                     throw new UnsupportedOperationException();
                 },
-                i -> {}));
+                i -> {},
+                i -> {
+                    throw new UnsupportedOperationException();
+                }));
+    }
+
+    default ListenableFuture<PartitionedConsumption<OuterPositionIterator>> startOuterOperator(OptionalInt lookupJoinsCount)
+    {
+        return immediateFuture(new PartitionedConsumption<>(
+                1,
+                ImmutableList.of(1),
+                i -> immediateFuture(null),
+                i -> {},
+                i -> immediateFuture(null)));
     }
 
     /**

presto-main/src/main/java/io/prestosql/operator/LookupSourceProvider.java

@@ -13,6 +13,7 @@
  */
 package io.prestosql.operator;
 
+import java.util.function.BiPredicate;
 import java.util.function.Function;
 import java.util.function.IntPredicate;
 
@@ -33,5 +34,10 @@ public interface LookupSourceProvider
         long spillEpoch();
 
         IntPredicate getSpillMask();
+
+        default BiPredicate<Integer, Long> getSpillMatcher()
+        {
+            return (a, b) -> true;
+        }
     }
 }

presto-main/src/main/java/io/prestosql/operator/OuterPositionIterator.java

@@ -13,9 +13,17 @@
  */
 package io.prestosql.operator;
 
+import com.google.common.util.concurrent.ListenableFuture;
 import io.prestosql.spi.PageBuilder;
 
+import static com.google.common.util.concurrent.Futures.immediateFuture;
+
 public interface OuterPositionIterator
 {
     boolean appendToNext(PageBuilder pageBuilder, int outputChannelOffset);
+
+    default ListenableFuture<OuterPositionIterator> getNextBatch()
+    {
+        return immediateFuture(null);
+    }
 }

presto-main/src/main/java/io/prestosql/operator/PartitionedConsumption.java

@@ -36,6 +36,7 @@ import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.util.concurrent.Futures.allAsList;
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
+import static java.util.Collections.emptyIterator;
 import static java.util.Objects.requireNonNull;
 
 @ThreadSafe
@@ -46,9 +47,10 @@ public final class PartitionedConsumption<T>
     @Nullable
     private List<Partition<T>> partitions;
 
-    public PartitionedConsumption(int consumersCount, Iterable<Integer> partitionNumbers, IntFunction<ListenableFuture<T>> loader, IntConsumer disposer)
+    public PartitionedConsumption(int consumersCount, Iterable<Integer> partitionNumbers, IntFunction<ListenableFuture<T>> loader, IntConsumer disposer,
+                                  IntFunction<ListenableFuture<?>> disposed)
     {
-        this(consumersCount, immediateFuture(null), partitionNumbers, loader, disposer);
+        this(consumersCount, immediateFuture(null), partitionNumbers, loader, disposer, disposed);
     }
 
     public PartitionedConsumption(
@@ -56,18 +58,20 @@ public final class PartitionedConsumption<T>
             ListenableFuture<?> activator,
             Iterable<Integer> partitionNumbers,
             IntFunction<ListenableFuture<T>> loader,
-            IntConsumer disposer)
+            IntConsumer disposer,
+            IntFunction<ListenableFuture<?>> disposed)
     {
         checkArgument(consumersCount > 0, "consumersCount must be positive");
         this.consumersCount = consumersCount;
-        this.partitions = createPartitions(activator, partitionNumbers, loader, disposer);
+        this.partitions = createPartitions(activator, partitionNumbers, loader, disposer, disposed);
     }
 
     private List<Partition<T>> createPartitions(
             ListenableFuture<?> activator,
             Iterable<Integer> partitionNumbers,
             IntFunction<ListenableFuture<T>> loader,
-            IntConsumer disposer)
+            IntConsumer disposer,
+            IntFunction<ListenableFuture<?>> disposed)
     {
         requireNonNull(partitionNumbers, "partitionNumbers is null");
         requireNonNull(loader, "loader is null");
@@ -78,7 +82,7 @@ public final class PartitionedConsumption<T>
         for (Integer partitionNumber : partitionNumbers) {
             Partition<T> partition = new Partition<>(consumersCount, partitionNumber, loader, partitionActivator, disposer);
             partitionList.add(partition);
-            partitionActivator = partition.released;
+            partitionActivator = disposed.apply(partitionNumber);
         }
         return partitionList.build();
     }
@@ -88,13 +92,15 @@ public final class PartitionedConsumption<T>
         return consumersCount;
     }
 
-    public Iterator<Partition<T>> beginConsumption()
+    public synchronized Iterator<Partition<T>> beginConsumption()
     {
-        Queue<Partition<T>> partitionQueue = new ArrayDeque<>(requireNonNull(this.partitions, "partitionQueue is already null"));
-        if (consumed.incrementAndGet() >= consumersCount) {
+        if (consumed.getAndIncrement() >= consumersCount) {
             // Unreference futures to allow GC
             this.partitions = null;
+            return emptyIterator();
         }
+
+        Queue<Partition<T>> partitionQueue = new ArrayDeque<>(requireNonNull(this.partitions, "partitionQueue is already null"));
         return new AbstractIterator<Partition<T>>()
         {
             @Override
@@ -116,7 +122,7 @@ public final class PartitionedConsumption<T>
         private final int partitionNumber;
         private final SettableFuture<?> requested;
         private final ListenableFuture<T> loaded;
-        private final SettableFuture<?> released;
+        private final IntConsumer disposer;
 
         @GuardedBy("this")
         private int pendingReleases;
@@ -134,8 +140,7 @@ public final class PartitionedConsumption<T>
                     allAsList(requested, previousReleased),
                     ignored -> loader.apply(partitionNumber),
                     directExecutor());
-            this.released = SettableFuture.create();
-            released.addListener(() -> disposer.accept(partitionNumber), directExecutor());
+            this.disposer = disposer;
             this.pendingReleases = consumersCount;
         }
 
@@ -156,7 +161,7 @@ public final class PartitionedConsumption<T>
             pendingReleases--;
             checkState(pendingReleases >= 0);
             if (pendingReleases == 0) {
-                released.set(null);
+                disposer.accept(partitionNumber);
             }
         }
     }

presto-main/src/main/java/io/prestosql/operator/PartitionedLookupSource.java

@@ -13,28 +13,40 @@
  */
 package io.prestosql.operator;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.io.Closer;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
 import io.prestosql.operator.exchange.LocalPartitionGenerator;
 import io.prestosql.spi.Page;
 import io.prestosql.spi.PageBuilder;
 import io.prestosql.spi.type.Type;
+import org.roaringbitmap.RoaringBitmap;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
 
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.NotThreadSafe;
 
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
-import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Verify.verify;
 import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.google.common.util.concurrent.Futures.immediateFuture;
+import static io.airlift.concurrent.MoreFutures.whenAnyComplete;
 import static java.lang.Integer.numberOfTrailingZeros;
 import static java.lang.Math.toIntExact;
 
@@ -46,7 +58,7 @@ public class PartitionedLookupSource
             List<Type> hashChannelTypes, boolean outer, Object restoredJoinPositions)
     {
         if (outer) {
-            OuterPositionTracker.Factory outerPositionTrackerFactory = new OuterPositionTracker.Factory(partitions, restoredJoinPositions);
+            OuterPositionTrackerFactory outerPositionTrackerFactory = new OuterPositionTrackerFactory(partitions, restoredJoinPositions);
 
             return new TrackingLookupSourceSupplier()
             {
@@ -67,10 +79,21 @@ public class PartitionedLookupSource
                     return outerPositionTrackerFactory.getOuterPositionIterator();
                 }
 
+                @Override
+                public ListenableFuture<?> setOuterPartitionReady(int partition)
+                {
+                    return outerPositionTrackerFactory.setPartitionReady(partition);
+                }
+
                 @Override
                 public Object captureJoinPositions()
                 {
-                    return outerPositionTrackerFactory.captureJoinPositions();
+                    try {
+                        return outerPositionTrackerFactory.captureJoinPositions();
+                    }
+                    catch (IOException e) {
+                        throw new UncheckedIOException(e);
+                    }
                 }
 
                 @Override
@@ -220,6 +243,18 @@ public class PartitionedLookupSource
         closed = true;
     }
 
+    public void setPartitionLookup(LookupSource lookupSource, int partition)
+    {
+        verify(partition < lookupSources.length);
+        verify(lookupSources[partition] instanceof PartitionedLookupSourceFactory.SpilledLookupSource);
+        verify(!(lookupSource instanceof PartitionedLookupSourceFactory.SpilledLookupSource));
+
+        this.lookupSources[partition] = lookupSource;
+        if (outerPositionTracker != null) {
+            lookupSource.getJoinPositionCount();
+        }
+    }
+
     private int decodePartition(long partitionedJoinPosition)
     {
         return (int) (partitionedJoinPosition & partitionMask);
@@ -239,7 +274,9 @@ public class PartitionedLookupSource
             implements OuterPositionIterator
     {
         private final LookupSource[] lookupSources;
-        private final boolean[][] visitedPositions;
+        private final RoaringBitmap[] visitedPositions;
+        private final OuterPositionTrackerFactory outerPositionTrackerFactory;
+        private final int[] partitionNumbers;
 
         @GuardedBy("this")
         private int currentSource;
@@ -247,18 +284,22 @@ public class PartitionedLookupSource
         @GuardedBy("this")
         private int currentPosition;
 
-        public PartitionedLookupOuterPositionIterator(LookupSource[] lookupSources, boolean[][] visitedPositions)
+        public PartitionedLookupOuterPositionIterator(LookupSource[] lookupSources, RoaringBitmap[] visitedPositions,
+                                                      int[] partitionNumbers, OuterPositionTrackerFactory outerPositionTrackerFactory)
         {
             this.lookupSources = lookupSources;
             this.visitedPositions = visitedPositions;
+            this.partitionNumbers = partitionNumbers;
+            this.outerPositionTrackerFactory = outerPositionTrackerFactory;
         }
 
         @Override
         public synchronized boolean appendToNext(PageBuilder pageBuilder, int outputChannelOffset)
         {
             while (currentSource < lookupSources.length) {
-                while (currentPosition < visitedPositions[currentSource].length) {
-                    if (!visitedPositions[currentSource][currentPosition]) {
+                long visitedPosCount = lookupSources[currentSource].getJoinPositionCount();
+                while (currentPosition < visitedPosCount) {
+                    if (!visitedPositions[currentSource].contains(currentPosition)) {
                         lookupSources[currentSource].appendTo(currentPosition, pageBuilder, outputChannelOffset);
                         currentPosition++;
                         return true;
@@ -266,10 +307,18 @@ public class PartitionedLookupSource
                     currentPosition++;
                 }
                 currentPosition = 0;
+                outerPositionTrackerFactory.setPartitionDone(partitionNumbers[currentSource]);
                 currentSource++;
             }
+
             return false;
         }
+
+        @Override
+        public ListenableFuture<OuterPositionIterator> getNextBatch()
+        {
+            return outerPositionTrackerFactory.getNextReady();
+        }
     }
 
     /**
@@ -286,67 +335,249 @@ public class PartitionedLookupSource
      * getVisitedPositions() is guaranteed by accessing AtomicLong referenceCount
      * variables in those two methods.
      */
-    private static class OuterPositionTracker
+    public static class OuterPositionTrackerFactory
     {
-        public static class Factory
+        private final List<LookupSource> lookupSources;
+        private final List<RoaringBitmap> visitedPositions;
+        private final ReentrantReadWriteLock[] locks;
+        private final AtomicBoolean[] finished;
+        private final AtomicLong[] referenceCount;
+        private final List<SettableFuture<OuterPositionIterator>> partitionReady;
+        private final List<SettableFuture<?>> partitionDone;
+
+        public OuterPositionTrackerFactory(List<Supplier<LookupSource>> partitions, Object restoredJoinPositions)
         {
-            private final LookupSource[] lookupSources;
-            private final boolean[][] visitedPositions;
-            private final AtomicBoolean finished = new AtomicBoolean();
-            private final AtomicLong referenceCount = new AtomicLong();
+            this.lookupSources = partitions.stream()
+                    .map(Supplier::get)
+                    .collect(Collectors.toList());
 
-            public Factory(List<Supplier<LookupSource>> partitions, Object restoredJoinPositions)
-            {
-                this.lookupSources = partitions.stream()
-                        .map(Supplier::get)
-                        .toArray(LookupSource[]::new);
+            visitedPositions = new ArrayList<>();
+            if (restoredJoinPositions != null) {
+                restoreJoinPositions(restoredJoinPositions);
+            }
 
-                if (restoredJoinPositions != null) {
-                    visitedPositions = (boolean[][]) restoredJoinPositions;
+            finished = new AtomicBoolean[lookupSources.size()];
+            referenceCount = new AtomicLong[lookupSources.size()];
+
+            partitionReady = new ArrayList<>();
+            partitionDone = new ArrayList<>();
+            locks = new ReentrantReadWriteLock[lookupSources.size()];
+            for (int i = 0; i < partitions.size(); i++) {
+                finished[i] = new AtomicBoolean();
+                referenceCount[i] = new AtomicLong();
+
+                partitionReady.add(SettableFuture.create());
+                partitionDone.add(SettableFuture.create());
+                if (!(partitions.get(i).get() instanceof PartitionedLookupSourceFactory.SpilledLookupSource)) {
+                    partitionReady.get(i).set(null);
                 }
-                else {
-                    visitedPositions = Arrays.stream(this.lookupSources)
-                            .map(LookupSource::getJoinPositionCount)
-                            .map(Math::toIntExact)
-                            .map(boolean[]::new)
-                            .toArray(boolean[][]::new);
-                }
-            }
-
-            public OuterPositionTracker create()
-            {
-                return new OuterPositionTracker(visitedPositions, finished, referenceCount);
-            }
-
-            public OuterPositionIterator getOuterPositionIterator()
-            {
-                // touching atomic values ensures memory visibility between commit and getVisitedPositions
-                verify(referenceCount.get() == 0);
-                finished.set(true);
-                return new PartitionedLookupOuterPositionIterator(lookupSources, visitedPositions);
-            }
-
-            public Object captureJoinPositions()
-            {
-                return visitedPositions;
-            }
-
-            public void restoreJoinPositions(Object state)
-            {
-                boolean[][] joinPositions = (boolean[][]) state;
-                for (int i = 0; i < joinPositions.length; i++) {
-                    checkState(joinPositions[i].length == visitedPositions[i].length);
-                    System.arraycopy(joinPositions[i], 0, visitedPositions[i], 0, joinPositions[i].length);
+
+                if (restoredJoinPositions == null) {
+                    visitedPositions.add(new RoaringBitmap());
                 }
+                locks[i] = new ReentrantReadWriteLock();
             }
         }
 
-        private final boolean[][] visitedPositions; // shared across multiple operators/drivers
+        public OuterPositionTracker create()
+        {
+            return new InMemoryOuterPositionTracker(visitedPositions, locks, finished, referenceCount);
+        }
+
+        public OuterPositionIterator getOuterPositionIterator()
+        {
+            int[] selectedPartitions = new int[lookupSources.size()];
+            int count = 0;
+            for (int i = 0; i < lookupSources.size(); i++) {
+                if (partitionReady.get(i).isDone()) {
+                    if (!partitionDone.get(i).isDone()) {
+                        if (lookupSources.get(i).getJoinPositionCount() <= 0
+                                || lookupSources.get(i).getJoinPositionCount() <= visitedPositions.get(i).getCardinality()) {
+                            setPartitionDone(i);
+                            continue;
+                        }
+                        selectedPartitions[count++] = i;
+                    }
+                }
+            }
+
+            LookupSource[] ls = new LookupSource[count];
+            RoaringBitmap[] rb = new RoaringBitmap[count];
+            for (int i = 0; i < count; i++) {
+                ls[i] = lookupSources.get(selectedPartitions[i]);
+                rb[i] = visitedPositions.get(selectedPartitions[i]);
+
+                // touching atomic values ensures memory visibility between commit and getVisitedPositions
+                verify(referenceCount[selectedPartitions[i]].get() == 0);
+                finished[selectedPartitions[i]].set(true);
+            }
+
+            return new PartitionedLookupOuterPositionIterator(ls, rb, selectedPartitions, this);
+        }
+
+        protected synchronized ListenableFuture<OuterPositionIterator> getNextReady()
+        {
+            ImmutableList.Builder<ListenableFuture<OuterPositionIterator>> builder = ImmutableList.builder();
+
+            int objs = 0;
+            for (int i = 0; i < lookupSources.size(); i++) {
+                if (!partitionDone.get(i).isDone()) {
+                    builder.add(partitionReady.get(i));
+                    objs++;
+                }
+            }
+
+            if (objs > 0) {
+                return whenAnyComplete(builder.build());
+            }
+
+            return immediateFuture(null);
+        }
+
+        protected synchronized void setOuterPositionIterator(int partitionNumber)
+        {
+            verify(partitionNumber < lookupSources.size());
+            verify(!partitionReady.get(partitionNumber).isDone());
+
+            partitionReady.get(partitionNumber)
+                    .set(new PartitionedLookupOuterPositionIterator(
+                            new LookupSource[] {lookupSources.get(partitionNumber)},
+                            new RoaringBitmap[] {visitedPositions.get(partitionNumber)},
+                            new int[] {0},
+                            this));
+        }
+
+        public Object captureJoinPositions() throws IOException
+        {
+            ByteArrayOutputStream bos = new ByteArrayOutputStream();
+            DataOutputStream dos = new DataOutputStream(bos);
+            for (RoaringBitmap rr : visitedPositions) {
+                rr.serialize(dos);
+            }
+            dos.close();
+            return bos.toByteArray();
+        }
+
+        public void restoreJoinPositions(Object state)
+        {
+            ByteBuffer bb = ByteBuffer.wrap((byte[]) state);
+            visitedPositions.clear();
+            for (int i = 0; i < lookupSources.size(); i++) {
+                ImmutableRoaringBitmap bm = new ImmutableRoaringBitmap(bb);
+                visitedPositions.add(new RoaringBitmap(bm));
+                bb.position(bb.position() + visitedPositions.get(i).serializedSizeInBytes());
+            }
+        }
+
+        public void setPartitionDone(int partition)
+        {
+            verify(partition < partitionDone.size());
+            partitionDone.get(partition).set(null);
+
+            locks[partition].writeLock().lock();
+            try {
+                visitedPositions.get(partition).clear();
+            }
+            finally {
+                locks[partition].writeLock().unlock();
+            }
+        }
+
+        public ListenableFuture<?> setPartitionReady(int partition)
+        {
+            verify(partition < lookupSources.size());
+            verify(!partitionReady.get(partition).isDone());
+
+            locks[partition].writeLock().lock();
+            try {
+                if (lookupSources.get(partition).getJoinPositionCount() <= 0
+                        || lookupSources.get(partition).getJoinPositionCount() <= visitedPositions.get(partition).getCardinality()) {
+                    setPartitionDone(partition); /* all matched in this partition; skip it! */
+                }
+                else {
+                    partitionReady.get(partition)
+                            .set(new PartitionedLookupOuterPositionIterator(
+                                    new LookupSource[]{lookupSources.get(partition)},
+                                    new RoaringBitmap[]{visitedPositions.get(partition)},
+                                    new int[]{0},
+                                    this));
+                }
+                return partitionDone.get(partition);
+            }
+            finally {
+                locks[partition].writeLock().unlock();
+            }
+        }
+    }
+
+    public interface OuterPositionTracker
+    {
+        void positionVisited(int partitioned, int position);
+
+        void commit();
+    }
+
+    private static class InMemoryOuterPositionTracker
+            implements OuterPositionTracker
+    {
+        private final RoaringBitmap[] visitedPositions; // shared across multiple operators/drivers
+        private final ReentrantReadWriteLock[] locks;
+        private final AtomicBoolean[] finished; // shared across multiple operators/drivers
+        private final AtomicLong[] referenceCount; // shared across multiple operators/drivers
+        private boolean[] written; // unique per each operator/driver
+
+        private InMemoryOuterPositionTracker(List<RoaringBitmap> visitedPositions, ReentrantReadWriteLock[] locks, AtomicBoolean[] finished, AtomicLong[] referenceCount)
+        {
+            this.visitedPositions = visitedPositions.toArray(new RoaringBitmap[visitedPositions.size()]);
+            this.locks = locks;
+            this.finished = finished;
+            this.referenceCount = referenceCount;
+            this.written = new boolean[visitedPositions.size()];
+        }
+
+        /**
+         * No synchronization here, because it would be very expensive. Check comment above.
+         */
+        @Override
+        public void positionVisited(int partition, int position)
+        {
+            verify(partition < referenceCount.length);
+            if (!written[partition]) {
+                written[partition] = true;
+                verify(!finished[partition].get());
+                referenceCount[partition].incrementAndGet();
+            }
+
+            locks[partition].writeLock().lock();
+            try {
+                visitedPositions[partition].add(position);
+            }
+            finally {
+                locks[partition].writeLock().unlock();
+            }
+        }
+
+        @Override
+        public void commit()
+        {
+            for (int i = 0; i < written.length; i++) {
+                if (written[i]) {
+                    // touching atomic values ensures memory visibility between commit and getVisitedPositions
+                    referenceCount[i].decrementAndGet();
+                }
+            }
+        }
+    }
+
+    private static class SpillableOuterPositionTracker
+            implements OuterPositionTracker
+    {
+        private final RoaringBitmap[] visitedPositions; // shared across multiple operators/drivers
         private final AtomicBoolean finished; // shared across multiple operators/drivers
         private final AtomicLong referenceCount; // shared across multiple operators/drivers
         private boolean written; // unique per each operator/driver
 
-        private OuterPositionTracker(boolean[][] visitedPositions, AtomicBoolean finished, AtomicLong referenceCount)
+        private SpillableOuterPositionTracker(RoaringBitmap[] visitedPositions, AtomicBoolean finished, AtomicLong referenceCount)
         {
             this.visitedPositions = visitedPositions;
             this.finished = finished;
@@ -356,16 +587,18 @@ public class PartitionedLookupSource
         /**
          * No synchronization here, because it would be very expensive. Check comment above.
          */
-        public void positionVisited(int partition, int position)
+        @Override
+        public void positionVisited(int partitioned, int position)
         {
             if (!written) {
                 written = true;
                 verify(!finished.get());
                 referenceCount.incrementAndGet();
             }
-            visitedPositions[partition][position] = true;
+            visitedPositions[partitioned].add(position); /* Todo: Trigger spill if needed */
         }
 
+        @Override
         public void commit()
         {
             if (written) {

presto-main/src/main/java/io/prestosql/operator/PartitionedLookupSourceFactory.java

@@ -16,6 +16,8 @@ package io.prestosql.operator;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.hash.BloomFilter;
+import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import io.prestosql.operator.LookupSourceProvider.LookupSourceLease;
@@ -38,9 +40,11 @@ import java.util.OptionalInt;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.BiPredicate;
 import java.util.function.Function;
 import java.util.function.IntPredicate;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
@@ -62,6 +66,7 @@ public final class PartitionedLookupSourceFactory
     private final Map<Symbol, Integer> layout;
     private final List<Type> hashChannelTypes;
     private final boolean outer;
+    private final boolean spillEnabledForOuter;
     private final SpilledLookupSource spilledLookupSource;
 
     private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
@@ -78,7 +83,7 @@ public final class PartitionedLookupSourceFactory
     private int partitionsSet;
 
     @GuardedBy("lock")
-    private SpillingInfo spillingInfo = new SpillingInfo(0, ImmutableSet.of());
+    private SpillingInfo spillingInfo = new SpillingInfo(0, ImmutableSet.of(), ImmutableMap.of());
 
     @GuardedBy("lock")
     private final Map<Integer, SpilledLookupSourceHandle> spilledPartitions = new HashMap<>();
@@ -98,6 +103,13 @@ public final class PartitionedLookupSourceFactory
     @GuardedBy("lock")
     private final SettableFuture<PartitionedConsumption<Supplier<LookupSource>>> partitionedConsumption = SettableFuture.create();
 
+    @GuardedBy("lock")
+    private PartitionedLookupSource postConsumptionOuterSource;
+
+    private final SettableFuture<?> probeInMemFinish = SettableFuture.create();
+
+    PartitionedLookupSource.OuterPositionTrackerFactory outerPositions;
+
     /**
      * Cached LookupSource on behalf of LookupJoinOperator (represented by SpillAwareLookupSourceProvider). LookupSource instantiation has non-negligible cost.
      * <p>
@@ -122,7 +134,7 @@ public final class PartitionedLookupSourceFactory
      */
     private Object restoredJoinPositions;
 
-    public PartitionedLookupSourceFactory(List<Type> types, List<Type> outputTypes, List<Type> hashChannelTypes, int partitionCount, Map<Symbol, Integer> layout, boolean outer)
+    public PartitionedLookupSourceFactory(List<Type> types, List<Type> outputTypes, List<Type> hashChannelTypes, int partitionCount, Map<Symbol, Integer> layout, boolean outer, boolean spillEnabledForOuter)
     {
         checkArgument(Integer.bitCount(partitionCount) == 1, "partitionCount must be a power of 2");
 
@@ -133,6 +145,7 @@ public final class PartitionedLookupSourceFactory
         checkArgument(partitionCount > 0);
         this.partitions = (Supplier<LookupSource>[]) new Supplier<?>[partitionCount];
         this.outer = outer;
+        this.spillEnabledForOuter = spillEnabledForOuter;
         spilledLookupSource = new SpilledLookupSource(outputTypes.size());
     }
 
@@ -232,14 +245,18 @@ public final class PartitionedLookupSourceFactory
 
         lock.writeLock().lock();
         try {
-            if (destroyed.isDone()) {
+            if (partitionedConsumption.isDone() || partitionsNoLongerNeeded.isDone()) {
                 spilledLookupSourceHandle.dispose();
                 return;
             }
 
             checkState(!spilledPartitions.containsKey(partitionIndex), "Partition already set as spilled");
             spilledPartitions.put(partitionIndex, spilledLookupSourceHandle);
-            spillingInfo = new SpillingInfo(spillingInfo.spillEpoch() + 1, spilledPartitions.keySet());
+            Map<Integer, BloomFilter<Long>> spillBlooms = spilledPartitions.entrySet().stream()
+                    .filter(e -> e.getValue().getSpillBloom().isPresent())
+                    .collect(Collectors.toMap(x -> x.getKey(),
+                            x -> x.getValue().getSpillBloom().get()));
+            spillingInfo = new SpillingInfo(spillingInfo.spillEpoch() + 1, spilledPartitions.keySet(), spillBlooms);
 
             if (partitions[partitionIndex] != null) {
                 // Was present and now it's spilled
@@ -258,9 +275,12 @@ public final class PartitionedLookupSourceFactory
                  * and to prevent probe side accessing the partition.
                  */
                 verify(!completed, "lookupSourceSupplier already exist when completing");
-                verify(!outer, "It is not possible to reset lookupSourceSupplier which is tracking for outer join");
                 verify(partitions.length > 1, "Spill occurred when only one partition");
-                lookupSourceSupplier = createPartitionedLookupSourceSupplier(ImmutableList.copyOf(partitions), hashChannelTypes, outer, restoredJoinPositions);
+
+                /* Reinitialize the spilled partition specific outerVisitedPositions */
+                Object capturedStates = lookupSourceSupplier.captureJoinPositions();
+                lookupSourceSupplier = createPartitionedLookupSourceSupplier(ImmutableList.copyOf(partitions),
+                        hashChannelTypes, outer, capturedStates);
                 closeCachedLookupSources();
             }
             else {
@@ -293,7 +313,11 @@ public final class PartitionedLookupSourceFactory
 
             if (partitionsSet != 1) {
                 List<Supplier<LookupSource>> partitionList = ImmutableList.copyOf(this.partitions);
-                this.lookupSourceSupplier = createPartitionedLookupSourceSupplier(partitionList, hashChannelTypes, outer, restoredJoinPositions);
+                Object capturedStates = restoredJoinPositions;
+                if (lookupSourceSupplier != null) {
+                    capturedStates = lookupSourceSupplier.captureJoinPositions();
+                }
+                this.lookupSourceSupplier = createPartitionedLookupSourceSupplier(partitionList, hashChannelTypes, outer, capturedStates);
             }
             else if (outer) {
                 this.lookupSourceSupplier = createOuterLookupSourceSupplier(partitions[0], restoredJoinPositions);
@@ -320,6 +344,7 @@ public final class PartitionedLookupSourceFactory
     {
         lock.writeLock().lock();
         try {
+            checkState(!partitionedConsumption.isDone(), "All probe operators are already finished.");
             if (!spillingInfo.hasSpilled()) {
                 finishedProbeOperators++;
                 return immediateFuture(new PartitionedConsumption<>(
@@ -328,30 +353,20 @@ public final class PartitionedLookupSourceFactory
                         i -> {
                             throw new UnsupportedOperationException();
                         },
-                        i -> {}));
+                        i -> {},
+                        i -> {
+                            throw new UnsupportedOperationException();
+                        }));
             }
 
             int operatorsCount = lookupJoinsCount
                     .orElseThrow(() -> new IllegalStateException("A fixed distribution is required for JOIN when spilling is enabled"));
-            checkState(finishedProbeOperators < operatorsCount, "%s probe operators finished out of %s declared", finishedProbeOperators + 1, operatorsCount);
 
             if (!partitionedConsumptionParticipants.isPresent()) {
                 // This is the first probe to finish after anything has been spilled.
-                partitionedConsumptionParticipants = OptionalInt.of(operatorsCount - finishedProbeOperators);
+                partitionedConsumptionParticipants = OptionalInt.of(finishedProbeOperators);
             }
-
             finishedProbeOperators++;
-            if (finishedProbeOperators == operatorsCount) {
-                // We can dispose partitions now since as right outer is not supported with spill
-                freePartitions();
-                verify(!partitionedConsumption.isDone());
-                partitionedConsumption.set(new PartitionedConsumption<>(
-                        partitionedConsumptionParticipants.getAsInt(),
-                        spilledPartitions.keySet(),
-                        this::loadSpilledLookupSource,
-                        this::disposeSpilledLookupSource));
-            }
-
             return partitionedConsumption;
         }
         finally {
@@ -361,14 +376,45 @@ public final class PartitionedLookupSourceFactory
 
     private ListenableFuture<Supplier<LookupSource>> loadSpilledLookupSource(int partitionNumber)
     {
+        if (outer) {
+            verify(partitionsSet != 1);
+            ListenableFuture<Supplier<LookupSource>> lookupSupplierFuture = getSpilledLookupSourceHandle(partitionNumber).getLookupSource();
+            return Futures.transformAsync(lookupSupplierFuture,
+                    lookupSource -> getUpdatedPartitionedLookupSourceSupplier(lookupSource, partitionNumber),
+                    directExecutor());
+        }
         return getSpilledLookupSourceHandle(partitionNumber).getLookupSource();
     }
 
+    private ListenableFuture<Supplier<LookupSource>> getUpdatedPartitionedLookupSourceSupplier(Supplier<LookupSource> lookupSource, int partitionNumber)
+    {
+        postConsumptionOuterSource = (PartitionedLookupSource) this.lookupSourceSupplier.getLookupSource();
+        postConsumptionOuterSource.setPartitionLookup(lookupSource.get(), partitionNumber);
+        return immediateFuture(() -> postConsumptionOuterSource);
+    }
+
+    private void loadOuterIterator(int partitionNumber)
+    {
+        if (postConsumptionOuterSource != null) {
+            /* commit all the partitions visited! */
+            postConsumptionOuterSource.close();
+            postConsumptionOuterSource = null;
+        }
+
+        this.lookupSourceSupplier.setOuterPartitionReady(partitionNumber)
+                .addListener(() -> disposeSpilledLookupSource(partitionNumber), directExecutor());
+    }
+
     private void disposeSpilledLookupSource(int partitionNumber)
     {
         getSpilledLookupSourceHandle(partitionNumber).dispose();
     }
 
+    private ListenableFuture<?> spilledLookupSourceDisposed(int partitionNumber)
+    {
+        return getSpilledLookupSourceHandle(partitionNumber).getDisposeCompleted();
+    }
+
     private SpilledLookupSourceHandle getSpilledLookupSourceHandle(int partitionNumber)
     {
         lock.readLock().lock();
@@ -380,6 +426,59 @@ public final class PartitionedLookupSourceFactory
         }
     }
 
+    @Override
+    public ListenableFuture<?> whenMemProbeFinishes()
+    {
+        lock.writeLock().lock();
+        try {
+            verify(!partitionedConsumption.isDone());
+            if (!spillingInfo.hasSpilled()) {
+                partitionedConsumption.set(new PartitionedConsumption<>(
+                        1,
+                        emptyList(),
+                        i -> {
+                            throw new UnsupportedOperationException();
+                        },
+                        i -> {},
+                        i -> {
+                            throw new UnsupportedOperationException();
+                        }));
+
+                return partitionedConsumption;
+            }
+
+            if (outer) {
+                postConsumptionOuterSource = (PartitionedLookupSource) this.lookupSourceSupplier.getLookupSource();
+                partitionedConsumption.set(new PartitionedConsumption<>(
+                        finishedProbeOperators - partitionedConsumptionParticipants.getAsInt(),
+                        spilledPartitions.keySet(),
+                        this::loadSpilledLookupSource,
+                        this::loadOuterIterator,
+                        this::spilledLookupSourceDisposed));
+            }
+            else {
+                freePartitions();
+                partitionedConsumption.set(new PartitionedConsumption<>(
+                        finishedProbeOperators - partitionedConsumptionParticipants.getAsInt(),
+                        spilledPartitions.keySet(),
+                        this::loadSpilledLookupSource,
+                        this::disposeSpilledLookupSource,
+                        this::spilledLookupSourceDisposed));
+            }
+
+            return partitionedConsumption;
+        }
+        finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+    @Override
+    public boolean isOuterEarlyStartEnabled()
+    {
+        return spillEnabledForOuter;
+    }
+
     @Override
     public OuterPositionIterator getOuterPositionIterator()
     {
@@ -403,6 +502,9 @@ public final class PartitionedLookupSourceFactory
         lock.writeLock().lock();
         try {
             freePartitions();
+            if (postConsumptionOuterSource != null) {
+                postConsumptionOuterSource.close();
+            }
             spilledPartitions.values().forEach(SpilledLookupSourceHandle::dispose);
 
             // Setting destroyed must be last because it's a part of the state exposed by isDestroyed() without synchronization.
@@ -519,9 +621,15 @@ public final class PartitionedLookupSourceFactory
         {
             return spillingInfo.getSpillMask();
         }
+
+        @Override
+        public BiPredicate<Integer, Long> getSpillMatcher()
+        {
+            return spillingInfo.getSpillMatcher();
+        }
     }
 
-    private static class SpilledLookupSource
+    protected static class SpilledLookupSource
             implements LookupSource
     {
         private final int channelCount;
@@ -603,11 +711,13 @@ public final class PartitionedLookupSourceFactory
     {
         private final long spillEpoch;
         private final Set<Integer> spilledPartitions;
+        private final ImmutableMap<Integer, BloomFilter<Long>> spillBlooms;
 
-        SpillingInfo(long spillEpoch, Set<Integer> spilledPartitions)
+        SpillingInfo(long spillEpoch, Set<Integer> spilledPartitions, Map<Integer, BloomFilter<Long>> spillBlooms)
         {
             this.spillEpoch = spillEpoch;
             this.spilledPartitions = ImmutableSet.copyOf(requireNonNull(spilledPartitions, "spilledPartitions is null"));
+            this.spillBlooms = ImmutableMap.copyOf(spillBlooms);
         }
 
         boolean hasSpilled()
@@ -624,6 +734,16 @@ public final class PartitionedLookupSourceFactory
         {
             return spilledPartitions::contains;
         }
+
+        BiPredicate<Integer, Long> getSpillMatcher()
+        {
+            return (partition, rawHash) -> {
+                if (!spillBlooms.containsKey(partition)) {
+                    return true;
+                }
+                return spillBlooms.get(partition).mightContain(rawHash);
+            };
+        }
     }
 
     private static class Marker

presto-main/src/main/java/io/prestosql/operator/SpilledLookupSourceHandle.java

@@ -14,6 +14,7 @@
 package io.prestosql.operator;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.hash.BloomFilter;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 
@@ -21,6 +22,7 @@ import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.ThreadSafe;
 
+import java.util.Optional;
 import java.util.function.Supplier;
 
 import static com.google.common.base.Preconditions.checkState;
@@ -35,7 +37,7 @@ final class SpilledLookupSourceHandle
         SPILLED,
         UNSPILLING,
         PRODUCED,
-        DISPOSED
+        DISPOSE_REQUESTED
     }
 
     @GuardedBy("this")
@@ -48,14 +50,27 @@ final class SpilledLookupSourceHandle
     private SettableFuture<Supplier<LookupSource>> unspilledLookupSource;
 
     private final SettableFuture<?> disposeRequested = SettableFuture.create();
+    private final SettableFuture<?> disposeCompleted = SettableFuture.create();
 
     private final ListenableFuture<?> unspillingOrDisposeRequested = whenAnyComplete(ImmutableList.of(unspillingRequested, disposeRequested));
 
+    private final Optional<BloomFilter<Long>> spillBloom;
+
+    public SpilledLookupSourceHandle(BloomFilter<Long> bloom)
+    {
+        spillBloom = Optional.ofNullable(bloom);
+    }
+
     public SettableFuture<?> getUnspillingRequested()
     {
         return unspillingRequested;
     }
 
+    public Optional<BloomFilter<Long>> getSpillBloom()
+    {
+        return spillBloom;
+    }
+
     public synchronized ListenableFuture<Supplier<LookupSource>> getLookupSource()
     {
         assertState(State.SPILLED);
@@ -70,7 +85,7 @@ final class SpilledLookupSourceHandle
     {
         requireNonNull(lookupSource, "lookupSource is null");
 
-        if (state == State.DISPOSED) {
+        if (state == State.DISPOSE_REQUESTED) {
             return;
         }
 
@@ -85,7 +100,7 @@ final class SpilledLookupSourceHandle
     {
         disposeRequested.set(null);
         unspilledLookupSource = null; // let the memory go
-        setState(State.DISPOSED);
+        setState(State.DISPOSE_REQUESTED);
     }
 
     public SettableFuture<?> getDisposeRequested()
@@ -93,6 +108,17 @@ final class SpilledLookupSourceHandle
         return disposeRequested;
     }
 
+    public synchronized void setDisposeCompleted()
+    {
+        assertState(State.DISPOSE_REQUESTED);
+        disposeCompleted.set(null);
+    }
+
+    public SettableFuture<?> getDisposeCompleted()
+    {
+        return disposeCompleted;
+    }
+
     public ListenableFuture<?> getUnspillingOrDisposeRequested()
     {
         return unspillingOrDisposeRequested;

presto-main/src/main/java/io/prestosql/operator/TrackingLookupSourceSupplier.java

@@ -13,6 +13,8 @@
  */
 package io.prestosql.operator;
 
+import com.google.common.util.concurrent.ListenableFuture;
+
 import java.util.function.Supplier;
 
 import static java.util.Objects.requireNonNull;
@@ -23,6 +25,15 @@ public interface TrackingLookupSourceSupplier
 
     OuterPositionIterator getOuterPositionIterator();
 
+    default ListenableFuture<?> setOuterPartitionReady(int partition)
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    default void updateUnspilledPositions(int partition, long positions)
+    {
+    }
+
     Object captureJoinPositions();
 
     void restoreJoinPositions(Object state);
@@ -47,13 +58,14 @@ public interface TrackingLookupSourceSupplier
             @Override
             public Object captureJoinPositions()
             {
-                throw new UnsupportedOperationException();
+                /* do nothing */
+                return null;
             }
 
             @Override
             public void restoreJoinPositions(Object state)
             {
-                throw new UnsupportedOperationException();
+                /* do nothing*/
             }
         };
     }

presto-main/src/main/java/io/prestosql/spiller/GenericPartitioningSpiller.java

@@ -39,6 +39,8 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.function.BiPredicate;
 import java.util.function.IntPredicate;
 import java.util.function.Predicate;
 
@@ -50,7 +52,7 @@ import static java.util.Objects.requireNonNull;
 
 @ThreadSafe
 @RestorableConfig(uncapturedFields = {"types", "partitionFunction", "closer",
-        "spillerFactory", "spillContext", "memoryContext", "pageBuilders"})
+        "spillerFactory", "spillContext", "memoryContext", "pageBuilders", "getRawHash"})
 public class GenericPartitioningSpiller
         implements PartitioningSpiller
 {
@@ -63,6 +65,7 @@ public class GenericPartitioningSpiller
 
     private final List<PageBuilder> pageBuilders;
     private final List<Optional<SingleStreamSpiller>> spillers;
+    private final BiFunction<Integer, Page, Long> getRawHash;
 
     private boolean readingStarted;
     private final Set<Integer> spilledPartitions = new HashSet<>();
@@ -72,7 +75,8 @@ public class GenericPartitioningSpiller
             PartitionFunction partitionFunction,
             SpillContext spillContext,
             AggregatedMemoryContext memoryContext,
-            SingleStreamSpillerFactory spillerFactory)
+            SingleStreamSpillerFactory spillerFactory,
+            BiFunction<Integer, Page, Long> getRawHash)
     {
         requireNonNull(spillContext, "spillContext is null");
 
@@ -93,6 +97,7 @@ public class GenericPartitioningSpiller
             spillers.add(Optional.empty());
         }
         this.pageBuilders = tmpPageBuilders.build();
+        this.getRawHash = getRawHash;
     }
 
     @Override
@@ -112,19 +117,25 @@ public class GenericPartitioningSpiller
 
     @Override
     public synchronized PartitioningSpillResult partitionAndSpill(Page page, IntPredicate spillPartitionMask)
+    {
+        return partitionAndSpill(page, spillPartitionMask, (ign1, ign2) -> true);
+    }
+
+    @Override
+    public PartitioningSpillResult partitionAndSpill(Page page, IntPredicate spillPartitionMask, BiPredicate<Integer, Long> spillPartitionMatcher)
     {
         requireNonNull(page, "page is null");
         requireNonNull(spillPartitionMask, "spillPartitionMask is null");
         checkArgument(page.getChannelCount() == types.size(), "Wrong page channel count, expected %s but got %s", types.size(), page.getChannelCount());
 
         checkState(!readingStarted, "reading already started");
-        IntArrayList unspilledPositions = partitionPage(page, spillPartitionMask);
+        IntArrayList unspilledPositions = partitionPage(page, spillPartitionMask, spillPartitionMatcher);
         ListenableFuture<?> future = flushFullBuilders();
 
         return new PartitioningSpillResult(future, page.getPositions(unspilledPositions.elements(), 0, unspilledPositions.size()));
     }
 
-    private synchronized IntArrayList partitionPage(Page page, IntPredicate spillPartitionMask)
+    private synchronized IntArrayList partitionPage(Page page, IntPredicate spillPartitionMask, BiPredicate<Integer, Long> spillPartitionMatcher)
     {
         IntArrayList unspilledPositions = new IntArrayList();
 
@@ -136,6 +147,10 @@ public class GenericPartitioningSpiller
                 continue;
             }
 
+            if (getRawHash != null && !spillPartitionMatcher.test(partition, getRawHash.apply(position, page))) {
+                continue;
+            }
+
             spilledPartitions.add(partition);
             PageBuilder pageBuilder = pageBuilders.get(partition);
             pageBuilder.declarePosition();

presto-main/src/main/java/io/prestosql/spiller/GenericPartitioningSpillerFactory.java

@@ -17,9 +17,11 @@ import com.google.inject.Inject;
 import io.prestosql.memory.context.AggregatedMemoryContext;
 import io.prestosql.operator.PartitionFunction;
 import io.prestosql.operator.SpillContext;
+import io.prestosql.spi.Page;
 import io.prestosql.spi.type.Type;
 
 import java.util.List;
+import java.util.function.BiFunction;
 
 import static java.util.Objects.requireNonNull;
 
@@ -41,6 +43,12 @@ public class GenericPartitioningSpillerFactory
             SpillContext spillContext,
             AggregatedMemoryContext memoryContext)
     {
-        return new GenericPartitioningSpiller(types, partitionFunction, spillContext, memoryContext, singleStreamSpillerFactory);
+        return new GenericPartitioningSpiller(types, partitionFunction, spillContext, memoryContext, singleStreamSpillerFactory, null);
+    }
+
+    @Override
+    public PartitioningSpiller create(List<Type> types, PartitionFunction partitionFunction, SpillContext spillContext, AggregatedMemoryContext memoryContext, BiFunction<Integer, Page, Long> getRawHash)
+    {
+        return new GenericPartitioningSpiller(types, partitionFunction, spillContext, memoryContext, singleStreamSpillerFactory, getRawHash);
     }
 }

presto-main/src/main/java/io/prestosql/spiller/PartitioningSpiller.java

@@ -20,6 +20,7 @@ import io.prestosql.spi.snapshot.Restorable;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Iterator;
+import java.util.function.BiPredicate;
 import java.util.function.IntPredicate;
 
 import static java.util.Objects.requireNonNull;
@@ -35,6 +36,8 @@ public interface PartitioningSpiller
      */
     PartitioningSpillResult partitionAndSpill(Page page, IntPredicate spillPartitionMask);
 
+    PartitioningSpillResult partitionAndSpill(Page page, IntPredicate spillPartitionMask, BiPredicate<Integer, Long> spillPartitionMatcher);
+
     /**
      * Returns iterator of previously spilled pages from given partition. Callers are expected to call
      * this method once. Calling multiple times can results in undefined behavior.

presto-main/src/main/java/io/prestosql/spiller/PartitioningSpillerFactory.java

@@ -16,9 +16,11 @@ package io.prestosql.spiller;
 import io.prestosql.memory.context.AggregatedMemoryContext;
 import io.prestosql.operator.PartitionFunction;
 import io.prestosql.operator.SpillContext;
+import io.prestosql.spi.Page;
 import io.prestosql.spi.type.Type;
 
 import java.util.List;
+import java.util.function.BiFunction;
 
 public interface PartitioningSpillerFactory
 {
@@ -28,6 +30,16 @@ public interface PartitioningSpillerFactory
             SpillContext spillContext,
             AggregatedMemoryContext memoryContext);
 
+    default PartitioningSpiller create(
+            List<Type> types,
+            PartitionFunction partitionFunction,
+            SpillContext spillContext,
+            AggregatedMemoryContext memoryContext,
+            BiFunction<Integer, Page, Long> getRawHash)
+    {
+        return create(types, partitionFunction, spillContext, memoryContext);
+    }
+
     static PartitioningSpillerFactory unsupportedPartitioningSpillerFactory()
     {
         return (types, partitionFunction, spillContext, memoryContext) -> {

presto-main/src/main/java/io/prestosql/sql/analyzer/FeaturesConfig.java

@@ -118,6 +118,8 @@ public class FeaturesConfig
     private boolean spillOrderBy = true;
     private boolean nonBlockingSpill;
     private boolean spillWindowOperator = true;
+    private boolean spillBuildForOuterJoinEnabled;
+    private boolean innerJoinSpillFilterEnabled;
     private DataSize aggregationOperatorUnspillMemoryLimit = new DataSize(4, DataSize.Unit.MEGABYTE);
     private List<Path> spillerSpillPaths = ImmutableList.of();
     private int spillerThreads = 4;
@@ -727,6 +729,30 @@ public class FeaturesConfig
         return nonBlockingSpill;
     }
 
+    public boolean isSpillBuildForOuterJoinEnabled()
+    {
+        return spillBuildForOuterJoinEnabled;
+    }
+
+    @Config("experimental.spill-build-for-outer-join-enabled")
+    public FeaturesConfig setSpillBuildForOuterJoinEnabled(boolean spillBuildForOuterJoinEnabled)
+    {
+        this.spillBuildForOuterJoinEnabled = spillBuildForOuterJoinEnabled;
+        return this;
+    }
+
+    public boolean isInnerJoinSpillFilterEnabled()
+    {
+        return innerJoinSpillFilterEnabled;
+    }
+
+    @Config("experimental.inner-join-spill-filter-enabled")
+    public FeaturesConfig setInnerJoinSpillFilterEnabled(boolean innerJoinSpillFilterEnabled)
+    {
+        this.innerJoinSpillFilterEnabled = innerJoinSpillFilterEnabled;
+        return this;
+    }
+
     @Config("experimental.spill-order-by")
     public FeaturesConfig setSpillOrderBy(boolean spillOrderBy)
     {

presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java

@@ -273,6 +273,7 @@ import static io.prestosql.SystemSessionProperties.isCrossRegionDynamicFilterEna
 import static io.prestosql.SystemSessionProperties.isEnableDynamicFiltering;
 import static io.prestosql.SystemSessionProperties.isNonBlockingSpillOrderby;
 import static io.prestosql.SystemSessionProperties.isSpillEnabled;
+import static io.prestosql.SystemSessionProperties.isSpillForOuterJoinEnabled;
 import static io.prestosql.SystemSessionProperties.isSpillOrderBy;
 import static io.prestosql.SystemSessionProperties.isSpillReuseExchange;
 import static io.prestosql.SystemSessionProperties.isSpillWindowOperator;
@@ -2440,7 +2441,9 @@ public class LocalExecutionPlanner
             PhysicalOperation probeSource = probeNode.accept(this, context);
 
             // Plan build
-            boolean spillEnabled = isSpillEnabled(session) && node.isSpillable().orElseThrow(() -> new IllegalArgumentException("spillable not yet set"));
+            boolean spillEnabled = isSpillEnabled(session)
+                    && node.isSpillable().orElseThrow(() -> new IllegalArgumentException("spillable not yet set"))
+                    && probeSource.getPipelineExecutionStrategy() == UNGROUPED_EXECUTION;
             JoinBridgeManager<PartitionedLookupSourceFactory> lookupSourceFactory =
                     createLookupSourceFactory(node, buildNode, buildSymbols, buildHashSymbol, probeSource, context, spillEnabled);
 
@@ -2517,6 +2520,12 @@ public class LocalExecutionPlanner
 
             boolean buildOuter = node.getType() == RIGHT || node.getType() == FULL;
             int taskCount = buildContext.getDriverInstanceCount().orElse(1);
+            /* Spill can take outer */
+            boolean canOuterSpill = isSpillForOuterJoinEnabled(session);
+            boolean spillAllowed = spillEnabled;
+            if (buildOuter && spillEnabled) {
+                spillAllowed = canOuterSpill;
+            }
 
             Optional<JoinFilterFunctionFactory> filterFunctionFactory = node.getFilter()
                     .map(filterExpression -> compileJoinFilterFunction(
@@ -2559,7 +2568,8 @@ public class LocalExecutionPlanner
                                     .collect(toImmutableList()),
                             taskCount,
                             buildSource.getLayout(),
-                            buildOuter),
+                            buildOuter,
+                            canOuterSpill),
                     buildOutputTypes);
 
             ImmutableList.Builder<OperatorFactory> factoriesBuilder = new ImmutableList.Builder();
@@ -2600,7 +2610,7 @@ public class LocalExecutionPlanner
                     searchFunctionFactories,
                     10_000,
                     pagesIndexFactory,
-                    spillEnabled && !buildOuter && taskCount > 1,
+                    spillAllowed && taskCount > 1,
                     singleStreamSpillerFactory);
 
             factoriesBuilder.add(hashBuilderOperatorFactory);

presto-main/src/test/java/io/prestosql/operator/BenchmarkHashBuildAndJoinOperators.java

@@ -294,7 +294,7 @@ public class BenchmarkHashBuildAndJoinOperators
                         .collect(toImmutableList()),
                 1,
                 requireNonNull(ImmutableMap.of(), "layout is null"),
-                false));
+                false, false));
         HashBuilderOperatorFactory hashBuilderOperatorFactory = new HashBuilderOperatorFactory(
                 HASH_BUILD_OPERATOR_ID,
                 TEST_PLAN_NODE_ID,

presto-main/src/test/java/io/prestosql/operator/OperatorAssertion.java

@@ -479,6 +479,7 @@ public final class OperatorAssertion
     public static List<Page> toPagesCompareStateSimple(OperatorFactory operatorFactory, DriverContext driverContext, List<Page> input, boolean revokeMemoryWhenAddingPages, Map<String, Object> expectedMapping)
     {
         try (Operator operator = operatorFactory.createOperator(driverContext)) {
+            operatorFactory.noMoreOperators(driverContext.getLifespan());
             operatorFactory.noMoreOperators();
             return toPagesCompareStateSimple(operator, input, revokeMemoryWhenAddingPages, expectedMapping);
         }

presto-main/src/test/java/io/prestosql/operator/TestHashJoinOperator.java

@@ -51,11 +51,14 @@ import io.prestosql.sql.gen.JoinFilterFunctionCompiler.JoinFilterFunctionFactory
 import io.prestosql.testing.MaterializedResult;
 import io.prestosql.testing.MaterializedRow;
 import io.prestosql.testing.TestingTaskContext;
+import org.roaringbitmap.RoaringBitmap;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
+import java.nio.ByteBuffer;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -564,6 +567,7 @@ public class TestHashJoinOperator
         checkState(buildOperatorCount == whenSpill.size());
         LookupSourceFactory lookupSourceFactory = lookupSourceFactoryManager.getJoinBridge(Lifespan.taskWide());
 
+        boolean closed = false;
         try (Operator joinOperator = joinOperatorFactory.createOperator(joinDriverContext)) {
             // build lookup source
             ListenableFuture<LookupSourceProvider> lookupSourceProvider = lookupSourceFactory.createLookupSourceProvider();
@@ -593,6 +597,10 @@ public class TestHashJoinOperator
                 runDriverInThread(executor, buildDriver);
             }
 
+            joinOperatorFactory.noMoreOperators(joinDriverContext.getLifespan());
+            joinOperatorFactory.noMoreOperators();
+            closed = true;
+
             ValuesOperatorFactory valuesOperatorFactory = new ValuesOperatorFactory(17, new PlanNodeId("values"), probePages.build());
 
             PageBuffer pageBuffer = new PageBuffer(10);
@@ -635,7 +643,10 @@ public class TestHashJoinOperator
             assertEqualsIgnoreOrder(getProperColumns(joinOperator, concat(probePages.getTypes(), buildPages.getTypes()), probePages, actualPages).getMaterializedRows(), expected.getMaterializedRows());
         }
         finally {
-            joinOperatorFactory.noMoreOperators();
+            if (!closed) {
+                joinOperatorFactory.noMoreOperators(joinDriverContext.getLifespan());
+                joinOperatorFactory.noMoreOperators();
+            }
         }
     }
 
@@ -1512,7 +1523,7 @@ public class TestHashJoinOperator
                         .collect(toImmutableList()),
                 partitionCount,
                 requireNonNull(ImmutableMap.of(), "layout is null"),
-                outer);
+                outer, false);
         JoinBridgeManager<PartitionedLookupSourceFactory> lookupSourceFactoryManager = new JoinBridgeManager<>(
                 outer,
                 UNGROUPED_EXECUTION,
@@ -1925,9 +1936,52 @@ public class TestHashJoinOperator
             matched = Booleans.asList(positions).stream().filter(e -> e).count();
         }
         else {
-            boolean[][] positions = (boolean[][]) state;
-            matched = Arrays.stream(positions).flatMap(array -> Booleans.asList(array).stream()).filter(e -> e).count();
+            ByteBuffer bb = ByteBuffer.wrap((byte[]) state);
+            List<RoaringBitmap> visitedPositions = new ArrayList<>();
+            for (int i = 0; i < (parallelBuild ? PARTITION_COUNT : 1); i++) {
+                ImmutableRoaringBitmap bm = new ImmutableRoaringBitmap(bb);
+                visitedPositions.add(new RoaringBitmap(bm));
+                bb.position(bb.position() + visitedPositions.get(i).serializedSizeInBytes());
+            }
+
+            matched = visitedPositions.stream().mapToLong(rr -> rr.getCardinality()).sum();
         }
         assertEquals(matched, 5);
     }
+
+    @Test(timeOut = 30_000)
+    public void testBuildGracefulSpill()
+            throws Exception
+    {
+        TaskStateMachine taskStateMachine = new TaskStateMachine(new TaskId("query", 0, 0), executor);
+        TaskContext taskContext = TestingTaskContext.createTaskContext(executor, scheduledExecutor, TEST_SESSION, taskStateMachine);
+
+        // build factory
+        RowPagesBuilder buildPages = rowPagesBuilder(ImmutableList.of(VARCHAR, BIGINT))
+                .addSequencePage(4, 20, 200);
+
+        DummySpillerFactory buildSpillerFactory = new DummySpillerFactory();
+
+        BuildSideSetup buildSideSetup = setupBuildSide(true, taskContext, Ints.asList(0), buildPages, Optional.empty(), true, buildSpillerFactory);
+        instantiateBuildDrivers(buildSideSetup, taskContext);
+
+        JoinBridgeManager<PartitionedLookupSourceFactory> lookupSourceFactoryManager = buildSideSetup.getLookupSourceFactoryManager();
+        PartitionedLookupSourceFactory lookupSourceFactory = lookupSourceFactoryManager.getJoinBridge(Lifespan.taskWide());
+
+        // finish probe before any build partition is spilled
+        lookupSourceFactory.finishProbeOperator(OptionalInt.of(1));
+        lookupSourceFactory.whenMemProbeFinishes();
+
+        // spill build partition after probe is finished
+        HashBuilderOperator hashBuilderOperator = buildSideSetup.getBuildOperators().get(0);
+        hashBuilderOperator.startMemoryRevoke().get();
+        hashBuilderOperator.finishMemoryRevoke();
+        hashBuilderOperator.finish();
+
+        // hash builder operator should not deadlock waiting for spilled lookup source to be disposed
+        hashBuilderOperator.isBlocked().get();
+
+        lookupSourceFactory.destroy();
+        assertTrue(hashBuilderOperator.isFinished());
+    }
 }

presto-main/src/test/java/io/prestosql/sql/analyzer/TestFeaturesConfig.java

@@ -81,6 +81,8 @@ public class TestFeaturesConfig
                 .setRe2JDfaRetries(5)
                 .setSpillEnabled(false)
                 .setNonBlockingSpill(false)
+                .setSpillBuildForOuterJoinEnabled(false)
+                .setInnerJoinSpillFilterEnabled(false)
                 .setSpillOrderBy(true)
                 .setSpillWindowOperator(true)
                 .setAggregationOperatorUnspillMemoryLimit(DataSize.valueOf("4MB"))
@@ -197,6 +199,8 @@ public class TestFeaturesConfig
                 .put("re2j.dfa-retries", "42")
                 .put("experimental.spill-enabled", "true")
                 .put("experimental.spill-non-blocking-orderby", "true")
+                .put("experimental.spill-build-for-outer-join-enabled", "true")
+                .put("experimental.inner-join-spill-filter-enabled", "true")
                 .put("experimental.spill-order-by", "false")
                 .put("experimental.spill-window-operator", "false")
                 .put("experimental.aggregation-operator-unspill-memory-limit", "100MB")
@@ -299,6 +303,8 @@ public class TestFeaturesConfig
                 .setRe2JDfaRetries(42)
                 .setSpillEnabled(true)
                 .setNonBlockingSpill(true)
+                .setSpillBuildForOuterJoinEnabled(true)
+                .setInnerJoinSpillFilterEnabled(true)
                 .setSpillOrderBy(false)
                 .setSpillWindowOperator(false)
                 .setAggregationOperatorUnspillMemoryLimit(DataSize.valueOf("100MB"))