zstan commented on code in PR #13311:
URL: https://github.com/apache/ignite/pull/13311#discussion_r3704618499
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/LimitNode.java:
##########
@@ -85,38 +90,46 @@ public LimitNode(
/** {@inheritDoc} */
@Override public void push(Row row) throws Exception {
- if (waiting == -1)
+ if (waiting == NOT_WAITING)
return;
- ++rowsProcessed;
-
--waiting;
- checkState();
-
- if (rowsProcessed > offset) {
- if (fetchNode == null || (fetchNode != null && rowsProcessed <=
fetch + offset))
- downstream().push(row);
+ if (rowsProcessed >= offset && hasMoreData()) {
+ // this two rows can`t be swapped, cause if all requested rows
have been pushed it will trigger further request call.
+ --requested;
+ downstream().push(row);
}
- if (fetch > 0 && rowsProcessed == fetch + offset && waiting > 0)
+ ++rowsProcessed;
+
+ // There several cases are possible:
+ // 1) requested = 512, limit = 1, offset = not defined: need to pass
1 row and call end()
+ // 2) requested = 512, limit = 512, offset = not defined: just need
to pass all rows without end() call
+ // 3) requested = 512, limit = 512, offset = 1: need to request
initially 512 and further 1 row
+ if (!hasMoreData() && requested > 0)
end();
+
+ if (waiting == 0 && requested > 0)
+ source().request(waiting = requested);
}
/** {@inheritDoc} */
@Override public void end() throws Exception {
- if (waiting == -1)
+ if (waiting == NOT_WAITING)
return;
assert downstream() != null;
- waiting = -1;
+ waiting = NOT_WAITING;
downstream().end();
Review Comment:
done
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
