Close spiller in Operator#close() for window spilling
@@ -613,10 +613,7 @@ public class WindowOperator
inMemoryPagesIndexWithHashStrategies.pagesIndex.clear();
currentSpillGroupRowPage = Optional.empty();
- if (spiller.isPresent()) {
- spiller.get().close();
- spiller = Optional.empty();
- }
+ closeSpiller();
updateMemoryUsage(false);
resetPagesIndex = false;
@@ -626,6 +623,7 @@ public class WindowOperator
if (finishing && inMemoryPagesIndexWithHashStrategies.pagesIndex.getPositionCount() == 0 && !spiller.isPresent()) {
localRevocableMemoryContext.close();
localUserMemoryContext.close();
+ closeSpiller();
return TransformationState.finished();
}
@@ -644,6 +642,12 @@ public class WindowOperator
return needsMoreData();
}
+ void closeSpiller()
+ {
+ spiller.ifPresent(Spiller::close);
+ spiller = Optional.empty();
+ }
+
TransformationState<WorkProcessor<PagesIndexWithHashStrategies>> fullGroupBuffered()
{
// Convert revocable memory to user memory as inMemoryPagesIndexWithHashStrategies holds on to memory so we no longer can revoke
@@ -861,5 +865,6 @@ public class WindowOperator
public void close()
{
driverWindowInfo.set(Optional.of(windowInfo.build()));
+ spillablePagesToPagesIndexes.ifPresent(SpillablePagesToPagesIndexes::closeSpiller);
}
}
