ESQL: Track Blocks from topn
#99887
ESQL: Track Blocks from topn
#99887
Conversation
This creates tracked `Block`s from topn and closes the blocks passed to it.
elasticsearchmachine
added
the
Team:QL (Deprecated)
|
Pinging @elastic/es-ql (Team:QL) |
|
Pinging @elastic/elasticsearch-esql (:Query Languages/ES|QL) |
| @@ -95,7 +95,7 @@ protected final BigArrays nonBreakingBigArrays() { | |||
| /** | |||
| * A {@link DriverContext} with a nonBreakingBigArrays. | |||
| */ | |||
| protected final DriverContext driverContext() { | |||
| protected DriverContext driverContext() { // TODO make this final and return a breaking block factory | |||
There was a problem hiding this comment.
++ Agreed. Once we get further.
| assertThat(e.getMessage(), equalTo(MockBigArrays.ERROR_MESSAGE)); | ||
| assertThat(bigArrays.breakerService().getBreaker(CircuitBreaker.REQUEST).getUsed(), equalTo(0L)); | ||
| } finally { | ||
| Releasables.close(() -> Iterators.map(input.iterator(), p -> () -> p.releaseBlocks())); |
There was a problem hiding this comment.
Boom! my head exploded!! ;-) So this is inferred to Iterators.<Page, Releasable>map(... Maybe just use a method reference here to avoid the extra lambda arrows ( as suggested by your favourite IDE! )
Iterators.map(input.iterator(), p -> p::releaseBlocks)
There was a problem hiding this comment.
That kept not working. Let me try again.
| * The input blocks don't count against the memory usage for the limited operator that we | ||
| * build. | ||
| */ | ||
| List<Page> input = CannedSourceOperator.collectPages(simpleInput(driverContext().blockFactory(), between(1_000, 10_000))); |
There was a problem hiding this comment.
We want this blockFactory to never circuit break, right? Just do the accountancy.
There was a problem hiding this comment.
Yeah. Right now we actually don't assert that we release all the blocks either. I'll grab that in a follow up. The trouble, I think, is just a missing try block around processing the page. I think it's that simple, but I want to dig more into the failure cases in a follow up.
| @@ -139,10 +153,11 @@ protected final List<Page> oneDriverPerPageList(Iterator<List<Page>> source, Sup | |||
| } | |||
|
|
|||
| private void assertSimple(DriverContext context, int size) { | |||
| List<Page> input = CannedSourceOperator.collectPages(simpleInput(size)); | |||
| List<Page> input = CannedSourceOperator.collectPages(simpleInput(context.blockFactory(), size)); | |||
| List<Page> inputClone = CannedSourceOperator.deepCopyOf(input); | |||
There was a problem hiding this comment.
why do we clone here?
There was a problem hiding this comment.
Adding a comment:
// Clone the input so that the operator can close it, then, later, we can read it again to build the assertion.
This creates tracked
Blocks from topn and closes the blocks passed to it.