You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "ivandasch (via GitHub)" <gi...@apache.org> on 2023/09/12 09:36:27 UTC
[GitHub] [ignite] ivandasch opened a new pull request, #10928: IGNITE-12455 SQL Calcite: Partition pruning
ivandasch opened a new pull request, #10928:
URL: https://github.com/apache/ignite/pull/10928
Thank you for submitting the pull request to the Apache Ignite.
In order to streamline the review of the contribution
we ask you to ensure the following steps have been taken:
### The Contribution Checklist
- [ ] There is a single JIRA ticket related to the pull request.
- [ ] The web-link to the pull request is attached to the JIRA ticket.
- [ ] The JIRA ticket has the _Patch Available_ state.
- [ ] The pull request body describes changes that have been made.
The description explains _WHAT_ and _WHY_ was made instead of _HOW_.
- [ ] The pull request title is treated as the final commit message.
The following pattern must be used: `IGNITE-XXXX Change summary` where `XXXX` - number of JIRA issue.
- [ ] A reviewer has been mentioned through the JIRA comments
(see [the Maintainers list](https://cwiki.apache.org/confluence/display/IGNITE/How+to+Contribute#HowtoContribute-ReviewProcessandMaintainers))
- [ ] The pull request has been checked by the Teamcity Bot and
the `green visa` attached to the JIRA ticket (see [TC.Bot: Check PR](https://mtcga.gridgain.com/prs.html))
### Notes
- [How to Contribute](https://cwiki.apache.org/confluence/display/IGNITE/How+to+Contribute)
- [Coding abbreviation rules](https://cwiki.apache.org/confluence/display/IGNITE/Abbreviation+Rules)
- [Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Coding+Guidelines)
- [Apache Ignite Teamcity Bot](https://cwiki.apache.org/confluence/display/IGNITE/Apache+Ignite+Teamcity+Bot)
If you need any help, please email dev@ignite.apache.org or ask anу advice on http://asf.slack.com _#ignite_ channel.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] sonarcloud[bot] commented on pull request #10928: IGNITE-12455 SQL Calcite: Partition pruning
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10928:
URL: https://github.com/apache/ignite/pull/10928#issuecomment-1715379078
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_ignite&pullRequest=10928)
[](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_ignite&pullRequest=10928&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_ignite&pullRequest=10928&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_ignite&pullRequest=10928&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=CODE_SMELL) [13 Code Smells](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_duplicated_lines_density&view=list)
 The version of Java (11.0.20.1) you have used to run this analysis is deprecated and we will stop accepting it soon. Please update to at least Java 17.
Read more [here](https://docs.sonarcloud.io/appendices/scanner-environment/)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-12455 SQL Calcite: Partition pruning [ignite]
Posted by "ivandasch (via GitHub)" <gi...@apache.org>.
ivandasch merged PR #10928:
URL: https://github.com/apache/ignite/pull/10928
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] ivandasch commented on a diff in pull request #10928: IGNITE-12455 SQL Calcite: Partition pruning
Posted by "ivandasch (via GitHub)" <gi...@apache.org>.
ivandasch commented on code in PR #10928:
URL: https://github.com/apache/ignite/pull/10928#discussion_r1341483486
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/TypeUtils.java:
##########
@@ -498,4 +505,46 @@ public static RexNode toRexLiteral(Object dfltVal, RelDataType type, DataContext
return rexBuilder.makeLiteral(dfltVal, type, true);
}
+
+ /**
+ * @return New object of specific type.
+ */
+ public static Object createObject(GridCacheContext<?, ?> cctx, String typeName, Class<?> typeCls) throws IgniteException {
Review Comment:
It can be used later for complex keys. It may be redundant right now, but it is also a kind of refactoring.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] sonarcloud[bot] commented on pull request #10928: IGNITE-12455 SQL Calcite: Partition pruning
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10928:
URL: https://github.com/apache/ignite/pull/10928#issuecomment-1721846343
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_ignite&pullRequest=10928)
[](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_ignite&pullRequest=10928&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_ignite&pullRequest=10928&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_ignite&pullRequest=10928&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=CODE_SMELL) [13 Code Smells](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_duplicated_lines_density&view=list)
 The version of Java (11.0.20.1) you have used to run this analysis is deprecated and we will stop accepting it soon. Please update to at least Java 17.
Read more [here](https://docs.sonarcloud.io/appendices/scanner-environment/)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] sonarcloud[bot] commented on pull request #10928: IGNITE-12455 SQL Calcite: Partition pruning
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10928:
URL: https://github.com/apache/ignite/pull/10928#issuecomment-1715597331
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_ignite&pullRequest=10928)
[](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_ignite&pullRequest=10928&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_ignite&pullRequest=10928&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_ignite&pullRequest=10928&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=CODE_SMELL) [12 Code Smells](https://sonarcloud.io/project/issues?id=apache_ignite&pullRequest=10928&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_ignite&pullRequest=10928&metric=new_duplicated_lines_density&view=list)
 The version of Java (11.0.20.1) you have used to run this analysis is deprecated and we will stop accepting it soon. Please update to at least Java 17.
Read more [here](https://docs.sonarcloud.io/appendices/scanner-environment/)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] alex-plekhanov commented on a diff in pull request #10928: IGNITE-12455 SQL Calcite: Partition pruning
Posted by "alex-plekhanov (via GitHub)" <gi...@apache.org>.
alex-plekhanov commented on code in PR #10928:
URL: https://github.com/apache/ignite/pull/10928#discussion_r1340968143
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/partition/PartitionNode.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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.partition;
+
+import java.util.Collection;
+
+/** */
+public interface PartitionNode {
+ /** */
+ Collection<Integer> apply(PartitionPruningContext ctx);
Review Comment:
`@Nullable`?
And let's describe meaning of null value in comments.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/partition/PartitionNode.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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.partition;
+
+import java.util.Collection;
+
+/** */
+public interface PartitionNode {
+ /** */
+ Collection<Integer> apply(PartitionPruningContext ctx);
+
+ /** */
+ default int cacheId() {
+ return Integer.MIN_VALUE;
Review Comment:
`MIN_VALUE` can be real cache id, let's use `CU.UNDEFINED_CACHE_ID`.
Also method can be moved to `PartitionSingleNode` (or field `cacheId` can be used instead).
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PartitionExtractor.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * 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;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+import java.util.stream.Collectors;
+import com.google.common.primitives.Primitives;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexDynamicParam;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionAllNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionLiteralNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNoneNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionOperandNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionParameterNode;
+import org.apache.ignite.internal.processors.query.calcite.prepare.Fragment;
+import org.apache.ignite.internal.processors.query.calcite.prepare.IgniteRelShuttle;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
+import org.apache.ignite.internal.processors.query.calcite.rel.ProjectableFilterableTableScan;
+import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
+import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+
+/** */
+public class PartitionExtractor extends IgniteRelShuttle {
+ /** */
+ private final IgniteTypeFactory typeFactory;
+
+ /** */
+ private final Deque<PartitionNode> stack = new ArrayDeque<>();
+
+ /** */
+ public PartitionExtractor(IgniteTypeFactory typeFactory) {
+ this.typeFactory = typeFactory;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteIndexScan rel) {
+ processScan(rel);
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteTableScan rel) {
+ processScan(rel);
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteRel processNode(IgniteRel rel) {
+ IgniteRel res = super.processNode(rel);
+
+ List<PartitionNode> operands = new ArrayList<>();
+ for (int i = 0; i < rel.getInputs().size(); ++i)
+ operands.add(stack.pop());
+
+ if (rel instanceof IgniteUnionAll)
+ stack.push(PartitionOperandNode.createOrOperandNode(operands));
+ else
+ stack.push(PartitionOperandNode.createAndOperandNode(operands));
Review Comment:
Why `createAndOperandNode` here? For example, if we have `JOIN` with two inputs on different partitions, we need data from both partitions, but not intersection of this partitions. Or if we have some SetOp other than `UnionAll` (`MINUS` for example) we also need all partitions. I.e.:
```
SELECT val FROM t WHERE id in (1, 2, 3)
MINUS
SELECT val FROM t WHERE id = 4
```
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PartitionExtractor.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * 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;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+import java.util.stream.Collectors;
+import com.google.common.primitives.Primitives;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexDynamicParam;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionAllNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionLiteralNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNoneNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionOperandNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionParameterNode;
+import org.apache.ignite.internal.processors.query.calcite.prepare.Fragment;
+import org.apache.ignite.internal.processors.query.calcite.prepare.IgniteRelShuttle;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
+import org.apache.ignite.internal.processors.query.calcite.rel.ProjectableFilterableTableScan;
+import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
+import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+
+/** */
+public class PartitionExtractor extends IgniteRelShuttle {
+ /** */
+ private final IgniteTypeFactory typeFactory;
+
+ /** */
+ private final Deque<PartitionNode> stack = new ArrayDeque<>();
+
+ /** */
+ public PartitionExtractor(IgniteTypeFactory typeFactory) {
+ this.typeFactory = typeFactory;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteIndexScan rel) {
+ processScan(rel);
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteTableScan rel) {
+ processScan(rel);
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteRel processNode(IgniteRel rel) {
+ IgniteRel res = super.processNode(rel);
+
Review Comment:
```
if (rel.getInputs().size() <= 1)
return res;
```
?
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/TypeUtils.java:
##########
@@ -498,4 +505,46 @@ public static RexNode toRexLiteral(Object dfltVal, RelDataType type, DataContext
return rexBuilder.makeLiteral(dfltVal, type, true);
}
+
+ /**
+ * @return New object of specific type.
+ */
+ public static Object createObject(GridCacheContext<?, ?> cctx, String typeName, Class<?> typeCls) throws IgniteException {
Review Comment:
Looks like redundant change, used only by CacheTableDescriptorImpl and was moved from this class
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PartitionExtractor.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * 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;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+import java.util.stream.Collectors;
+import com.google.common.primitives.Primitives;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexDynamicParam;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionAllNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionLiteralNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNoneNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionOperandNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionParameterNode;
+import org.apache.ignite.internal.processors.query.calcite.prepare.Fragment;
+import org.apache.ignite.internal.processors.query.calcite.prepare.IgniteRelShuttle;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
+import org.apache.ignite.internal.processors.query.calcite.rel.ProjectableFilterableTableScan;
+import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
+import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+
+/** */
+public class PartitionExtractor extends IgniteRelShuttle {
+ /** */
+ private final IgniteTypeFactory typeFactory;
+
+ /** */
+ private final Deque<PartitionNode> stack = new ArrayDeque<>();
+
+ /** */
+ public PartitionExtractor(IgniteTypeFactory typeFactory) {
+ this.typeFactory = typeFactory;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteIndexScan rel) {
+ processScan(rel);
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteTableScan rel) {
+ processScan(rel);
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteRel processNode(IgniteRel rel) {
+ IgniteRel res = super.processNode(rel);
+
+ List<PartitionNode> operands = new ArrayList<>();
+ for (int i = 0; i < rel.getInputs().size(); ++i)
+ operands.add(stack.pop());
+
+ if (rel instanceof IgniteUnionAll)
+ stack.push(PartitionOperandNode.createOrOperandNode(operands));
+ else
+ stack.push(PartitionOperandNode.createAndOperandNode(operands));
+
+ return res;
+ }
+
+ /** */
+ public PartitionNode go(Fragment fragment) {
+ if (!(fragment.root() instanceof IgniteSender))
+ return PartitionAllNode.INSTANCE;
+
+ if (fragment.mapping() == null || !fragment.mapping().colocated())
+ return PartitionAllNode.INSTANCE;
+
+ visit(fragment.root());
+
+ return stack.isEmpty() ? PartitionAllNode.INSTANCE : stack.pop();
+ }
+
+ /** */
+ private void processScan(IgniteRel rel) {
+ assert rel instanceof ProjectableFilterableTableScan;
+
+ RexNode condition = ((ProjectableFilterableTableScan)rel).condition();
+
+ if (!rel.distribution().function().affinity() || rel.distribution().getKeys().size() != 1) {
+ stack.push(PartitionAllNode.INSTANCE);
+
+ return;
+ }
+
+ ImmutableIntList keys = ImmutableIntList.copyOf(rel.distribution().getKeys());
+ IgniteTable tbl = rel.getTable().unwrap(IgniteTable.class);
+ RelDataType rowType = tbl.getRowType(typeFactory);
+ int cacheId = ((CacheTableDescriptor)tbl.descriptor()).cacheInfo().cacheId();
+
+ List<Class<?>> types = new ArrayList<>(rowType.getFieldCount());
+ for (RelDataTypeField field : rowType.getFieldList()) {
+ if (QueryUtils.KEY_FIELD_NAME.equals(field.getName()))
+ keys = keys.append(field.getIndex());
+
+ types.add(Primitives.wrap((Class<?>)typeFactory.getJavaClass(field.getType())));
+ }
+
+ PartitionNode partNode = processCondition(condition, types, keys, cacheId);
Review Comment:
We should also take into account `requiredColumns`.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/partition/PartitionOperandNode.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.partition;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/** */
+public class PartitionOperandNode implements PartitionNode {
+ /** */
+ private final Operand op;
+
+ /** */
+ private final List<PartitionNode> operands;
+
+ /** */
+ private PartitionOperandNode(Operand op, List<PartitionNode> operands) {
+ this.op = op;
+ this.operands = Collections.unmodifiableList(operands);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Collection<Integer> apply(PartitionPruningContext ctx) {
+ Set<Integer> allParts = null;
+
+ if (op == Operand.AND) {
+ for (PartitionNode operand : operands) {
+ Collection<Integer> parts = operand.apply(ctx);
+
+ if (parts == null)
+ continue;
+
+ if (allParts == null)
+ allParts = new HashSet<>(parts);
+ else
+ allParts.retainAll(parts);
+ }
+ }
+ else {
+ for (PartitionNode operand: operands) {
+ Collection<Integer> parts = operand.apply(ctx);
+
+ if (parts == null)
Review Comment:
Shouldn't we return `null` in this case? Null only possible for PartitionAllNode, in case of `OR` operand we also should return all nodes (`null`). `PartitionAllNode` is filtered out by the `optimize` method, but this method never called.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/partition/PartitionNode.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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.partition;
+
+import java.util.Collection;
+
+/** */
+public interface PartitionNode {
+ /** */
+ Collection<Integer> apply(PartitionPruningContext ctx);
+
+ /** */
+ default int cacheId() {
+ return Integer.MIN_VALUE;
+ }
+
+ /** */
+ default PartitionNode optimize() {
Review Comment:
Never used.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PartitionExtractor.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * 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;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+import java.util.stream.Collectors;
+import com.google.common.primitives.Primitives;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexDynamicParam;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionAllNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionLiteralNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNoneNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionOperandNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionParameterNode;
+import org.apache.ignite.internal.processors.query.calcite.prepare.Fragment;
+import org.apache.ignite.internal.processors.query.calcite.prepare.IgniteRelShuttle;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
+import org.apache.ignite.internal.processors.query.calcite.rel.ProjectableFilterableTableScan;
+import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
+import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+
+/** */
+public class PartitionExtractor extends IgniteRelShuttle {
+ /** */
+ private final IgniteTypeFactory typeFactory;
+
+ /** */
+ private final Deque<PartitionNode> stack = new ArrayDeque<>();
+
+ /** */
+ public PartitionExtractor(IgniteTypeFactory typeFactory) {
+ this.typeFactory = typeFactory;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteIndexScan rel) {
+ processScan(rel);
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteTableScan rel) {
+ processScan(rel);
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteRel processNode(IgniteRel rel) {
+ IgniteRel res = super.processNode(rel);
+
+ List<PartitionNode> operands = new ArrayList<>();
+ for (int i = 0; i < rel.getInputs().size(); ++i)
+ operands.add(stack.pop());
+
+ if (rel instanceof IgniteUnionAll)
+ stack.push(PartitionOperandNode.createOrOperandNode(operands));
+ else
+ stack.push(PartitionOperandNode.createAndOperandNode(operands));
+
+ return res;
+ }
+
+ /** */
+ public PartitionNode go(Fragment fragment) {
+ if (!(fragment.root() instanceof IgniteSender))
+ return PartitionAllNode.INSTANCE;
+
+ if (fragment.mapping() == null || !fragment.mapping().colocated())
+ return PartitionAllNode.INSTANCE;
+
+ visit(fragment.root());
+
+ return stack.isEmpty() ? PartitionAllNode.INSTANCE : stack.pop();
+ }
+
+ /** */
+ private void processScan(IgniteRel rel) {
+ assert rel instanceof ProjectableFilterableTableScan;
+
+ RexNode condition = ((ProjectableFilterableTableScan)rel).condition();
+
+ if (!rel.distribution().function().affinity() || rel.distribution().getKeys().size() != 1) {
+ stack.push(PartitionAllNode.INSTANCE);
+
+ return;
+ }
+
+ ImmutableIntList keys = ImmutableIntList.copyOf(rel.distribution().getKeys());
+ IgniteTable tbl = rel.getTable().unwrap(IgniteTable.class);
+ RelDataType rowType = tbl.getRowType(typeFactory);
+ int cacheId = ((CacheTableDescriptor)tbl.descriptor()).cacheInfo().cacheId();
+
+ List<Class<?>> types = new ArrayList<>(rowType.getFieldCount());
+ for (RelDataTypeField field : rowType.getFieldList()) {
+ if (QueryUtils.KEY_FIELD_NAME.equals(field.getName()))
+ keys = keys.append(field.getIndex());
+
+ types.add(Primitives.wrap((Class<?>)typeFactory.getJavaClass(field.getType())));
+ }
+
+ PartitionNode partNode = processCondition(condition, types, keys, cacheId);
+
+ stack.push(partNode);
+ }
+
+ /** */
+ private PartitionNode processCondition(RexNode condition, List<Class<?>> types, ImmutableIntList keys, int cacheId) {
+ if (!(condition instanceof RexCall))
+ return PartitionAllNode.INSTANCE;
+
+ SqlKind opKind = condition.getKind();
+ List<RexNode> operands = ((RexCall)condition).getOperands();
+
+ switch (opKind) {
+ case IS_NULL: {
+ RexNode left = operands.get(0);
+
+ if (!left.isA(SqlKind.LOCAL_REF))
+ return PartitionAllNode.INSTANCE;
+
+ int idx = ((RexLocalRef)left).getIndex();
+
+ if (!keys.contains(idx))
+ return PartitionAllNode.INSTANCE;
+ else
+ return PartitionNoneNode.INSTANCE;
+ }
+ case EQUALS: {
+ if (operands.size() != 2)
+ return PartitionAllNode.INSTANCE;
+
+ RexNode left = operands.get(0);
+ RexNode right = operands.get(1);
+
+ if (!left.isA(SqlKind.LOCAL_REF))
Review Comment:
What about conditions like `? = t.a`?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-12455 SQL Calcite: Partition pruning [ignite]
Posted by "alex-plekhanov (via GitHub)" <gi...@apache.org>.
alex-plekhanov commented on code in PR #10928:
URL: https://github.com/apache/ignite/pull/10928#discussion_r1356282219
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/partition/PartitionNode.java:
##########
@@ -18,18 +18,25 @@
package org.apache.ignite.internal.processors.query.calcite.exec.partition;
import java.util.Collection;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.jetbrains.annotations.Nullable;
/** */
public interface PartitionNode {
- /** */
- Collection<Integer> apply(PartitionPruningContext ctx);
+ /**
+ * @param ctx Partition pruning context.
+ * @return Collection of partitions after pruning or {@code null} if all partition required.
Review Comment:
Typo: `all partitions`
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PartitionExtractor.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.List;
+import java.util.stream.Collectors;
+import com.google.common.primitives.Primitives;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexDynamicParam;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionAllNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionLiteralNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNoneNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionOperandNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionParameterNode;
+import org.apache.ignite.internal.processors.query.calcite.prepare.Fragment;
+import org.apache.ignite.internal.processors.query.calcite.prepare.IgniteRelShuttle;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.ProjectableFilterableTableScan;
+import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
+import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+
+/** */
+public class PartitionExtractor extends IgniteRelShuttle {
+ /** */
+ private final IgniteTypeFactory typeFactory;
+
+ /** */
+ private final Deque<PartitionNode> stack = new ArrayDeque<>();
+
+ /** */
+ public PartitionExtractor(IgniteTypeFactory typeFactory) {
+ this.typeFactory = typeFactory;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteIndexScan rel) {
+ processScan(rel);
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteTableScan rel) {
+ processScan(rel);
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteRel processNode(IgniteRel rel) {
+ IgniteRel res = super.processNode(rel);
+
+ if (rel.getInputs().isEmpty())
+ return res;
+
+ List<PartitionNode> operands = new ArrayList<>();
+ for (int i = 0; i < rel.getInputs().size(); ++i) {
+ if (stack.isEmpty())
+ break;
+
+ operands.add(stack.pop());
+ }
+
+ stack.push(PartitionOperandNode.createOrOperandNode(operands));
+
+ return res;
+ }
+
+ /** */
+ public PartitionNode go(Fragment fragment) {
+ if (!(fragment.root() instanceof IgniteSender))
+ return PartitionAllNode.INSTANCE;
+
+ if (fragment.mapping() == null || !fragment.mapping().colocated())
+ return PartitionAllNode.INSTANCE;
+
+ visit(fragment.root());
+
+ if (stack.isEmpty())
+ return PartitionAllNode.INSTANCE;
+
+ PartitionNode res = stack.pop();
+
+ return res.optimize();
Review Comment:
It optimizes only root level statement, for example, if you have condition like `AND(OR(=(ID,0),=(1,1)), =(VAL,1))`, this condition can't be optimized only using root level. Perhaps we should optimize when building PartitionOperandNode (in `createAndOperandNode` and `createOrOperandNode`) methods.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PartitionExtractor.java:
##########
@@ -84,18 +84,15 @@ public PartitionExtractor(IgniteTypeFactory typeFactory) {
@Override protected IgniteRel processNode(IgniteRel rel) {
IgniteRel res = super.processNode(rel);
- if (rel.getInputs().isEmpty())
- return res;
+ if (rel.getInputs().size() > 1) {
+ List<PartitionNode> operands = new ArrayList<>();
+ for (int i = 0; i < rel.getInputs().size(); ++i)
+ operands.add(stack.pop());
- List<PartitionNode> operands = new ArrayList<>();
- for (int i = 0; i < rel.getInputs().size(); ++i) {
- if (stack.isEmpty())
- break;
-
- operands.add(stack.pop());
+ stack.push(PartitionOperandNode.createOrOperandNode(operands));
}
-
- stack.push(PartitionOperandNode.createOrOperandNode(operands));
+ else if (rel.getInputs().isEmpty())
Review Comment:
Condition is redundant (just else is enough)
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/PartitionPruneTest.java:
##########
@@ -0,0 +1,557 @@
+/*
+ * 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.integration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.query.calcite.message.QueryStartRequest;
+import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
+import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.junit.Test;
+
+/** */
+public class PartitionPruneTest extends AbstractBasicIntegrationTest {
+ /** */
+ private static final int ENTRIES_COUNT = 10000;
+
+ /** */
+ private static final LongAdder INTERCEPTED_START_REQUEST_COUNT = new LongAdder();
+
+ /** */
+ private static final ConcurrentSkipListSet<Integer> INTERCEPTED_PARTS = new ConcurrentSkipListSet<>();
+
+ /** */
+ private static final ConcurrentSkipListSet<ClusterNode> INTERCEPTED_NODES = new ConcurrentSkipListSet<>(
+ Comparator.comparing(ClusterNode::id));
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
+
+ cfg.setCommunicationSpi(new TcpCommunicationSpi() {
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC) {
+ assert msg != null;
+
+ if (msg instanceof GridIoMessage) {
+ GridIoMessage msg0 = (GridIoMessage)msg;
+
+ if (msg0.message() instanceof QueryStartRequest) {
+ INTERCEPTED_START_REQUEST_COUNT.increment();
+ INTERCEPTED_NODES.add(node);
+
+ QueryStartRequest startReq = (QueryStartRequest)msg0.message();
+
+ assertNotNull(startReq.fragmentDescription());
+
+ FragmentMapping mapping = startReq.fragmentDescription().mapping();
+
+ assertNotNull(mapping);
+
+ List<ColocationGroup> groups = U.field(mapping, "colocationGroups");
+
+ assertEquals(1, F.size(groups));
+
+ int[] parts = F.first(groups).partitions(node.id());
+
+ if (!F.isEmpty(parts)) {
+ for (int part: parts)
+ INTERCEPTED_PARTS.add(part);
+ }
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+ });
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ sql("CREATE TABLE T1(ID INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID)) WITH cache_name=t1_cache,backups=1");
+ sql("CREATE TABLE T2(ID INT, AK INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID, AK)) WITH " +
+ "cache_name=t2_cache,backups=1,affinity_key=AK");
+ sql("CREATE TABLE DICT(ID INT PRIMARY KEY, IDX_VAL VARCHAR, VAL VARCHAR) WITH template=replicated,cache_name=dict_cache");
+
+ sql("CREATE INDEX T1_IDX ON T1(IDX_VAL)");
+ sql("CREATE INDEX T2_IDX ON T2(IDX_VAL)");
+ sql("CREATE INDEX DICT_IDX ON DICT(IDX_VAL)");
+
+ Stream.of("T1", "DICT").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
+ sb.append("(").append(i).append(", ")
+ .append("'name_").append(i).append("', ")
+ .append("'name_").append(i).append("')");
+
+ if (i < ENTRIES_COUNT - 1)
+ sb.append(",");
+ }
+
+ sql(sb.toString());
+
+ assertEquals(ENTRIES_COUNT, client.getOrCreateCache(tableName + "_CACHE").size(CachePeekMode.PRIMARY));
+ });
+
+ Stream.of("T2").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, AK, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
+ sb.append("(").append(i).append(", ")
+ .append(i).append(",")
+ .append("'name_").append(i).append("', ")
+ .append("'name_").append(i).append("')");
+
+ if (i < ENTRIES_COUNT - 1)
+ sb.append(",");
+ }
+
+ sql(sb.toString());
+
+ assertEquals(ENTRIES_COUNT, client.getOrCreateCache(tableName + "_CACHE").size(CachePeekMode.PRIMARY));
+ });
+
+ sql("ANALYZE PUBLIC.T1(ID), PUBLIC.T2(ID,AK), PUBLIC.DICT(ID) WITH \"NULLS=0,DISTINCT=10000,TOTAL=10000\"");
+
+ clearIntercepted();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ clearIntercepted();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int nodeCount() {
+ return 8;
+ }
+
+ /** */
+ @Test
+ public void testSimple() {
+ execute("select count(*) from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals(1L, res.get(0).get(0));
+ },
+ 123);
+
+ execute("select * from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select * from T1 where ? = T1.ID",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select VAL, IDX_VAL, ID from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123, 123);
+ }
+
+ /** */
+ @Test
+ public void testNullsInCondition() {
+ execute("select * from T1 where T1.ID is NULL",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ });
+
+ execute("select * from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, new Object[]{ null });
+
+ execute("select * from T1 where T1.ID is NULL and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, 123);
+
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, null, 123);
+
+ execute("select * from T1 where T1.ID is NULL or T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ }, 123);
+
+ execute("select * from T1 where T1.ID = ? or T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ }, null, 123);
+ }
+
+ /** */
+ @Test
+ public void testEmptyConditions() {
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518);
+
+ execute("select * from T1 where T1.ID = ? and (T1.ID = ? OR T1.ID = ? OR T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+
+ execute("select * from T1 where (T1.ID = ? OR T1.ID = ?) AND (T1.ID = ? OR T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+
+ execute("select * from T1 where (T1.ID = ? AND T1.ID = ?) OR (T1.ID = ? AND T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+ }
+
+ /** */
+ @Test
+ public void testSelectIn() {
+ IntStream.of(2, 6).forEach(i -> {
+ testSelect(i, true, "_KEY");
+ testSelect(i, true, "ID");
+ });
+ }
+
+
+ /** */
+ @Test
+ public void testSelectOr() {
+ testSelect(1, false, "_KEY");
+ testSelect(1, false, "ID");
+
+ IntStream.of(2, 6).forEach(i -> {
+ testSelect(i, false, "_KEY");
+ testSelect(i, false, "ID");
+ });
+ }
+
+ /** */
+ @Test
+ public void testSimpleJoin() {
+ // Key (not alias).
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T1.ID = ?",
+ (res) -> {
+// assertPartitions(partition("T1_CACHE", 123));
+// assertNodes(node("T1_CACHE", 123));
+ assertEquals(1, res.size());
+ assertEquals(123, res.get(0).get(0));
+ },
+ 123
+ );
+
+ // Key (alias).
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T1._KEY = ?",
+ (res) -> {
+// assertPartitions(partition("T1_CACHE", 125));
+// assertNodes(node("T1_CACHE", 125));
+
+ assertEquals(1, res.size());
+ assertEquals(125, res.get(0).get(0));
+ },
+ 125
+ );
+
+ // Non-affinity key.
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T2.ID = ?",
+ (res) -> {
+ assertEquals(1, res.size());
+ assertEquals(125, res.get(0).get(0));
+ },
+ 125
+ );
+
+ // Affinity key.
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T2.AK = ?",
+ (res) -> {
+// assertPartitions(partition("T2_CACHE", 125));
+ assertEquals(1, res.size());
+ assertEquals(125, res.get(0).get(0));
+ },
+ 125
+ );
+ }
+
+ /** */
+ private void testSelect(int sz, boolean withIn, String column) {
+ assertTrue(sz >= 1);
+ int[] values = ThreadLocalRandom.current().ints(0, ENTRIES_COUNT).distinct().limit(sz).toArray();
+
+ StringBuilder query;
+
+ if (!withIn || sz == 1)
+ query = new StringBuilder("select * from T1 where ");
+ else
+ query = new StringBuilder("select * from T1 where T1.").append(column).append(" in (");
+
+ for (int i = 0; i < sz; ++i) {
+ if (!withIn || sz == 1)
+ query.append("T1.").append(column).append("= ?");
+ else
+ query.append('?');
+
+ if (sz == 1)
+ break;
+
+ if (i == sz - 1)
+ query.append(!withIn ? "" : ")");
+ else
+ query.append(!withIn ? " OR " : ", ");
+ }
+
+ execute(query.toString(),
+ res -> {
+ assertPartitions(IntStream.of(values).map(i -> partition("T1_CACHE", i)).toArray());
+ assertNodes(IntStream.of(values).mapToObj(i -> node("T1_CACHE", i)).toArray(ClusterNode[]::new));
+
+ assertEquals(values.length, res.size());
+
+ assertEquals(
+ IntStream.of(values).sorted().boxed().collect(Collectors.toList()),
+ res.stream().map(row -> row.get(0)).sorted().collect(Collectors.toList())
+ );
+ },
+ IntStream.of(values).boxed().toArray(Integer[]::new));
+ }
+
+ /** */
+ public void execute(String sql, Consumer<List<List<?>>> resConsumer, Object... args) {
+ log.info(">>> TEST COMBINATION: \"" + sql + "\"");
+
+ // Execute query as is.
+ log.info("Execute \"" + sql + "\" with args " + Arrays.toString(args));
+
+ List<List<?>> res = sql(sql, args);
+
+ resConsumer.accept(res);
+ clearIntercepted();
+
+ // Start filling arguments recursively.
+ if (args != null && args.length > 0)
+ executeCombinations0(sql, resConsumer, new HashSet<>(), args);
+ }
+
+ /** */
+ private void executeCombinations0(
+ String sql,
+ Consumer<List<List<?>>> resConsumer,
+ Set<String> executedSqls,
+ Object... args
+ ) {
+ assert args != null && args.length > 0;
+
+ // Get argument positions.
+ List<Integer> paramPoss = new ArrayList<>();
+
+ int pos = 0;
+
+ while (true) {
+ int paramPos = sql.indexOf('?', pos);
+
+ if (paramPos == -1)
+ break;
+
+ paramPoss.add(paramPos);
+
+ pos = paramPos + 1;
+ }
+
+ for (int i = 0; i < args.length; i++) {
+ // Prepare new SQL and arguments.
+ int paramPos = paramPoss.get(i);
+
+ String newSql = sql.substring(0, paramPos) + (args[i] instanceof String ? "'" + args[i] + "'" : args[i])
+ + sql.substring(paramPos + 1);
+
+ Object[] newArgs = new Object[args.length - 1];
+
+ int newArgsPos = 0;
+
+ for (int j = 0; j < args.length; j++) {
+ if (j != i)
+ newArgs[newArgsPos++] = args[j];
+ }
+
+ // Execute if this combination was never executed before.
+ if (executedSqls.add(newSql)) {
+ log.info("Execute sql \"" + newSql + "\"");
+
+ List<List<?>> res = sql(newSql, newArgs);
+
+ resConsumer.accept(res);
+ clearIntercepted();
+ }
+
+ // Continue recursively.
+ if (newArgs.length > 0)
+ executeCombinations0(newSql, resConsumer, executedSqls, newArgs);
+ }
+ }
+
+ /** */
+ protected static void assertPartitions(int... expParts) {
+ Collection<Integer> expParts0 = new TreeSet<>();
+
+ if (!F.isEmpty(expParts)) {
+ for (int expPart : expParts)
+ expParts0.add(expPart);
+ }
+
+ TreeSet<Integer> actualParts = new TreeSet<>(INTERCEPTED_PARTS);
+
+ assertEquals("Unexpected partitions [exp=" + Arrays.toString(expParts) + ", actual=" + actualParts + ']',
Review Comment:
`Arrays.toString(expParts)` -> `expParts0` (output will be sorted)
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/PartitionPruneTest.java:
##########
@@ -0,0 +1,557 @@
+/*
+ * 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.integration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.query.calcite.message.QueryStartRequest;
+import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
+import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.junit.Test;
+
+/** */
+public class PartitionPruneTest extends AbstractBasicIntegrationTest {
+ /** */
+ private static final int ENTRIES_COUNT = 10000;
+
+ /** */
+ private static final LongAdder INTERCEPTED_START_REQUEST_COUNT = new LongAdder();
+
+ /** */
+ private static final ConcurrentSkipListSet<Integer> INTERCEPTED_PARTS = new ConcurrentSkipListSet<>();
+
+ /** */
+ private static final ConcurrentSkipListSet<ClusterNode> INTERCEPTED_NODES = new ConcurrentSkipListSet<>(
+ Comparator.comparing(ClusterNode::id));
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
+
+ cfg.setCommunicationSpi(new TcpCommunicationSpi() {
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC) {
+ assert msg != null;
+
+ if (msg instanceof GridIoMessage) {
+ GridIoMessage msg0 = (GridIoMessage)msg;
+
+ if (msg0.message() instanceof QueryStartRequest) {
+ INTERCEPTED_START_REQUEST_COUNT.increment();
+ INTERCEPTED_NODES.add(node);
+
+ QueryStartRequest startReq = (QueryStartRequest)msg0.message();
+
+ assertNotNull(startReq.fragmentDescription());
+
+ FragmentMapping mapping = startReq.fragmentDescription().mapping();
+
+ assertNotNull(mapping);
+
+ List<ColocationGroup> groups = U.field(mapping, "colocationGroups");
Review Comment:
Add getter for colocationGroups instead of reflection?
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/PartitionPruneTest.java:
##########
@@ -0,0 +1,557 @@
+/*
+ * 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.integration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.query.calcite.message.QueryStartRequest;
+import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
+import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.junit.Test;
+
+/** */
+public class PartitionPruneTest extends AbstractBasicIntegrationTest {
+ /** */
+ private static final int ENTRIES_COUNT = 10000;
+
+ /** */
+ private static final LongAdder INTERCEPTED_START_REQUEST_COUNT = new LongAdder();
+
+ /** */
+ private static final ConcurrentSkipListSet<Integer> INTERCEPTED_PARTS = new ConcurrentSkipListSet<>();
+
+ /** */
+ private static final ConcurrentSkipListSet<ClusterNode> INTERCEPTED_NODES = new ConcurrentSkipListSet<>(
+ Comparator.comparing(ClusterNode::id));
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
+
+ cfg.setCommunicationSpi(new TcpCommunicationSpi() {
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC) {
+ assert msg != null;
+
+ if (msg instanceof GridIoMessage) {
+ GridIoMessage msg0 = (GridIoMessage)msg;
+
+ if (msg0.message() instanceof QueryStartRequest) {
+ INTERCEPTED_START_REQUEST_COUNT.increment();
+ INTERCEPTED_NODES.add(node);
+
+ QueryStartRequest startReq = (QueryStartRequest)msg0.message();
+
+ assertNotNull(startReq.fragmentDescription());
+
+ FragmentMapping mapping = startReq.fragmentDescription().mapping();
+
+ assertNotNull(mapping);
+
+ List<ColocationGroup> groups = U.field(mapping, "colocationGroups");
+
+ assertEquals(1, F.size(groups));
+
+ int[] parts = F.first(groups).partitions(node.id());
+
+ if (!F.isEmpty(parts)) {
+ for (int part: parts)
+ INTERCEPTED_PARTS.add(part);
+ }
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+ });
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ sql("CREATE TABLE T1(ID INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID)) WITH cache_name=t1_cache,backups=1");
+ sql("CREATE TABLE T2(ID INT, AK INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID, AK)) WITH " +
+ "cache_name=t2_cache,backups=1,affinity_key=AK");
+ sql("CREATE TABLE DICT(ID INT PRIMARY KEY, IDX_VAL VARCHAR, VAL VARCHAR) WITH template=replicated,cache_name=dict_cache");
+
+ sql("CREATE INDEX T1_IDX ON T1(IDX_VAL)");
+ sql("CREATE INDEX T2_IDX ON T2(IDX_VAL)");
+ sql("CREATE INDEX DICT_IDX ON DICT(IDX_VAL)");
+
+ Stream.of("T1", "DICT").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
+ sb.append("(").append(i).append(", ")
+ .append("'name_").append(i).append("', ")
+ .append("'name_").append(i).append("')");
+
+ if (i < ENTRIES_COUNT - 1)
+ sb.append(",");
+ }
+
+ sql(sb.toString());
+
+ assertEquals(ENTRIES_COUNT, client.getOrCreateCache(tableName + "_CACHE").size(CachePeekMode.PRIMARY));
+ });
+
+ Stream.of("T2").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, AK, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
Review Comment:
10000 -> ENTRIES_COUNT
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/PartitionPruneTest.java:
##########
@@ -0,0 +1,557 @@
+/*
+ * 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.integration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.query.calcite.message.QueryStartRequest;
+import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
+import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.junit.Test;
+
+/** */
+public class PartitionPruneTest extends AbstractBasicIntegrationTest {
+ /** */
+ private static final int ENTRIES_COUNT = 10000;
+
+ /** */
+ private static final LongAdder INTERCEPTED_START_REQUEST_COUNT = new LongAdder();
+
+ /** */
+ private static final ConcurrentSkipListSet<Integer> INTERCEPTED_PARTS = new ConcurrentSkipListSet<>();
+
+ /** */
+ private static final ConcurrentSkipListSet<ClusterNode> INTERCEPTED_NODES = new ConcurrentSkipListSet<>(
+ Comparator.comparing(ClusterNode::id));
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
+
+ cfg.setCommunicationSpi(new TcpCommunicationSpi() {
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC) {
+ assert msg != null;
+
+ if (msg instanceof GridIoMessage) {
+ GridIoMessage msg0 = (GridIoMessage)msg;
+
+ if (msg0.message() instanceof QueryStartRequest) {
+ INTERCEPTED_START_REQUEST_COUNT.increment();
+ INTERCEPTED_NODES.add(node);
+
+ QueryStartRequest startReq = (QueryStartRequest)msg0.message();
+
+ assertNotNull(startReq.fragmentDescription());
+
+ FragmentMapping mapping = startReq.fragmentDescription().mapping();
+
+ assertNotNull(mapping);
+
+ List<ColocationGroup> groups = U.field(mapping, "colocationGroups");
+
+ assertEquals(1, F.size(groups));
+
+ int[] parts = F.first(groups).partitions(node.id());
+
+ if (!F.isEmpty(parts)) {
+ for (int part: parts)
+ INTERCEPTED_PARTS.add(part);
+ }
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+ });
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ sql("CREATE TABLE T1(ID INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID)) WITH cache_name=t1_cache,backups=1");
+ sql("CREATE TABLE T2(ID INT, AK INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID, AK)) WITH " +
+ "cache_name=t2_cache,backups=1,affinity_key=AK");
+ sql("CREATE TABLE DICT(ID INT PRIMARY KEY, IDX_VAL VARCHAR, VAL VARCHAR) WITH template=replicated,cache_name=dict_cache");
+
+ sql("CREATE INDEX T1_IDX ON T1(IDX_VAL)");
+ sql("CREATE INDEX T2_IDX ON T2(IDX_VAL)");
+ sql("CREATE INDEX DICT_IDX ON DICT(IDX_VAL)");
+
+ Stream.of("T1", "DICT").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
Review Comment:
10000 -> ENTRIES_COUNT
Let's also configure LongQueryWarningTimeout, to avoid logging of this query as long-running and reduce log size.
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/PartitionPruneTest.java:
##########
@@ -0,0 +1,557 @@
+/*
+ * 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.integration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.query.calcite.message.QueryStartRequest;
+import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
+import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.junit.Test;
+
+/** */
+public class PartitionPruneTest extends AbstractBasicIntegrationTest {
+ /** */
+ private static final int ENTRIES_COUNT = 10000;
+
+ /** */
+ private static final LongAdder INTERCEPTED_START_REQUEST_COUNT = new LongAdder();
+
+ /** */
+ private static final ConcurrentSkipListSet<Integer> INTERCEPTED_PARTS = new ConcurrentSkipListSet<>();
+
+ /** */
+ private static final ConcurrentSkipListSet<ClusterNode> INTERCEPTED_NODES = new ConcurrentSkipListSet<>(
+ Comparator.comparing(ClusterNode::id));
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
+
+ cfg.setCommunicationSpi(new TcpCommunicationSpi() {
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC) {
+ assert msg != null;
+
+ if (msg instanceof GridIoMessage) {
+ GridIoMessage msg0 = (GridIoMessage)msg;
+
+ if (msg0.message() instanceof QueryStartRequest) {
+ INTERCEPTED_START_REQUEST_COUNT.increment();
+ INTERCEPTED_NODES.add(node);
+
+ QueryStartRequest startReq = (QueryStartRequest)msg0.message();
+
+ assertNotNull(startReq.fragmentDescription());
+
+ FragmentMapping mapping = startReq.fragmentDescription().mapping();
+
+ assertNotNull(mapping);
+
+ List<ColocationGroup> groups = U.field(mapping, "colocationGroups");
+
+ assertEquals(1, F.size(groups));
+
+ int[] parts = F.first(groups).partitions(node.id());
+
+ if (!F.isEmpty(parts)) {
+ for (int part: parts)
+ INTERCEPTED_PARTS.add(part);
+ }
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+ });
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ sql("CREATE TABLE T1(ID INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID)) WITH cache_name=t1_cache,backups=1");
+ sql("CREATE TABLE T2(ID INT, AK INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID, AK)) WITH " +
+ "cache_name=t2_cache,backups=1,affinity_key=AK");
+ sql("CREATE TABLE DICT(ID INT PRIMARY KEY, IDX_VAL VARCHAR, VAL VARCHAR) WITH template=replicated,cache_name=dict_cache");
+
+ sql("CREATE INDEX T1_IDX ON T1(IDX_VAL)");
+ sql("CREATE INDEX T2_IDX ON T2(IDX_VAL)");
+ sql("CREATE INDEX DICT_IDX ON DICT(IDX_VAL)");
+
+ Stream.of("T1", "DICT").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
+ sb.append("(").append(i).append(", ")
+ .append("'name_").append(i).append("', ")
+ .append("'name_").append(i).append("')");
+
+ if (i < ENTRIES_COUNT - 1)
+ sb.append(",");
+ }
+
+ sql(sb.toString());
+
+ assertEquals(ENTRIES_COUNT, client.getOrCreateCache(tableName + "_CACHE").size(CachePeekMode.PRIMARY));
+ });
+
+ Stream.of("T2").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, AK, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
+ sb.append("(").append(i).append(", ")
+ .append(i).append(",")
+ .append("'name_").append(i).append("', ")
+ .append("'name_").append(i).append("')");
+
+ if (i < ENTRIES_COUNT - 1)
+ sb.append(",");
+ }
+
+ sql(sb.toString());
+
+ assertEquals(ENTRIES_COUNT, client.getOrCreateCache(tableName + "_CACHE").size(CachePeekMode.PRIMARY));
+ });
+
+ sql("ANALYZE PUBLIC.T1(ID), PUBLIC.T2(ID,AK), PUBLIC.DICT(ID) WITH \"NULLS=0,DISTINCT=10000,TOTAL=10000\"");
+
+ clearIntercepted();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ clearIntercepted();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int nodeCount() {
+ return 8;
+ }
+
+ /** */
+ @Test
+ public void testSimple() {
+ execute("select count(*) from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals(1L, res.get(0).get(0));
+ },
+ 123);
+
+ execute("select * from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select * from T1 where ? = T1.ID",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select VAL, IDX_VAL, ID from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123, 123);
+ }
+
+ /** */
+ @Test
+ public void testNullsInCondition() {
+ execute("select * from T1 where T1.ID is NULL",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ });
+
+ execute("select * from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, new Object[]{ null });
+
+ execute("select * from T1 where T1.ID is NULL and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, 123);
+
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, null, 123);
+
+ execute("select * from T1 where T1.ID is NULL or T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ }, 123);
+
+ execute("select * from T1 where T1.ID = ? or T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ }, null, 123);
+ }
+
+ /** */
+ @Test
+ public void testEmptyConditions() {
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518);
+
+ execute("select * from T1 where T1.ID = ? and (T1.ID = ? OR T1.ID = ? OR T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+
+ execute("select * from T1 where (T1.ID = ? OR T1.ID = ?) AND (T1.ID = ? OR T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+
+ execute("select * from T1 where (T1.ID = ? AND T1.ID = ?) OR (T1.ID = ? AND T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+ }
+
+ /** */
+ @Test
+ public void testSelectIn() {
+ IntStream.of(2, 6).forEach(i -> {
+ testSelect(i, true, "_KEY");
+ testSelect(i, true, "ID");
+ });
+ }
+
+
+ /** */
+ @Test
+ public void testSelectOr() {
+ testSelect(1, false, "_KEY");
+ testSelect(1, false, "ID");
+
+ IntStream.of(2, 6).forEach(i -> {
+ testSelect(i, false, "_KEY");
+ testSelect(i, false, "ID");
+ });
+ }
+
+ /** */
+ @Test
+ public void testSimpleJoin() {
Review Comment:
Let's add test with multiple imputs, but not join (union or other set op for example)
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/PartitionPruneTest.java:
##########
@@ -0,0 +1,557 @@
+/*
+ * 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.integration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.query.calcite.message.QueryStartRequest;
+import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
+import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.junit.Test;
+
+/** */
+public class PartitionPruneTest extends AbstractBasicIntegrationTest {
+ /** */
+ private static final int ENTRIES_COUNT = 10000;
+
+ /** */
+ private static final LongAdder INTERCEPTED_START_REQUEST_COUNT = new LongAdder();
+
+ /** */
+ private static final ConcurrentSkipListSet<Integer> INTERCEPTED_PARTS = new ConcurrentSkipListSet<>();
+
+ /** */
+ private static final ConcurrentSkipListSet<ClusterNode> INTERCEPTED_NODES = new ConcurrentSkipListSet<>(
+ Comparator.comparing(ClusterNode::id));
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
+
+ cfg.setCommunicationSpi(new TcpCommunicationSpi() {
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC) {
+ assert msg != null;
+
+ if (msg instanceof GridIoMessage) {
+ GridIoMessage msg0 = (GridIoMessage)msg;
+
+ if (msg0.message() instanceof QueryStartRequest) {
+ INTERCEPTED_START_REQUEST_COUNT.increment();
+ INTERCEPTED_NODES.add(node);
+
+ QueryStartRequest startReq = (QueryStartRequest)msg0.message();
+
+ assertNotNull(startReq.fragmentDescription());
+
+ FragmentMapping mapping = startReq.fragmentDescription().mapping();
+
+ assertNotNull(mapping);
+
+ List<ColocationGroup> groups = U.field(mapping, "colocationGroups");
+
+ assertEquals(1, F.size(groups));
+
+ int[] parts = F.first(groups).partitions(node.id());
+
+ if (!F.isEmpty(parts)) {
+ for (int part: parts)
+ INTERCEPTED_PARTS.add(part);
+ }
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+ });
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ sql("CREATE TABLE T1(ID INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID)) WITH cache_name=t1_cache,backups=1");
+ sql("CREATE TABLE T2(ID INT, AK INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID, AK)) WITH " +
+ "cache_name=t2_cache,backups=1,affinity_key=AK");
+ sql("CREATE TABLE DICT(ID INT PRIMARY KEY, IDX_VAL VARCHAR, VAL VARCHAR) WITH template=replicated,cache_name=dict_cache");
+
+ sql("CREATE INDEX T1_IDX ON T1(IDX_VAL)");
+ sql("CREATE INDEX T2_IDX ON T2(IDX_VAL)");
+ sql("CREATE INDEX DICT_IDX ON DICT(IDX_VAL)");
+
+ Stream.of("T1", "DICT").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
+ sb.append("(").append(i).append(", ")
+ .append("'name_").append(i).append("', ")
+ .append("'name_").append(i).append("')");
+
+ if (i < ENTRIES_COUNT - 1)
+ sb.append(",");
+ }
+
+ sql(sb.toString());
+
+ assertEquals(ENTRIES_COUNT, client.getOrCreateCache(tableName + "_CACHE").size(CachePeekMode.PRIMARY));
+ });
+
+ Stream.of("T2").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, AK, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
+ sb.append("(").append(i).append(", ")
+ .append(i).append(",")
+ .append("'name_").append(i).append("', ")
+ .append("'name_").append(i).append("')");
+
+ if (i < ENTRIES_COUNT - 1)
+ sb.append(",");
+ }
+
+ sql(sb.toString());
+
+ assertEquals(ENTRIES_COUNT, client.getOrCreateCache(tableName + "_CACHE").size(CachePeekMode.PRIMARY));
+ });
+
+ sql("ANALYZE PUBLIC.T1(ID), PUBLIC.T2(ID,AK), PUBLIC.DICT(ID) WITH \"NULLS=0,DISTINCT=10000,TOTAL=10000\"");
+
+ clearIntercepted();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ clearIntercepted();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int nodeCount() {
+ return 8;
+ }
+
+ /** */
+ @Test
+ public void testSimple() {
+ execute("select count(*) from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals(1L, res.get(0).get(0));
+ },
+ 123);
+
+ execute("select * from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select * from T1 where ? = T1.ID",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select VAL, IDX_VAL, ID from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123, 123);
+ }
+
+ /** */
+ @Test
+ public void testNullsInCondition() {
+ execute("select * from T1 where T1.ID is NULL",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ });
+
+ execute("select * from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, new Object[]{ null });
+
+ execute("select * from T1 where T1.ID is NULL and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, 123);
+
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, null, 123);
+
+ execute("select * from T1 where T1.ID is NULL or T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ }, 123);
+
+ execute("select * from T1 where T1.ID = ? or T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ }, null, 123);
+ }
+
+ /** */
+ @Test
+ public void testEmptyConditions() {
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518);
+
+ execute("select * from T1 where T1.ID = ? and (T1.ID = ? OR T1.ID = ? OR T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+
+ execute("select * from T1 where (T1.ID = ? OR T1.ID = ?) AND (T1.ID = ? OR T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+
+ execute("select * from T1 where (T1.ID = ? AND T1.ID = ?) OR (T1.ID = ? AND T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+ }
+
+ /** */
+ @Test
+ public void testSelectIn() {
+ IntStream.of(2, 6).forEach(i -> {
+ testSelect(i, true, "_KEY");
+ testSelect(i, true, "ID");
+ });
+ }
+
+
+ /** */
+ @Test
+ public void testSelectOr() {
+ testSelect(1, false, "_KEY");
+ testSelect(1, false, "ID");
+
+ IntStream.of(2, 6).forEach(i -> {
+ testSelect(i, false, "_KEY");
+ testSelect(i, false, "ID");
+ });
+ }
+
+ /** */
+ @Test
+ public void testSimpleJoin() {
+ // Key (not alias).
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T1.ID = ?",
+ (res) -> {
+// assertPartitions(partition("T1_CACHE", 123));
+// assertNodes(node("T1_CACHE", 123));
+ assertEquals(1, res.size());
+ assertEquals(123, res.get(0).get(0));
+ },
+ 123
+ );
+
+ // Key (alias).
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T1._KEY = ?",
+ (res) -> {
+// assertPartitions(partition("T1_CACHE", 125));
+// assertNodes(node("T1_CACHE", 125));
+
+ assertEquals(1, res.size());
+ assertEquals(125, res.get(0).get(0));
+ },
+ 125
+ );
+
+ // Non-affinity key.
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T2.ID = ?",
+ (res) -> {
+ assertEquals(1, res.size());
+ assertEquals(125, res.get(0).get(0));
+ },
+ 125
+ );
+
+ // Affinity key.
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T2.AK = ?",
+ (res) -> {
+// assertPartitions(partition("T2_CACHE", 125));
+ assertEquals(1, res.size());
+ assertEquals(125, res.get(0).get(0));
+ },
+ 125
+ );
+ }
+
+ /** */
+ private void testSelect(int sz, boolean withIn, String column) {
+ assertTrue(sz >= 1);
+ int[] values = ThreadLocalRandom.current().ints(0, ENTRIES_COUNT).distinct().limit(sz).toArray();
+
+ StringBuilder query;
+
+ if (!withIn || sz == 1)
+ query = new StringBuilder("select * from T1 where ");
+ else
+ query = new StringBuilder("select * from T1 where T1.").append(column).append(" in (");
+
+ for (int i = 0; i < sz; ++i) {
+ if (!withIn || sz == 1)
+ query.append("T1.").append(column).append("= ?");
+ else
+ query.append('?');
+
+ if (sz == 1)
+ break;
+
+ if (i == sz - 1)
+ query.append(!withIn ? "" : ")");
+ else
+ query.append(!withIn ? " OR " : ", ");
+ }
+
+ execute(query.toString(),
+ res -> {
+ assertPartitions(IntStream.of(values).map(i -> partition("T1_CACHE", i)).toArray());
+ assertNodes(IntStream.of(values).mapToObj(i -> node("T1_CACHE", i)).toArray(ClusterNode[]::new));
+
+ assertEquals(values.length, res.size());
+
+ assertEquals(
+ IntStream.of(values).sorted().boxed().collect(Collectors.toList()),
+ res.stream().map(row -> row.get(0)).sorted().collect(Collectors.toList())
+ );
+ },
+ IntStream.of(values).boxed().toArray(Integer[]::new));
+ }
+
+ /** */
+ public void execute(String sql, Consumer<List<List<?>>> resConsumer, Object... args) {
+ log.info(">>> TEST COMBINATION: \"" + sql + "\"");
+
+ // Execute query as is.
+ log.info("Execute \"" + sql + "\" with args " + Arrays.toString(args));
+
+ List<List<?>> res = sql(sql, args);
+
+ resConsumer.accept(res);
+ clearIntercepted();
+
+ // Start filling arguments recursively.
+ if (args != null && args.length > 0)
+ executeCombinations0(sql, resConsumer, new HashSet<>(), args);
+ }
+
+ /** */
+ private void executeCombinations0(
+ String sql,
+ Consumer<List<List<?>>> resConsumer,
+ Set<String> executedSqls,
+ Object... args
+ ) {
+ assert args != null && args.length > 0;
+
+ // Get argument positions.
+ List<Integer> paramPoss = new ArrayList<>();
+
+ int pos = 0;
+
+ while (true) {
+ int paramPos = sql.indexOf('?', pos);
+
+ if (paramPos == -1)
+ break;
+
+ paramPoss.add(paramPos);
+
+ pos = paramPos + 1;
+ }
+
+ for (int i = 0; i < args.length; i++) {
+ // Prepare new SQL and arguments.
+ int paramPos = paramPoss.get(i);
+
+ String newSql = sql.substring(0, paramPos) + (args[i] instanceof String ? "'" + args[i] + "'" : args[i])
+ + sql.substring(paramPos + 1);
+
+ Object[] newArgs = new Object[args.length - 1];
+
+ int newArgsPos = 0;
+
+ for (int j = 0; j < args.length; j++) {
+ if (j != i)
+ newArgs[newArgsPos++] = args[j];
+ }
+
+ // Execute if this combination was never executed before.
+ if (executedSqls.add(newSql)) {
+ log.info("Execute sql \"" + newSql + "\"");
+
+ List<List<?>> res = sql(newSql, newArgs);
+
+ resConsumer.accept(res);
+ clearIntercepted();
+ }
+
+ // Continue recursively.
+ if (newArgs.length > 0)
+ executeCombinations0(newSql, resConsumer, executedSqls, newArgs);
+ }
+ }
+
+ /** */
+ protected static void assertPartitions(int... expParts) {
+ Collection<Integer> expParts0 = new TreeSet<>();
+
+ if (!F.isEmpty(expParts)) {
+ for (int expPart : expParts)
+ expParts0.add(expPart);
+ }
+
+ TreeSet<Integer> actualParts = new TreeSet<>(INTERCEPTED_PARTS);
+
+ assertEquals("Unexpected partitions [exp=" + Arrays.toString(expParts) + ", actual=" + actualParts + ']',
+ expParts0, actualParts);
+ }
+
+ /** */
+ protected int partition(String cacheName, Object key) {
+ return client.affinity(cacheName).partition(key);
+ }
+
+ /** */
+ protected ClusterNode node(String cacheName, Object key) {
+ return G.allGrids().stream()
+ .filter(ign -> ign.affinity(cacheName).isPrimary(ign.cluster().localNode(), key))
+ .map(ign -> ign.cluster().localNode()).findFirst().orElse(null);
+ }
+
+ /** */
+ protected static void assertNodes(ClusterNode... expNodes) {
+ Collection<ClusterNode> expNodes0 = new TreeSet<>(Comparator.comparing(ClusterNode::id));
+
+ if (!F.isEmpty(expNodes)) {
Review Comment:
Redundant braces for one line statement
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/PartitionPruneTest.java:
##########
@@ -0,0 +1,557 @@
+/*
+ * 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.integration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.query.calcite.message.QueryStartRequest;
+import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
+import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.junit.Test;
+
+/** */
+public class PartitionPruneTest extends AbstractBasicIntegrationTest {
+ /** */
+ private static final int ENTRIES_COUNT = 10000;
+
+ /** */
+ private static final LongAdder INTERCEPTED_START_REQUEST_COUNT = new LongAdder();
+
+ /** */
+ private static final ConcurrentSkipListSet<Integer> INTERCEPTED_PARTS = new ConcurrentSkipListSet<>();
+
+ /** */
+ private static final ConcurrentSkipListSet<ClusterNode> INTERCEPTED_NODES = new ConcurrentSkipListSet<>(
+ Comparator.comparing(ClusterNode::id));
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
+
+ cfg.setCommunicationSpi(new TcpCommunicationSpi() {
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC) {
+ assert msg != null;
+
+ if (msg instanceof GridIoMessage) {
+ GridIoMessage msg0 = (GridIoMessage)msg;
+
+ if (msg0.message() instanceof QueryStartRequest) {
+ INTERCEPTED_START_REQUEST_COUNT.increment();
+ INTERCEPTED_NODES.add(node);
+
+ QueryStartRequest startReq = (QueryStartRequest)msg0.message();
+
+ assertNotNull(startReq.fragmentDescription());
+
+ FragmentMapping mapping = startReq.fragmentDescription().mapping();
+
+ assertNotNull(mapping);
+
+ List<ColocationGroup> groups = U.field(mapping, "colocationGroups");
+
+ assertEquals(1, F.size(groups));
+
+ int[] parts = F.first(groups).partitions(node.id());
+
+ if (!F.isEmpty(parts)) {
+ for (int part: parts)
+ INTERCEPTED_PARTS.add(part);
+ }
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+ });
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ sql("CREATE TABLE T1(ID INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID)) WITH cache_name=t1_cache,backups=1");
+ sql("CREATE TABLE T2(ID INT, AK INT, IDX_VAL VARCHAR, VAL VARCHAR, PRIMARY KEY(ID, AK)) WITH " +
+ "cache_name=t2_cache,backups=1,affinity_key=AK");
+ sql("CREATE TABLE DICT(ID INT PRIMARY KEY, IDX_VAL VARCHAR, VAL VARCHAR) WITH template=replicated,cache_name=dict_cache");
+
+ sql("CREATE INDEX T1_IDX ON T1(IDX_VAL)");
+ sql("CREATE INDEX T2_IDX ON T2(IDX_VAL)");
+ sql("CREATE INDEX DICT_IDX ON DICT(IDX_VAL)");
+
+ Stream.of("T1", "DICT").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
+ sb.append("(").append(i).append(", ")
+ .append("'name_").append(i).append("', ")
+ .append("'name_").append(i).append("')");
+
+ if (i < ENTRIES_COUNT - 1)
+ sb.append(",");
+ }
+
+ sql(sb.toString());
+
+ assertEquals(ENTRIES_COUNT, client.getOrCreateCache(tableName + "_CACHE").size(CachePeekMode.PRIMARY));
+ });
+
+ Stream.of("T2").forEach(tableName -> {
+ StringBuilder sb = new StringBuilder("INSERT INTO ").append(tableName)
+ .append("(ID, AK, IDX_VAL, VAL) VALUES ");
+
+ for (int i = 0; i < 10000; ++i) {
+ sb.append("(").append(i).append(", ")
+ .append(i).append(",")
+ .append("'name_").append(i).append("', ")
+ .append("'name_").append(i).append("')");
+
+ if (i < ENTRIES_COUNT - 1)
+ sb.append(",");
+ }
+
+ sql(sb.toString());
+
+ assertEquals(ENTRIES_COUNT, client.getOrCreateCache(tableName + "_CACHE").size(CachePeekMode.PRIMARY));
+ });
+
+ sql("ANALYZE PUBLIC.T1(ID), PUBLIC.T2(ID,AK), PUBLIC.DICT(ID) WITH \"NULLS=0,DISTINCT=10000,TOTAL=10000\"");
+
+ clearIntercepted();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ clearIntercepted();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int nodeCount() {
+ return 8;
+ }
+
+ /** */
+ @Test
+ public void testSimple() {
+ execute("select count(*) from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals(1L, res.get(0).get(0));
+ },
+ 123);
+
+ execute("select * from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select * from T1 where ? = T1.ID",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select VAL, IDX_VAL, ID from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123);
+
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ },
+ 123, 123);
+ }
+
+ /** */
+ @Test
+ public void testNullsInCondition() {
+ execute("select * from T1 where T1.ID is NULL",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ });
+
+ execute("select * from T1 where T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, new Object[]{ null });
+
+ execute("select * from T1 where T1.ID is NULL and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, 123);
+
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ }, null, 123);
+
+ execute("select * from T1 where T1.ID is NULL or T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ }, 123);
+
+ execute("select * from T1 where T1.ID = ? or T1.ID = ?",
+ res -> {
+ assertPartitions(partition("T1_CACHE", 123));
+ assertNodes(node("T1_CACHE", 123));
+
+ assertEquals(1, res.size());
+ assertEquals("name_123", res.get(0).get(1));
+ }, null, 123);
+ }
+
+ /** */
+ @Test
+ public void testEmptyConditions() {
+ execute("select * from T1 where T1.ID = ? and T1.ID = ?",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518);
+
+ execute("select * from T1 where T1.ID = ? and (T1.ID = ? OR T1.ID = ? OR T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+
+ execute("select * from T1 where (T1.ID = ? OR T1.ID = ?) AND (T1.ID = ? OR T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+
+ execute("select * from T1 where (T1.ID = ? AND T1.ID = ?) OR (T1.ID = ? AND T1.ID = ?)",
+ res -> {
+ assertPartitions();
+ assertNodes();
+
+ assertTrue(res.isEmpty());
+ },
+ 123, 518, 781, 295);
+ }
+
+ /** */
+ @Test
+ public void testSelectIn() {
+ IntStream.of(2, 6).forEach(i -> {
+ testSelect(i, true, "_KEY");
+ testSelect(i, true, "ID");
+ });
+ }
+
+
+ /** */
+ @Test
+ public void testSelectOr() {
+ testSelect(1, false, "_KEY");
+ testSelect(1, false, "ID");
+
+ IntStream.of(2, 6).forEach(i -> {
+ testSelect(i, false, "_KEY");
+ testSelect(i, false, "ID");
+ });
+ }
+
+ /** */
+ @Test
+ public void testSimpleJoin() {
+ // Key (not alias).
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T1.ID = ?",
+ (res) -> {
+// assertPartitions(partition("T1_CACHE", 123));
+// assertNodes(node("T1_CACHE", 123));
+ assertEquals(1, res.size());
+ assertEquals(123, res.get(0).get(0));
+ },
+ 123
+ );
+
+ // Key (alias).
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T1._KEY = ?",
+ (res) -> {
+// assertPartitions(partition("T1_CACHE", 125));
+// assertNodes(node("T1_CACHE", 125));
+
+ assertEquals(1, res.size());
+ assertEquals(125, res.get(0).get(0));
+ },
+ 125
+ );
+
+ // Non-affinity key.
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T2.ID = ?",
+ (res) -> {
+ assertEquals(1, res.size());
+ assertEquals(125, res.get(0).get(0));
+ },
+ 125
+ );
+
+ // Affinity key.
+ execute("SELECT * FROM T1 INNER JOIN T2 ON T1.ID = T2.AK WHERE T2.AK = ?",
+ (res) -> {
+// assertPartitions(partition("T2_CACHE", 125));
+ assertEquals(1, res.size());
+ assertEquals(125, res.get(0).get(0));
+ },
+ 125
+ );
+ }
+
+ /** */
+ private void testSelect(int sz, boolean withIn, String column) {
+ assertTrue(sz >= 1);
+ int[] values = ThreadLocalRandom.current().ints(0, ENTRIES_COUNT).distinct().limit(sz).toArray();
+
+ StringBuilder query;
+
+ if (!withIn || sz == 1)
+ query = new StringBuilder("select * from T1 where ");
+ else
+ query = new StringBuilder("select * from T1 where T1.").append(column).append(" in (");
+
+ for (int i = 0; i < sz; ++i) {
+ if (!withIn || sz == 1)
+ query.append("T1.").append(column).append("= ?");
+ else
+ query.append('?');
+
+ if (sz == 1)
+ break;
+
+ if (i == sz - 1)
+ query.append(!withIn ? "" : ")");
+ else
+ query.append(!withIn ? " OR " : ", ");
+ }
+
+ execute(query.toString(),
+ res -> {
+ assertPartitions(IntStream.of(values).map(i -> partition("T1_CACHE", i)).toArray());
+ assertNodes(IntStream.of(values).mapToObj(i -> node("T1_CACHE", i)).toArray(ClusterNode[]::new));
+
+ assertEquals(values.length, res.size());
+
+ assertEquals(
+ IntStream.of(values).sorted().boxed().collect(Collectors.toList()),
+ res.stream().map(row -> row.get(0)).sorted().collect(Collectors.toList())
+ );
+ },
+ IntStream.of(values).boxed().toArray(Integer[]::new));
+ }
+
+ /** */
+ public void execute(String sql, Consumer<List<List<?>>> resConsumer, Object... args) {
+ log.info(">>> TEST COMBINATION: \"" + sql + "\"");
+
+ // Execute query as is.
+ log.info("Execute \"" + sql + "\" with args " + Arrays.toString(args));
+
+ List<List<?>> res = sql(sql, args);
+
+ resConsumer.accept(res);
+ clearIntercepted();
+
+ // Start filling arguments recursively.
+ if (args != null && args.length > 0)
+ executeCombinations0(sql, resConsumer, new HashSet<>(), args);
+ }
+
+ /** */
+ private void executeCombinations0(
+ String sql,
+ Consumer<List<List<?>>> resConsumer,
+ Set<String> executedSqls,
+ Object... args
+ ) {
+ assert args != null && args.length > 0;
+
+ // Get argument positions.
+ List<Integer> paramPoss = new ArrayList<>();
+
+ int pos = 0;
+
+ while (true) {
+ int paramPos = sql.indexOf('?', pos);
+
+ if (paramPos == -1)
+ break;
+
+ paramPoss.add(paramPos);
+
+ pos = paramPos + 1;
+ }
+
+ for (int i = 0; i < args.length; i++) {
+ // Prepare new SQL and arguments.
+ int paramPos = paramPoss.get(i);
+
+ String newSql = sql.substring(0, paramPos) + (args[i] instanceof String ? "'" + args[i] + "'" : args[i])
+ + sql.substring(paramPos + 1);
+
+ Object[] newArgs = new Object[args.length - 1];
+
+ int newArgsPos = 0;
+
+ for (int j = 0; j < args.length; j++) {
+ if (j != i)
+ newArgs[newArgsPos++] = args[j];
+ }
+
+ // Execute if this combination was never executed before.
+ if (executedSqls.add(newSql)) {
+ log.info("Execute sql \"" + newSql + "\"");
+
+ List<List<?>> res = sql(newSql, newArgs);
+
+ resConsumer.accept(res);
+ clearIntercepted();
+ }
+
+ // Continue recursively.
+ if (newArgs.length > 0)
+ executeCombinations0(newSql, resConsumer, executedSqls, newArgs);
+ }
+ }
+
+ /** */
+ protected static void assertPartitions(int... expParts) {
+ Collection<Integer> expParts0 = new TreeSet<>();
+
+ if (!F.isEmpty(expParts)) {
+ for (int expPart : expParts)
+ expParts0.add(expPart);
+ }
+
+ TreeSet<Integer> actualParts = new TreeSet<>(INTERCEPTED_PARTS);
+
+ assertEquals("Unexpected partitions [exp=" + Arrays.toString(expParts) + ", actual=" + actualParts + ']',
+ expParts0, actualParts);
+ }
+
+ /** */
+ protected int partition(String cacheName, Object key) {
+ return client.affinity(cacheName).partition(key);
+ }
+
+ /** */
+ protected ClusterNode node(String cacheName, Object key) {
+ return G.allGrids().stream()
+ .filter(ign -> ign.affinity(cacheName).isPrimary(ign.cluster().localNode(), key))
+ .map(ign -> ign.cluster().localNode()).findFirst().orElse(null);
+ }
+
+ /** */
+ protected static void assertNodes(ClusterNode... expNodes) {
+ Collection<ClusterNode> expNodes0 = new TreeSet<>(Comparator.comparing(ClusterNode::id));
+
+ if (!F.isEmpty(expNodes)) {
+ expNodes0.addAll(Arrays.asList(expNodes));
+ }
+
+ TreeSet<ClusterNode> actualNodes = new TreeSet<>(INTERCEPTED_NODES);
+
+ assertEquals("Unexpected nodes [exp=" + Arrays.toString(expNodes) + ", actual=" + actualNodes + ']',
Review Comment:
`Arrays.toString(expNodes)` -> `expNodes0` (output will be sorted)
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org