During deeply nested group-by execution, the merge buffer for the inner group-by result should be held during the outer group-by execution

[jihoonson]

For deeply nested group-by execution, at most two merge buffers should be acquired on the broker side for the currently accumulating sequence and the underlying input sequence which is already accumulated. With the group-by strategy v2, this intermediate result is stored on a merge buffer.

Obviously, the merge buffer needs to be held until the current sequence accumulation is completed. However, in the current implementation (see CombiningSequence), the input sequence (CombiningSequence.baseSequence) is first accumulated, and then the current sequence is accumulated. The merge buffer for the input sequence is released immediately once it's accumulation is completed (BaseSequence.IteratorMaker.cleanup()), so it is not guaranteed to hold the intermediate result stored on that merge buffer until the current sequence is completely accumulated.

[jon-wei]

looking into this

[jihoonson]

I found that this rarely happens only when nested queries have a simple count aggregation like select count(*) from (select d1, sum(m1) from foo group by d1) t. And in this case, it seems normal because the simple count aggregation needs just merging inputs instead of accumulating them.
I'll close this issue.

[gianm]

@jihoonson are you saying the behavior you saw is not actually a bug?

[jihoonson]

Yes, it seems normal.
More precisely speaking, in CombiningSequence.accumulation(), baseSequence.accumulate(null, combiningAccumulator) consumes only a single row because the simple count aggregation always produces a single row. And in this case, CombiningAccumulator.accumulate() always executes the below code which doesn't have to acquire a merge buffer.

if (prevValue == null) {
  return mergeFn.apply(t, null);
}

Thus, even though the merge buffer of the input sequence is released before the current sequence is accumulated, the intermediate result of the input sequence accumulation is kept as lastValue which is the result of baseSequence.accumulate(null, combiningAccumulator) in CombiningSequence.accumulation().

[jihoonson]

You can reproduce this by checking out https://github.com/jihoonson/druid/tree/merge-buffer-debug and running GroupByQueryMergeBufferTest. The tests in this class of this branch emit some logs which show what happens in sequence.