diff --git a/hetu-docs/en/admin/properties.md b/hetu-docs/en/admin/properties.md index 3b021eb85..e94c2dfe8 100644 --- a/hetu-docs/en/admin/properties.md +++ b/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` diff --git a/presto-benchmark/src/main/java/io/prestosql/benchmark/HashBuildAndJoinBenchmark.java b/presto-benchmark/src/main/java/io/prestosql/benchmark/HashBuildAndJoinBenchmark.java index ce9f42589..aa7f87c9d 100644 --- a/presto-benchmark/src/main/java/io/prestosql/benchmark/HashBuildAndJoinBenchmark.java +++ b/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"), diff --git a/presto-benchmark/src/main/java/io/prestosql/benchmark/HashBuildBenchmark.java b/presto-benchmark/src/main/java/io/prestosql/benchmark/HashBuildBenchmark.java index 517090240..96a75fa73 100644 --- a/presto-benchmark/src/main/java/io/prestosql/benchmark/HashBuildBenchmark.java +++ b/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"), diff --git a/presto-benchmark/src/main/java/io/prestosql/benchmark/HashJoinBenchmark.java b/presto-benchmark/src/main/java/io/prestosql/benchmark/HashJoinBenchmark.java index b6ececb80..667d02c32 100644 --- a/presto-benchmark/src/main/java/io/prestosql/benchmark/HashJoinBenchmark.java +++ b/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"), diff --git a/presto-main/pom.xml b/presto-main/pom.xml index 557024e1c..e1d300a75 100644 --- a/presto-main/pom.xml +++ b/presto-main/pom.xml @@ -579,6 +579,12 @@ org.objenesis objenesis + + + org.roaringbitmap + RoaringBitmap + 0.9.0 + diff --git a/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java b/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java index ec44250c8..c812f921b 100644 --- a/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java +++ b/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); diff --git a/presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java b/presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java index 2f352a8a9..90d1cf75d 100644 --- a/presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java +++ b/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); diff --git a/presto-main/src/main/java/io/prestosql/operator/HashBuilderOperator.java b/presto-main/src/main/java/io/prestosql/operator/HashBuilderOperator.java index 4e4bac855..d0ac3bac5 100644 --- a/presto-main/src/main/java/io/prestosql/operator/HashBuilderOperator.java +++ b/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> lookupSourceNotNeeded = Optional.empty(); - private final SpilledLookupSourceHandle spilledLookupSourceHandle = new SpilledLookupSourceHandle(); + private final SpilledLookupSourceHandle spilledLookupSourceHandle; private Optional spiller = Optional.empty(); private ListenableFuture spillInProgress = NOT_BLOCKED; private Optional>> unspillInProgress = Optional.empty(); @@ -248,6 +258,7 @@ public class HashBuilderOperator private OptionalLong lookupSourceChecksum = OptionalLong.empty(); private Optional finishMemoryRevoke = Optional.empty(); + private BloomFilter 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() + { + private Iterator 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; } } diff --git a/presto-main/src/main/java/io/prestosql/operator/JoinBridge.java b/presto-main/src/main/java/io/prestosql/operator/JoinBridge.java index c10168c28..0588f1e36 100644 --- a/presto-main/src/main/java/io/prestosql/operator/JoinBridge.java +++ b/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; + } } diff --git a/presto-main/src/main/java/io/prestosql/operator/JoinBridgeManager.java b/presto-main/src/main/java/io/prestosql/operator/JoinBridgeManager.java index abbe03cea..755e0cda9 100644 --- a/presto-main/src/main/java/io/prestosql/operator/JoinBridgeManager.java +++ b/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 return internalJoinBridgeDataManager.getOuterPositionsFuture(lifespan); } + public void probeOperatorFinished(Lifespan lifespan) + { + initializeIfNecessary(); + internalJoinBridgeDataManager.probeOperatorFinish(lifespan); + } + private static InternalJoinBridgeDataManager internalJoinBridgeDataManager( PipelineExecutionStrategy probeExecutionStrategy, PipelineExecutionStrategy buildExecutionStrategy, @@ -213,6 +221,8 @@ public class JoinBridgeManager void outerOperatorCreated(Lifespan lifespan); void outerOperatorClosed(Lifespan lifespan); + + void probeOperatorFinish(Lifespan lifespan); } // 1 probe, 1 lookup source @@ -252,6 +262,7 @@ public class JoinBridgeManager public void probeOperatorFactoryClosed(Lifespan lifespan) { checkArgument(Lifespan.taskWide().equals(lifespan)); + joinLifecycle.releaseForProbeInMem(); joinLifecycle.releaseForProbe(); } @@ -289,6 +300,13 @@ public class JoinBridgeManager 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 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 data(lifespan).joinLifecycle.releaseForOuter(); } + @Override + public void probeOperatorFinish(Lifespan lifespan) + { + checkArgument(!Lifespan.taskWide().equals(lifespan)); + data(lifespan).joinLifecycle.releaseForProbeInMem(); + } + private JoinBridgeAndLifecycle data(Lifespan lifespan) { checkArgument(!Lifespan.taskWide().equals(lifespan)); @@ -421,6 +447,7 @@ public class JoinBridgeManager @Override public void probeOperatorFactoryClosedForAllLifespans() { + joinLifecycle.releaseForProbeInMem(); joinLifecycle.releaseForProbe(); } @@ -464,11 +491,18 @@ public class JoinBridgeManager 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 // * 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 private void retainForProbe() { probeReferenceCount.retain(); + probeInMemReferenceCount.retain(); } private void releaseForProbe() @@ -516,6 +554,11 @@ public class JoinBridgeManager { outerReferenceCount.release(); } + + public void releaseForProbeInMem() + { + probeInMemReferenceCount.release(); + } } private static class FreezeOnReadCounter diff --git a/presto-main/src/main/java/io/prestosql/operator/LookupJoinOperator.java b/presto-main/src/main/java/io/prestosql/operator/LookupJoinOperator.java index 3eff7d65d..923d476e7 100644 --- a/presto-main/src/main/java/io/prestosql/operator/LookupJoinOperator.java +++ b/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 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 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 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 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 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 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; diff --git a/presto-main/src/main/java/io/prestosql/operator/LookupJoinOperatorFactory.java b/presto-main/src/main/java/io/prestosql/operator/LookupJoinOperatorFactory.java index 7a8a94667..fbc58bd87 100644 --- a/presto-main/src/main/java/io/prestosql/operator/LookupJoinOperatorFactory.java +++ b/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 diff --git a/presto-main/src/main/java/io/prestosql/operator/LookupOuterOperator.java b/presto-main/src/main/java/io/prestosql/operator/LookupOuterOperator.java index 7af08e794..56156ce7c 100644 --- a/presto-main/src/main/java/io/prestosql/operator/LookupOuterOperator.java +++ b/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 outerPositionsFuture; + private ListenableFuture outerPositionsFuture; private final List 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; } diff --git a/presto-main/src/main/java/io/prestosql/operator/LookupSourceFactory.java b/presto-main/src/main/java/io/prestosql/operator/LookupSourceFactory.java index 7b9255fb1..5a1bbef77 100644 --- a/presto-main/src/main/java/io/prestosql/operator/LookupSourceFactory.java +++ b/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> startOuterOperator(OptionalInt lookupJoinsCount) + { + return immediateFuture(new PartitionedConsumption<>( + 1, + ImmutableList.of(1), + i -> immediateFuture(null), + i -> {}, + i -> immediateFuture(null))); } /** diff --git a/presto-main/src/main/java/io/prestosql/operator/LookupSourceProvider.java b/presto-main/src/main/java/io/prestosql/operator/LookupSourceProvider.java index 7843487db..7c61c6d0c 100644 --- a/presto-main/src/main/java/io/prestosql/operator/LookupSourceProvider.java +++ b/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 getSpillMatcher() + { + return (a, b) -> true; + } } } diff --git a/presto-main/src/main/java/io/prestosql/operator/OuterPositionIterator.java b/presto-main/src/main/java/io/prestosql/operator/OuterPositionIterator.java index 63bde20ec..bf7f01544 100644 --- a/presto-main/src/main/java/io/prestosql/operator/OuterPositionIterator.java +++ b/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 getNextBatch() + { + return immediateFuture(null); + } } diff --git a/presto-main/src/main/java/io/prestosql/operator/PartitionedConsumption.java b/presto-main/src/main/java/io/prestosql/operator/PartitionedConsumption.java index 89006d507..b2caffaaf 100644 --- a/presto-main/src/main/java/io/prestosql/operator/PartitionedConsumption.java +++ b/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 @Nullable private List> partitions; - public PartitionedConsumption(int consumersCount, Iterable partitionNumbers, IntFunction> loader, IntConsumer disposer) + public PartitionedConsumption(int consumersCount, Iterable partitionNumbers, IntFunction> loader, IntConsumer disposer, + IntFunction> 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 ListenableFuture activator, Iterable partitionNumbers, IntFunction> loader, - IntConsumer disposer) + IntConsumer disposer, + IntFunction> 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> createPartitions( ListenableFuture activator, Iterable partitionNumbers, IntFunction> loader, - IntConsumer disposer) + IntConsumer disposer, + IntFunction> disposed) { requireNonNull(partitionNumbers, "partitionNumbers is null"); requireNonNull(loader, "loader is null"); @@ -78,7 +82,7 @@ public final class PartitionedConsumption for (Integer partitionNumber : partitionNumbers) { Partition 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 return consumersCount; } - public Iterator> beginConsumption() + public synchronized Iterator> beginConsumption() { - Queue> 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> partitionQueue = new ArrayDeque<>(requireNonNull(this.partitions, "partitionQueue is already null")); return new AbstractIterator>() { @Override @@ -116,7 +122,7 @@ public final class PartitionedConsumption private final int partitionNumber; private final SettableFuture requested; private final ListenableFuture loaded; - private final SettableFuture released; + private final IntConsumer disposer; @GuardedBy("this") private int pendingReleases; @@ -134,8 +140,7 @@ public final class PartitionedConsumption 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 pendingReleases--; checkState(pendingReleases >= 0); if (pendingReleases == 0) { - released.set(null); + disposer.accept(partitionNumber); } } } diff --git a/presto-main/src/main/java/io/prestosql/operator/PartitionedLookupSource.java b/presto-main/src/main/java/io/prestosql/operator/PartitionedLookupSource.java index 45169c649..1c48e789d 100644 --- a/presto-main/src/main/java/io/prestosql/operator/PartitionedLookupSource.java +++ b/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 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 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 lookupSources; + private final List visitedPositions; + private final ReentrantReadWriteLock[] locks; + private final AtomicBoolean[] finished; + private final AtomicLong[] referenceCount; + private final List> partitionReady; + private final List> partitionDone; + + public OuterPositionTrackerFactory(List> 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> 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 getNextReady() + { + ImmutableList.Builder> 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 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) { diff --git a/presto-main/src/main/java/io/prestosql/operator/PartitionedLookupSourceFactory.java b/presto-main/src/main/java/io/prestosql/operator/PartitionedLookupSourceFactory.java index d18a5759f..a0841bede 100644 --- a/presto-main/src/main/java/io/prestosql/operator/PartitionedLookupSourceFactory.java +++ b/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 layout; private final List 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 spilledPartitions = new HashMap<>(); @@ -98,6 +103,13 @@ public final class PartitionedLookupSourceFactory @GuardedBy("lock") private final SettableFuture>> 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. *

@@ -122,7 +134,7 @@ public final class PartitionedLookupSourceFactory */ private Object restoredJoinPositions; - public PartitionedLookupSourceFactory(List types, List outputTypes, List hashChannelTypes, int partitionCount, Map layout, boolean outer) + public PartitionedLookupSourceFactory(List types, List outputTypes, List hashChannelTypes, int partitionCount, Map 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[]) 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> 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> 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> loadSpilledLookupSource(int partitionNumber) { + if (outer) { + verify(partitionsSet != 1); + ListenableFuture> lookupSupplierFuture = getSpilledLookupSourceHandle(partitionNumber).getLookupSource(); + return Futures.transformAsync(lookupSupplierFuture, + lookupSource -> getUpdatedPartitionedLookupSourceSupplier(lookupSource, partitionNumber), + directExecutor()); + } return getSpilledLookupSourceHandle(partitionNumber).getLookupSource(); } + private ListenableFuture> getUpdatedPartitionedLookupSourceSupplier(Supplier 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 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 spilledPartitions; + private final ImmutableMap> spillBlooms; - SpillingInfo(long spillEpoch, Set spilledPartitions) + SpillingInfo(long spillEpoch, Set spilledPartitions, Map> 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 getSpillMatcher() + { + return (partition, rawHash) -> { + if (!spillBlooms.containsKey(partition)) { + return true; + } + return spillBlooms.get(partition).mightContain(rawHash); + }; + } } private static class Marker diff --git a/presto-main/src/main/java/io/prestosql/operator/SpilledLookupSourceHandle.java b/presto-main/src/main/java/io/prestosql/operator/SpilledLookupSourceHandle.java index ddcc278ae..98f72e6ad 100644 --- a/presto-main/src/main/java/io/prestosql/operator/SpilledLookupSourceHandle.java +++ b/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> 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> spillBloom; + + public SpilledLookupSourceHandle(BloomFilter bloom) + { + spillBloom = Optional.ofNullable(bloom); + } + public SettableFuture getUnspillingRequested() { return unspillingRequested; } + public Optional> getSpillBloom() + { + return spillBloom; + } + public synchronized ListenableFuture> 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; diff --git a/presto-main/src/main/java/io/prestosql/operator/TrackingLookupSourceSupplier.java b/presto-main/src/main/java/io/prestosql/operator/TrackingLookupSourceSupplier.java index 9abfb0650..07784a9ce 100644 --- a/presto-main/src/main/java/io/prestosql/operator/TrackingLookupSourceSupplier.java +++ b/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*/ } }; } diff --git a/presto-main/src/main/java/io/prestosql/spiller/GenericPartitioningSpiller.java b/presto-main/src/main/java/io/prestosql/spiller/GenericPartitioningSpiller.java index 0fecbd09b..9f0976dfb 100644 --- a/presto-main/src/main/java/io/prestosql/spiller/GenericPartitioningSpiller.java +++ b/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 pageBuilders; private final List> spillers; + private final BiFunction getRawHash; private boolean readingStarted; private final Set spilledPartitions = new HashSet<>(); @@ -72,7 +75,8 @@ public class GenericPartitioningSpiller PartitionFunction partitionFunction, SpillContext spillContext, AggregatedMemoryContext memoryContext, - SingleStreamSpillerFactory spillerFactory) + SingleStreamSpillerFactory spillerFactory, + BiFunction 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 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 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(); diff --git a/presto-main/src/main/java/io/prestosql/spiller/GenericPartitioningSpillerFactory.java b/presto-main/src/main/java/io/prestosql/spiller/GenericPartitioningSpillerFactory.java index 941f2508f..93138d7f8 100644 --- a/presto-main/src/main/java/io/prestosql/spiller/GenericPartitioningSpillerFactory.java +++ b/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 types, PartitionFunction partitionFunction, SpillContext spillContext, AggregatedMemoryContext memoryContext, BiFunction getRawHash) + { + return new GenericPartitioningSpiller(types, partitionFunction, spillContext, memoryContext, singleStreamSpillerFactory, getRawHash); } } diff --git a/presto-main/src/main/java/io/prestosql/spiller/PartitioningSpiller.java b/presto-main/src/main/java/io/prestosql/spiller/PartitioningSpiller.java index 5b5e1415d..3ab781e13 100644 --- a/presto-main/src/main/java/io/prestosql/spiller/PartitioningSpiller.java +++ b/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 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. diff --git a/presto-main/src/main/java/io/prestosql/spiller/PartitioningSpillerFactory.java b/presto-main/src/main/java/io/prestosql/spiller/PartitioningSpillerFactory.java index 406798da6..eec38fd45 100644 --- a/presto-main/src/main/java/io/prestosql/spiller/PartitioningSpillerFactory.java +++ b/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 types, + PartitionFunction partitionFunction, + SpillContext spillContext, + AggregatedMemoryContext memoryContext, + BiFunction getRawHash) + { + return create(types, partitionFunction, spillContext, memoryContext); + } + static PartitioningSpillerFactory unsupportedPartitioningSpillerFactory() { return (types, partitionFunction, spillContext, memoryContext) -> { diff --git a/presto-main/src/main/java/io/prestosql/sql/analyzer/FeaturesConfig.java b/presto-main/src/main/java/io/prestosql/sql/analyzer/FeaturesConfig.java index 71f94a606..774375411 100644 --- a/presto-main/src/main/java/io/prestosql/sql/analyzer/FeaturesConfig.java +++ b/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 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) { diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java b/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java index 7b567fa48..2fae9cdfc 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java +++ b/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 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 filterFunctionFactory = node.getFilter() .map(filterExpression -> compileJoinFilterFunction( @@ -2559,7 +2568,8 @@ public class LocalExecutionPlanner .collect(toImmutableList()), taskCount, buildSource.getLayout(), - buildOuter), + buildOuter, + canOuterSpill), buildOutputTypes); ImmutableList.Builder 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); diff --git a/presto-main/src/test/java/io/prestosql/operator/BenchmarkHashBuildAndJoinOperators.java b/presto-main/src/test/java/io/prestosql/operator/BenchmarkHashBuildAndJoinOperators.java index e1f9eb90a..46c1d6af7 100644 --- a/presto-main/src/test/java/io/prestosql/operator/BenchmarkHashBuildAndJoinOperators.java +++ b/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, diff --git a/presto-main/src/test/java/io/prestosql/operator/OperatorAssertion.java b/presto-main/src/test/java/io/prestosql/operator/OperatorAssertion.java index f27ecdc96..06bc7b43c 100644 --- a/presto-main/src/test/java/io/prestosql/operator/OperatorAssertion.java +++ b/presto-main/src/test/java/io/prestosql/operator/OperatorAssertion.java @@ -479,6 +479,7 @@ public final class OperatorAssertion public static List toPagesCompareStateSimple(OperatorFactory operatorFactory, DriverContext driverContext, List input, boolean revokeMemoryWhenAddingPages, Map expectedMapping) { try (Operator operator = operatorFactory.createOperator(driverContext)) { + operatorFactory.noMoreOperators(driverContext.getLifespan()); operatorFactory.noMoreOperators(); return toPagesCompareStateSimple(operator, input, revokeMemoryWhenAddingPages, expectedMapping); } diff --git a/presto-main/src/test/java/io/prestosql/operator/TestHashJoinOperator.java b/presto-main/src/test/java/io/prestosql/operator/TestHashJoinOperator.java index e11c9698f..8401f49cb 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestHashJoinOperator.java +++ b/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 = 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 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 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 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()); + } } diff --git a/presto-main/src/test/java/io/prestosql/sql/analyzer/TestFeaturesConfig.java b/presto-main/src/test/java/io/prestosql/sql/analyzer/TestFeaturesConfig.java index 98424827d..fbf683a90 100644 --- a/presto-main/src/test/java/io/prestosql/sql/analyzer/TestFeaturesConfig.java +++ b/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"))