IGNITE-28882: window execution refactoring

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<Row> partCmp = expressionFactory.comparator(TraitUtils.createCollation(grpKeys));
 
+        RowFactory<Row> rowFactory = ctx.rowHandler().factory(ctx.getTypeFactory(), outType);
+
         List<AggregateCall> calls = grp.getAggregateCalls(rel);
-        Supplier<WindowPartition<Row>> partFactory = new WindowPartitionFactory<>(ctx, grp, calls, inputType);
+        assert !calls.isEmpty() : "Window aggregate calls should not be empty";
 
-        RowFactory<Row> rowFactory = ctx.rowHandler().factory(ctx.getTypeFactory(), outType);
+        WindowPartitionFactory<Row> factory = new WindowPartitionFactory<>(ctx);
 
-        WindowNode<Row> node = new WindowNode<>(ctx, outType, partCmp, partFactory, rowFactory);
+        SingleNode<Row> windowNode;
+        if (WindowFunctions.streamable(grp)) {
+            StreamingWindowPartition<Row> partition = factory.newStreamingPartition(grp, calls, inputType);
+            windowNode = new StreamingWindowNode<>(ctx, outType, partCmp, partition, rowFactory);
+        }
+        else {
+            BufferingWindowPartition<Row> partition = factory.newBufferingPartition(grp, calls, inputType);
+            windowNode = new BufferingWindowNode<>(ctx, outType, partCmp, partition, rowFactory);
+        }
 
         Node<Row> input = visit(rel.getInput());
 
-        node.register(input);
+        windowNode.register(input);
 
-        return node;
+        return windowNode;
     }
 
     /** {@inheritDoc} */

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<Row> extends WindowPartitionBase<Row> {
-    /** Rows in partition. */
-    private final List<Row> buf;
+public final class BufferingWindowPartition<Row> extends WindowPartitionBase<Row> {
+    /** */
+    private final ExecutionContext<Row> ctx;
 
-    /** Frame within partition. */
-    private final WindowPartitionFrame<Row> frame;
+    /** */
+    private final Window.Group grp;
 
     /** */
-    private RowTracker<Row> memoryTracker;
+    private final RelDataType inputRowType;
+
+    /** Slices in partition. */
+    private final Deque<FrameHolder> 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<Row> extends WindowPartitionBase<Row> {
         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<Row> factory, Consumer<Row> output) {
+    /**
+     * Appends rows to the partition.
+     * Important: the buffer must not be modified after being passed to this method.
+     **/
+    public void appendPartition(List<Row> buf, Runnable onBufRemoved) {
         if (buf.isEmpty())
             return;
 
-        List<WindowFunctionWrapper<Row>> accumulators = createWrappers();
-        Object[] accResults = new Object[accumulators.size()];
+        WindowPartitionFrame<Row> 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<Row> factory) {
+        if (currFrame == null || currFrame.consumed()) {
+            if (currFrame != null)
+                currFrame.release();
+            currFrame = frames.pollFirst();
+        }
 
-            int accIdx = 0;
-            for (WindowFunctionWrapper<Row> 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<Row> memoryTracker) {
-        this.memoryTracker = memoryTracker;
+        ready = 0;
     }
 
     /** Creates frame for partition. */
@@ -118,15 +131,70 @@ private static <Row> WindowPartitionFrame<Row> 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<Row> 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<WindowFunctionWrapper<Row>> accumulators;
+
+        /** */
+        private Object[] accResults;
+
+        /** */
+        private FrameHolder(WindowPartitionFrame<Row> frame, Runnable removed) {
+            this.frame = frame;
+            onFrameRemoved = removed;
+        }
+
+        /** */
+        Row nextRow(RowHandler.RowFactory<Row> 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<Row> 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();
+        }
     }
 }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/RangeWindowPartitionFrame.java

@@ -103,7 +103,7 @@ final class RangeWindowPartitionFrame<Row> extends WindowPartitionFrame<Row> {
         if (lowerBoundRow == null)
             cachedStartIdx = 0;
         else
-            cachedStartIdx = bsearchBound(lowerBoundRow, buf, true);
+            cachedStartIdx = bsearchBound(lowerBoundRow, true);
 
         return cachedStartIdx;
     }
@@ -121,29 +121,20 @@ final class RangeWindowPartitionFrame<Row> extends WindowPartitionFrame<Row> {
         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<Row> 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))

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/window/RowWindowPartitionFrame.java

@@ -95,13 +95,6 @@ final class RowWindowPartitionFrame<Row> extends WindowPartitionFrame<Row> {
             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);