WindowOperatorQueryFrameProcessor: Fix frame writer capacity issues + adhere to FrameProcessor's contract

[Akshat-Jain]

Description

Currently, we had a bug in the WindowOperatorQueryFrameProcessor where the frame writer's capacity could get reached for larger queries, causing it to not output all of the result rows.

For example, the following query gives incomplete output rows when we use maxNumTasks=2, compared to when we use more workers.

select trip_id, row_number() over(partition by trip_id) as c1 from "trips_xaa" where __time < TIMESTAMP '2013-09-20 11:15:25' group by trip_id
-- This gives 20479 rows with maxNumTasks=2
-- This gives 30341 rows with maxNumTasks=5
-- This gives 30341 rows with maxNumTasks=11

This PR fixes the above issue by maintaining the state of last rowId flushed to output channel, and triggering another iteration of runIncrementally() method if frame writer has rows pending flush to the output channel.

The above is done keeping in mind FrameProcessor's contract which enforces that we should write only a single frame to each output channel in any given iteration of runIncrementally().

For manual testing, I've been verifying the behavior with queries like the following:

select trip_id, row_number() over(partition by trip_id) as c1 from "trips_xaa" where __time < TIMESTAMP '2016-10-20 11:15:25' group by trip_id
-- 49795247 rows inputted, 49795247 rows outputted by window stage (with maxNumTasks=2 and maxNumTasks=11)

select c1, count(c1) from (select trip_id, row_number() over(partition by trip_id) as c1 from "trips_xaa" where __time < TIMESTAMP '2016-10-20 11:15:25' group by trip_id) group by c1
-- [1, 49795247] with maxNumTasks=2 and maxNumTasks=11

Additionally, I have added a test WindowOperatorQueryFrameProcessorTest#testFrameWriterReachingCapacity() which was previously writing less number of rows to the output channel, but is writing the complete set of rows.

---

This PR has:

• been self-reviewed.
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[kgyrtkirk]

this seems like a bugfix - please remove the optimization; that should be reviewed separetly

[Akshat-Jain]

this seems like a bugfix - please remove the optimization; that should be reviewed separetly

@kgyrtkirk I think we anyway should get the optimization also merged for Druid 31 (since it's significant), hence bundled it together with this patch because it was already updating a lot of the frame writer logic.

If it helps, this part of the diff corresponds to the optimization:

[kgyrtkirk]

remove the performance enhancement ; and place it in a separate PR - also add a benchmark to help keep track of performance improvements

[Akshat-Jain]

remove the performance enhancement ; and place it in a separate PR - also add a benchmark to help keep track of performance improvements

@kgyrtkirk I was hoping to keep all changes related to frame writer together, to ensure that everything works fine with both of them together. Instead of a bug fix PR, I'm looking at this PR as more of a "revamp frame writer logic in WindowOperatorQueryFrameProcessor" PR.

Splitting it into multiple PRs is certainly doable, but I'd still have to continue testing them coupled together locally, which is unnecessary and would cause unnecessary delays (we are hoping to get this in for Druid 31 release).

Regarding the benchmark - I agree that it would be great to have it. It's on my list of todos, but it's not a trivial task and would require some time. So until then, benchmark shouldn't block any performance improvements / logic changes.

Thoughts? cc: @cryptoe

[kgyrtkirk]

what's happening in this PR right now is pretty convoluted - that's why I'm asking to remove any performance enhancements ; because I don't think the fix for the bug is right

[Akshat-Jain]

@kgyrtkirk Fair enough, working on splitting up the PR.

[Akshat-Jain]

@kgyrtkirk Have removed the optimization diff from this PR. Have opened a separate PR for it: #17211, appreciate your reviews on that PR as well. Thanks!

[kgyrtkirk]

looks good; just a minor comment :)

[kgyrtkirk]

thank you for the updates!
+1

[kgyrtkirk]

its just performance: but the creation of this rac is an O(n) operation; regardless where the rowId stands.
that's why it would have been better to just pack all these things into an inner-workhorse class....when that will be done this should be taken into account.