diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java index 1e93eecaf7c29..e72d4886de7ed 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java @@ -51,9 +51,12 @@ import org.apache.ignite.internal.processors.query.calcite.exec.exp.RangeIterable; import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AccumulatorWrapper; import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AggregateType; -import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.WindowPartition; +import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.BufferingWindowPartition; +import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.StreamingWindowPartition; +import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.WindowFunctions; import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.WindowPartitionFactory; import org.apache.ignite.internal.processors.query.calcite.exec.rel.AbstractSetOpNode; +import org.apache.ignite.internal.processors.query.calcite.exec.rel.BufferingWindowNode; import org.apache.ignite.internal.processors.query.calcite.exec.rel.CollectNode; import org.apache.ignite.internal.processors.query.calcite.exec.rel.CorrelatedNestedLoopJoinNode; import org.apache.ignite.internal.processors.query.calcite.exec.rel.FilterNode; @@ -73,12 +76,13 @@ import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanNode; import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanStorageNode; import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanTableRowNode; +import org.apache.ignite.internal.processors.query.calcite.exec.rel.SingleNode; import org.apache.ignite.internal.processors.query.calcite.exec.rel.SortAggregateNode; import org.apache.ignite.internal.processors.query.calcite.exec.rel.SortNode; +import org.apache.ignite.internal.processors.query.calcite.exec.rel.StreamingWindowNode; import org.apache.ignite.internal.processors.query.calcite.exec.rel.TableSpoolNode; import org.apache.ignite.internal.processors.query.calcite.exec.rel.UncollectNode; import org.apache.ignite.internal.processors.query.calcite.exec.rel.UnionAllNode; -import org.apache.ignite.internal.processors.query.calcite.exec.rel.WindowNode; import org.apache.ignite.internal.processors.query.calcite.metadata.AffinityService; import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup; import org.apache.ignite.internal.processors.query.calcite.prepare.bounds.SearchBounds; @@ -961,18 +965,28 @@ else if (rel instanceof Intersect) assert collation.getFieldCollations().size() >= grpKeys.size(); Comparator partCmp = expressionFactory.comparator(TraitUtils.createCollation(grpKeys)); + RowFactory rowFactory = ctx.rowHandler().factory(ctx.getTypeFactory(), outType); + List calls = grp.getAggregateCalls(rel); - Supplier> partFactory = new WindowPartitionFactory<>(ctx, grp, calls, inputType); + assert !calls.isEmpty() : "Window aggregate calls should not be empty"; - RowFactory rowFactory = ctx.rowHandler().factory(ctx.getTypeFactory(), outType); + WindowPartitionFactory factory = new WindowPartitionFactory<>(ctx); - WindowNode node = new WindowNode<>(ctx, outType, partCmp, partFactory, rowFactory); + SingleNode windowNode; + if (WindowFunctions.streamable(grp)) { + StreamingWindowPartition partition = factory.newStreamingPartition(grp, calls, inputType); + windowNode = new StreamingWindowNode<>(ctx, outType, partCmp, partition, rowFactory); + } + else { + BufferingWindowPartition partition = factory.newBufferingPartition(grp, calls, inputType); + windowNode = new BufferingWindowNode<>(ctx, outType, partCmp, partition, rowFactory); + } Node input = visit(rel.getInput()); - node.register(input); + windowNode.register(input); - return node; + return windowNode; } /** {@inheritDoc} */ diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/BufferingWindowPartition.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/BufferingWindowPartition.java index 98aafe087ead8..fa7ea36ee6cc9 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/BufferingWindowPartition.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/BufferingWindowPartition.java @@ -17,26 +17,35 @@ package org.apache.ignite.internal.processors.query.calcite.exec.exp.window; -import java.util.ArrayList; +import java.util.ArrayDeque; import java.util.Comparator; +import java.util.Deque; import java.util.List; -import java.util.function.Consumer; import org.apache.calcite.rel.core.Window; import org.apache.calcite.rel.type.RelDataType; import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext; import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler; -import org.apache.ignite.internal.processors.query.calcite.exec.tracker.RowTracker; +import org.jetbrains.annotations.Nullable; /** Buffering implementation of the ROWS / RANGE window partition. */ -final class BufferingWindowPartition extends WindowPartitionBase { - /** Rows in partition. */ - private final List buf; +public final class BufferingWindowPartition extends WindowPartitionBase { + /** */ + private final ExecutionContext ctx; - /** Frame within partition. */ - private final WindowPartitionFrame frame; + /** */ + private final Window.Group grp; /** */ - private RowTracker memoryTracker; + private final RelDataType inputRowType; + + /** Slices in partition. */ + private final Deque frames; + + /** Number of rows, can be evaluted right now. */ + private int ready; + + /** Current frame for row evaluation. */ + private FrameHolder currFrame; /** */ BufferingWindowPartition( @@ -48,60 +57,64 @@ final class BufferingWindowPartition extends WindowPartitionBase { RelDataType inputRowType ) { super(peerCmp, funcFactory, rowFactory); - buf = new ArrayList<>(); - frame = createFrame(ctx, grp, peerCmp, inputRowType, buf); - } - - /** {@inheritDoc} */ - @Override public void add(Row row) { - buf.add(row); - onRowAdded(row); + this.ctx = ctx; + this.grp = grp; + this.inputRowType = inputRowType; + frames = new ArrayDeque<>(); } - /** {@inheritDoc} */ - @Override public void evalTo(RowHandler.RowFactory factory, Consumer output) { + /** + * Appends rows to the partition. + * Important: the buffer must not be modified after being passed to this method. + **/ + public void appendPartition(List buf, Runnable onBufRemoved) { if (buf.isEmpty()) return; - List> accumulators = createWrappers(); - Object[] accResults = new Object[accumulators.size()]; + WindowPartitionFrame frame = createFrame(ctx, grp, peerCmp, inputRowType, buf); + ready += frame.size(); + frames.add(new FrameHolder(frame, onBufRemoved)); + } - int size = buf.size(); - Row prevRow = null; - int peerIdx = -1; - for (int rowIdx = 0; rowIdx < size; rowIdx++) { - Row currRow = buf.get(rowIdx); - if (isNewPeer(currRow, prevRow)) - peerIdx++; + /** + * Evaluates next row in the window partition. + * @return Result row or {@code null} if there are no more rows. + */ + public @Nullable Row nextRow(RowHandler.RowFactory factory) { + if (currFrame == null || currFrame.consumed()) { + if (currFrame != null) + currFrame.release(); + currFrame = frames.pollFirst(); + } - int accIdx = 0; - for (WindowFunctionWrapper acc : accumulators) { - Object accResult = acc.callBuffering(currRow, rowIdx, peerIdx, frame); - accResults[accIdx++] = accResult; - } + if (currFrame == null) { + assert ready == 0; + return null; + } - Row resultRow = createResultRow(factory, currRow, accResults); - output.accept(resultRow); + assert ready > 0; + Row resultRow = currFrame.nextRow(factory); + ready--; + return resultRow; + } - prevRow = currRow; - } + /** Returns the number of rows can be evaluted. */ + public int ready() { + return ready; } /** {@inheritDoc} */ @Override public void reset() { - buf.forEach(this::onRowRemoved); - buf.clear(); - frame.reset(); - } + FrameHolder holder; + while ((holder = frames.poll()) != null) + holder.release(); - /** {@inheritDoc} */ - @Override public boolean isStreaming() { - return false; - } + if (currFrame != null) { + currFrame.release(); + currFrame = null; + } - /** {@inheritDoc} */ - @Override public void attachMemoryTracker(RowTracker memoryTracker) { - this.memoryTracker = memoryTracker; + ready = 0; } /** Creates frame for partition. */ @@ -118,15 +131,70 @@ private static WindowPartitionFrame createFrame( return new RangeWindowPartitionFrame<>(buf, ctx, peerCmp, grp, inputRowType); } - /** Adds row to memory tracker. */ - private void onRowAdded(Row row) { - if (memoryTracker != null) - memoryTracker.onRowAdded(row); - } + /** */ + private final class FrameHolder { + /** */ + private final WindowPartitionFrame frame; - /** Removes row from memory tracker. */ - private void onRowRemoved(Row row) { - if (memoryTracker != null) - memoryTracker.onRowRemoved(row); + /** */ + private final Runnable onFrameRemoved; + + /** Index of the current row in the frame. */ + private int rowIdx; + + /** Index of the current peer in the frame. */ + private int peerIdx = -1; + + /** */ + private Row prevRow; + + /** */ + private List> accumulators; + + /** */ + private Object[] accResults; + + /** */ + private FrameHolder(WindowPartitionFrame frame, Runnable removed) { + this.frame = frame; + onFrameRemoved = removed; + } + + /** */ + Row nextRow(RowHandler.RowFactory factory) { + assert !consumed(); + + if (accumulators == null) { + accumulators = createWrappers(); + accResults = new Object[accumulators.size()]; + } + + Row currRow = frame.get(rowIdx); + if (isNewPeer(currRow, prevRow)) + peerIdx++; + + int accIdx = 0; + for (WindowFunctionWrapper acc : accumulators) { + Object accResult = acc.callBuffering(currRow, rowIdx, peerIdx, currFrame.frame); + accResults[accIdx++] = accResult; + } + + Row resultRow = createResultRow(factory, currRow, accResults); + + rowIdx++; + prevRow = currRow; + + return resultRow; + } + + /** */ + boolean consumed() { + return rowIdx == frame.size(); + } + + /** */ + void release() { + onFrameRemoved.run(); + } } } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/RangeWindowPartitionFrame.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/RangeWindowPartitionFrame.java index 48bd52ef503a4..8a88ec160ee84 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/RangeWindowPartitionFrame.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/RangeWindowPartitionFrame.java @@ -103,7 +103,7 @@ final class RangeWindowPartitionFrame extends WindowPartitionFrame { if (lowerBoundRow == null) cachedStartIdx = 0; else - cachedStartIdx = bsearchBound(lowerBoundRow, buf, true); + cachedStartIdx = bsearchBound(lowerBoundRow, true); return cachedStartIdx; } @@ -121,29 +121,20 @@ final class RangeWindowPartitionFrame extends WindowPartitionFrame { if (upperBoundRow == null) cachedEndIdx = size() - 1; else - cachedEndIdx = bsearchBound(upperBoundRow, buf, false); + cachedEndIdx = bsearchBound(upperBoundRow, false); return cachedEndIdx; } - /** {@inheritDoc} */ - @Override public void reset() { - // Reseting index cache. - cachedStartPeerIdx = -1; - cachedEndPeerIdx = -1; - cachedStartRowIdx = -1; - cachedEndRowIdx = -1; - } - /** Binary search bound. */ - private int bsearchBound(Row row, List buf, boolean lower) { + private int bsearchBound(Row row, boolean lower) { int start = 0; - int end = buf.size() - 1; + int end = size() - 1; while (start <= end) { int mid = (start + end) / 2; - Row midRow = buf.get(mid); + Row midRow = get(mid); int cmp = compareRowPeer(midRow, row); if (cmp > 0 || (lower && cmp == 0)) diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/RowWindowPartitionFrame.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/RowWindowPartitionFrame.java index 84712e3fc09b0..f8d44cae010b2 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/RowWindowPartitionFrame.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/RowWindowPartitionFrame.java @@ -95,13 +95,6 @@ final class RowWindowPartitionFrame extends WindowPartitionFrame { return applyOffset(rowIdx, cachedEndOffset, size() - 1); } - /** {@inheritDoc} */ - @Override protected void reset() { - // Reseting index cache. - cachedStartRowIdx = -1; - cachedEndRowIdx = -1; - } - /** */ private static int applyOffset(int rowIdx, int offset, int cap) { int idx = Math.addExact(rowIdx, offset); diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/StreamWindowPartition.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/StreamingWindowPartition.java similarity index 63% rename from modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/StreamWindowPartition.java rename to modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/StreamingWindowPartition.java index 6459c8ecff9aa..cb0cd12ec0bb7 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/StreamWindowPartition.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/StreamingWindowPartition.java @@ -19,18 +19,13 @@ import java.util.Comparator; import java.util.List; -import java.util.function.Consumer; import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler; -import org.apache.ignite.internal.processors.query.calcite.exec.tracker.RowTracker; /** Non-buffering implementation of the ROWS / RANGE window partition. */ -final class StreamWindowPartition extends WindowPartitionBase { +public final class StreamingWindowPartition extends WindowPartitionBase { /** */ private Row prevRow; - /** */ - private Row currRow; - /** */ private int rowIdx = -1; @@ -44,7 +39,7 @@ final class StreamWindowPartition extends WindowPartitionBase { private Object[] accResults; /** */ - StreamWindowPartition( + StreamingWindowPartition( Comparator peerCmp, WindowFunctionFactory funcFactory, RowHandler.RowFactory rowFactory @@ -52,55 +47,42 @@ final class StreamWindowPartition extends WindowPartitionBase { super(peerCmp, funcFactory, rowFactory); } - /** {@inheritDoc} */ - @Override public void add(Row row) { - assert currRow == null : "StreamingWindowPartition can only hold one row"; - currRow = row; - } - - /** {@inheritDoc} */ - @Override public void evalTo(RowHandler.RowFactory factory, Consumer output) { - if (currRow == null) - return; - + /** Evaluates window functions for the given row. */ + public Row eval(Row row, RowHandler.RowFactory factory) { if (accumulators == null) { accumulators = createWrappers(); accResults = new Object[accumulators.size()]; } rowIdx++; - if (isNewPeer(currRow, prevRow)) + if (isNewPeer(row, prevRow)) peerIdx++; int accIdx = 0; for (WindowFunctionWrapper acc : accumulators) { - Object accResult = acc.callStreaming(currRow, rowIdx, peerIdx); + Object accResult = acc.callStreaming(row, rowIdx, peerIdx); accResults[accIdx++] = accResult; } - Row resultRow = createResultRow(factory, currRow, accResults); - output.accept(resultRow); + Row resultRow = createResultRow(factory, row, accResults); + prevRow = row; + return resultRow; + } + + /** Returns true if any of the window functions is an accumulator stores incomig row. */ + public boolean hasAggAccumulators() { + List> accumulators = this.accumulators; + if (accumulators == null) + accumulators = createWrappers(); - prevRow = currRow; - currRow = null; + return accumulators.stream().anyMatch(WindowFunctionWrapper::isAggAccumulator); } /** {@inheritDoc} */ @Override public void reset() { - currRow = null; prevRow = null; rowIdx = -1; peerIdx = -1; accumulators = null; } - - /** {@inheritDoc} */ - @Override public boolean isStreaming() { - return true; - } - - /** {@inheritDoc} */ - @Override public void attachMemoryTracker(RowTracker memoryTracker) { - // Streaming partition does not use memory tracker. - } } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowFunctionFactory.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowFunctionFactory.java index 493e8d75c8802..3a34159a0cf53 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowFunctionFactory.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowFunctionFactory.java @@ -28,6 +28,7 @@ import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AccumulatorWrapper; import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AccumulatorsFactoryBase; import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AggregateType; +import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.IterableAccumulator; import org.apache.ignite.internal.processors.query.calcite.util.Commons; import org.jetbrains.annotations.NotNull; @@ -163,6 +164,11 @@ private final class FunctionWrapper implements WindowFunctionWrapper { Object result = windowFunction.call(accRow, rowIdx, peerIdx, frame); return outAdapter.apply(result); } + + /** {@inheritDoc} */ + @Override public boolean isAggAccumulator() { + return false; + } } /** */ @@ -237,6 +243,11 @@ else if (frameEnd != end && end >= 0) { return acc.end(); } + /** {@inheritDoc} */ + @Override public boolean isAggAccumulator() { + return accumulator() instanceof IterableAccumulator; + } + /** */ private AccumulatorWrapper accumulator() { if (accHolder != null) diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowFunctionWrapper.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowFunctionWrapper.java index 138ef14ae6ce8..e80fd49179526 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowFunctionWrapper.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowFunctionWrapper.java @@ -26,4 +26,7 @@ interface WindowFunctionWrapper { /** Performs window function computation for the specified row inside the buffered frame. */ @Nullable Object callBuffering(Row row, int rowIdx, int peerIdx, WindowPartitionFrame frame); + + /** Returns true if the window function has an accumulator stores incomig row. */ + boolean isAggAccumulator(); } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartition.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartition.java index 0f875faa11f52..fb276ea8ff3c3 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartition.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartition.java @@ -17,24 +17,8 @@ package org.apache.ignite.internal.processors.query.calcite.exec.exp.window; -import java.util.function.Consumer; -import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler; -import org.apache.ignite.internal.processors.query.calcite.exec.tracker.RowTracker; - /** Partition of rows in window function calculation. */ public interface WindowPartition { - /** Adding row to the window partition. */ - void add(Row row); - - /** Evaluates partition to an output. */ - void evalTo(RowHandler.RowFactory factory, Consumer output); - /** Reset current window partition (i.e., on partition restart). */ void reset(); - - /** Returns {@code true} if partition is streaming. */ - boolean isStreaming(); - - /** Sets memory tracker for current partition. */ - void attachMemoryTracker(RowTracker memoryTracker); } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionBase.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionBase.java index 41aa3eb58c973..258088db91dee 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionBase.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionBase.java @@ -25,7 +25,7 @@ /** Base implementation of window partition. */ abstract class WindowPartitionBase implements WindowPartition { /** Comparator for computing the peer index. */ - private final Comparator peerCmp; + protected final Comparator peerCmp; /** */ private final WindowFunctionFactory funcFactory; diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionFactory.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionFactory.java index 88d0db30e9cf1..95eba7262245a 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionFactory.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionFactory.java @@ -19,7 +19,6 @@ import java.util.Comparator; import java.util.List; -import java.util.function.Supplier; import org.apache.calcite.rel.core.AggregateCall; import org.apache.calcite.rel.core.Window; import org.apache.calcite.rel.type.RelDataType; @@ -28,56 +27,55 @@ import org.apache.ignite.internal.processors.query.calcite.util.Commons; /** Factory to create {@link WindowPartitionBase} factory from {@link Window.Group}. */ -public final class WindowPartitionFactory implements Supplier> { - /** */ - private final WindowFunctionFactory funcFactory; - +public final class WindowPartitionFactory { /** */ private final ExecutionContext ctx; /** */ - private final Window.Group grp; + public WindowPartitionFactory(ExecutionContext ctx) { + this.ctx = ctx; + } - /** */ - private final RelDataType inputRowType; + /** Creates streaming window partition for provieded group, aggragates and input row type, */ + public StreamingWindowPartition newStreamingPartition(Window.Group grp, List calls, + RelDataType inputRowType) { - /** */ - private final RowHandler.RowFactory rowFactory; + assert WindowFunctions.streamable(grp); - /** */ - private final Comparator peerCmp; + return createPartition(grp, calls, inputRowType, StreamingWindowPartition::new); + } - /** */ - private final boolean streamable; + /** Creates buffering window partition for provieded group, aggragates and input row type, */ + public BufferingWindowPartition newBufferingPartition(Window.Group grp, List calls, + RelDataType inputRowType) { - /** */ - public WindowPartitionFactory( - ExecutionContext ctx, - Window.Group grp, - List calls, - RelDataType inputRowType - ) { - this.ctx = ctx; - this.grp = grp; - this.inputRowType = inputRowType; + return createPartition(grp, calls, inputRowType, (peerCmp, funcFactory, rowFactory) + -> new BufferingWindowPartition<>(peerCmp, funcFactory, rowFactory, ctx, grp, inputRowType)); + } - List aggTypes = Commons.transform(calls, AggregateCall::getType); - rowFactory = ctx.rowHandler().factory(Commons.typeFactory(), aggTypes); + /** */ + private > T createPartition(Window.Group grp, List calls, + RelDataType inputRowType, PartitionCreator creator) { + Comparator peerCmp; if (grp.isRows) // peer comparator in meaningless in rows frame. peerCmp = null; else peerCmp = ctx.expressionFactory().comparator(grp.collation()); - streamable = WindowFunctions.streamable(grp); - funcFactory = new WindowFunctionFactory<>(ctx, grp, calls, inputRowType); + WindowFunctionFactory funcFactory = new WindowFunctionFactory<>(ctx, grp, calls, inputRowType); + + List aggTypes = Commons.transform(calls, AggregateCall::getType); + RowHandler.RowFactory rowFactory = ctx.rowHandler().factory(Commons.typeFactory(), aggTypes); + + return creator.create(peerCmp, funcFactory, rowFactory); } - /** {@inheritDoc} */ - @Override public WindowPartition get() { - if (streamable) - return new StreamWindowPartition<>(peerCmp, funcFactory, rowFactory); - else - return new BufferingWindowPartition<>(peerCmp, funcFactory, rowFactory, ctx, grp, inputRowType); + /** */ + @FunctionalInterface + private interface PartitionCreator> { + /** */ + T create(Comparator peerCmp, WindowFunctionFactory funcFactory, RowHandler.RowFactory rowFactory); } + } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionFrame.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionFrame.java index 25225f215ae46..c69ef5f4d9912 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionFrame.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/WindowPartitionFrame.java @@ -17,17 +17,16 @@ package org.apache.ignite.internal.processors.query.calcite.exec.exp.window; -import java.util.Collections; import java.util.List; /** Rows frame within window partition. */ abstract class WindowPartitionFrame { /** Holds immutable refrence to buffered window partition rows. */ - protected final List buf; + private final List buf; /** */ WindowPartitionFrame(List buf) { - this.buf = Collections.unmodifiableList(buf); + this.buf = buf; } /** Returns row from partition by index. */ @@ -56,7 +55,4 @@ final int size(int rowIdx, int peerIdx) { final int size() { return buf.size(); } - - /** Resets current frame. */ - protected abstract void reset(); } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/WindowNode.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/BufferingWindowNode.java similarity index 55% rename from modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/WindowNode.java rename to modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/BufferingWindowNode.java index ae4e0527eb3fd..7a44ea8fce9ca 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/WindowNode.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/BufferingWindowNode.java @@ -17,29 +17,34 @@ package org.apache.ignite.internal.processors.query.calcite.exec.rel; -import java.util.ArrayDeque; +import java.util.ArrayList; import java.util.Comparator; -import java.util.Deque; -import java.util.function.Supplier; +import java.util.List; import org.apache.calcite.rel.type.RelDataType; import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext; import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler; -import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.WindowPartition; +import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.BufferingWindowPartition; +import org.apache.ignite.internal.processors.query.calcite.exec.tracker.RowTracker; import org.apache.ignite.internal.util.typedef.F; +import static org.apache.ignite.internal.processors.query.calcite.exec.rel.MemoryTrackingNode.DFLT_ROW_OVERHEAD; + /** Window node. */ -public class WindowNode extends MemoryTrackingNode implements SingleNode, Downstream { +public class BufferingWindowNode extends AbstractNode implements SingleNode, Downstream { /** */ private final Comparator partCmp; /** */ - private final Supplier> partFactory; + private final BufferingWindowPartition part; /** */ private final RowHandler.RowFactory rowFactory; /** */ - private WindowPartition part; + private List inBuf; + + /** */ + private RowTracker inBufTracker; /** */ private int requested; @@ -50,39 +55,39 @@ public class WindowNode extends MemoryTrackingNode implements SingleNo /** */ private Row prevRow; - /** */ - private final Deque outBuf = new ArrayDeque<>(IN_BUFFER_SIZE); - /** */ private boolean inLoop; /** */ - public WindowNode( + public BufferingWindowNode( ExecutionContext ctx, RelDataType rowType, Comparator partCmp, - Supplier> partFactory, + BufferingWindowPartition part, RowHandler.RowFactory rowFactory ) { - super(ctx, rowType, DFLT_ROW_OVERHEAD); + super(ctx, rowType); this.partCmp = partCmp; - this.partFactory = partFactory; + this.part = part; this.rowFactory = rowFactory; } /** {@inheritDoc} */ @Override public void request(int rowsCnt) throws Exception { assert !F.isEmpty(sources()) && sources().size() == 1; - assert rowsCnt > 0; + assert rowsCnt > 0 && requested == 0; checkState(); requested = rowsCnt; - if (waiting == 0 && outBuf.isEmpty()) + if (inLoop) + return; + + if (part.ready() > 0) + context().execute(this::flush, this::onError); + else if (waiting == 0) source().request(waiting = IN_BUFFER_SIZE); - else if (!inLoop) - flush(); } /** {@inheritDoc} */ @@ -94,24 +99,29 @@ else if (!inLoop) waiting--; - if (part == null) { - part = partFactory.get(); - part.attachMemoryTracker(nodeMemoryTracker); - } - else if (prevRow != null && partCmp != null && partCmp.compare(prevRow, row) != 0) { - part.evalTo(rowFactory, this::appendToOutBuf); - part.reset(); + if (prevRow != null && partCmp != null && partCmp.compare(prevRow, row) != 0) { + part.appendPartition(inBuf, inBufTracker::reset); + inBuf = null; + inBufTracker = null; } - part.add(row); + if (inBuf == null) { + inBuf = new ArrayList<>(IN_BUFFER_SIZE); + inBufTracker = context().createNodeMemoryTracker(DFLT_ROW_OVERHEAD); + } - if (part.isStreaming()) - part.evalTo(rowFactory, this::appendToOutBuf); + inBuf.add(row); + inBufTracker.onRowAdded(row); prevRow = row; - if (!inLoop) + if (inLoop) + return; + + if (part.ready() > 0) flush(); + else if (waiting == 0) + context().execute(() -> source().request(waiting = IN_BUFFER_SIZE), this::onError); } /** {@inheritDoc} */ @@ -124,9 +134,11 @@ else if (prevRow != null && partCmp != null && partCmp.compare(prevRow, row) != checkState(); - if (part != null) { - part.evalTo(rowFactory, this::appendToOutBuf); - part.reset(); + // append partition for remaining rows if any. + if (!F.isEmpty(inBuf)) { + part.appendPartition(inBuf, inBufTracker::reset); + inBuf = null; + inBufTracker = null; } flush(); @@ -136,12 +148,22 @@ else if (prevRow != null && partCmp != null && partCmp.compare(prevRow, row) != @Override protected void rewindInternal() { requested = 0; waiting = 0; - if (part != null) { - part.reset(); - part = null; + prevRow = null; + part.reset(); + if (inBuf != null) { + inBufTracker.reset(); + inBuf = null; + inBufTracker = null; } - outBuf.clear(); - nodeMemoryTracker.reset(); + } + + /** {@inheritDoc} */ + @Override protected void closeInternal() { + part.reset(); + if (inBuf != null) + inBufTracker.reset(); + + super.closeInternal(); } /** {@inheritDoc} */ @@ -152,32 +174,37 @@ else if (prevRow != null && partCmp != null && partCmp.compare(prevRow, row) != return this; } - /** */ - private void appendToOutBuf(Row row) { - outBuf.add(row); - nodeMemoryTracker.onRowAdded(row); - } - /** */ private void flush() throws Exception { inLoop = true; try { - while (requested > 0 && !outBuf.isEmpty()) { + int processed = 0; + while (requested > 0) { + Row result = part.nextRow(rowFactory); + if (result == null) + break; + requested--; - Row row = outBuf.poll(); - nodeMemoryTracker.onRowRemoved(row); - downstream().push(row); + downstream().push(result); + + processed++; + + if (processed == IN_BUFFER_SIZE && requested > 0) { + // Allow others to do their job. + context().execute(this::flush, this::onError); + return; + } } } finally { inLoop = false; } - if (waiting == 0 && requested > 0 && outBuf.isEmpty()) + if (waiting == 0 && requested > 0) context().execute(() -> source().request(waiting = IN_BUFFER_SIZE), this::onError); - if (waiting < 0 && requested > 0 && outBuf.isEmpty()) { + if (waiting < 0 && requested > 0 && part.ready() == 0) { requested = 0; downstream().end(); } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/StreamingWindowNode.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/StreamingWindowNode.java new file mode 100644 index 0000000000000..8dc9f105bf805 --- /dev/null +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/StreamingWindowNode.java @@ -0,0 +1,113 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.query.calcite.exec.rel; + +import java.util.Comparator; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext; +import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler; +import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.StreamingWindowPartition; +import org.apache.ignite.internal.util.typedef.F; + +/** Non-buffering implementation of Window node. */ +public class StreamingWindowNode extends MemoryTrackingNode implements SingleNode, Downstream { + /** */ + private final Comparator partCmp; + + /** */ + private final RowHandler.RowFactory rowFactory; + + /** */ + private final StreamingWindowPartition part; + + /** */ + private final boolean hasAggAccum; + + /** */ + private Row prevRow; + + /** */ + public StreamingWindowNode( + ExecutionContext ctx, + RelDataType rowType, + Comparator partCmp, + StreamingWindowPartition part, + RowHandler.RowFactory rowFactory + ) { + super(ctx, rowType, DFLT_ROW_OVERHEAD); + this.partCmp = partCmp; + this.part = part; + this.rowFactory = rowFactory; + hasAggAccum = part.hasAggAccumulators(); + } + + /** {@inheritDoc} */ + @Override public void request(int rowsCnt) throws Exception { + assert !F.isEmpty(sources()) && sources().size() == 1; + assert rowsCnt > 0; + + checkState(); + + source().request(rowsCnt); + } + + /** {@inheritDoc} */ + @Override public void push(Row row) throws Exception { + assert downstream() != null; + + checkState(); + + if (prevRow != null && partCmp != null && partCmp.compare(prevRow, row) != 0) { + part.reset(); + nodeMemoryTracker.reset(); + } + + Row result = part.eval(row, rowFactory); + + if (hasAggAccum) + nodeMemoryTracker.onRowAdded(row); + + prevRow = row; + + downstream().push(result); + } + + /** {@inheritDoc} */ + @Override public void end() throws Exception { + assert downstream() != null; + + checkState(); + + downstream().end(); + } + + /** {@inheritDoc} */ + @Override protected void rewindInternal() { + prevRow = null; + part.reset(); + nodeMemoryTracker.reset(); + } + + /** {@inheritDoc} */ + @Override protected Downstream requestDownstream(int idx) { + if (idx != 0) + throw new IndexOutOfBoundsException(); + + return this; + } +} diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/WindowConverterRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/WindowConverterRule.java index d550d98182595..4cea7c737f3e6 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/WindowConverterRule.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/WindowConverterRule.java @@ -106,6 +106,7 @@ private static Window.Group replaceAggCallOrdinal(Window.Group grp) { for (int i = 0; i < calls.size(); i++) { Window.RexWinAggCall aggCall = calls.get(i); Window.RexWinAggCall newCall = new Window.RexWinAggCall( + aggCall.pos, (SqlAggFunction)aggCall.op, aggCall.type, aggCall.operands, diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/WindowExecutionTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/WindowExecutionTest.java index bf409bde84b22..5486d52dfca7e 100644 --- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/WindowExecutionTest.java +++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/WindowExecutionTest.java @@ -23,7 +23,6 @@ import java.util.Comparator; import java.util.List; import java.util.UUID; -import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.IntStream; import org.apache.calcite.rel.RelCollations; @@ -36,21 +35,21 @@ import org.apache.calcite.rex.RexWindowExclusion; import org.apache.calcite.sql.SqlAggFunction; import org.apache.calcite.sql.fun.SqlStdOperatorTable; +import org.apache.calcite.sql.parser.SqlParserPos; import org.apache.calcite.util.ImmutableBitSet; import org.apache.calcite.util.ImmutableIntList; import org.apache.ignite.internal.processors.query.calcite.exec.ArrayRowHandler; import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext; import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler; import org.apache.ignite.internal.processors.query.calcite.exec.exp.IgniteRexBuilder; -import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.WindowFunctions; -import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.WindowPartition; +import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.BufferingWindowPartition; +import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.StreamingWindowPartition; import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.WindowPartitionFactory; import org.apache.ignite.internal.processors.query.calcite.sql.fun.IgniteOwnSqlOperatorTable; import org.apache.ignite.internal.processors.query.calcite.trait.TraitUtils; import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory; import org.apache.ignite.internal.processors.query.calcite.util.TypeUtils; import org.apache.ignite.internal.util.typedef.F; -import org.junit.Assert; import org.junit.Before; import org.junit.Test; @@ -65,12 +64,15 @@ public class WindowExecutionTest extends AbstractExecutionTest { public void testRowNumber() { checkWindow(rowNumber(), true, new Object[][] {{1}, {2}, {1}, {2}, {3}, {1}}); + checkWindow(rowNumber(), false, + new Object[][] {{1}, {2}, {1}, {2}, {3}, {1}}); } /** dense_rank() over (partition by {0} order by {1}). */ @Test public void testDenseRank() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.DENSE_RANK, relIntType, F.asList(), @@ -90,12 +92,15 @@ public void testDenseRank() { checkWindow(grp, true, new Object[][] {{1}, {1}, {1}, {2}, {3}, {1}}); + checkWindow(grp, false, + new Object[][] {{1}, {1}, {1}, {2}, {3}, {1}}); } /** rank() over (partition by {0} order by {1}). */ @Test public void testRank() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.RANK, relIntType, F.asList(), @@ -115,12 +120,15 @@ public void testRank() { checkWindow(grp, true, new Object[][] {{1}, {1}, {1}, {2}, {3}, {1}}); + checkWindow(grp, false, + new Object[][] {{1}, {1}, {1}, {2}, {3}, {1}}); } /** percent_rank() over (partition by {0} order by {1}). */ @Test public void testRercentRank() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.PERCENT_RANK, relDoubleType, F.asList(), @@ -146,6 +154,7 @@ public void testRercentRank() { @Test public void testCumeDist() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.CUME_DIST, relDoubleType, F.asList(), @@ -171,6 +180,7 @@ public void testCumeDist() { @Test public void testFirstValue() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.FIRST_VALUE, relIntType, F.asList(rexBuilder.makeInputRef(relIntType, 1)), @@ -196,6 +206,7 @@ public void testFirstValue() { @Test public void testLastValue() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.LAST_VALUE, relIntType, F.asList(rexBuilder.makeInputRef(relIntType, 2)), @@ -221,6 +232,7 @@ public void testLastValue() { @Test public void testNTile() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.NTILE, relIntType, F.asList(rexBuilder.makeInputRef(relIntType, 3)), @@ -246,6 +258,7 @@ public void testNTile() { @Test public void testNth() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.NTH_VALUE, typeFactory.createTypeWithNullability(relIntType, true), F.asList( @@ -274,6 +287,7 @@ public void testNth() { @Test public void testLag1() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, IgniteOwnSqlOperatorTable.LAG, typeFactory.createTypeWithNullability(relIntType, true), F.asList(rexBuilder.makeInputRef(relIntType, 2)), @@ -299,6 +313,7 @@ public void testLag1() { @Test public void testLag2() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, IgniteOwnSqlOperatorTable.LAG, typeFactory.createTypeWithNullability(relIntType, true), F.asList( @@ -327,6 +342,7 @@ public void testLag2() { @Test public void testLag3() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, IgniteOwnSqlOperatorTable.LAG, typeFactory.createTypeWithNullability(relIntType, true), F.asList( @@ -356,6 +372,7 @@ public void testLag3() { @Test public void testLead1() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, IgniteOwnSqlOperatorTable.LEAD, typeFactory.createTypeWithNullability(relIntType, true), F.asList(rexBuilder.makeInputRef(relIntType, 2)), @@ -381,6 +398,7 @@ public void testLead1() { @Test public void testLead2() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, IgniteOwnSqlOperatorTable.LEAD, typeFactory.createTypeWithNullability(relIntType, true), F.asList( @@ -409,6 +427,7 @@ public void testLead2() { @Test public void testLead3() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, IgniteOwnSqlOperatorTable.LEAD, typeFactory.createTypeWithNullability(relIntType, true), F.asList( @@ -439,6 +458,8 @@ public void testLead3() { public void testCountRowsBetweenUnboundedPrescendingAndCurrentRow() { checkWindow(count(true, UNBOUNDED_PRECEDING, CURRENT_ROW), true, new Object[][] {{1}, {2}, {1}, {2}, {3}, {1}}); + checkWindow(count(true, UNBOUNDED_PRECEDING, CURRENT_ROW), false, + new Object[][] {{1}, {2}, {1}, {2}, {3}, {1}}); } /** count(*) over (partition by {0} rows between unbounded prescending and unbounded following). */ @@ -480,6 +501,8 @@ public void testCountRangeBetween2PrescendingAnd1Following() { public void testSumRowsAndRowNumberToCurrentRow() { checkWindow(sumRowsAndRowNumber(CURRENT_ROW), true, new Object[][] {{1, 1}, {2, 2}, {2, 1}, {4, 2}, {6, 3}, {3, 1}}); + checkWindow(sumRowsAndRowNumber(CURRENT_ROW), false, + new Object[][] {{1, 1}, {2, 2}, {2, 1}, {4, 2}, {6, 3}, {3, 1}}); } /** @@ -530,9 +553,8 @@ private void checkWindow( Node input, Object[][] expRes ) { - Assert.assertEquals(streaming, WindowFunctions.streamable(grp)); - WindowNode window = createWindowNode(ctx, grp, input); + Node window = createWindowNode(ctx, grp, input, streaming); int resFldShift = input.rowType().getFieldCount(); @@ -550,7 +572,8 @@ private void checkWindow( } /** */ - private WindowNode createWindowNode(ExecutionContext ctx, Window.Group grp, Node input) { + private Node createWindowNode(ExecutionContext ctx, Window.Group grp, + Node input, boolean streaming) { Class[] outFields = new Class[input.rowType().getFieldCount() + grp.aggCalls.size()]; Arrays.fill(outFields, int.class); RelDataType outRowType = TypeUtils.createRowType(typeFactory, outFields); @@ -560,16 +583,18 @@ private WindowNode createWindowNode(ExecutionContext ctx, Wi List aggCalls = toAggCall(grp); - Supplier> partitionFactory = - new WindowPartitionFactory<>(ctx, grp, aggCalls, input.rowType()); + WindowPartitionFactory partitionFactory = new WindowPartitionFactory<>(ctx); + + SingleNode window; + if (streaming) { + StreamingWindowPartition partition = partitionFactory.newStreamingPartition(grp, aggCalls, input.rowType()); + window = new StreamingWindowNode<>(ctx, outRowType, partCmp, partition, rowFactory()); + } + else { + BufferingWindowPartition partition = partitionFactory.newBufferingPartition(grp, aggCalls, input.rowType()); + window = new BufferingWindowNode<>(ctx, outRowType, partCmp, partition, rowFactory()); + } - WindowNode window = new WindowNode<>( - ctx, - outRowType, - partCmp, - partitionFactory, - rowFactory() - ); window.register(input); return window; } @@ -632,6 +657,7 @@ private Node createSeqInputNode(ExecutionContext ctx, int si /** row_number() over (partition by {0}). */ private static Window.Group rowNumber() { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.ROW_NUMBER, relIntType, F.asList(), @@ -655,6 +681,7 @@ private static Window.Group rowNumber() { */ private static Window.Group count(boolean rows, RexWindowBound lower, RexWindowBound upper) { Window.RexWinAggCall aggCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.COUNT, relIntType, F.asList(rexBuilder.makeInputRef(relIntType, 0)), @@ -679,6 +706,7 @@ private static Window.Group count(boolean rows, RexWindowBound lower, RexWindowB */ private static Window.Group sumRowsAndRowNumber(RexWindowBound upper) { Window.RexWinAggCall sumCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.SUM, relIntType, F.asList(rexBuilder.makeInputRef(relIntType, 0)), @@ -687,6 +715,7 @@ private static Window.Group sumRowsAndRowNumber(RexWindowBound upper) { false ); Window.RexWinAggCall rowNumberCall = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.ROW_NUMBER, relIntType, F.asList(),