Skip to content

Commit

Permalink
ScanQueryFrameProcessor: Close CursorHolders as we go along.
Browse files Browse the repository at this point in the history
The change in apache#16533 added CursorHolders to the processor-level Closer.
This is problematic when running on an input channel: it means we started
keeping around all CursorHolders for all frames we process and closing them
when the channel is complete, rather than closing them as we go along.

This patch restores the prior behavior, where resources are closed as we go.
  • Loading branch information
gianm committed Sep 25, 2024
1 parent 0ec7eb3 commit 32b7e98
Showing 1 changed file with 17 additions and 5 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@ public class ScanQueryFrameProcessor extends BaseLeafFrameProcessor
private final VirtualColumns frameWriterVirtualColumns;
private final Closer closer = Closer.create();

private CursorHolder cursorHolder;
private Cursor cursor;
private Segment segment;
private final SimpleSettableOffset cursorOffset = new SimpleAscendingOffset(Integer.MAX_VALUE);
Expand Down Expand Up @@ -156,6 +157,7 @@ public ReturnOrAwait<Object> runIncrementally(final IntSet readableInputs) throw
@Override
public void cleanup() throws IOException
{
closer.register(cursorHolder);
closer.register(frameWriter);
closer.register(super::cleanup);
closer.close();
Expand Down Expand Up @@ -302,16 +304,16 @@ protected ReturnOrAwait<Unit> runWithInputChannel(
);
}

final CursorHolder cursorHolder = closer.register(
cursorFactory.makeCursorHolder(ScanQueryEngine.makeCursorBuildSpec(query, null))
);
final CursorHolder nextCursorHolder =
cursorFactory.makeCursorHolder(ScanQueryEngine.makeCursorBuildSpec(query, null));
final Cursor nextCursor = cursorHolder.asCursor();

if (nextCursor == null) {
// no cursor
nextCursorHolder.close();
return ReturnOrAwait.returnObject(Unit.instance());
}
final long rowsFlushed = setNextCursor(nextCursor, frameSegment);
final long rowsFlushed = setNextCursor(nextCursorHolder, nextCursor, frameSegment);

if (rowsFlushed > 0) {
return ReturnOrAwait.runAgain();
Expand Down Expand Up @@ -415,9 +417,19 @@ private long flushFrameWriter() throws IOException
}
}

private long setNextCursor(final Cursor cursor, final Segment segment) throws IOException
private long setNextCursor(
final CursorHolder cursorHolder,
final Cursor cursor,
final Segment segment
) throws IOException
{
final long rowsFlushed = flushFrameWriter();
if (this.cursorHolder != null) {
// Close here, don't add to the processor-level Closer, to avoid leaking CursorHolders. We may generate many
// CursorHolders per instance of this processor, and we need to close them as we go, not all at the end.
this.cursorHolder.close();
}
this.cursorHolder = cursorHolder;
this.cursor = cursor;
this.segment = segment;
this.cursorOffset.reset();
Expand Down

0 comments on commit 32b7e98

Please sign in to comment.