korlov42 commented on a change in pull request #9095:
URL: https://github.com/apache/ignite/pull/9095#discussion_r640586896
##########
File path:
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MinusNode.java
##########
@@ -17,246 +17,38 @@
package org.apache.ignite.internal.processors.query.calcite.exec.rel;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
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.RowHandler.RowFactory;
import
org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AggregateType;
import
org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.GroupKey;
-import org.apache.ignite.internal.util.typedef.F;
/**
* Execution node for MINUS (EXCEPT) operator.
*/
-public class MinusNode<Row> extends AbstractNode<Row> {
+public class MinusNode<Row> extends AbstractSetOpNode<Row> {
/** */
- private final AggregateType type;
-
- /** */
- private final boolean all;
-
- /** */
- private final RowFactory<Row> rowFactory;
-
- /** */
- private final Grouping grouping;
-
- /** */
- private int requested;
-
- /** */
- private int waiting;
-
- /** Current source index. */
- private int curSrcIdx;
-
- /** */
- private boolean inLoop;
-
- /**
- * @param ctx Execution context.
- */
public MinusNode(ExecutionContext<Row> ctx, RelDataType rowType,
AggregateType type, boolean all,
RowFactory<Row> rowFactory) {
- super(ctx, rowType);
-
- this.all = all;
- this.type = type;
- this.rowFactory = rowFactory;
-
- grouping = new Grouping();
- }
-
- /** {@inheritDoc} */
- @Override public void request(int rowsCnt) throws Exception {
- assert !F.isEmpty(sources());
- assert rowsCnt > 0 && requested == 0;
- assert waiting <= 0;
-
- checkState();
-
- requested = rowsCnt;
-
- if (waiting == 0)
- sources().get(curSrcIdx).request(waiting = IN_BUFFER_SIZE);
- else if (!inLoop)
- context().execute(this::flush, this::onError);
- }
-
- /** */
- public void push(Row row, int idx) throws Exception {
- assert downstream() != null;
- assert waiting > 0;
-
- checkState();
-
- waiting--;
-
- grouping.add(row, idx);
-
- if (waiting == 0)
- sources().get(curSrcIdx).request(waiting = IN_BUFFER_SIZE);
+ super(ctx, rowType, type, all, rowFactory, new MinusGrouping<>(ctx,
rowFactory, type, all));
}
/** */
- public void end(int idx) throws Exception {
- assert downstream() != null;
- assert waiting > 0;
- assert curSrcIdx == idx;
-
- checkState();
-
- if (type == AggregateType.SINGLE && grouping.isEmpty())
- curSrcIdx = sources().size(); // Skip subsequent sources.
- else
- curSrcIdx++;
-
- if (curSrcIdx >= sources().size()) {
- waiting = -1;
-
- flush();
- }
- else
- sources().get(curSrcIdx).request(waiting);
- }
-
- /** {@inheritDoc} */
- @Override protected void rewindInternal() {
- requested = 0;
- waiting = 0;
- grouping.groups.clear();
- }
-
- /** {@inheritDoc} */
- @Override protected Downstream<Row> requestDownstream(int idx) {
- return new Downstream<Row>() {
- @Override public void push(Row row) throws Exception {
- MinusNode.this.push(row, idx);
- }
-
- @Override public void end() throws Exception {
- MinusNode.this.end(idx);
- }
-
- @Override public void onError(Throwable e) {
- MinusNode.this.onError(e);
- }
- };
- }
-
- /** */
- private void flush() throws Exception {
- if (isClosed())
- return;
-
- checkState();
-
- assert waiting == -1;
-
- int processed = 0;
-
- inLoop = true;
-
- try {
- while (requested > 0 && !grouping.isEmpty()) {
- int toSnd = Math.min(requested, IN_BUFFER_SIZE - processed);
-
- for (Row row : grouping.getRows(toSnd)) {
- checkState();
-
- requested--;
- downstream().push(row);
-
- 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 (requested > 0) {
- requested = 0;
-
- downstream().end();
- }
- }
-
- /** */
- private class Grouping {
- /**
- * Value in this map will always have 2 elements, first - count of
keys in the first set, second - count of
- * keys in all sets except first.
- */
- private final Map<GroupKey, int[]> groups = new HashMap<>();
-
- /** */
- private final RowHandler<Row> hnd;
-
- /** */
- private Grouping() {
- hnd = context().rowHandler();
- }
-
+ private static class MinusGrouping<Row> extends Grouping<Row> {
/** */
- private void add(Row row, int setIdx) {
- if (type == AggregateType.REDUCE) {
- assert setIdx == 0 : "Unexpected set index: " + setIdx;
-
- addOnReducer(row);
- }
- else if (type == AggregateType.MAP)
- addOnMapper(row, setIdx);
- else
- addOnSingle(row, setIdx);
- }
-
- /**
- * @param cnt Number of rows.
- *
- * @return Actually sent rows number.
- */
- private List<Row> getRows(int cnt) {
- if (F.isEmpty(groups))
- return Collections.emptyList();
- else if (type == AggregateType.MAP)
- return getOnMapper(cnt);
- else
- return getOnSingleOrReducer(cnt);
+ private MinusGrouping(ExecutionContext<Row> ctx, RowFactory<Row>
rowFactory, AggregateType type, boolean all) {
+ super(ctx, rowFactory, type, all);
}
/** */
Review comment:
```suggestion
/** {@inheritDoc} */
```
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
