alex-plekhanov commented on a change in pull request #9095:
URL: https://github.com/apache/ignite/pull/9095#discussion_r640656426
##########
File path:
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MinusNode.java
##########
@@ -274,94 +66,21 @@ private void addOnSingle(Row row, int setIdx) {
}
Review comment:
Didn't get about "comment". Can you please clarify what do you mean?
Fixed logic for "all = false"
##########
File path:
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MinusNode.java
##########
@@ -274,94 +66,21 @@ private void addOnSingle(Row row, int setIdx) {
}
/** */
Review comment:
Fixed
##########
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:
Fixed
##########
File path:
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/IntersectNode.java
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.query.calcite.exec.rel;
+
+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.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;
+
+/**
+ * Execution node for INTERSECT operator.
+ */
+public class IntersectNode<Row> extends AbstractSetOpNode<Row> {
+ /** */
+ public IntersectNode(ExecutionContext<Row> ctx, RelDataType rowType,
AggregateType type, boolean all,
+ RowFactory<Row> rowFactory, int inputsCnt) {
+ super(ctx, rowType, type, all, rowFactory, new
IntersectGrouping<>(ctx, rowFactory, type, all, inputsCnt));
+ }
+
+ /** */
+ private static class IntersectGrouping<Row> extends Grouping<Row> {
+ /** Inputs count. */
+ private final int inputsCnt;
+
+ /** */
+ private IntersectGrouping(ExecutionContext<Row> ctx, RowFactory<Row>
rowFactory, AggregateType type,
+ boolean all, int inputsCnt) {
+ super(ctx, rowFactory, type, all);
+
+ this.inputsCnt = inputsCnt;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void endOfSet(int setIdx) {
+ if (type == AggregateType.SINGLE && rowsCnt == 0)
+ groups.clear();
+
+ super.endOfSet(setIdx);
+ }
+
+ /** */
Review comment:
Fixed
##########
File path:
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/IntersectNode.java
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.query.calcite.exec.rel;
+
+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.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;
+
+/**
+ * Execution node for INTERSECT operator.
+ */
+public class IntersectNode<Row> extends AbstractSetOpNode<Row> {
+ /** */
+ public IntersectNode(ExecutionContext<Row> ctx, RelDataType rowType,
AggregateType type, boolean all,
+ RowFactory<Row> rowFactory, int inputsCnt) {
+ super(ctx, rowType, type, all, rowFactory, new
IntersectGrouping<>(ctx, rowFactory, type, all, inputsCnt));
+ }
+
+ /** */
+ private static class IntersectGrouping<Row> extends Grouping<Row> {
+ /** Inputs count. */
+ private final int inputsCnt;
+
+ /** */
+ private IntersectGrouping(ExecutionContext<Row> ctx, RowFactory<Row>
rowFactory, AggregateType type,
+ boolean all, int inputsCnt) {
+ super(ctx, rowFactory, type, all);
+
+ this.inputsCnt = inputsCnt;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void endOfSet(int setIdx) {
+ if (type == AggregateType.SINGLE && rowsCnt == 0)
+ groups.clear();
+
+ super.endOfSet(setIdx);
+ }
+
+ /** */
+ @Override protected void addOnSingle(Row row, int setIdx) {
+ int[] cntrs;
+
+ GroupKey key = key(row);
+
+ if (setIdx == 0) {
+ cntrs = groups.computeIfAbsent(key, k -> new int[inputsCnt]);
+
+ cntrs[0]++;
+ }
+ else {
+ cntrs = groups.get(key);
+
+ if (cntrs != null) {
+ if (cntrs[setIdx - 1] == 0)
+ groups.remove(key);
+ else
+ cntrs[setIdx]++;
+ }
+ }
+ }
+
+ /** */
Review comment:
Fixed
--
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]
