Skip to content

Commit 9cc287c

Browse files
ozinovevoleg-zinovev
ozinovev
authored and
oleg-zinovev
committed
IGNITE-28882: window execution refactoring
1 parent b4dc819 commit 9cc287c

14 files changed

Lines changed: 446 additions & 243 deletions

File tree

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java‎

Lines changed: 21 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -51,9 +51,12 @@
5151
import org.apache.ignite.internal.processors.query.calcite.exec.exp.RangeIterable;
5252
import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AccumulatorWrapper;
5353
import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AggregateType;
54-
import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.WindowPartition;
54+
import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.BufferingWindowPartition;
55+
import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.StreamingWindowPartition;
56+
import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.WindowFunctions;
5557
import org.apache.ignite.internal.processors.query.calcite.exec.exp.window.WindowPartitionFactory;
5658
import org.apache.ignite.internal.processors.query.calcite.exec.rel.AbstractSetOpNode;
59+
import org.apache.ignite.internal.processors.query.calcite.exec.rel.BufferingWindowNode;
5760
import org.apache.ignite.internal.processors.query.calcite.exec.rel.CollectNode;
5861
import org.apache.ignite.internal.processors.query.calcite.exec.rel.CorrelatedNestedLoopJoinNode;
5962
import org.apache.ignite.internal.processors.query.calcite.exec.rel.FilterNode;
@@ -73,12 +76,13 @@
7376
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanNode;
7477
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanStorageNode;
7578
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanTableRowNode;
79+
import org.apache.ignite.internal.processors.query.calcite.exec.rel.SingleNode;
7680
import org.apache.ignite.internal.processors.query.calcite.exec.rel.SortAggregateNode;
7781
import org.apache.ignite.internal.processors.query.calcite.exec.rel.SortNode;
82+
import org.apache.ignite.internal.processors.query.calcite.exec.rel.StreamingWindowNode;
7883
import org.apache.ignite.internal.processors.query.calcite.exec.rel.TableSpoolNode;
7984
import org.apache.ignite.internal.processors.query.calcite.exec.rel.UncollectNode;
8085
import org.apache.ignite.internal.processors.query.calcite.exec.rel.UnionAllNode;
81-
import org.apache.ignite.internal.processors.query.calcite.exec.rel.WindowNode;
8286
import org.apache.ignite.internal.processors.query.calcite.metadata.AffinityService;
8387
import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
8488
import org.apache.ignite.internal.processors.query.calcite.prepare.bounds.SearchBounds;
@@ -961,18 +965,28 @@ else if (rel instanceof Intersect)
961965
assert collation.getFieldCollations().size() >= grpKeys.size();
962966
Comparator<Row> partCmp = expressionFactory.comparator(TraitUtils.createCollation(grpKeys));
963967

968+
RowFactory<Row> rowFactory = ctx.rowHandler().factory(ctx.getTypeFactory(), outType);
969+
964970
List<AggregateCall> calls = grp.getAggregateCalls(rel);
965-
Supplier<WindowPartition<Row>> partFactory = new WindowPartitionFactory<>(ctx, grp, calls, inputType);
971+
assert !calls.isEmpty() : "Window aggregate calls should not be empty";
966972

967-
RowFactory<Row> rowFactory = ctx.rowHandler().factory(ctx.getTypeFactory(), outType);
973+
WindowPartitionFactory<Row> factory = new WindowPartitionFactory<>(ctx);
968974

969-
WindowNode<Row> node = new WindowNode<>(ctx, outType, partCmp, partFactory, rowFactory);
975+
SingleNode<Row> windowNode;
976+
if (WindowFunctions.streamable(grp)) {
977+
StreamingWindowPartition<Row> partition = factory.newStreamingPartition(grp, calls, inputType);
978+
windowNode = new StreamingWindowNode<>(ctx, outType, partCmp, partition, rowFactory);
979+
}
980+
else {
981+
BufferingWindowPartition<Row> partition = factory.newBufferingPartition(grp, calls, inputType);
982+
windowNode = new BufferingWindowNode<>(ctx, outType, partCmp, partition, rowFactory);
983+
}
970984

971985
Node<Row> input = visit(rel.getInput());
972986

973-
node.register(input);
987+
windowNode.register(input);
974988

975-
return node;
989+
return windowNode;
976990
}
977991

978992
/** {@inheritDoc} */

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

Lines changed: 120 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -17,26 +17,35 @@
1717

1818
package org.apache.ignite.internal.processors.query.calcite.exec.exp.window;
1919

20-
import java.util.ArrayList;
20+
import java.util.ArrayDeque;
2121
import java.util.Comparator;
22+
import java.util.Deque;
2223
import java.util.List;
23-
import java.util.function.Consumer;
2424
import org.apache.calcite.rel.core.Window;
2525
import org.apache.calcite.rel.type.RelDataType;
2626
import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
2727
import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler;
28-
import org.apache.ignite.internal.processors.query.calcite.exec.tracker.RowTracker;
28+
import org.jetbrains.annotations.Nullable;
2929

