Try index compaction before spilling in HashBuilderOperator
@@ -205,6 +205,8 @@ public class HashBuilderOperator
DISPOSED
}
+ private static final double INDEX_COMPACTION_ON_REVOCATION_TARGET = 0.8;
+
private final OperatorContext operatorContext;
private final PartitionedLookupSourceFactory lookupSourceFactory;
private final ListenableFuture<?> lookupSourceFactoryDestroyed;
@@ -353,8 +355,6 @@ public class HashBuilderOperator
index.addPage(page);
if (spillEnabled) {
- // TODO trySetRevocableMemoryReservation else index.compact()
-
operatorContext.setRevocableMemoryReservation(index.getEstimatedSize().toBytes());
}
else {
@@ -379,6 +379,14 @@ public class HashBuilderOperator
checkState(spillEnabled, "Spill not enabled, no revokable memory should be reserved");
if (state == State.CONSUMING_INPUT) {
+ long indexSizeBeforeCompaction = index.getEstimatedSize().toBytes();
+ index.compact();
+ long indexSizeAfterCompaction = index.getEstimatedSize().toBytes();
+ if (indexSizeAfterCompaction < indexSizeBeforeCompaction * INDEX_COMPACTION_ON_REVOCATION_TARGET) {
+ finishMemoryRevoke = Optional.of(() -> {});
+ return immediateFuture(null);
+ }
+
finishMemoryRevoke = Optional.of(() -> {
index.clear();
operatorContext.setMemoryReservation(index.getEstimatedSize().toBytes());
