You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2025/03/27 09:28:07 UTC
(ignite-3) 02/03: Make nodes produce batches
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-24923
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit d8db4d3dad58d84b437a9faac0cc36d9161783c6
Author: amashenkov <am...@apache.org>
AuthorDate: Thu Mar 27 01:17:01 2025 +0300
Make nodes produce batches
---
.../sql/engine/exec/rel/AbstractSetOpNode.java | 17 +--
.../sql/engine/exec/rel/AsyncRootNode.java | 10 +-
.../exec/rel/CorrelatedNestedLoopJoinNode.java | 41 ++---
.../internal/sql/engine/exec/rel/Downstream.java | 12 +-
.../internal/sql/engine/exec/rel/FilterNode.java | 13 +-
.../sql/engine/exec/rel/HashAggregateNode.java | 18 +--
.../internal/sql/engine/exec/rel/HashJoinNode.java | 133 +++++++++++------
.../ignite/internal/sql/engine/exec/rel/Inbox.java | 26 ++--
.../internal/sql/engine/exec/rel/LimitNode.java | 6 +-
.../sql/engine/exec/rel/MergeJoinNode.java | 142 ++++++++++--------
.../internal/sql/engine/exec/rel/ModifyNode.java | 2 +-
.../sql/engine/exec/rel/NestedLoopJoinNode.java | 166 +++++++++++++++------
.../ignite/internal/sql/engine/exec/rel/Node.java | 2 +-
.../internal/sql/engine/exec/rel/ScanNode.java | 12 +-
.../sql/engine/exec/rel/SortAggregateNode.java | 17 ++-
.../internal/sql/engine/exec/rel/SortNode.java | 33 ++--
.../sql/engine/exec/rel/StorageScanNode.java | 20 +--
.../sql/engine/exec/rel/TableSpoolNode.java | 16 +-
.../internal/sql/engine/exec/TestDownstream.java | 4 +-
.../sql/engine/exec/rel/AbstractNodeSelfTest.java | 3 +-
.../sql/engine/exec/rel/LimitExecutionTest.java | 5 +-
21 files changed, 436 insertions(+), 262 deletions(-)
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractSetOpNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractSetOpNode.java
index de23c9fdd79..5533312d949 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractSetOpNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractSetOpNode.java
@@ -130,8 +130,10 @@ public abstract class AbstractSetOpNode<RowT> extends AbstractNode<RowT> {
protected Downstream<RowT> requestDownstream(int idx) {
return new Downstream<RowT>() {
@Override
- public void push(RowT row) throws Exception {
- AbstractSetOpNode.this.push(row, idx);
+ public void push(List<RowT> rows) throws Exception {
+ for (RowT row : rows) {
+ AbstractSetOpNode.this.push(row, idx);
+ }
}
@Override
@@ -157,15 +159,12 @@ public abstract class AbstractSetOpNode<RowT> extends AbstractNode<RowT> {
if (requested > 0 && !grouping.isEmpty()) {
int toSnd = Math.min(requested, inBufSize - processed);
- for (RowT row : grouping.getRows(toSnd)) {
- requested--;
-
- downstream().push(row);
+ List<RowT> batch = grouping.getRows(toSnd);
+ requested -= batch.size();
- processed++;
- }
+ downstream().push(batch);
- if (processed >= inBufSize && requested > 0) {
+ if (requested > 0 && !grouping.isEmpty()) {
// Allow others to do their job.
this.execute(this::flush);
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AsyncRootNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AsyncRootNode.java
index dcc33bdd1ba..75beb7b6c3b 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AsyncRootNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AsyncRootNode.java
@@ -82,12 +82,16 @@ public class AsyncRootNode<InRowT, OutRowT> implements Downstream<InRowT>, Async
/** {@inheritDoc} */
@Override
- public void push(InRowT row) throws Exception {
+ public void push(List<InRowT> rows) throws Exception {
assert waiting > 0 : waiting;
- buff.add(converter.apply(row));
+ waiting -= rows.size();
- if (--waiting == 0) {
+ for (InRowT row : rows) {
+ buff.add(converter.apply(row));
+ }
+
+ if (waiting == 0) {
completePrefetchFuture(null);
flush();
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/CorrelatedNestedLoopJoinNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/CorrelatedNestedLoopJoinNode.java
index c489a96acb6..2c46c50c1f3 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/CorrelatedNestedLoopJoinNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/CorrelatedNestedLoopJoinNode.java
@@ -387,21 +387,14 @@ public class CorrelatedNestedLoopJoinNode<RowT> extends AbstractNode<RowT> {
assert state == State.IDLE;
state = State.IN_LOOP;
- int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && rightIdx < rightInBuf.size()) {
if (leftIdx == leftInBuf.size()) {
leftIdx = 0;
}
while (requested > 0 && leftIdx < leftInBuf.size()) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
-
- return;
- }
-
RowT left = leftInBuf.get(leftIdx);
RowT right = rightInBuf.get(rightIdx);
@@ -412,7 +405,7 @@ public class CorrelatedNestedLoopJoinNode<RowT> extends AbstractNode<RowT> {
RowT row = joinProjection.project(context(), left, right);
- downstream().push(row);
+ batch.add(row);
}
leftIdx++;
@@ -422,6 +415,16 @@ public class CorrelatedNestedLoopJoinNode<RowT> extends AbstractNode<RowT> {
rightInBuf.set(rightIdx++, null);
}
}
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0 && rightIdx < rightInBuf.size()) {
+ execute(this::join);
+
+ return;
+ }
+ }
} finally {
state = State.IDLE;
}
@@ -446,22 +449,26 @@ public class CorrelatedNestedLoopJoinNode<RowT> extends AbstractNode<RowT> {
state = State.IN_LOOP;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && notMatchedIdx < leftInBuf.size()) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
-
- return;
- }
-
requested--;
- downstream().push(joinProjection.project(context(), leftInBuf.get(notMatchedIdx), rightEmptyRow));
+ batch.add(joinProjection.project(context(), leftInBuf.get(notMatchedIdx), rightEmptyRow));
leftMatched.set(notMatchedIdx);
notMatchedIdx = leftMatched.nextClearBit(notMatchedIdx + 1);
}
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0 && notMatchedIdx < leftInBuf.size()) {
+ execute(this::join);
+
+ return;
+ }
+ }
} finally {
state = State.IDLE;
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Downstream.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Downstream.java
index 80bf6b4be06..9ef853ca45c 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Downstream.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Downstream.java
@@ -23,7 +23,7 @@ import java.util.List;
* Represents an abstract data consumer.
*
* <b>Note</b>: except several cases (like consumer node and mailboxes), {@link Node#request(int)},
- * {@link Downstream#push(Object)} and {@link Downstream#end()} methods should be used from one single thread.
+ * {@link Downstream#push(List)} and {@link Downstream#end()} methods should be used from one single thread.
*/
public interface Downstream<RowT> {
/**
@@ -32,14 +32,12 @@ public interface Downstream<RowT> {
* @param row Data row.
*/
@Deprecated
- void push(RowT row) throws Exception;
-
- default void push(List<RowT> batch) throws Exception {
- for (RowT row : batch) {
- push(row);
- }
+ default void push(RowT row) throws Exception {
+ push(List.of(row));
}
+ void push(List<RowT> batch) throws Exception;
+
/**
* Signals that data is over.
*/
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/FilterNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/FilterNode.java
index 6c3fdadd8b6..717766413ac 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/FilterNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/FilterNode.java
@@ -129,16 +129,19 @@ public class FilterNode<RowT> extends AbstractNode<RowT> implements SingleNode<R
private void filter() throws Exception {
inLoop = true;
try {
- int processed = 0;
+ List<RowT> batch = newBatch();
while (requested > 0 && !inBuf.isEmpty()) {
requested--;
- downstream().push(inBuf.remove());
+ batch.add(inBuf.remove());
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- if (processed++ >= inBufSize) {
- // Allow others to do their job.
+ if (requested > 0 && !inBuf.isEmpty()) {
execute(this::filter);
- break;
+ return;
}
}
} finally {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashAggregateNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashAggregateNode.java
index 121b4a45fb7..6e4aaff3881 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashAggregateNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashAggregateNode.java
@@ -188,25 +188,23 @@ public class HashAggregateNode<RowT> extends AbstractNode<RowT> implements Singl
while (requested > 0 && !groupingsQueue.isEmpty()) {
Grouping grouping = groupingsQueue.peek();
- int toSnd = Math.min(requested, inBufSize - processed);
+ int toSnd = Math.min(requested, inBufSize);
- for (RowT row : grouping.getRows(toSnd)) {
- requested--;
- downstream().push(row);
+ List<RowT> rows = grouping.getRows(toSnd);
+ requested -= rows.size();
- processed++;
+ if (grouping.isEmpty()) {
+ groupingsQueue.remove();
}
- if (processed >= inBufSize && requested > 0) {
+ downstream().push(rows);
+
+ if (requested > 0 && !groupingsQueue.isEmpty()) {
// allow others to do their job
this.execute(this::doFlush);
return;
}
-
- if (grouping.isEmpty()) {
- groupingsQueue.remove();
- }
}
} finally {
inLoop = false;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java
index 443f418ac24..dce80cccfc1 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java
@@ -197,6 +197,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
inLoop = true;
int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
// Proceed with next left row, if previous was fully processed.
if (!rightIt.hasNext()) {
@@ -211,6 +212,10 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
// Emits matched rows.
while (requested > 0 && rightIt.hasNext()) {
if (processed++ > inBufSize) {
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ }
+
// Allow others to do their job.
execute(this::join);
@@ -226,7 +231,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
--requested;
RowT row = outputProjection.project(context(), left, right);
- downstream().push(row);
+ batch.add(row);
}
if (!rightIt.hasNext()) {
@@ -234,15 +239,18 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
}
} else {
left = null;
+ processed++;
+ }
+ }
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- return;
- }
- }
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
+ execute(this::join);
+ return;
+ }
}
} finally {
inLoop = false;
@@ -288,6 +296,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
inLoop = true;
int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
// Proceed with next left row, if previous was fully processed.
if (!rightIt.hasNext()) {
@@ -306,19 +315,12 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
if (rightIt.hasNext()) {
// Emits matched rows.
while (requested > 0 && rightIt.hasNext()) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
-
- return;
- }
-
RowT right = rightIt.next();
--requested;
RowT row = outputProjection.project(context(), left, right);
- downstream().push(row);
+ batch.add(row);
}
}
@@ -326,6 +328,16 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
left = null;
}
}
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
+ execute(this::join);
+
+ return;
+ }
+ }
} finally {
inLoop = false;
}
@@ -398,6 +410,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
inLoop = true;
int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
// Proceed with next left row, if previous was fully processed.
if (!rightIt.hasNext()) {
@@ -412,6 +425,10 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
// Emits matched rows.
while (requested > 0 && rightIt.hasNext()) {
if (processed++ > inBufSize) {
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ }
+
// Allow others to do their job.
execute(this::join);
@@ -423,7 +440,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
--requested;
RowT row = outputProjection.project(context(), left, right);
- downstream().push(row);
+ batch.add(row);
}
if (!rightIt.hasNext()) {
@@ -431,15 +448,18 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
}
} else {
left = null;
+ processed++;
+ }
+ }
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- return;
- }
- }
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
+ execute(this::join);
+ return;
+ }
}
} finally {
inLoop = false;
@@ -459,15 +479,19 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
RowT emptyLeft = leftRowFactory.create();
+ List<RowT> batch = newBatch();
while (requested > 0 && rightIt.hasNext()) {
RowT right = rightIt.next();
RowT row = outputProjection.project(context(), emptyLeft, right);
--requested;
- downstream().push(row);
+ batch.add(row);
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- if (processed++ > inBufSize) {
- // Allow others to do their job.
+ if (requested > 0 && rightIt.hasNext()) {
execute(this::join);
return;
@@ -538,6 +562,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
inLoop = true;
int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
// Proceed with next left row, if previous was fully processed.
if (!rightIt.hasNext()) {
@@ -557,6 +582,10 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
// Emits matched rows.
while (requested > 0 && rightIt.hasNext()) {
if (processed++ > inBufSize) {
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ }
+
// Allow others to do their job.
execute(this::join);
@@ -568,7 +597,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
--requested;
RowT row = outputProjection.project(context(), left, right);
- downstream().push(row);
+ batch.add(row);
}
if (!rightIt.hasNext()) {
@@ -576,15 +605,18 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
}
} else {
left = null;
+ processed++;
+ }
+ }
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- return;
- }
- }
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
+ execute(this::join);
+ return;
+ }
}
} finally {
inLoop = false;
@@ -604,15 +636,19 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
RowT emptyLeft = leftRowFactory.create();
+ List<RowT> batch = newBatch();
while (requested > 0 && rightIt.hasNext()) {
RowT right = rightIt.next();
RowT row = outputProjection.project(context(), emptyLeft, right);
--requested;
- downstream().push(row);
+ batch.add(row);
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- if (processed++ > inBufSize) {
- // Allow others to do their job.
+ if (requested > 0 && rightIt.hasNext()) {
execute(this::join);
return;
@@ -675,6 +711,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
inLoop = true;
int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
// Proceed with next left row, if previous was fully processed.
if (!rightIt.hasNext()) {
@@ -698,6 +735,10 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
}
if (processed++ > inBufSize) {
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ }
+
// Allow others to do their job.
execute(this::join);
@@ -710,22 +751,25 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
if (anyMatched) {
requested--;
- downstream().push(left);
+ batch.add(left);
rightIt = Collections.emptyIterator();
}
if (!rightIt.hasNext()) {
left = null;
+ processed++;
}
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- if (processed++ > inBufSize) {
- // Allow others to do their job.
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
execute(this::join);
return;
}
-
}
} finally {
inLoop = false;
@@ -760,8 +804,8 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
protected void join() throws Exception {
if (waitingRight == NOT_WAITING) {
inLoop = true;
- int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
left = leftInBuf.remove();
@@ -770,13 +814,16 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
if (rightRows.isEmpty()) {
requested--;
- downstream().push(left);
+ batch.add(left);
}
left = null;
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- if (processed++ > inBufSize) {
- // Allow others to do their job.
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
execute(this::join);
return;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
index 3501dc307e9..34a98b0dcdd 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
@@ -244,15 +244,15 @@ public class Inbox<RowT> extends AbstractNode<RowT> implements Mailbox<RowT>, Si
}
}
- int processed = 0;
inLoop = true;
try {
+ List<RowT> batch = newBatch();
loop:
while (requested > 0 && !heap.isEmpty()) {
RemoteSource<RowT> source = heap.poll().right;
requested--;
- downstream().push(source.remove());
+ batch.add(source.remove());
State state = source.check();
@@ -270,10 +270,13 @@ public class Inbox<RowT> extends AbstractNode<RowT> implements Mailbox<RowT>, Si
default:
throw unexpected(state);
}
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- if (processed++ >= inBufSize) {
- // Allow others to do their job.
- execute(this::push);
+ if (requested > 0 && !heap.isEmpty()) {
+ execute(this::pushOrdered);
return;
}
@@ -296,9 +299,9 @@ public class Inbox<RowT> extends AbstractNode<RowT> implements Mailbox<RowT>, Si
int idx = 0;
int noProgress = 0;
- int processed = 0;
inLoop = true;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && !remoteSources.isEmpty()) {
RemoteSource<RowT> source = remoteSources.get(idx);
@@ -310,7 +313,7 @@ public class Inbox<RowT> extends AbstractNode<RowT> implements Mailbox<RowT>, Si
case READY:
noProgress = 0;
requested--;
- downstream().push(source.remove());
+ batch.add(source.remove());
break;
case WAITING:
@@ -329,10 +332,13 @@ public class Inbox<RowT> extends AbstractNode<RowT> implements Mailbox<RowT>, Si
if (idx == remoteSources.size()) {
idx = 0;
}
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- if (processed++ >= inBufSize) {
- // Allow others to do their job.
- execute(this::push);
+ if (requested > 0 && !remoteSources.isEmpty()) {
+ execute(this::pushUnordered);
return;
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/LimitNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/LimitNode.java
index 0f7fa951887..9eb7910346a 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/LimitNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/LimitNode.java
@@ -113,12 +113,12 @@ public class LimitNode<RowT> extends AbstractNode<RowT> implements SingleNode<Ro
waiting -= batch.size();
- int skip = rowsProcessed > offset ? 0 : offset - rowsProcessed;
+ int skip = offset > rowsProcessed ? offset - rowsProcessed : 0;
int count = fetchNode == null ? batch.size() - skip : Math.min(fetch + offset - rowsProcessed, batch.size() - skip);
- rowsProcessed += count;
+ rowsProcessed += batch.size();
- if(count > 0) {
+ if (count > 0) {
downstream().push(batch.subList(skip, skip + count));
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinNode.java
index 9ccbc0416e6..0cc0b9d6b96 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinNode.java
@@ -328,19 +328,11 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
/** {@inheritDoc} */
@Override
protected void join() throws Exception {
- int processed = 0;
inLoop = true;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty()) && (right != null || !rightInBuf.isEmpty()
|| rightMaterialization != null)) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
-
- return;
- }
-
-
if (left == null) {
left = leftInBuf.remove();
}
@@ -426,7 +418,17 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
}
requested--;
- downstream().push(row);
+ batch.add(row);
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0) {
+ execute(this::join);
+
+ return;
+ }
}
} finally {
inLoop = false;
@@ -504,18 +506,11 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
/** {@inheritDoc} */
@Override
protected void join() throws Exception {
- int processed = 0;
inLoop = true;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty()) && (right != null || !rightInBuf.isEmpty()
|| rightMaterialization != null || waitingRight == NOT_WAITING)) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
-
- return;
- }
-
if (left == null) {
left = leftInBuf.remove();
@@ -545,7 +540,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
row = outputProjection.project(context(), left, rightRowFactory.create());
requested--;
- downstream().push(row);
+ batch.add(row);
left = null;
@@ -559,7 +554,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
row = outputProjection.project(context(), left, rightRowFactory.create());
requested--;
- downstream().push(row);
+ batch.add(row);
}
left = null;
@@ -623,7 +618,17 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
}
requested--;
- downstream().push(row);
+ batch.add(row);
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0) {
+ execute(this::join);
+
+ return;
+ }
}
} finally {
inLoop = false;
@@ -699,18 +704,11 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
/** {@inheritDoc} */
@Override
protected void join() throws Exception {
- int processed = 0;
inLoop = true;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && !(left == null && leftInBuf.isEmpty() && waitingLeft != NOT_WAITING)
&& (right != null || !rightInBuf.isEmpty() || rightMaterialization != null)) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
-
- return;
- }
-
if (left == null && !leftInBuf.isEmpty()) {
left = leftInBuf.remove();
}
@@ -741,7 +739,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
row = outputProjection.project(context(), leftRowFactory.create(), right);
requested--;
- downstream().push(row);
+ batch.add(row);
}
right = null;
@@ -765,7 +763,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
row = outputProjection.project(context(), leftRowFactory.create(), right);
requested--;
- downstream().push(row);
+ batch.add(row);
}
right = null;
@@ -830,7 +828,17 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
}
requested--;
- downstream().push(row);
+ batch.add(row);
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0) {
+ execute(this::join);
+
+ return;
+ }
}
} finally {
inLoop = false;
@@ -915,17 +923,11 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
/** {@inheritDoc} */
@Override
protected void join() throws Exception {
- int processed = 0;
inLoop = true;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && !(left == null && leftInBuf.isEmpty() && waitingLeft != NOT_WAITING)
&& !(right == null && rightInBuf.isEmpty() && rightMaterialization == null && waitingRight != NOT_WAITING)) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
-
- return;
- }
if (left == null && !leftInBuf.isEmpty()) {
left = leftInBuf.remove();
@@ -960,7 +962,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
row = outputProjection.project(context(), leftRowFactory.create(), right);
requested--;
- downstream().push(row);
+ batch.add(row);
}
right = null;
@@ -973,7 +975,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
row = outputProjection.project(context(), left, rightRowFactory.create());
requested--;
- downstream().push(row);
+ batch.add(row);
}
left = null;
@@ -991,7 +993,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
row = outputProjection.project(context(), left, rightRowFactory.create());
requested--;
- downstream().push(row);
+ batch.add(row);
}
left = null;
@@ -1007,7 +1009,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
row = outputProjection.project(context(), leftRowFactory.create(), right);
requested--;
- downstream().push(row);
+ batch.add(row);
}
right = null;
@@ -1075,7 +1077,17 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
}
requested--;
- downstream().push(row);
+ batch.add(row);
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0) {
+ execute(this::join);
+
+ return;
+ }
}
} finally {
inLoop = false;
@@ -1125,17 +1137,10 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
/** {@inheritDoc} */
@Override
protected void join() throws Exception {
- int processed = 0;
inLoop = true;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty()) && (right != null || !rightInBuf.isEmpty())) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
-
- return;
- }
-
if (left == null) {
left = leftInBuf.remove();
}
@@ -1157,10 +1162,20 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
}
requested--;
- downstream().push(left);
+ batch.add(left);
left = null;
}
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0) {
+ execute(this::join);
+
+ return;
+ }
+ }
} finally {
inLoop = false;
}
@@ -1209,18 +1224,11 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
/** {@inheritDoc} */
@Override
protected void join() throws Exception {
- int processed = 0;
inLoop = true;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())
&& !(right == null && rightInBuf.isEmpty() && waitingRight != NOT_WAITING)) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
-
- return;
- }
-
if (left == null) {
left = leftInBuf.remove();
}
@@ -1244,10 +1252,20 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
}
requested--;
- downstream().push(left);
+ batch.add(left);
left = null;
}
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0) {
+ execute(this::join);
+
+ return;
+ }
+ }
} finally {
inLoop = false;
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java
index 03025c67e61..88938fce9c3 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java
@@ -243,7 +243,7 @@ public class ModifyNode<RowT> extends AbstractNode<RowT> implements SingleNode<R
assert downstream() != null;
if (waiting == NOT_WAITING && requested > 0 && !inFlightUpdate && rows.isEmpty()) {
- downstream().push(context().rowHandler().factory(MODIFY_RESULT).create(updatedRows));
+ downstream().push(List.of(context().rowHandler().factory(MODIFY_RESULT).create(updatedRows)));
requested = 0;
downstream().end();
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java
index 1d1f8319de3..38826f20f02 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java
@@ -209,6 +209,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
inLoop = true;
int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
if (left == null) {
left = leftInBuf.remove();
@@ -216,6 +217,9 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
while (requested > 0 && rightIdx < rightMaterialized.size()) {
if (processed++ > inBufSize) {
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ }
// Allow others to do their job.
execute(this::join);
@@ -228,7 +232,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
requested--;
RowT row = outputProjection.project(context(), left, rightMaterialized.get(rightIdx - 1));
- downstream().push(row);
+ batch.add(row);
}
if (rightIdx == rightMaterialized.size()) {
@@ -236,6 +240,16 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
rightIdx = 0;
}
}
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
+ execute(this::join);
+
+ return;
+ }
+ }
} finally {
inLoop = false;
}
@@ -291,6 +305,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
inLoop = true;
int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
if (left == null) {
left = leftInBuf.remove();
@@ -300,6 +315,9 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
while (requested > 0 && rightIdx < rightMaterialized.size()) {
if (processed++ > inBufSize) {
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ }
// Allow others to do their job.
execute(this::join);
@@ -314,7 +332,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
matched = true;
RowT row = outputProjection.project(context(), left, rightMaterialized.get(rightIdx - 1));
- downstream().push(row);
+ batch.add(row);
}
if (rightIdx == rightMaterialized.size()) {
@@ -324,9 +342,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
requested--;
wasPushed = true;
- downstream().push(outputProjection.project(context(), left, rightRowFactory.create()));
-
- processed++;
+ batch.add(outputProjection.project(context(), left, rightRowFactory.create()));
}
if (matched || wasPushed) {
@@ -334,9 +350,12 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
rightIdx = 0;
}
}
+ }
- if (processed > inBufSize) {
- // Allow others to do their job.
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
execute(this::join);
return;
@@ -424,6 +443,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
inLoop = true;
int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
if (left == null) {
left = leftInBuf.remove();
@@ -431,6 +451,9 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
while (requested > 0 && rightIdx < rightMaterialized.size()) {
if (processed++ > inBufSize) {
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ }
// Allow others to do their job.
execute(this::join);
@@ -447,7 +470,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
rightNotMatchedIndexes.clear(rightIdx - 1);
RowT joined = outputProjection.project(context(), left, right);
- downstream().push(joined);
+ batch.add(joined);
}
if (rightIdx == rightMaterialized.size()) {
@@ -455,6 +478,16 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
rightIdx = 0;
}
}
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
+ execute(this::join);
+
+ return;
+ }
+ }
} finally {
inLoop = false;
}
@@ -466,19 +499,23 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
inLoop = true;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && rightNotMatchedIt.hasNext()) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
-
- return;
- }
-
int rowIdx = rightNotMatchedIt.nextInt();
RowT row = outputProjection.project(context(), leftRowFactory.create(), rightMaterialized.get(rowIdx));
requested--;
- downstream().push(row);
+ batch.add(row);
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0 && rightNotMatchedIt.hasNext()) {
+ execute(this::join);
+
+ return;
+ }
}
} finally {
inLoop = false;
@@ -559,6 +596,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
inLoop = true;
int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
if (left == null) {
left = leftInBuf.remove();
@@ -568,6 +606,9 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
while (requested > 0 && rightIdx < rightMaterialized.size()) {
if (processed++ > inBufSize) {
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ }
// Allow others to do their job.
execute(this::join);
@@ -585,7 +626,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
rightNotMatchedIndexes.clear(rightIdx - 1);
RowT joined = outputProjection.project(context(), left, right);
- downstream().push(joined);
+ batch.add(joined);
}
if (rightIdx == rightMaterialized.size()) {
@@ -595,7 +636,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
requested--;
wasPushed = true;
- downstream().push(outputProjection.project(context(), left, rightRowFactory.create()));
+ batch.add(outputProjection.project(context(), left, rightRowFactory.create()));
processed++;
}
@@ -604,9 +645,12 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
rightIdx = 0;
}
}
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- if (processed >= inBufSize) {
- // Allow others to do their job.
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
execute(this::join);
return;
@@ -623,15 +667,19 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
inLoop = true;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && rightNotMatchedIt.hasNext()) {
int rowIdx = rightNotMatchedIt.nextInt();
RowT row = outputProjection.project(context(), leftRowFactory.create(), rightMaterialized.get(rowIdx));
requested--;
- downstream().push(row);
+ batch.add(row);
+ }
- if (processed++ >= inBufSize) {
- // Allow others to do their job.
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+
+ if (requested > 0 && rightNotMatchedIt.hasNext()) {
execute(this::join);
return;
@@ -696,36 +744,55 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
@Override
protected void join() throws Exception {
if (waitingRight == NOT_WAITING) {
- while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
- if (left == null) {
- left = leftInBuf.remove();
- }
+ inLoop = true;
+ try {
+ int processed = 0;
+ List<RowT> batch = newBatch();
+ while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
+ if (left == null) {
+ left = leftInBuf.remove();
+ }
- boolean matched = false;
+ boolean matched = false;
- int processed = 0;
- while (!matched && requested > 0 && rightIdx < rightMaterialized.size()) {
- if (processed++ > inBufSize) {
- // Allow others to do their job.
- execute(this::join);
+ while (!matched && requested > 0 && rightIdx < rightMaterialized.size()) {
+ if (processed++ > inBufSize) {
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ }
+ // Allow others to do their job.
+ execute(this::join);
- return;
+ return;
+ }
+
+ if (!cond.test(left, rightMaterialized.get(rightIdx++))) {
+ continue;
+ }
+
+ requested--;
+ batch.add(left);
+
+ matched = true;
}
- if (!cond.test(left, rightMaterialized.get(rightIdx++))) {
- continue;
+ if (matched || rightIdx == rightMaterialized.size()) {
+ left = null;
+ rightIdx = 0;
}
+ }
- requested--;
- downstream().push(left);
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- matched = true;
- }
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
+ execute(this::join);
- if (matched || rightIdx == rightMaterialized.size()) {
- left = null;
- rightIdx = 0;
+ return;
+ }
}
+ } finally {
+ inLoop = false;
}
}
@@ -760,6 +827,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
inLoop = true;
int processed = 0;
try {
+ List<RowT> batch = newBatch();
while (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
if (left == null) {
left = leftInBuf.remove();
@@ -769,6 +837,9 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
while (rightIdx < rightMaterialized.size()) {
if (processed++ > inBufSize) {
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ }
// Allow others to do their job.
execute(this::join);
@@ -783,14 +854,17 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
if (!matched) {
requested--;
- downstream().push(left);
+ batch.add(left);
}
left = null;
rightIdx = 0;
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
- if (rightMaterialized.isEmpty() && processed++ >= inBufSize) {
- // Allow others to do their job.
+ if (requested > 0 && (left != null || !leftInBuf.isEmpty())) {
execute(this::join);
return;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Node.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Node.java
index 9f9cfa0961f..b4ea0749592 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Node.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Node.java
@@ -25,7 +25,7 @@ import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
* Represents a node of execution tree.
*
* <b>Note</b>: except several cases (like consumer node and mailboxes), {@link Node#request(int)}, {@link Node#close()},
- * {@link Downstream#push(Object)} and {@link Downstream#end()} methods should be used from one single thread.
+ * {@link Downstream#push(List)} and {@link Downstream#end()} methods should be used from one single thread.
*/
public interface Node<RowT> extends AutoCloseable {
/**
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ScanNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ScanNode.java
index 8b5d6f59489..899ca10c82f 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ScanNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ScanNode.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.sql.engine.exec.rel;
+import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
@@ -109,15 +110,16 @@ public class ScanNode<RowT> extends AbstractNode<RowT> implements SingleNode<Row
inst = func.createInstance(context());
}
- int processed = 0;
+ ArrayList<RowT> batch = new ArrayList<>(inBufSize);
while (requested > 0 && inst.hasNext()) {
requested--;
- downstream().push(inst.next());
+ batch.add(inst.next());
+ }
- if (++processed == inBufSize) {
- // allow others to do their job
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ if (requested > 0) {
this.execute(this::push);
-
return;
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/SortAggregateNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/SortAggregateNode.java
index a7af91f6128..e6413272f06 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/SortAggregateNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/SortAggregateNode.java
@@ -244,10 +244,21 @@ public class SortAggregateNode<RowT> extends AbstractNode<RowT> implements Singl
private void flush() throws Exception {
inLoop = true;
try {
- while (requested > 0 && !outBuf.isEmpty()) {
- requested--;
+ if (requested > 0 && !outBuf.isEmpty()) {
+ List<RowT> batch = newBatch(Math.min(inBufSize, outBuf.size()));
+ while (requested > 0 && !outBuf.isEmpty()) {
+ requested--;
- downstream().push(outBuf.poll());
+ batch.add(outBuf.poll());
+ }
+
+ downstream().push(batch);
+
+ if (requested > 0 && !outBuf.isEmpty()) {
+ execute(this::flush);
+
+ return;
+ }
}
} finally {
inLoop = false;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/SortNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/SortNode.java
index 3f9ba93968c..f2c78c0ecdf 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/SortNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/SortNode.java
@@ -19,10 +19,11 @@ package org.apache.ignite.internal.sql.engine.exec.rel;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
-import java.util.ArrayList;
+import java.util.ArrayDeque;
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;
+import java.util.Queue;
import java.util.function.Supplier;
import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
import org.apache.ignite.internal.util.BoundedPriorityQueue;
@@ -47,7 +48,7 @@ public class SortNode<RowT> extends AbstractNode<RowT> implements SingleNode<Row
private final int limit;
/** Reverse-ordered rows in case of limited sort. */
- private List<RowT> reversed;
+ private ArrayDeque<RowT> reversed;
/**
* Constructor.
@@ -90,9 +91,7 @@ public class SortNode<RowT> extends AbstractNode<RowT> implements SingleNode<Row
requested = 0;
waiting = 0;
rows.clear();
- if (reversed != null) {
- reversed.clear();
- }
+ reversed = null;
}
/** {@inheritDoc} */
@@ -173,11 +172,11 @@ public class SortNode<RowT> extends AbstractNode<RowT> implements SingleNode<Row
// Prepare final order (reversed).
if (limit > 0 && !rows.isEmpty()) {
if (reversed == null) {
- reversed = new ArrayList<>(rows.size());
+ reversed = new ArrayDeque<>(rows.size());
}
while (!rows.isEmpty()) {
- reversed.add(rows.poll());
+ reversed.addFirst(rows.poll());
if (++processed >= inBufSize) {
// Allow the others to do their job.
@@ -186,24 +185,28 @@ public class SortNode<RowT> extends AbstractNode<RowT> implements SingleNode<Row
return;
}
}
-
- processed = 0;
}
- while (requested > 0 && !(reversed == null ? rows.isEmpty() : reversed.isEmpty())) {
- requested--;
+ Queue<RowT> queue = reversed == null ? rows : reversed;
+
+ if (requested > 0 && !queue.isEmpty()) {
+ List<RowT> batch = newBatch();
+ while (requested > 0 && !queue.isEmpty()) {
+ requested--;
+
+ batch.add(queue.poll());
+ }
- downstream().push(reversed == null ? rows.poll() : reversed.remove(reversed.size() - 1));
+ downstream().push(batch);
- if (++processed >= inBufSize && requested > 0) {
- // allow others to do their job
+ if (requested > 0 && !queue.isEmpty()) {
this.execute(this::flush);
return;
}
}
- if (reversed == null ? rows.isEmpty() : reversed.isEmpty()) {
+ if (queue.isEmpty()) {
if (requested > 0) {
downstream().end();
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/StorageScanNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/StorageScanNode.java
index 41e5494f059..9bc47733207 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/StorageScanNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/StorageScanNode.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.sql.engine.exec.rel;
+import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.Flow;
@@ -121,17 +122,10 @@ public abstract class StorageScanNode<RowT> extends AbstractNode<RowT> {
private void push() throws Exception {
if (requested > 0 && !inBuff.isEmpty()) {
- int processed = 0;
inLoop = true;
try {
+ List<RowT> batch = new ArrayList<>(Math.min(requested, inBufSize));
while (requested > 0 && !inBuff.isEmpty()) {
- if (processed++ >= inBufSize) {
- // Allow others to do their job.
- execute(this::push);
-
- return;
- }
-
RowT row = inBuff.poll();
if (filters != null && !filters.test(row)) {
@@ -143,7 +137,15 @@ public abstract class StorageScanNode<RowT> extends AbstractNode<RowT> {
}
requested--;
- downstream().push(row);
+ batch.add(row);
+ }
+
+ if (!batch.isEmpty()) {
+ downstream().push(batch);
+ if (requested > 0) {
+ execute(this::push);
+ return;
+ }
}
} finally {
inLoop = false;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/TableSpoolNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/TableSpoolNode.java
index 9453c943197..4276f6b5afd 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/TableSpoolNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/TableSpoolNode.java
@@ -112,18 +112,16 @@ public class TableSpoolNode<RowT> extends AbstractNode<RowT> implements SingleNo
int processed = 0;
inLoop = true;
try {
- while (requested > 0 && rowIdx < rows.size()) {
- if (processed++ >= inBufSize) {
- // Allow others to do their job
- this.execute(this::doPush);
+ int count = Math.min(requested, rows.size() - rowIdx);
- return;
- }
+ if (count > 0) {
+ List<RowT> batch = newBatch(count);
+ batch.addAll(rows.subList(rowIdx, rowIdx + count));
- downstream().push(rows.get(rowIdx));
+ requested -= count;
+ rowIdx += count;
- rowIdx++;
- requested--;
+ downstream().push(batch);
}
} finally {
inLoop = false;
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TestDownstream.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TestDownstream.java
index a0eaf8433f3..156f2f98208 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TestDownstream.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TestDownstream.java
@@ -34,8 +34,8 @@ public class TestDownstream<T> implements Downstream<T> {
private final CompletableFuture<List<T>> completion = new CompletableFuture<>();
@Override
- public void push(T row) {
- rows.add(row);
+ public void push(List<T> batch) {
+ rows.addAll(batch);
}
@Override
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractNodeSelfTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractNodeSelfTest.java
index 934e6048b6a..598eb353578 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractNodeSelfTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractNodeSelfTest.java
@@ -24,6 +24,7 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
@@ -143,7 +144,7 @@ public class AbstractNodeSelfTest extends BaseIgniteAbstractTest {
}
@Override
- public void push(T row) {
+ public void push(List<T> batch) {
throw new UnsupportedOperationException();
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/LimitExecutionTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/LimitExecutionTest.java
index b726beaa890..dc51bd14c04 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/LimitExecutionTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/LimitExecutionTest.java
@@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
@@ -210,9 +211,11 @@ public class LimitExecutionTest extends AbstractExecutionTest<Object[]> {
int r = requested.getAndAdd(rowsCnt);
this.execute(() -> {
+ ArrayList<Object[]> batch = new ArrayList<>(rowsCnt);
for (int i = 0; i < rowsCnt; i++) {
- downstream().push(new Object[]{r + i});
+ batch.add(new Object[]{r + i});
}
+ downstream().push(batch);
});
}
}