3030
/** Buffering implementation of the ROWS / RANGE window partition. */
31-
final class BufferingWindowPartition<Row> extends WindowPartitionBase<Row> {
32-
/** Rows in partition. */
33-
private final List<Row> buf;
31+
public final class BufferingWindowPartition<Row> extends WindowPartitionBase<Row> {
32+
/** */
33+
private final ExecutionContext<Row> ctx;
3434

35-
/** Frame within partition. */
36-
private final WindowPartitionFrame<Row> frame;
35+
/** */
36+
private final Window.Group grp;
3737

3838
/** */
39-
private RowTracker<Row> memoryTracker;
39+
private final RelDataType inputRowType;
40+
41+
/** Slices in partition. */
42+
private final Deque<FrameHolder> frames;
43+
44+
/** Number of rows, can be evaluted right now. */
45+
private int ready;
46+
47+
/** Current frame for row evaluation. */
48+
private FrameHolder currFrame;
4049

4150
/** */
4251
BufferingWindowPartition(
@@ -48,60 +57,58 @@ final class BufferingWindowPartition<Row> extends WindowPartitionBase<Row> {
4857
RelDataType inputRowType
4958
) {
5059
super(peerCmp, funcFactory, rowFactory);
51-
buf = new ArrayList<>();
52-
frame = createFrame(ctx, grp, peerCmp, inputRowType, buf);
53-
}
54-
55-
/** {@inheritDoc} */
56-
@Override public void add(Row row) {
57-
buf.add(row);
58-
onRowAdded(row);
60+
this.ctx = ctx;
61+
this.grp = grp;
62+
this.inputRowType = inputRowType;
63+
frames = new ArrayDeque<>();
5964
}
6065

61-
/** {@inheritDoc} */
62-
@Override public void evalTo(RowHandler.RowFactory<Row> factory, Consumer<Row> output) {
66+
/**
67+
* Appends rows to the partition.
68+
* Important: the buffer must not be modified after being passed to this method.
69+
**/
70+
public void appendPartition(List<Row> buf, Runnable onBufRemoved) {
6371
if (buf.isEmpty())
6472
return;
6573

66-
List<WindowFunctionWrapper<Row>> accumulators = createWrappers();
67-
Object[] accResults = new Object[accumulators.size()];
68-
69-
int size = buf.size();
70-
Row prevRow = null;
71-
int peerIdx = -1;
72-
for (int rowIdx = 0; rowIdx < size; rowIdx++) {
73-
Row currRow = buf.get(rowIdx);
74-
if (isNewPeer(currRow, prevRow))
75-
peerIdx++;
76-
77-
int accIdx = 0;
78-
for (WindowFunctionWrapper<Row> acc : accumulators) {
79-
Object accResult = acc.callBuffering(currRow, rowIdx, peerIdx, frame);
80-
accResults[accIdx++] = accResult;
81-
}
74+
WindowPartitionFrame<Row> frame = createFrame(ctx, grp, peerCmp, inputRowType, buf);
75+
ready += frame.size();
76+
frames.add(new FrameHolder(frame, onBufRemoved));
77+
}
8278

83-
Row resultRow = createResultRow(factory, currRow, accResults);
84-
output.accept(resultRow);
79+
/**
80+
* Evaluates next row in the window partition.
81+
* @return Result row or {@code null} if there are no more rows.
82+
*/
83+
public @Nullable Row nextRow(RowHandler.RowFactory<Row> factory) {
84+
if (currFrame == null || currFrame.consumed()) {
85+
if (currFrame != null)
86+
currFrame.release();
87+
currFrame = frames.pollFirst();
88+
}
8589

86-
prevRow = currRow;
90+
if (currFrame == null) {
91+
assert ready == 0;
92+
return null;
8793
}
88-
}
8994

90-
/** {@inheritDoc} */
91-
@Override public void reset() {
92-
buf.forEach(this::onRowRemoved);
93-
buf.clear();
94-
frame.reset();
95+
assert ready > 0;
96+
Row resultRow = currFrame.nextRow(factory);
97+
ready--;
98+
return resultRow;
9599
}
96100

97-
/** {@inheritDoc} */
98-
@Override public boolean isStreaming() {
99-
return false;
101+
/** Returns the number of rows can be evaluted. */
102+
public int ready() {
103+
return ready;
100104
}
101105

102106
/** {@inheritDoc} */
103-
@Override public void attachMemoryTracker(RowTracker<Row> memoryTracker) {
104-
this.memoryTracker = memoryTracker;
107+
@Override public void reset() {
108+
FrameHolder holder;
109+
while ((holder = frames.poll()) != null)
110+
holder.release();
111+
ready = 0;
105112
}
106113

107114
/** Creates frame for partition. */
@@ -118,15 +125,70 @@ private static <Row> WindowPartitionFrame<Row> createFrame(
118125
return new RangeWindowPartitionFrame<>(buf, ctx, peerCmp, grp, inputRowType);
119126
}
120127

121-
/** Adds row to memory tracker. */
122-
private void onRowAdded(Row row) {
123-
if (memoryTracker != null)
124-
memoryTracker.onRowAdded(row);
125-
}
128+
/** */
129+
private final class FrameHolder {
130+
/** */
131+
private final WindowPartitionFrame<Row> frame;
126132

127-
/** Removes row from memory tracker. */
128-
private void onRowRemoved(Row row) {
129-
if (memoryTracker != null)
130-
memoryTracker.onRowRemoved(row);
133+
/** */
134+
private final Runnable onFrameRemoved;
135+
136+
/** Index of the current row in the frame. */
137+
private int rowIdx;
138+
139+
/** Index of the current peer in the frame. */
140+
private int peerIdx = -1;
141+
142+
/** */
143+
private Row prevRow;
144+
145+
/** */
146+
private List<WindowFunctionWrapper<Row>> accumulators;
147+
148+
/** */
149+
private Object[] accResults;
150+
151+
/** */
152+
private FrameHolder(WindowPartitionFrame<Row> frame, Runnable removed) {
153+
this.frame = frame;
154+
onFrameRemoved = removed;
155+
}
156+
157+
/** */
158+
Row nextRow(RowHandler.RowFactory<Row> factory) {
159+
assert !consumed();
160+
161+
if (accumulators == null) {
162+
accumulators = createWrappers();
163+
accResults = new Object[accumulators.size()];
164+
}
165+
166+
Row currRow = frame.get(rowIdx);
167+
if (isNewPeer(currRow, prevRow))
168+
peerIdx++;
169+
170+
int accIdx = 0;
171+
for (WindowFunctionWrapper<Row> acc : accumulators) {
172+
Object accResult = acc.callBuffering(currRow, rowIdx, peerIdx, currFrame.frame);
173+
accResults[accIdx++] = accResult;
174+
}
175+
176+
Row resultRow = createResultRow(factory, currRow, accResults);
177+
178+
rowIdx++;
179+
prevRow = currRow;
180+
181+
return resultRow;
182+
}
183+
184+
/** */
185+
boolean consumed() {
186+
return rowIdx == frame.size();
187+
}
188+
189+
/** */
190+
void release() {
191+
onFrameRemoved.run();
192+
}
131193
}
132194
}

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

Lines changed: 5 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,7 @@ final class RangeWindowPartitionFrame<Row> extends WindowPartitionFrame<Row> {
103103
if (lowerBoundRow == null)
104104
cachedStartIdx = 0;
105105
else
106-
cachedStartIdx = bsearchBound(lowerBoundRow, buf, true);
106+
cachedStartIdx = bsearchBound(lowerBoundRow, true);
107107

108108
return cachedStartIdx;
109109
}
@@ -121,29 +121,20 @@ final class RangeWindowPartitionFrame<Row> extends WindowPartitionFrame<Row> {
121121
if (upperBoundRow == null)
122122
cachedEndIdx = size() - 1;
123123
else
124-
cachedEndIdx = bsearchBound(upperBoundRow, buf, false);
124+
cachedEndIdx = bsearchBound(upperBoundRow, false);
125125

126126
return cachedEndIdx;
127127
}
128128

129-
/** {@inheritDoc} */
130-
@Override public void reset() {
131-
// Reseting index cache.
132-
cachedStartPeerIdx = -1;
133-
cachedEndPeerIdx = -1;
134-
cachedStartRowIdx = -1;
135-
cachedEndRowIdx = -1;
136-
}
137-
138129
/** Binary search bound. */
139-
private int bsearchBound(Row row, List<Row> buf, boolean lower) {
130+
private int bsearchBound(Row row, boolean lower) {
140131
int start = 0;
141-
int end = buf.size() - 1;
132+
int end = size() - 1;
142133

143134
while (start <= end) {
144135
int mid = (start + end) / 2;
145136

146-
Row midRow = buf.get(mid);
137+
Row midRow = get(mid);
147138
int cmp = compareRowPeer(midRow, row);
148139

149140
if (cmp > 0 || (lower && cmp == 0))

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

Lines changed: 0 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -95,13 +95,6 @@ final class RowWindowPartitionFrame<Row> extends WindowPartitionFrame<Row> {
9595
return applyOffset(rowIdx, cachedEndOffset, size() - 1);
9696
}
9797

98-
/** {@inheritDoc} */
99-
@Override protected void reset() {
100-
// Reseting index cache.
101-
cachedStartRowIdx = -1;
102-
cachedEndRowIdx = -1;
103-
}
104-
10598
/** */
10699
private static int applyOffset(int rowIdx, int offset, int cap) {
107100
int idx = Math.addExact(rowIdx, offset);

0 commit comments

Comments
 (0)