Re: [PR] Simplify WindowOperatorQueryFrameProcessor to run all operators on entire partition (druid)

["cryptoe (via GitHub)" <gi...@apache.org>]

cryptoe commented on code in PR #16823:
URL: https://github.com/apache/druid/pull/16823#discussion_r1701324231


##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/WindowOperatorQueryFrameProcessor.java:
##########
@@ -137,205 +103,44 @@ public List<WritableFrameChannel> outputChannels()
   public ReturnOrAwait<Object> runIncrementally(IntSet readableInputs)
   {
     /*
-     *
-     * PARTITION BY A ORDER BY B
-     *
-     * Frame 1   -> rac1
-     * A  B
-     * 1, 2
-     * 1, 3
-     * 2, 1 --> key changed
-     * 2, 2
-     *
-     *
-     * Frame 2 -> rac2
-     * 3, 1 --> key changed
-     * 3, 2
-     * 3, 3
-     * 3, 4
-     *
-     * Frame 3 -> rac3
-     *
-     * 3, 5
-     * 3, 6
-     * 4, 1 --> key changed
-     * 4, 2
-     *
-     * In case of empty OVER clause, all these racs need to be added to a single rows and columns
-     * to be processed. The way we can do this is to use a ConcatRowsAndColumns
-     * ConcatRC [rac1, rac2, rac3]
-     * Run all ops on this
-     *
-     *
-     * The flow would look like:
-     * 1. Validate if the operator doesn't have any OVER() clause with PARTITION BY for this stage.
-     * 2. If 1 is true make a giant rows and columns (R&C) using concat as shown above
-     *    Let all operators run amok on that R&C
-     * 3. If 1 is false
-     *    Read a frame
-     *    keep the older row in a class variable
-     *    check row by row and compare current with older row to check if partition boundary is reached
-     *    when frame partition by changes
-     *    create R&C for those particular set of columns, they would have the same partition key
-     *    output will be a single R&C
-     *    write to output channel
-     *
-     *
+     * We add all the RACs to a single ConcatRowsAndColumns and pass it to the list of operators to process.
      *  Future thoughts: {@link https://github.com/apache/druid/issues/16126}
-     *
-     *  1. We are writing 1 partition to each frame in this way. In case of high cardinality data
-     *      we will be making a large number of small frames. We can have a check to keep size of frame to a value
-     *      say 20k rows and keep on adding to the same pending frame and not create a new frame
-     *
-     *  2. Current approach with R&C and operators materialize a single R&C for processing. In case of data
-     *     with low cardinality a single R&C might be too big to consume. Same for the case of empty OVER() clause
-     *     Most of the window operations like SUM(), RANK(), RANGE() etc. can be made with 2 passes of the data.
-     *     We might think to reimplement them in the MSQ way so that we do not have to materialize so much data
      */
 
-    if (partitionColumnNames.isEmpty()) {
-      // If we do not have any OVER() clause with PARTITION BY for this stage.
-      // Bring all data to a single executor for processing.
-      // Convert each frame to RAC.
-      // Concatenate all the racs to make a giant RAC.
-      // Let all operators run on the giant RAC until channel is finished.
-      if (inputChannel.canRead()) {
-        final Frame frame = inputChannel.read();
-        convertRowFrameToRowsAndColumns(frame);
-      } else if (inputChannel.isFinished()) {
-        runAllOpsOnMultipleRac(frameRowsAndCols);
-        return ReturnOrAwait.returnObject(Unit.instance());
-      } else {
-        return ReturnOrAwait.awaitAll(inputChannels().size());
-      }
-      return ReturnOrAwait.runAgain();
+    // Bring all data to a single executor for processing.

Review Comment:
   Probably a few things to check : 
   * Should we adjust the num rows per partition to some X number if window query is encountered. 
   * What are the guardrails if we go OOM. Can we fail gracefully instead of getting an annoying OOM error. 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org