Skip to content

Improve memory for HashBuilderOperator unspill #9269

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
JunhyungSong wants to merge 1 commit into from
Closed

Improve memory for HashBuilderOperator unspill #9269

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 20 additions & 5 deletions core/trino-main/src/main/java/io/trino/operator/join/HashBuilderOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import com.google.common.io.Closer;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.log.Logger;
import io.trino.execution.Lifespan;
import io.trino.memory.context.LocalMemoryContext;
import io.trino.operator.DriverContext;
Expand Down Expand Up @@ -52,13 +53,16 @@
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static io.airlift.concurrent.MoreFutures.checkSuccess;
import static io.airlift.concurrent.MoreFutures.getDone;
import static io.airlift.units.DataSize.succinctBytes;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;

@ThreadSafe
public class HashBuilderOperator
implements Operator
{
private static final Logger log = Logger.get(HashBuilderOperator.class);

public static class HashBuilderOperatorFactory
implements OperatorFactory
{
Expand Down Expand Up @@ -556,23 +560,33 @@ private void finishLookupSourceUnspilling()
{
checkState(state == State.INPUT_UNSPILLING);
if (!unspillInProgress.get().isDone()) {
// Pages have not be unspilled yet.
// Pages have not been unspilled yet.
return;
}

// Use Queue so that Pages already consumed by Index are not retained by us.
Queue<Page> pages = new ArrayDeque<>(getDone(unspillInProgress.get()));
long memoryRetainedByRemainingPages = pages.stream()
unspillInProgress = Optional.empty();
long sizeOfUnSpilledPages = pages.stream()
.mapToLong(Page::getSizeInBytes)
.sum();
long retainedSizeOfUnSpilledPages = pages.stream()
.mapToLong(Page::getRetainedSizeInBytes)
.sum();
localUserMemoryContext.setBytes(memoryRetainedByRemainingPages + index.getEstimatedSize().toBytes());
log.debug(
"Unspilling for operator %s, unspilled partition %d, sizeOfUnSpilledPages %s, retainedSizeOfUnSpilledPages %s",
operatorContext,
partitionIndex,
succinctBytes(sizeOfUnSpilledPages),
succinctBytes(retainedSizeOfUnSpilledPages));
localUserMemoryContext.setBytes(retainedSizeOfUnSpilledPages + index.getEstimatedSize().toBytes());

while (!pages.isEmpty()) {
Page next = pages.remove();
index.addPage(next);
// There is no attempt to compact index, since unspilled pages are unlikely to have blocks with retained size > logical size.
memoryRetainedByRemainingPages -= next.getRetainedSizeInBytes();
localUserMemoryContext.setBytes(memoryRetainedByRemainingPages + index.getEstimatedSize().toBytes());
retainedSizeOfUnSpilledPages -= next.getRetainedSizeInBytes();
localUserMemoryContext.setBytes(retainedSizeOfUnSpilledPages + index.getEstimatedSize().toBytes());
}

LookupSourceSupplier partition = buildLookupSource();
Expand Down Expand Up @@ -633,6 +647,7 @@ public void close()
// close() can be called in any state, due for example to query failure, and must clean resource up unconditionally

lookupSourceSupplier = null;
unspillInProgress = Optional.empty();
state = State.CLOSED;
finishMemoryRevoke = finishMemoryRevoke.map(ifPresent -> () -> {});

Expand Down