Not sure I follow this. What is the special casing here w.r.t single value v/s multi value ?

This comment is copied over from AggregateOperator (since both use the AggregationUtils) and basically discusses that if only a single input is part of the aggregation, then the Merger::initialize() function today doesn't cast it as a double, so the output won't be a double. For other cases it does use the double value:

    default Object initialize(Object other, DataSchema.ColumnDataType dataType) {
      // TODO: Initialize as a double so that if only one row is returned it matches the type when many rows are
      //       returned
      return other == null ? dataType.getNullPlaceholder() : other;
    }

Example merger function:

  private static Object mergeMax(Object left, Object right) {
    return Math.max(((Number) left).doubleValue(), ((Number) right).doubleValue());
  }

I tried to fix this but ran into a bunch of test issues so decided to try and fix the initialize to also use doubleValue() in a separate PR if possible. Also wanted context on why the code was written in this way. Hope this clarifies the comment.

@walterddr quick question in case you have context. Why do we initialize the aggregation output without using doubleValue() but use doubleValue() in the aggregation? Just wanted to understand the context and see if I can make a fix for this in a follow up PR

(nit) is it possible to encapsulate the input into an POJO type object ? Might look cleaner / easy to read / review.

Not necessary for this PR though

yeah sure let me explore this as a future improvement

Hmm not sure I understand this. I thought FRAME is not supported in the initial implementation based on the phases laid out in design doc.

These checks are for default frames (just to ensure we aren't getting inputs where the frames are overridden)

• Without ORDER BY: default frame: UNBOUNDED PRECEDING to UNBOUNDED FOLLOWING
• With ORDER BY: default frame: UNBOUNDED PRECEDING to CURRENT ROW

Also whether ROWS or RANGE is the default depends on the default frame so that's why I check for that too.

Why is 2nd check (after OR) needed ?

This is a temporary check because today we don't support window functions with ORDER BY where the ORDER BY key is different from the PARTITION BY key. We do support queries where the keys are the same, e.g. OVER(PARTITION BY key1 ORDER BY key1).

For queries such as OVER(PARTITION BY key1 ORDER BY key1), the default frame lands up getting set to UNBOUNDED PRECEDING to CURRENT ROW. This check here ensures that if the upperBound is indeed CURRENT ROW we should check that it's a query where the PARTITION BY and ORDER BY key is the same.

(nit) multi stage engine's query planning is likely to be heap heavy so we may want to double check if int is really needed or can we do away with short.

not a big deal in this PR but something to do in follow-up if folks agree

Great suggestion! I'll revisit these once I start work on frame support as I want to look into whether there is a recommendation on the frame limits or not. Will take this as an action item for frame support.

let's follow up later but IMO the most heavy part is not the memory size but the latency overhead :-P

Agree! we'll need to test this out for performance for sure

The state that whether or not we can produce output block should be maintained and dictated by the piece of code that consumed the input block - consumeInputBlocks

Generally speaking, whenever an operator consumes an input block, before exiting that function it should be able to determine whether or not it is ready to produce output block and accordingly change the local state which can then be leveraged in getNextBlock.

^^ might just be more intuitive but nothing wrong with the current approach

I've tried to keep the operator flow similar to AggregateOperator for now. This flag _hasReturnedWindowAggregateBlock is meant to differentiate the scenario where we finished consumption and haven't yet sent the output block, vs. the scenario where we finished consumption and sent the output block already but now just need to send a final EOS. In both cases the consumeInputBlocks will indicate there is nothing more to send.

We can revisit this logic for all operators at once if needed.

Sounds good. FYI @walterddr

for now the logic is kind of there to guard the EOS as @somandal mentioned. we can revisit in general later. please kindly file an issue. (I am yet to understand what's the confusion and how can we improve it, but will look into it more generally once we file the issue with more context)

Not really related to this PR but probably something we can improve generally

I am thinking that if we do the processing of operators in off heap buffers, then can we do something like this to avoid memcpy ?

Say we setup 3 ops in an op chain

Op1 -> Op2 -> Op3 from upstream to downstream

When the operators are setup, they get an input container (aka record batch with schema)

Op3's input container = Op2's output container
Op2's input container = Op1's output container

When the OpChain is setup, we call something like setup() on operator

Output setup (Input) {
   // create Output container and return it
   // the caller will use the returned Output as Input o setup the next downstream operator
}

We produce into Output but that does not require memcpy since it is direct buffers

Great suggestion. Let's look into this as a separate change.

Sounds good. FYI @walterddr

i dont understand why we need to copy in the first place. window over is 1-in-1-out in the row perspective. it only caches the row it gets, accumulate the results for the additional columns, and attach the result back to the row when it finishes.

i agree we will have overhead for heap memory b/c we need to buffer everything. but not sure why we need a mem copy, it felt like if we are using List<Object> instead of Object[] we can completely avoid this copy yes?

We need to create an output in the format of List<Object[]>, where each Object[] contains the contents of a single row. The output Object[] row size is larger than the input row's size since we need to store the aggregated columns. Using List<Object> instead of Object[] can avoid the copy as it can be appended to and added to the output result.

what is the other function we've supported? can we add a comment for this? (i think it is BOOL_AND and BOOL_OR)

but IMO, BOOL_AND and BOOL_OR are just MIN(booleanCol) and MAX(booleanCol) :-P i didn't understand why we needed that in the first place lol

Ah good point, I think we just saw that PostgreSQL has support for all aggregation functions and thought why not add what we have (at least the basic ones). Hope it's alright to leave this here. I've updated to add a comment here.

commented previously not sure isRows is a clear enough flag. let's name it isRowBased or use a enum

enum WindowBase {
  ROW, RANGE
}

done - I renamed it here, and store it as an enum in WindowAggregateOperator. If it's okay I'll create a util to store the enum in a common place for use in both places later as a followup?

never mind, updated this to add the enum directly into WindowNode and access it in WindowAggregateOperator.

SWEET refactoring!!! Thank you!!!

suggest javadoc

done

let's follow up later but IMO the most heavy part is not the memory size but the latency overhead :-P

i dont understand why we need to copy in the first place. window over is 1-in-1-out in the row perspective. it only caches the row it gets, accumulate the results for the additional columns, and attach the result back to the row when it finishes.

i agree we will have overhead for heap memory b/c we need to buffer everything. but not sure why we need a mem copy, it felt like if we are using List<Object> instead of Object[] we can completely avoid this copy yes?

for now the logic is kind of there to guard the EOS as @somandal mentioned. we can revisit in general later. please kindly file an issue. (I am yet to understand what's the confusion and how can we improve it, but will look into it more generally once we file the issue with more context)