IGNITE-10307: SQL: Partition pruning for joins. This closes #5774.
authordevozerov <vozerov@gridgain.com>
Thu, 17 Jan 2019 15:47:59 +0000 (18:47 +0300)
committerdevozerov <vozerov@gridgain.com>
Thu, 17 Jan 2019 15:47:59 +0000 (18:47 +0300)
24 files changed:
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionAffinityFunctionType.java [new file with mode: 0644]
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionAllNode.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionCompositeNode.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionConstantNode.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionExtractor.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionGroupNode.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionJoinCondition.java [new file with mode: 0644]
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionJoinGroup.java [new file with mode: 0644]
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionNode.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionNoneNode.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionParameterNode.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionResult.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionSingleNode.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTable.java [new file with mode: 0644]
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTableAffinityDescriptor.java [new file with mode: 0644]
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTableDescriptor.java [deleted file]
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTableModel.java [new file with mode: 0644]
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQuerySplitter.java
modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/BetweenOperationExtractPartitionSelfTest.java
modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/JoinPartitionPruningSelfTest.java [new file with mode: 0644]
modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java

index 0c1b1d6..5e00aea 100644 (file)
@@ -92,7 +92,8 @@ import org.apache.ignite.internal.processors.query.QueryIndexDescriptorImpl;
 import org.apache.ignite.internal.processors.query.RunningQueryManager;
 import org.apache.ignite.internal.processors.query.SqlClientContext;
 import org.apache.ignite.internal.processors.query.UpdateSourceIterator;
-import org.apache.ignite.internal.processors.query.h2.affinity.PartitionNode;
+import org.apache.ignite.internal.processors.query.h2.affinity.PartitionExtractor;
+import org.apache.ignite.internal.processors.query.h2.affinity.PartitionResult;
 import org.apache.ignite.internal.processors.query.h2.database.H2TreeClientIndex;
 import org.apache.ignite.internal.processors.query.h2.database.H2TreeIndex;
 import org.apache.ignite.internal.processors.query.h2.database.io.H2ExtrasInnerIO;
@@ -139,6 +140,7 @@ import org.apache.ignite.internal.sql.command.SqlSetStreamingCommand;
 import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashMap;
 import org.apache.ignite.internal.util.GridEmptyCloseableIterator;
 import org.apache.ignite.internal.util.GridSpinBusyLock;
+import org.apache.ignite.internal.util.IgniteUtils;
 import org.apache.ignite.internal.util.lang.GridCloseableIterator;
 import org.apache.ignite.internal.util.lang.GridPlainRunnable;
 import org.apache.ignite.internal.util.lang.IgniteInClosure2X;
@@ -248,6 +250,9 @@ public class IgniteH2Indexing implements GridQueryIndexing {
     /** */
     private DdlStatementsProcessor ddlProc;
 
+    /** Partition extractor. */
+    private PartitionExtractor partExtractor;
+
     /** */
     private final RunningQueryManager runningQueryMgr = new RunningQueryManager();
 
@@ -2006,46 +2011,27 @@ public class IgniteH2Indexing implements GridQueryIndexing {
         boolean cursorCreated = false;
 
         try {
-            // TODO: Use intersection (https://issues.apache.org/jira/browse/IGNITE-10567)
-            int partitions[] = qry.getPartitions();
-
-            if (partitions == null && twoStepQry.derivedPartitions() != null) {
-                try {
-                    PartitionNode partTree = twoStepQry.derivedPartitions().tree();
-
-                    Collection<Integer> partitions0 = partTree.apply(qry.getArgs());
-
-                    if (F.isEmpty(partitions0))
-                        partitions = new int[0];
-                    else {
-                        partitions = new int[partitions0.size()];
-
-                        int i = 0;
-
-                        for (Integer part : partitions0)
-                            partitions[i++] = part;
-                    }
-
-                    if (partitions.length == 0) { //here we know that result of requested query is empty
-                        return new QueryCursorImpl<List<?>>(new Iterable<List<?>>() {
-                            @Override public Iterator<List<?>> iterator() {
-                                return new Iterator<List<?>>() {
-                                    @Override public boolean hasNext() {
-                                        return false;
-                                    }
+            // When explicit partitions are set, there must be an owning cache they should be applied to.
+            int explicitParts[] = qry.getPartitions();
+            PartitionResult derivedParts = twoStepQry.derivedPartitions();
+
+            int parts[] = calculatePartitions(explicitParts, derivedParts, qry.getArgs());
+
+            if (parts != null && parts.length == 0) {
+                return new QueryCursorImpl<>(new Iterable<List<?>>() {
+                    @Override public Iterator<List<?>> iterator() {
+                        return new Iterator<List<?>>() {
+                            @Override public boolean hasNext() {
+                                return false;
+                            }
 
-                                    @Override public List<?> next() {
-                                        return null;
-                                    }
-                                };
+                            @SuppressWarnings("IteratorNextCanNotThrowNoSuchElementException")
+                            @Override public List<?> next() {
+                                return null;
                             }
-                        });
+                        };
                     }
-                }
-                catch (IgniteCheckedException e) {
-                    throw new CacheException("Failed to calculate derived partitions: [qry=" + qry.getSql() +
-                        ", params=" + Arrays.deepToString(qry.getArgs()) + "]", e);
-                }
+                });
             }
 
             Iterable<List<?>> iter = runQueryTwoStep(
@@ -2057,7 +2043,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
                 qry.getTimeout(),
                 cancel,
                 qry.getArgs(),
-                partitions,
+                parts,
                 qry.isLazy(),
                 mvccTracker
             );
@@ -2079,6 +2065,43 @@ public class IgniteH2Indexing implements GridQueryIndexing {
     }
 
     /**
+     * Calculate partitions for the query.
+     *
+     * @param explicitParts Explicit partitions provided in SqlFieldsQuery.partitions property.
+     * @param derivedParts Derived partitions found during partition pruning.
+     * @param args Arguments.
+     * @return Calculated partitions or {@code null} if failed to calculate and there should be a broadcast.
+     */
+    @SuppressWarnings("ZeroLengthArrayAllocation")
+    private int[] calculatePartitions(int[] explicitParts, PartitionResult derivedParts, Object[] args) {
+        if (!F.isEmpty(explicitParts))
+            return explicitParts;
+        else if (derivedParts != null) {
+            try {
+                Collection<Integer> realParts = derivedParts.tree().apply(args);
+
+                if (F.isEmpty(realParts))
+                    return IgniteUtils.EMPTY_INTS;
+                else {
+                    int[] realParts0 = new int[realParts.size()];
+
+                    int i = 0;
+
+                    for (Integer realPart : realParts)
+                        realParts0[i++] = realPart;
+
+                    return realParts0;
+                }
+            }
+            catch (IgniteCheckedException e) {
+                throw new CacheException("Failed to calculate derived partitions for query.", e);
+            }
+        }
+
+        return null;
+    }
+
+    /**
      * Do initial parsing of the statement and create query caches, if needed.
      * @param c Connection.
      * @param sqlQry Query.
@@ -2375,6 +2398,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
         dmlProc = new DmlStatementsProcessor(ctx, this);
         ddlProc = new DdlStatementsProcessor(ctx, schemaMgr);
 
+        partExtractor = new PartitionExtractor(this);
+
         if (JdbcUtils.serializer != null)
             U.warn(log, "Custom H2 serialization is already configured, will override.");
 
@@ -2670,6 +2695,13 @@ public class IgniteH2Indexing implements GridQueryIndexing {
     }
 
     /**
+     * @return Partition extractor.
+     */
+    public PartitionExtractor partitionExtractor() {
+        return partExtractor;
+    }
+
+    /**
      * Collect cache identifiers from two-step query.
      *
      * @param mainCacheId Id of main cache.
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionAffinityFunctionType.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionAffinityFunctionType.java
new file mode 100644 (file)
index 0000000..4c88fcb
--- /dev/null
@@ -0,0 +1,48 @@
+/*
+ * 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.h2.affinity;
+
+/**
+ * Affinity function type.
+ */
+public enum PartitionAffinityFunctionType {
+    /** Custom affintiy function. */
+    CUSTOM(0),
+
+    /** Rendezvous affinity function. */
+    RENDEZVOUS(1);
+
+    /** Value. */
+    private final int val;
+
+    /**
+     * Constructor.
+     *
+     * @param val Value.
+     */
+    PartitionAffinityFunctionType(int val) {
+        this.val = val;
+    }
+
+    /**
+     * @return Value.
+     */
+    public int value() {
+        return val;
+    }
+}
index 842d82c..30860f5 100644 (file)
@@ -41,6 +41,11 @@ public class PartitionAllNode implements PartitionNode {
     }
 
     /** {@inheritDoc} */
+    @Override public int joinGroup() {
+        return PartitionTableModel.GRP_NONE;
+    }
+
+    /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(PartitionAllNode.class, this);
     }
index 2cb330f..45ceaaf 100644 (file)
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.query.h2.affinity;
 
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.S;
 
 import java.util.Collection;
@@ -65,6 +66,8 @@ public class PartitionCompositeNode implements PartitionNode {
                 return null;
 
             // (A, B) and (B, C) -> (B)
+            leftParts = new HashSet<>(leftParts);
+
             leftParts.retainAll(rightParts);
         }
         else {
@@ -77,6 +80,8 @@ public class PartitionCompositeNode implements PartitionNode {
                 return leftParts;
 
             // (A, B) or (B, C) -> (A, B, C)
+            leftParts = new HashSet<>(leftParts);
+
             leftParts.addAll(rightParts);
         }
 
@@ -84,6 +89,12 @@ public class PartitionCompositeNode implements PartitionNode {
     }
 
     /** {@inheritDoc} */
+    @Override public int joinGroup() {
+        // Similar to group node, we cannot cache join group value here as it may be changed dynamically.
+        return left.joinGroup();
+    }
+
+    /** {@inheritDoc} */
     @Override public PartitionNode optimize() {
         PartitionNode left = this.left;
         PartitionNode right = this.right;
@@ -103,9 +114,15 @@ public class PartitionCompositeNode implements PartitionNode {
             return optimizeSpecial(right, left);
 
         // If one of child nodes cannot be optimized, nothing can be done further.
-        // Note that we cannot return "this" here because left or right parts might have been optimized.
-        if (left instanceof PartitionCompositeNode || right instanceof PartitionCompositeNode)
+        // Note that we cannot return "this" here because left or right parts might have been changed.
+        if (left instanceof PartitionCompositeNode || right instanceof PartitionCompositeNode) {
+            // Should be "NONE" for AND in fact, but this would violate current non-collocated join semantics as
+            // explained in "optimizeSimpleAnd" method below.
+            if (left.joinGroup() != right.joinGroup())
+                return PartitionAllNode.INSTANCE;
+
             return new PartitionCompositeNode(left, right, op);
+        }
 
         // Try optimizing composite nodes.
         if (left instanceof PartitionGroupNode)
@@ -182,6 +199,11 @@ public class PartitionCompositeNode implements PartitionNode {
     private PartitionNode optimizeGroupAnd(PartitionGroupNode left, PartitionNode right) {
         assert op == PartitionCompositeNodeOperator.AND;
 
+        // Should be "NONE" for AND in fact, but this would violate current non-collocated join semantics as
+        // explained in "optimizeSimpleAnd" method below.
+        if (left.joinGroup() != right.joinGroup())
+            return PartitionAllNode.INSTANCE;
+
         // Optimistic check whether both sides are equal.
         if (right instanceof PartitionGroupNode) {
             PartitionGroupNode right0 = (PartitionGroupNode)right;
@@ -206,22 +228,50 @@ public class PartitionCompositeNode implements PartitionNode {
             }
 
             if (rightConsts != null) {
-                // {A, B) and (B, C) -> (B).
-                consts.retainAll(rightConsts);
-
-                if (consts.isEmpty())
-                    // {A, B) and (C, D) -> NONE.
-                    return PartitionNoneNode.INSTANCE;
-                else if (consts.size() == 1)
+                // Try to merge nodes if they belong to the same table.
+                boolean sameTbl = true;
+                String curTblAlias = null;
+
+                for (PartitionSingleNode curConst : consts) {
+                    if (curTblAlias == null)
+                        curTblAlias = curConst.table().alias();
+                    else if (!F.eq(curTblAlias, curConst.table().alias())) {
+                        sameTbl = false;
+
+                        break;
+                    }
+                }
+
+                if (sameTbl) {
+                    for (PartitionSingleNode curConst : rightConsts) {
+                        if (curTblAlias == null)
+                            curTblAlias = curConst.table().alias();
+                        else if (!F.eq(curTblAlias, curConst.table().alias())) {
+                            sameTbl = false;
+
+                            break;
+                        }
+                    }
+                }
+
+                if (sameTbl) {
                     // {A, B) and (B, C) -> (B).
-                    return consts.iterator().next();
-                else
-                    // {A, B, C) and (B, C, D) -> (B, C).
-                    return new PartitionGroupNode(consts);
+                    consts.retainAll(rightConsts);
+
+                    if (consts.isEmpty())
+                        // {A, B) and (C, D) -> NONE.
+                        return PartitionNoneNode.INSTANCE;
+                    else if (consts.size() == 1)
+                        // {A, B) and (B, C) -> (B).
+                        return consts.iterator().next();
+                    else
+                        // {A, B, C) and (B, C, D) -> (B, C).
+                        return new PartitionGroupNode(consts);
+                }
             }
         }
 
-        // Otherwise it is a mixed set of concrete partitions and arguments. Cancel optimization.
+        // Otherwise it is a mixed set of concrete partitions and arguments possibly from different caches.
         // Note that in fact we can optimize expression to certain extent (e.g. (A) and (B, :C) -> (A) and (:C)),
         // but resulting expression is always composite node still, which cannot be optimized on upper levels.
         // So we skip any fine-grained optimization in favor of simplicity.
@@ -238,6 +288,10 @@ public class PartitionCompositeNode implements PartitionNode {
     private PartitionNode optimizeGroupOr(PartitionGroupNode left, PartitionNode right) {
         assert op == PartitionCompositeNodeOperator.OR;
 
+        // Cannot merge disjunctive nodes if they belong to different join groups.
+        if (left.joinGroup() != right.joinGroup())
+            return PartitionAllNode.INSTANCE;
+
         HashSet<PartitionSingleNode> siblings = new HashSet<>(left.siblings());
 
         if (right instanceof PartitionSingleNode)
@@ -278,18 +332,28 @@ public class PartitionCompositeNode implements PartitionNode {
     private PartitionNode optimizeSimpleAnd(PartitionSingleNode left, PartitionSingleNode right) {
         assert op == PartitionCompositeNodeOperator.AND;
 
+        // Currently we do not merge such nodes because it may violate existing broken (!!!) join semantics.
+        // Normally, if we have two non-collocated partition sets, then this should be an empty set for collocated
+        // query mode. Unfortunately, current semantics of collocated query mode assume that even though both sides
+        // of expression are located on random nodes, there is a slight chance that they may accidentally reside on
+        // a single node and hence return some rows. We return "ALL" here to keep this broken semantics consistent
+        // irrespective of whether partition pruning is used or not. Once non-collocated joins are fixed, this
+        // condition will be changed to "NONE".
+        if (left.joinGroup() != right.joinGroup())
+            return PartitionAllNode.INSTANCE;
+
         // Check if both sides are equal.
         if (left.equals(right))
             // (X) and (X) -> X
             // (:X) and (:X) -> :X
             return left;
 
-        // If both sides are constants, and they are not equal, this is empty set.
-        if (left.constant() && right.constant())
+        // If both sides are constants from the same table and they are not equal, this is empty set.
+        if (left.constant() && right.constant() && F.eq(left.table().alias(), right.tbl.alias()))
             // X and Y -> NONE
             return PartitionNoneNode.INSTANCE;
 
-        // Otherwise it is a mixed set, cannot reduce.
+        // Otherwise this is a mixed set, cannot reduce.
         // X and :Y -> (X) AND (:Y)
         return new PartitionCompositeNode(left, right, PartitionCompositeNodeOperator.AND);
     }
@@ -304,7 +368,21 @@ public class PartitionCompositeNode implements PartitionNode {
     private PartitionNode optimizeSimpleOr(PartitionSingleNode left, PartitionSingleNode right) {
         assert op == PartitionCompositeNodeOperator.OR;
 
-        return left.equals(right) ? left : PartitionGroupNode.merge(left, right);
+        // Cannot merge disjunctive nodes if they belong to different join groups.
+        if (left.joinGroup() != right.joinGroup())
+            return PartitionAllNode.INSTANCE;
+
+        // (A) or (A) -> (A)
+        if (left.equals(right))
+            return left;
+
+        // (A) or (B) -> (A, B)
+        HashSet<PartitionSingleNode> nodes = new HashSet<>();
+
+        nodes.add(left);
+        nodes.add(right);
+
+        return new PartitionGroupNode(nodes);
     }
 
     /** {@inheritDoc} */
index 9efafe4..9e258ae 100644 (file)
@@ -29,11 +29,11 @@ public class PartitionConstantNode extends PartitionSingleNode {
     /**
      * Constructor.
      *
-     * @param resolver Resolver.
+     * @param tbl Table.
      * @param part Partition.
      */
-    public PartitionConstantNode(PartitionTableDescriptor resolver, int part) {
-        super(resolver);
+    public PartitionConstantNode(PartitionTable tbl, int part) {
+        super(tbl);
 
         this.part = part;
     }
index 12549ce..0898e63 100644 (file)
@@ -21,8 +21,12 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.IgniteSystemProperties;
 import org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery;
+import org.apache.ignite.internal.processors.query.h2.H2Utils;
 import org.apache.ignite.internal.processors.query.h2.IgniteH2Indexing;
 import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
 import org.apache.ignite.internal.processors.query.h2.sql.GridSqlAlias;
@@ -30,6 +34,7 @@ import org.apache.ignite.internal.processors.query.h2.sql.GridSqlAst;
 import org.apache.ignite.internal.processors.query.h2.sql.GridSqlColumn;
 import org.apache.ignite.internal.processors.query.h2.sql.GridSqlConst;
 import org.apache.ignite.internal.processors.query.h2.sql.GridSqlElement;
+import org.apache.ignite.internal.processors.query.h2.sql.GridSqlJoin;
 import org.apache.ignite.internal.processors.query.h2.sql.GridSqlOperation;
 import org.apache.ignite.internal.processors.query.h2.sql.GridSqlOperationType;
 import org.apache.ignite.internal.processors.query.h2.sql.GridSqlParameter;
@@ -41,6 +46,9 @@ import org.h2.table.Column;
 import org.h2.value.Value;
 import org.jetbrains.annotations.Nullable;
 
+import java.util.ArrayList;
+import java.util.Collections;
+
 /**
  * Partition tree extractor.
  */
@@ -84,14 +92,11 @@ public class PartitionExtractor {
 
         GridSqlSelect select = (GridSqlSelect)qry;
 
-        // Currently we can extract data only from a single table.
-        GridSqlTable tbl = unwrapTable(select.from());
-
-        if (tbl == null)
-            return null;
+        // Prepare table model.
+        PartitionTableModel tblModel = prepareTableModel(select.from());
 
         // Do extract.
-        PartitionNode tree = extractFromExpression(select.where());
+        PartitionNode tree = extractFromExpression(select.where(), tblModel, false);
 
         assert tree != null;
 
@@ -101,10 +106,8 @@ public class PartitionExtractor {
         if (tree instanceof PartitionAllNode)
             return null;
 
-        // Return.
-        PartitionTableDescriptor desc = descriptor(tbl.dataTable());
-
-        return new PartitionResult(desc, tree);
+        // Done.
+        return new PartitionResult(tree, tblModel.joinGroupAffinity(tree.joinGroup()));
     }
 
     /**
@@ -114,21 +117,25 @@ public class PartitionExtractor {
      * @return Partition result or {@code null} if nothing is resolved.
      */
     @SuppressWarnings("IfMayBeConditional")
-    public PartitionResult merge(List<GridCacheSqlQuery> qrys) {
+    public PartitionResult mergeMapQueries(List<GridCacheSqlQuery> qrys) {
         // Check if merge is possible.
-        PartitionTableDescriptor desc = null;
+        PartitionTableAffinityDescriptor aff = null;
 
         for (GridCacheSqlQuery qry : qrys) {
             PartitionResult qryRes = (PartitionResult)qry.derivedPartitions();
 
+            // Failed to get results for one query -> broadcast.
             if (qryRes == null)
-                // Failed to get results for one query -> broadcast.
                 return null;
 
-            if (desc == null)
-                desc = qryRes.descriptor();
-            else if (!F.eq(desc, qryRes.descriptor()))
-                // Queries refer to different tables, cannot merge -> broadcast.
+            // This only possible if query is resolved to "NONE". Will be skipped later during map request prepare.
+            if (qryRes.affinity() == null)
+                continue;
+
+            if (aff == null)
+                aff = qryRes.affinity();
+            else if (!aff.isCompatible(qryRes.affinity()))
+                // Queries refer to incompatible affinity groups, cannot merge -> broadcast.
                 return null;
         }
 
@@ -152,33 +159,232 @@ public class PartitionExtractor {
         if (tree instanceof PartitionAllNode)
             return null;
 
-        return new PartitionResult(desc, tree);
+        // If there is no affinity, then we assume "NONE" result.
+        assert aff != null || tree == PartitionNoneNode.INSTANCE;
+
+        return new PartitionResult(tree, aff);
     }
 
     /**
-     * Try unwrapping the table.
+     * Prepare table model.
      *
-     * @param from From.
-     * @return Table or {@code null} if not a table.
+     * @param from FROM clause.
+     * @return Join model.
      */
-    @Nullable private static GridSqlTable unwrapTable(GridSqlAst from) {
-        if (from instanceof GridSqlAlias)
-            from = from.child();
+    private PartitionTableModel prepareTableModel(GridSqlAst from) {
+        PartitionTableModel res = new PartitionTableModel();
+
+        prepareTableModel0(from, res);
 
-        if (from instanceof GridSqlTable)
-            return (GridSqlTable)from;
+        return res;
+    }
+
+    /**
+     * Prepare tables which will be used in join model.
+     *
+     * @param from From flag.
+     * @param model Table model.
+     * @return {@code True} if extracted tables successfully, {@code false} if failed to extract.
+     */
+    private List<PartitionTable> prepareTableModel0(GridSqlAst from, PartitionTableModel model) {
+        if (from instanceof GridSqlJoin) {
+            // Process JOIN recursively.
+            GridSqlJoin join = (GridSqlJoin)from;
+
+            List<PartitionTable> leftTbls = prepareTableModel0(join.leftTable(), model);
+            List<PartitionTable> rightTbls = prepareTableModel0(join.rightTable(), model);
+
+            if (join.isLeftOuter()) {
+                // "a LEFT JOIN b" is transformed into "a", and "b" is put into special stop-list.
+                // If a condition is met on "b" afterwards, we will ignore it.
+                for (PartitionTable rightTbl : rightTbls)
+                    model.addExcludedTable(rightTbl.alias());
+
+                return leftTbls;
+            }
+
+            // Extract equi-join or cross-join from condition. For normal INNER JOINs most likely we will have "1=1"
+            // cross join here, real join condition will be found in WHERE clause later.
+            PartitionJoinCondition cond = parseJoinCondition(join.on());
+
+            if (cond != null && !cond.cross())
+                model.addJoin(cond);
+
+            ArrayList<PartitionTable> res = new ArrayList<>(leftTbls.size() + rightTbls.size());
+
+            res.addAll(leftTbls);
+            res.addAll(rightTbls);
+
+            return res;
+        }
+
+        PartitionTable tbl = prepareTable(from, model);
+
+        return tbl != null ? Collections.singletonList(tbl) : Collections.emptyList();
+    }
+
+    /**
+     * Try parsing condition as simple JOIN codition. Only equijoins are supported for now, so anything more complex
+     * than "A.a = B.b" are not processed.
+     *
+     * @param on Initial AST.
+     * @return Join condition or {@code null} if not simple equijoin.
+     */
+    private static PartitionJoinCondition parseJoinCondition(GridSqlElement on) {
+        if (on instanceof GridSqlOperation) {
+            GridSqlOperation on0 = (GridSqlOperation)on;
+
+            if (on0.operationType() == GridSqlOperationType.EQUAL) {
+                // Check for cross-join first.
+                GridSqlConst leftConst = unwrapConst(on0.child(0));
+                GridSqlConst rightConst = unwrapConst(on0.child(1));
+
+                if (leftConst != null && rightConst != null) {
+                    try {
+                        int leftConstval = leftConst.value().getInt();
+                        int rightConstVal = rightConst.value().getInt();
+
+                        if (leftConstval == rightConstVal)
+                            return PartitionJoinCondition.CROSS;
+                    }
+                    catch (Exception ignore) {
+                        // No-op.
+                    }
+                }
+
+                // This is not cross-join, neither normal join between columns.
+                if (leftConst != null || rightConst != null)
+                    return null;
+
+                // Check for normal equi-join.
+                GridSqlColumn left = unwrapColumn(on0.child(0));
+                GridSqlColumn right = unwrapColumn(on0.child(1));
+
+                if (left != null && right != null) {
+                    String leftAlias = left.tableAlias();
+                    String rightAlias = right.tableAlias();
+
+                    String leftCol = left.columnName();
+                    String rightCol = right.columnName();
+
+                    return new PartitionJoinCondition(leftAlias, rightAlias, leftCol, rightCol);
+                }
+            }
+        }
 
         return null;
     }
 
     /**
+     * Prepare single table.
+     *
+     * @param from Expression.
+     * @param tblModel Table model.
+     * @return Added table or {@code null} if table is exlcuded from the model.
+     */
+    private static PartitionTable prepareTable(GridSqlAst from, PartitionTableModel tblModel) {
+        // Unwrap alias. We assume that every table must be aliased.
+        assert from instanceof GridSqlAlias;
+
+        String alias = ((GridSqlAlias)from).alias();
+
+        from = from.child();
+
+        if (from instanceof GridSqlTable) {
+            // Normal table.
+            GridSqlTable from0 = (GridSqlTable)from;
+
+            GridH2Table tbl0 = from0.dataTable();
+
+            // Unknown table type, e.g. temp table.
+            if (tbl0 == null) {
+                tblModel.addExcludedTable(alias);
+
+                return null;
+            }
+
+            String cacheName = tbl0.cacheName();
+
+            String affColName = null;
+            String secondAffColName = null;
+
+            for (Column col : tbl0.getColumns()) {
+                if (tbl0.isColumnForPartitionPruningStrict(col)) {
+                    if (affColName == null)
+                        affColName = col.getName();
+                    else {
+                        secondAffColName = col.getName();
+
+                        // Break as we cannot have more than two affinity key columns.
+                        break;
+                    }
+                }
+            }
+
+            PartitionTable tbl = new PartitionTable(alias, cacheName, affColName, secondAffColName);
+
+            PartitionTableAffinityDescriptor aff = affinityForCache(tbl0.cacheInfo().config());
+
+            if (aff == null) {
+                // Non-standard affinity, exclude table.
+                tblModel.addExcludedTable(alias);
+
+                return null;
+            }
+
+            tblModel.addTable(tbl, aff);
+
+            return tbl;
+        }
+        else {
+            // Subquery/union/view, etc.
+            assert alias != null;
+
+            tblModel.addExcludedTable(alias);
+
+            return null;
+        }
+    }
+
+    /**
+     * Prepare affinity identifier for cache.
+     *
+     * @param ccfg Cache configuration.
+     * @return Affinity identifier.
+     */
+    private static PartitionTableAffinityDescriptor affinityForCache(CacheConfiguration ccfg) {
+        // Partition could be extracted only from PARTITIONED caches.
+        if (ccfg.getCacheMode() != CacheMode.PARTITIONED)
+            return null;
+
+        PartitionAffinityFunctionType aff = ccfg.getAffinity().getClass().equals(RendezvousAffinityFunction.class) ?
+            PartitionAffinityFunctionType.RENDEZVOUS : PartitionAffinityFunctionType.CUSTOM;
+
+        boolean hasNodeFilter = ccfg.getNodeFilter() != null &&
+            !(ccfg.getNodeFilter() instanceof CacheConfiguration.IgniteAllNodesPredicate);
+
+        return new PartitionTableAffinityDescriptor(
+            aff,
+            ccfg.getAffinity().partitions(),
+            hasNodeFilter,
+            ccfg.getDataRegionName()
+        );
+    }
+
+    /**
      * Extract partitions from expression.
      *
      * @param expr Expression.
+     * @param tblModel Table model.
+     * @param disjunct Whether current processing frame is located under disjunction ("OR"). In this case we cannot
+     *                 rely on join expressions like (A.a = B.b) to build co-location model because another conflicting
+     *                 join expression on the same tables migth be located on the other side of the "OR".
+     *                 Example: "JOIN ON A.a = B.b OR A.a > B.b".
      * @return Partition tree.
      */
     @SuppressWarnings("EnumSwitchStatementWhichMissesCases")
-    private PartitionNode extractFromExpression(GridSqlAst expr) throws IgniteCheckedException {
+    private PartitionNode extractFromExpression(GridSqlAst expr, PartitionTableModel tblModel, boolean disjunct)
+        throws IgniteCheckedException {
         PartitionNode res = PartitionAllNode.INSTANCE;
 
         if (expr instanceof GridSqlOperation) {
@@ -186,22 +392,22 @@ public class PartitionExtractor {
 
             switch (op.operationType()) {
                 case AND:
-                    res = extractFromAnd(op);
+                    res = extractFromAnd(op, tblModel, disjunct);
 
                     break;
 
                 case OR:
-                    res = extractFromOr(op);
+                    res = extractFromOr(op, tblModel);
 
                     break;
 
                 case IN:
-                    res = extractFromIn(op);
+                    res = extractFromIn(op, tblModel);
 
                     break;
 
                 case EQUAL:
-                    res = extractFromEqual(op);
+                    res = extractFromEqual(op, tblModel, disjunct);
             }
         }
 
@@ -213,18 +419,21 @@ public class PartitionExtractor {
      * Extract partition information from AND.
      *
      * @param op Operation.
+     * @param tblModel Table model.
+     * @param disjunct Disjunction marker.
      * @return Partition.
      */
-    private PartitionNode extractFromAnd(GridSqlOperation op) throws IgniteCheckedException {
+    private PartitionNode extractFromAnd(GridSqlOperation op, PartitionTableModel tblModel, boolean disjunct)
+        throws IgniteCheckedException {
         assert op.size() == 2;
 
-        PartitionNode betweenNodes = tryExtractBetween(op);
+        PartitionNode betweenNodes = tryExtractBetween(op, tblModel);
 
         if (betweenNodes != null)
             return betweenNodes;
 
-        PartitionNode part1 = extractFromExpression(op.child(0));
-        PartitionNode part2 = extractFromExpression(op.child(1));
+        PartitionNode part1 = extractFromExpression(op.child(0), tblModel, disjunct);
+        PartitionNode part2 = extractFromExpression(op.child(1), tblModel, disjunct);
 
         return new PartitionCompositeNode(part1, part2, PartitionCompositeNodeOperator.AND);
     }
@@ -233,13 +442,16 @@ public class PartitionExtractor {
      * Extract partition information from OR.
      *
      * @param op Operation.
+     * @param tblModel Table model.
      * @return Partition.
      */
-    private PartitionNode extractFromOr(GridSqlOperation op) throws IgniteCheckedException {
+    private PartitionNode extractFromOr(GridSqlOperation op, PartitionTableModel tblModel)
+        throws IgniteCheckedException {
         assert op.size() == 2;
 
-        PartitionNode part1 = extractFromExpression(op.child(0));
-        PartitionNode part2 = extractFromExpression(op.child(1));
+        // Parse inner expressions recursively with disjuncion flag set.
+        PartitionNode part1 = extractFromExpression(op.child(0), tblModel, true);
+        PartitionNode part2 = extractFromExpression(op.child(1), tblModel, true);
 
         return new PartitionCompositeNode(part1, part2, PartitionCompositeNodeOperator.OR);
     }
@@ -248,9 +460,11 @@ public class PartitionExtractor {
      * Extract partition information from IN.
      *
      * @param op Operation.
+     * @param tblModel Table model.
      * @return Partition.
      */
-    private PartitionNode extractFromIn(GridSqlOperation op) throws IgniteCheckedException {
+    private PartitionNode extractFromIn(GridSqlOperation op, PartitionTableModel tblModel)
+        throws IgniteCheckedException {
         // Operation should contain at least two children: left (column) and right (const or column).
         if (op.size() < 2)
             return PartitionAllNode.INSTANCE;
@@ -258,11 +472,9 @@ public class PartitionExtractor {
         // Left operand should be column.
         GridSqlAst left = op.child();
 
-        GridSqlColumn leftCol;
+        GridSqlColumn leftCol = unwrapColumn(left);
 
-        if (left instanceof GridSqlColumn)
-            leftCol = (GridSqlColumn)left;
-        else
+        if (leftCol == null)
             return PartitionAllNode.INSTANCE;
 
         // Can work only with Ignite tables.
@@ -291,8 +503,8 @@ public class PartitionExtractor {
                 // set globally. Hence, returning null.
                 return PartitionAllNode.INSTANCE;
 
-            // Do extract.
-            PartitionSingleNode part = extractSingle(leftCol.column(), rightConst, rightParam);
+            // Extract.
+            PartitionSingleNode part = extractSingle(leftCol, rightConst, rightParam, tblModel);
 
             // Same thing as above: single unknown partition in disjunction defeats optimization.
             if (part == null)
@@ -308,19 +520,20 @@ public class PartitionExtractor {
      * Extract partition information from equality.
      *
      * @param op Operation.
+     * @param tblModel Table model.
+     * @param disjunct Disjunction flag. When set possible join expression will not be processed.
      * @return Partition.
      */
-    private PartitionNode extractFromEqual(GridSqlOperation op) throws IgniteCheckedException {
+    private PartitionNode extractFromEqual(GridSqlOperation op, PartitionTableModel tblModel, boolean disjunct)
+        throws IgniteCheckedException {
         assert op.operationType() == GridSqlOperationType.EQUAL;
 
         GridSqlElement left = op.child(0);
         GridSqlElement right = op.child(1);
 
-        GridSqlColumn leftCol;
+        GridSqlColumn leftCol = unwrapColumn(left);
 
-        if (left instanceof GridSqlColumn)
-            leftCol = (GridSqlColumn)left;
-        else
+        if (leftCol == null)
             return PartitionAllNode.INSTANCE;
 
         if (!(leftCol.column().getTable() instanceof GridH2Table))
@@ -337,10 +550,20 @@ public class PartitionExtractor {
             rightConst = null;
             rightParam = (GridSqlParameter)right;
         }
-        else
+        else {
+            if (right instanceof GridSqlColumn) {
+                if (!disjunct) {
+                    PartitionJoinCondition cond = parseJoinCondition(op);
+
+                    if (cond != null && !cond.cross())
+                        tblModel.addJoin(cond);
+                }
+
+            }
             return PartitionAllNode.INSTANCE;
+        }
 
-        PartitionSingleNode part = extractSingle(leftCol.column(), rightConst, rightParam);
+        PartitionSingleNode part = extractSingle(leftCol, rightConst, rightParam, tblModel);
 
         return part != null ? part : PartitionAllNode.INSTANCE;
     }
@@ -351,52 +574,81 @@ public class PartitionExtractor {
      * @param leftCol Left column.
      * @param rightConst Right constant.
      * @param rightParam Right parameter.
+     * @param tblModel Table model.
      * @return Partition or {@code null} if failed to extract.
      */
-    @Nullable private PartitionSingleNode extractSingle(Column leftCol, GridSqlConst rightConst,
-        GridSqlParameter rightParam) throws IgniteCheckedException {
+    @Nullable private PartitionSingleNode extractSingle(
+        GridSqlColumn leftCol,
+        GridSqlConst rightConst,
+        GridSqlParameter rightParam,
+        PartitionTableModel tblModel
+    ) throws IgniteCheckedException {
         assert leftCol != null;
-        assert leftCol.getTable() != null;
-        assert leftCol.getTable() instanceof GridH2Table;
 
-        GridH2Table tbl = (GridH2Table)leftCol.getTable();
+        Column leftCol0 = leftCol.column();
 
-        if (!tbl.isColumnForPartitionPruning(leftCol))
+        assert leftCol0.getTable() != null;
+        assert leftCol0.getTable() instanceof GridH2Table;
+
+        GridH2Table tbl = (GridH2Table)leftCol0.getTable();
+
+        if (!tbl.isColumnForPartitionPruning(leftCol0))
             return null;
 
-        PartitionTableDescriptor tblDesc = descriptor(tbl);
+        PartitionTable tbl0 = tblModel.table(leftCol.tableAlias());
+
+        // If table is in ignored set, then we cannot use it for partition extraction.
+        if (tbl0 == null)
+            return null;
 
         if (rightConst != null) {
-            int part = idx.kernalContext().affinity().partition(tbl.cacheName(), rightConst.value().getObject());
+            Object constVal = H2Utils.convert(rightConst.value().getObject(), idx, leftCol0.getType());
+
+            int part = idx.kernalContext().affinity().partition(tbl.cacheName(), constVal);
 
-            return new PartitionConstantNode(tblDesc, part);
+            return new PartitionConstantNode(tbl0, part);
         }
         else if (rightParam != null)
-            return new PartitionParameterNode(tblDesc, idx, rightParam.index(), leftCol.getType());
+            return new PartitionParameterNode(tbl0, idx, rightParam.index(), leftCol0.getType());
         else
             return null;
     }
 
     /**
-     * Get descriptor from table.
+     * Unwrap constant if possible.
+     *
+     * @param ast AST.
+     * @return Constant or {@code null} if not a constant.
+     */
+    @Nullable public static GridSqlConst unwrapConst(GridSqlAst ast) {
+        return ast instanceof GridSqlConst ? (GridSqlConst)ast : null;
+    }
+
+    /**
+     * Unwrap column if possible.
      *
-     * @param tbl Table.
-     * @return Descriptor.
+     * @param ast AST.
+     * @return Column or {@code null} if not a column.
      */
-    private static PartitionTableDescriptor descriptor(GridH2Table tbl) {
-        return new PartitionTableDescriptor(tbl.cacheName(), tbl.getName());
+    @Nullable public static GridSqlColumn unwrapColumn(GridSqlAst ast) {
+        if (ast instanceof GridSqlAlias)
+            ast = ast.child();
+
+        return ast instanceof GridSqlColumn ? (GridSqlColumn)ast : null;
     }
 
     /**
      * Try to extract partitions from {@code op} assuming that it's between operation or simple range.
      *
      * @param op Sql operation.
+     * @param tblModel Table model.
      * @return {@code PartitionSingleNode} if operation reduced to one partition,
      *   {@code PartitionGroupNode} if operation reduced to multiple partitions or null if operation is neither
      *   between nor simple range. Null also returns if it's not possible to extract partitions from given operation.
      * @throws IgniteCheckedException If failed.
      */
-    private PartitionNode tryExtractBetween(GridSqlOperation op) throws IgniteCheckedException {
+    private PartitionNode tryExtractBetween(GridSqlOperation op, PartitionTableModel tblModel)
+        throws IgniteCheckedException {
         // Between operation (or similar range) should contain exact two children.
         assert op.size() == 2;
 
@@ -487,11 +739,18 @@ public class PartitionExtractor {
 
         Set<PartitionSingleNode> parts = new HashSet<>();
 
-        PartitionTableDescriptor desc = descriptor(tbl);
+        PartitionTable tbl0 = tblModel.table(leftCol.tableAlias());
+
+        // If table is in ignored set, then we cannot use it for partition extraction.
+        if (tbl0 == null)
+            return null;
 
         for (long i = leftLongVal; i <= rightLongVal; i++) {
-            parts.add(new PartitionConstantNode(desc,
-                idx.kernalContext().affinity().partition((tbl).cacheName(), i)));
+            Object constVal = H2Utils.convert(i, idx, leftCol.column().getType());
+
+            int part = idx.kernalContext().affinity().partition(tbl0.cacheName(), constVal);
+
+            parts.add(new PartitionConstantNode(tbl0, part));
 
             if (parts.size() > maxPartsCntBetween)
                 return null;
index ef3a154..3d66439 100644 (file)
@@ -35,22 +35,6 @@ public class PartitionGroupNode implements PartitionNode {
     private final Set<PartitionSingleNode> siblings;
 
     /**
-     * Merge two simple nodes.
-     *
-     * @param node1 Node 1.
-     * @param node2 Node 2.
-     * @return Group node.
-     */
-    public static PartitionGroupNode merge(PartitionSingleNode node1, PartitionSingleNode node2) {
-        HashSet<PartitionSingleNode> nodes = new HashSet<>();
-
-        nodes.add(node1);
-        nodes.add(node2);
-
-        return new PartitionGroupNode(nodes);
-    }
-
-    /**
      * Constructor.
      *
      * @param siblings Partitions.
@@ -72,6 +56,13 @@ public class PartitionGroupNode implements PartitionNode {
         return res;
     }
 
+    /** {@inheritDoc} */
+    @Override public int joinGroup() {
+        // Note that we cannot cache join group in constructor. We have strong invariant that all siblings always
+        // belongs to the same group. However, number of this group may be changed during expression tree traversing.
+        return siblings.iterator().next().joinGroup();
+    }
+
     /**
      * @return Siblings
      */
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionJoinCondition.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionJoinCondition.java
new file mode 100644 (file)
index 0000000..244c301
--- /dev/null
@@ -0,0 +1,132 @@
+/*
+ * 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.h2.affinity;
+
+import org.apache.ignite.internal.util.typedef.F;
+
+/**
+ * Join condition.
+ */
+public class PartitionJoinCondition {
+    /** Cross JOIN. */
+    public static final PartitionJoinCondition CROSS = new PartitionJoinCondition(null, null, null, null, true);
+
+    /** Left alias. */
+    private final String leftAlias;
+
+    /** Right alias. */
+    private final String rightAlias;
+
+    /** Left column name. */
+    private final String leftCol;
+
+    /** Right column name. */
+    private final String rightCol;
+
+    /** Whether this is a cross-join. */
+    private final boolean cross;
+
+    /**
+     * Constructor.
+     *
+     * @param leftAlias Left alias.
+     * @param rightAlias Right alias.
+     * @param leftCol Left column name.
+     * @param rightCol Right column name.
+     */
+    public PartitionJoinCondition(String leftAlias, String rightAlias, String leftCol, String rightCol) {
+        this(leftAlias, rightAlias, leftCol, rightCol, false);
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param leftAlias Left alias.
+     * @param rightAlias Right alias.
+     * @param leftCol Left column name.
+     * @param rightCol Right column name.
+     * @param cross Whether this is a cross-join.
+     */
+    private PartitionJoinCondition(String leftAlias, String rightAlias, String leftCol, String rightCol,
+        boolean cross) {
+        this.leftAlias = leftAlias;
+        this.rightAlias = rightAlias;
+        this.leftCol = leftCol;
+        this.rightCol = rightCol;
+        this.cross = cross;
+    }
+
+    /**
+     * Left alias.
+     */
+    public String leftAlias() {
+        return leftAlias;
+    }
+
+    /**
+     * Right alias.
+     */
+    public String rightAlias() {
+        return rightAlias;
+    }
+
+    /**
+     * @return Left column.
+     */
+    public String leftColumn() {
+        return leftCol;
+    }
+
+    /**
+     * @return Right column.
+     */
+    public String rightColumn() {
+        return rightCol;
+    }
+
+    /**
+     * @return Wheter this is a cross-join.
+     */
+    public boolean cross() {
+        return cross;
+    }
+
+    /** {@inheritDoc} */
+    @Override public int hashCode() {
+        int res = leftAlias.hashCode();
+
+        res = 31 * res + rightAlias.hashCode();
+        res = 31 * res + leftCol.hashCode();
+        res = 31 * res + rightCol.hashCode();
+        res = 31 * res + Boolean.hashCode(cross);
+
+        return res;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean equals(Object obj) {
+        if (obj instanceof PartitionJoinCondition) {
+            PartitionJoinCondition other = (PartitionJoinCondition)obj;
+
+            return F.eq(leftAlias, other.leftAlias) && F.eq(rightAlias, other.rightAlias) &&
+                F.eq(leftCol, other.leftCol) && F.eq(rightCol, other.rightCol) && F.eq(cross, other.cross);
+        }
+
+        return false;
+    }
+}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionJoinGroup.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionJoinGroup.java
new file mode 100644 (file)
index 0000000..641d013
--- /dev/null
@@ -0,0 +1,81 @@
+/*
+ * 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.h2.affinity;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+
+/**
+ * Group of joined tables whose affinity function could be "merged".
+ */
+@SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
+public class PartitionJoinGroup {
+    /** Tables within a group. */
+    private final Collection<PartitionTable> tbls = Collections.newSetFromMap(new IdentityHashMap<>());
+
+    /** Affinity function descriptor. */
+    private final PartitionTableAffinityDescriptor affDesc;
+
+    /**
+     * Constructor.
+     *
+     * @param affDesc Affinity function descriptor.
+     */
+    public PartitionJoinGroup(PartitionTableAffinityDescriptor affDesc) {
+        this.affDesc = affDesc;
+    }
+
+    /**
+     * @return Tables in a group.
+     */
+    public Collection<PartitionTable> tables() {
+        return tbls;
+    }
+
+    /**
+     * Add table to the group.
+     *
+     * @param tbl Table.
+     * @return This for chaining.
+     */
+    public PartitionJoinGroup addTable(PartitionTable tbl) {
+        tbls.add(tbl);
+
+        return this;
+    }
+
+    /**
+     * Remove table from the group.
+     *
+     * @param tbl Table.
+     * @return If group is empty after removal.
+     */
+    public boolean removeTable(PartitionTable tbl) {
+        tbls.remove(tbl);
+
+        return tbls.isEmpty();
+    }
+
+    /**
+     * @return Affinity descriptor.
+     */
+    public PartitionTableAffinityDescriptor affinityDescriptor() {
+        return affDesc;
+    }
+}
index 238739c..7372fc2 100644 (file)
@@ -35,6 +35,11 @@ public interface PartitionNode {
     Collection<Integer> apply(Object... args) throws IgniteCheckedException;
 
     /**
+     * @return Join group for the given node.
+     */
+    int joinGroup();
+
+    /**
      * Try optimizing partition nodes into a simpler form.
      *
      * @return Optimized node or {@code this} if optimization failed.
index b3a1358..5d4b324 100644 (file)
@@ -42,6 +42,11 @@ public class PartitionNoneNode implements PartitionNode {
     }
 
     /** {@inheritDoc} */
+    @Override public int joinGroup() {
+        return PartitionTableModel.GRP_NONE;
+    }
+
+    /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(PartitionNoneNode.class, this);
     }
index 0624f2c..e9f4880 100644 (file)
@@ -45,7 +45,7 @@ public class PartitionParameterNode extends PartitionSingleNode {
      * @param idx Parameter index.
      * @param dataType Parameter data type.
      */
-    public PartitionParameterNode(PartitionTableDescriptor tbl, IgniteH2Indexing indexing, int idx,
+    public PartitionParameterNode(PartitionTable tbl, IgniteH2Indexing indexing, int idx,
         int dataType) {
         super(tbl);
 
index 13e7f87..daa14d3 100644 (file)
@@ -24,37 +24,36 @@ import org.apache.ignite.internal.util.typedef.internal.S;
  * Partition extraction result.
  */
 public class PartitionResult {
-    /** Descriptor. */
-    @GridToStringInclude
-    private final PartitionTableDescriptor desc;
-
     /** Tree. */
     @GridToStringInclude
     private final PartitionNode tree;
 
+    /** Affinity function. */
+    private final PartitionTableAffinityDescriptor aff;
+
     /**
      * Constructor.
      *
-     * @param desc Descriptor.
      * @param tree Tree.
+     * @param aff Affinity function.
      */
-    public PartitionResult(PartitionTableDescriptor desc, PartitionNode tree) {
-        this.desc = desc;
+    public PartitionResult(PartitionNode tree, PartitionTableAffinityDescriptor aff) {
         this.tree = tree;
+        this.aff = aff;
     }
 
     /**
-     * Descriptor.
+     * Tree.
      */
-    public PartitionTableDescriptor descriptor() {
-        return desc;
+    public PartitionNode tree() {
+        return tree;
     }
 
     /**
-     * Tree.
+     * @return Affinity function.
      */
-    public PartitionNode tree() {
-        return tree;
+    public PartitionTableAffinityDescriptor affinity() {
+        return aff;
     }
 
     /** {@inheritDoc} */
index caf966c..35e7d30 100644 (file)
@@ -30,14 +30,14 @@ import java.util.Collections;
 public abstract class PartitionSingleNode implements PartitionNode {
     /** Table descriptor. */
     @GridToStringExclude
-    protected final PartitionTableDescriptor tbl;
+    protected final PartitionTable tbl;
 
     /**
      * Constructor.
      *
      * @param tbl Table descriptor.
      */
-    protected PartitionSingleNode(PartitionTableDescriptor tbl) {
+    protected PartitionSingleNode(PartitionTable tbl) {
         this.tbl = tbl;
     }
 
@@ -59,17 +59,29 @@ public abstract class PartitionSingleNode implements PartitionNode {
      */
     public abstract boolean constant();
 
+    /** {@inheritDoc} */
+    @Override public int joinGroup() {
+        return tbl.joinGroup();
+    }
+
     /**
      * @return Partition for constant node, index for argument node.
      */
     public abstract int value();
 
+    /**
+     * @return Underlying table.
+     */
+    public PartitionTable table() {
+        return tbl;
+    }
+
     /** {@inheritDoc} */
     @Override public int hashCode() {
         int hash = (constant() ? 1 : 0);
 
         hash = 31 * hash + value();
-        hash = 31 * hash + tbl.hashCode();
+        hash = 31 * hash + tbl.alias().hashCode();
 
         return hash;
     }
@@ -84,6 +96,7 @@ public abstract class PartitionSingleNode implements PartitionNode {
 
         PartitionSingleNode other = (PartitionSingleNode)obj;
 
-        return F.eq(constant(), other.constant()) && F.eq(value(), other.value()) && F.eq(tbl, other.tbl);
+        return F.eq(constant(), other.constant()) && F.eq(value(), other.value()) &&
+            F.eq(tbl.alias(), other.tbl.alias());
     }
 }
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTable.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTable.java
new file mode 100644 (file)
index 0000000..1b996c1
--- /dev/null
@@ -0,0 +1,113 @@
+/*
+ * 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.h2.affinity;
+
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Single table with affinity info.
+ */
+public class PartitionTable {
+    /** Alias used in the query. */
+    private final String alias;
+
+    /** Cache name. */
+    private final String cacheName;
+
+    /** Affinity column name (if can be resolved). */
+    private final String affColName;
+
+    /** Second affinity column name (possible when _KEY is affinity column and an alias for this column exists. */
+    private final String secondAffColName;
+
+    /** Join group index. */
+    private int joinGrp;
+
+    /**
+     * Constructor.
+     *
+     * @param alias Unique alias.
+     * @param cacheName Cache name.
+     * @param affColName Affinity column name.
+     * @param secondAffColName Second affinity column name.
+     */
+    public PartitionTable(
+        String alias,
+        String cacheName,
+        @Nullable String affColName,
+        @Nullable String secondAffColName
+    ) {
+        this.alias = alias;
+        this.cacheName = cacheName;
+
+        if (affColName == null && secondAffColName != null) {
+            this.affColName = secondAffColName;
+            this.secondAffColName = null;
+        }
+        else {
+            this.affColName = affColName;
+            this.secondAffColName = secondAffColName;
+        }
+    }
+
+    /**
+     * @return Alias.
+     */
+    public String alias() {
+        return alias;
+    }
+
+    /**
+     * @return Cache name.
+     */
+    public String cacheName() {
+        return cacheName;
+    }
+
+    /**
+     * Check whether passed column is affinity column.
+     *
+     * @param colName Column name.
+     * @return {@code True} if affinity column.
+     */
+    @SuppressWarnings("BooleanMethodIsAlwaysInverted")
+    public boolean isAffinityColumn(String colName) {
+        return F.eq(colName, affColName) || F.eq(colName, secondAffColName);
+    }
+
+    /**
+     * @return Join group index.
+     */
+    public int joinGroup() {
+        return joinGrp;
+    }
+
+    /**
+     * @param joinGrp Join group index.
+     */
+    public void joinGroup(int joinGrp) {
+        this.joinGrp = joinGrp;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(PartitionTable.class, this);
+    }
+}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTableAffinityDescriptor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTableAffinityDescriptor.java
new file mode 100644 (file)
index 0000000..21dab9c
--- /dev/null
@@ -0,0 +1,97 @@
+/*
+ * 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.h2.affinity;
+
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+import java.io.Serializable;
+
+/**
+ * Affinity function descriptor. Used to compare affinity functions of two tables.
+ */
+public class PartitionTableAffinityDescriptor implements Serializable {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Affinity function type. */
+    private final PartitionAffinityFunctionType affFunc;
+
+    /** Number of partitions. */
+    private final int parts;
+
+    /** Whether node filter is set. */
+    private final boolean hasNodeFilter;
+
+    /** Data region name. */
+    private final String dataRegion;
+
+    /**
+     * Constructor.
+     *
+     * @param affFunc Affinity function type.
+     * @param parts Number of partitions.
+     * @param hasNodeFilter Whether node filter is set.
+     * @param dataRegion Data region.
+     */
+    public PartitionTableAffinityDescriptor(
+        PartitionAffinityFunctionType affFunc,
+        int parts,
+        boolean hasNodeFilter,
+        String dataRegion
+    ) {
+        this.affFunc = affFunc;
+        this.parts = parts;
+        this.hasNodeFilter = hasNodeFilter;
+        this.dataRegion = dataRegion;
+    }
+
+    /**
+     * Check is provided descriptor is compatible with this instance (i.e. can be used in the same co-location group).
+     *
+     * @param other Other descriptor.
+     * @return {@code True} if compatible.
+     */
+    @SuppressWarnings("BooleanMethodIsAlwaysInverted")
+    public boolean isCompatible(PartitionTableAffinityDescriptor other) {
+        if (other == null)
+            return false;
+
+        // Rendezvous affinity function is deterministic and doesn't depend on previous cluster view changes.
+        // In future other user affinity functions would be applicable as well if explicityl marked deterministic.
+        if (affFunc == PartitionAffinityFunctionType.RENDEZVOUS) {
+            // We cannot be sure that two caches are co-located if custom node filter is present.
+            // Nota that technically we may try to compare two filters. However, this adds unnecessary complexity
+            // and potential deserialization issues when SQL is called from client nodes or thin clients.
+            if (!hasNodeFilter) {
+                return
+                    other.affFunc == PartitionAffinityFunctionType.RENDEZVOUS &&
+                    !other.hasNodeFilter &&
+                    other.parts == parts &&
+                    F.eq(other.dataRegion, dataRegion);
+            }
+        }
+
+        return false;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(PartitionTableAffinityDescriptor.class, this);
+    }
+}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTableDescriptor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTableDescriptor.java
deleted file mode 100644 (file)
index b11e07e..0000000
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * 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.h2.affinity;
-
-import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.internal.S;
-
-/**
- * Partition resolver.
- */
-public class PartitionTableDescriptor {
-    /** Cache name. */
-    private final String cacheName;
-
-    /** Table name. */
-    private final String tblName;
-
-    /**
-     * Constructor.
-     *
-     * @param cacheName Cache name.
-     * @param tblName Table name.
-     */
-    public PartitionTableDescriptor(String cacheName, String tblName) {
-        this.cacheName = cacheName;
-        this.tblName = tblName;
-    }
-
-    /**
-     * @return Cache name.
-     */
-    public String cacheName() {
-        return cacheName;
-    }
-
-    /** {@inheritDoc} */
-    @Override public int hashCode() {
-        return 31 * cacheName.hashCode() + tblName.hashCode();
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean equals(Object o) {
-        if (o == this)
-            return true;
-
-        if (o.getClass() != getClass())
-            return false;
-
-        PartitionTableDescriptor other = (PartitionTableDescriptor)o;
-
-        return F.eq(cacheName, other.cacheName) && F.eq(tblName, other.tblName);
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(PartitionTableDescriptor.class, this);
-    }
-}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTableModel.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/affinity/PartitionTableModel.java
new file mode 100644 (file)
index 0000000..6393941
--- /dev/null
@@ -0,0 +1,157 @@
+/*
+ * 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.h2.affinity;
+
+import org.jetbrains.annotations.Nullable;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Partition join model. Describes how tables are joined with each other.
+ */
+public class PartitionTableModel {
+    /** Join group which could not be applied (e.g. for "ALL" case). */
+    public static final int GRP_NONE = -1;
+
+    /** All tables observed during parsing excluding outer. */
+    private final Map<String, PartitionTable> tbls = new HashMap<>();
+
+    /** Join groups. */
+    private final Map<Integer, PartitionJoinGroup> grps = new HashMap<>();
+
+    /** Talbes which are excluded from partition pruning calculation. */
+    private Set<String> excludedTblNames;
+
+    /** Group index generator */
+    private int grpIdxGen;
+
+    /**
+     * Add table.
+     *
+     * @param tbl Table.
+     * @param aff Affinity descriptor.
+     */
+    public void addTable(PartitionTable tbl, PartitionTableAffinityDescriptor aff) {
+        int grpIdx = grpIdxGen++;
+
+        tbl.joinGroup(grpIdx);
+
+        tbls.put(tbl.alias(), tbl);
+        grps.put(grpIdx, new PartitionJoinGroup(aff).addTable(tbl));
+    }
+
+    /**
+     * Get table by alias.
+     *
+     * @param alias Alias.
+     * @return Table or {@code null} if it cannot be used for partition pruning.
+     */
+    @Nullable public PartitionTable table(String alias) {
+        PartitionTable res = tbls.get(alias);
+
+        assert res != null || (excludedTblNames != null && excludedTblNames.contains(alias));
+
+        return res;
+    }
+
+    /**
+     * Add excluded table
+     *
+     * @param alias Alias.
+     */
+    public void addExcludedTable(String alias) {
+        PartitionTable tbl = tbls.remove(alias);
+
+        if (tbl != null) {
+            PartitionJoinGroup grp = grps.get(tbl.joinGroup());
+
+            assert grp != null;
+
+            if (grp.removeTable(tbl))
+                grps.remove(tbl.joinGroup());
+        }
+
+        if (excludedTblNames == null)
+            excludedTblNames = new HashSet<>();
+
+        excludedTblNames.add(alias);
+    }
+
+    /**
+     * Add equi-join condition. Two joined tables may possibly be merged into a single group.
+     *
+     * @param cond Condition.
+     */
+    public void addJoin(PartitionJoinCondition cond) {
+        PartitionTable leftTbl = tbls.get(cond.leftAlias());
+        PartitionTable rightTbl = tbls.get(cond.rightAlias());
+
+        assert leftTbl != null || (excludedTblNames != null && excludedTblNames.contains(cond.leftAlias()));
+        assert rightTbl != null || (excludedTblNames != null && excludedTblNames.contains(cond.rightAlias()));
+
+        // At least one tables is excluded, return.
+        if (leftTbl == null || rightTbl == null)
+            return;
+
+        // At least one column in condition is not affinity column, return.
+        if (!leftTbl.isAffinityColumn(cond.leftColumn()) || !rightTbl.isAffinityColumn(cond.rightColumn()))
+            return;
+
+        // Remember join group of the right table as it will be changed below.
+        int rightGrpId = rightTbl.joinGroup();
+
+        PartitionJoinGroup leftGrp = grps.get(leftTbl.joinGroup());
+        PartitionJoinGroup rightGrp = grps.get(rightGrpId);
+
+        assert leftGrp != null;
+        assert rightGrp != null;
+
+        // Groups are not compatible, return.
+        if (!leftGrp.affinityDescriptor().isCompatible(rightGrp.affinityDescriptor()))
+            return;
+
+        // Safe to merge groups.
+        for (PartitionTable tbl : rightGrp.tables()) {
+            tbl.joinGroup(leftTbl.joinGroup());
+
+            leftGrp.addTable(tbl);
+        }
+
+        grps.remove(rightGrpId);
+    }
+
+    /**
+     * Get affinity descriptor for the group.
+     *
+     * @param grpId Group ID.
+     * @return Affinity descriptor or {@code null} if there is no affinity descriptor (e.g. for "NONE" result).
+     */
+    @Nullable public PartitionTableAffinityDescriptor joinGroupAffinity(int grpId) {
+        if (grpId == GRP_NONE)
+            return null;
+
+        PartitionJoinGroup grp = grps.get(grpId);
+
+        assert grp != null;
+
+        return grp.affinityDescriptor();
+    }
+}
index b69a011..fff12e3 100644 (file)
@@ -62,7 +62,6 @@ import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.cache.CacheMode.PARTITIONED;
 import static org.apache.ignite.internal.processors.query.h2.opt.GridH2KeyValueRowOnheap.DEFAULT_COLUMNS_COUNT;
-import static org.apache.ignite.internal.processors.query.h2.opt.GridH2RowDescriptor.COL_NOT_EXISTS;
 
 /**
  * H2 Table implementation.
@@ -101,8 +100,8 @@ public class GridH2Table extends TableBase {
     /** */
     private final IndexColumn affKeyCol;
 
-    /** */
-    private final int affKeyColId;
+    /** Whether affinity key column is the whole cache key. */
+    private final boolean affKeyColIsKey;
 
     /** */
     private final LongAdder size = new LongAdder();
@@ -122,6 +121,9 @@ public class GridH2Table extends TableBase {
     /** Flag remove index or not when table will be destroyed. */
     private volatile boolean rmIndex;
 
+    /** Columns with thread-safe access. */
+    private volatile Column[] safeColumns;
+
     /**
      * Creates table.
      *
@@ -141,36 +143,8 @@ public class GridH2Table extends TableBase {
         this.desc = desc;
         this.cacheInfo = cacheInfo;
 
-        if (!desc.type().customAffinityKeyMapper()) {
-            String affKeyFieldName = desc.type().affinityKey();
-
-            if (affKeyFieldName != null) {
-                if (doesColumnExist(affKeyFieldName)) {
-                    int colId = getColumn(affKeyFieldName).getColumnId();
-
-                    if (desc.isKeyColumn(colId)) {
-                        affKeyCol = indexColumn(GridH2KeyValueRowOnheap.KEY_COL, SortOrder.ASCENDING);
-                        affKeyColId = GridH2KeyValueRowOnheap.KEY_COL;
-                    }
-                    else {
-                        affKeyCol = indexColumn(colId, SortOrder.ASCENDING);
-                        affKeyColId = colId;
-                    }
-                }
-                else {
-                    affKeyCol = null;
-                    affKeyColId = COL_NOT_EXISTS;
-                }
-            }
-            else {
-                affKeyCol = indexColumn(GridH2KeyValueRowOnheap.KEY_COL, SortOrder.ASCENDING);
-                affKeyColId = GridH2KeyValueRowOnheap.KEY_COL;
-            }
-        }
-        else {
-            affKeyCol = null;
-            affKeyColId = COL_NOT_EXISTS;
-        }
+        affKeyCol = calculateAffinityKeyColumn();
+        affKeyColIsKey = affKeyCol != null && desc.isKeyColumn(affKeyCol.column.getColumnId());
 
         this.rowFactory = rowFactory;
 
@@ -210,6 +184,36 @@ public class GridH2Table extends TableBase {
     }
 
     /**
+     * Calculate affinity key column which will be used for partition pruning and distributed joins.
+     *
+     * @return Affinity column or {@code null} if none can be used.
+     */
+    private IndexColumn calculateAffinityKeyColumn() {
+        // If custome affinity key mapper is set, we do not know how to convert _KEY to partition, return null.
+        if (desc.type().customAffinityKeyMapper())
+            return null;
+
+        String affKeyFieldName = desc.type().affinityKey();
+
+        // If explicit affinity key field is not set, then use _KEY.
+        if (affKeyFieldName == null)
+            return indexColumn(GridH2KeyValueRowOnheap.KEY_COL, SortOrder.ASCENDING);
+
+        // If explicit affinity key field is set, but is not found in the table, do not use anything.
+        if (!doesColumnExist(affKeyFieldName))
+            return null;
+
+        int colId = getColumn(affKeyFieldName).getColumnId();
+
+        // If affinity key column is either _KEY or it's alias (QueryEntity.keyFieldName), normalize it to _KEY.
+        if (desc.isKeyColumn(colId))
+            return indexColumn(GridH2KeyValueRowOnheap.KEY_COL, SortOrder.ASCENDING);
+
+        // Otherwise use column as is.
+        return indexColumn(colId, SortOrder.ASCENDING);
+    }
+
+    /**
      * @return {@code true} If this is a partitioned table.
      */
     public boolean isPartitioned() {
@@ -230,9 +234,65 @@ public class GridH2Table extends TableBase {
      * @return {@code True} if affinity key column.
      */
     public boolean isColumnForPartitionPruning(Column col) {
+        return isColumnForPartitionPruning0(col, false);
+    }
+
+    /**
+     * Check whether passed column could be used for partition transfer during partition pruning on joined tables and
+     * for external affinity calculation (e.g. on thin clients).
+     * <p>
+     * Note that it is different from {@link #isColumnForPartitionPruning(Column)} method in that not every column
+     * which qualifies for partition pruning can be used by thin clients or join partinion prunining logic.
+     * <p>
+     * Consider the following schema:
+     * <pre>
+     * CREATE TABLE dept (id PRIMARY KEY);
+     * CREATE TABLE emp (id, dept_id AFFINITY KEY, PRIMARY KEY(id, dept_id));
+     * </pre>
+     * For expression-based partition pruning on "emp" table on the <b>server side</b> we may use both "_KEY" and
+     * "dept_id" columns, as passing them through standard affinity workflow will yield the same result:
+     * dept_id -> part
+     * _KEY -> dept_id -> part
+     * <p>
+     * But we cannot use "_KEY" on thin client side, as it doesn't know how to extract affinity key field properly.
+     * Neither we can perform partition transfer in JOINs when "_KEY" is used.
+     * <p>
+     * This is OK as data is collocated, so we can merge partitions extracted from both tables:
+     * <pre>
+     * SELECT * FROM dept d INNER JOIN emp e ON d.id = e.dept_id WHERE e.dept_id=? AND d.id=?
+     * </pre>
+     * But this is not OK as joined data is not collocated, and tables form distinct collocation groups:
+     * <pre>
+     * SELECT * FROM dept d INNER JOIN emp e ON d.id = e._KEY WHERE e.dept_id=? AND d.id=?
+     * </pre>
+     * NB: The last query is not logically correct and will produce empty result. However, it is correct from SQL
+     * perspective, so we should make incorrect assumptions about partitions as it may make situation even worse.
+     *
+     * @param col Column.
+     * @return {@code True} if column could be used for partition extraction on both server and client sides and for
+     *     partition transfer in joins.
+     */
+    public boolean isColumnForPartitionPruningStrict(Column col) {
+        return isColumnForPartitionPruning0(col, true);
+    }
+
+    /**
+     * Internal logic to check whether column qualifies for partition extraction or not.
+     *
+     * @param col Column.
+     * @param strict Strict flag.
+     * @return {@code True} if column could be used for partition.
+     */
+    private boolean isColumnForPartitionPruning0(Column col, boolean strict) {
+        if (affKeyCol == null)
+            return false;
+
         int colId = col.getColumnId();
 
-        return colId == affKeyColId || desc.isKeyColumn(colId);
+        if (colId == affKeyCol.column.getColumnId())
+            return true;
+
+        return (affKeyColIsKey || !strict) && desc.isKeyColumn(colId);
     }
 
     /**
@@ -333,6 +393,7 @@ public class GridH2Table extends TableBase {
      *
      * @param exclusive Exclusive flag.
      */
+    @SuppressWarnings({"LockAcquiredButNotSafelyReleased", "CallToThreadYield"})
     private void lock(boolean exclusive) {
         Lock l = exclusive ? lock.writeLock() : lock.readLock();
 
@@ -980,12 +1041,14 @@ public class GridH2Table extends TableBase {
         lock(true);
 
         try {
-            int pos = columns.length;
+            Column[] safeColumns0 = safeColumns;
+
+            int pos = safeColumns0.length;
 
-            Column[] newCols = new Column[columns.length + cols.size()];
+            Column[] newCols = new Column[safeColumns0.length + cols.size()];
 
             // First, let's copy existing columns to new array
-            System.arraycopy(columns, 0, newCols, 0, columns.length);
+            System.arraycopy(safeColumns0, 0, newCols, 0, safeColumns0.length);
 
             // And now, let's add new columns
             for (QueryField col : cols) {
@@ -1026,13 +1089,16 @@ public class GridH2Table extends TableBase {
      * @param cols Columns.
      * @param ifExists If EXISTS flag.
      */
+    @SuppressWarnings("ForLoopReplaceableByForEach")
     public void dropColumns(List<String> cols, boolean ifExists) {
         assert !ifExists || cols.size() == 1;
 
         lock(true);
 
         try {
-            int size = columns.length;
+            Column[] safeColumns0 = safeColumns;
+
+            int size = safeColumns0.length;
 
             for (String name : cols) {
                 if (!doesColumnExist(name)) {
@@ -1052,8 +1118,8 @@ public class GridH2Table extends TableBase {
 
             int dst = 0;
 
-            for (int i = 0; i < columns.length; i++) {
-                Column column = columns[i];
+            for (int i = 0; i < safeColumns0.length; i++) {
+                Column column = safeColumns0[i];
 
                 for (String name : cols) {
                     if (F.eq(name, column.getName())) {
@@ -1084,7 +1150,16 @@ public class GridH2Table extends TableBase {
     }
 
     /** {@inheritDoc} */
+    @Override protected void setColumns(Column[] columns) {
+        this.safeColumns = columns;
+
+        super.setColumns(columns);
+    }
+
+    /** {@inheritDoc} */
     @Override public Column[] getColumns() {
+        Column[] safeColumns0 = safeColumns;
+
         Boolean insertHack = INSERT_HACK.get();
 
         if (insertHack != null && insertHack) {
@@ -1093,15 +1168,15 @@ public class GridH2Table extends TableBase {
             StackTraceElement elem = elems[2];
 
             if (F.eq(elem.getClassName(), Insert.class.getName()) && F.eq(elem.getMethodName(), "prepare")) {
-                Column[] columns0 = new Column[columns.length - 3];
+                Column[] columns0 = new Column[safeColumns0.length - 3];
 
-                System.arraycopy(columns, 3, columns0, 0, columns0.length);
+                System.arraycopy(safeColumns0, 3, columns0, 0, columns0.length);
 
                 return columns0;
             }
         }
 
-        return columns;
+        return safeColumns0;
     }
 
     /**
index fcb3424..818777c 100644 (file)
@@ -129,6 +129,9 @@ public class GridSqlQuerySplitter {
     private boolean collocatedGrpBy;
 
     /** */
+    private boolean distributedJoins;
+
+    /** */
     private IdentityHashMap<GridSqlAst, GridSqlAlias> uniqueFromAliases = new IdentityHashMap<>();
 
     /** Partition extractor. */
@@ -137,13 +140,15 @@ public class GridSqlQuerySplitter {
     /**
      * @param params Query parameters.
      * @param collocatedGrpBy If it is a collocated GROUP BY query.
-     * @param idx Indexing.
+     * @param distributedJoins Distributed joins flag.
+     * @param extractor Partition extractor.
      */
-    public GridSqlQuerySplitter(Object[] params, boolean collocatedGrpBy, IgniteH2Indexing idx) {
+    public GridSqlQuerySplitter(Object[] params, boolean collocatedGrpBy, boolean distributedJoins,
+        PartitionExtractor extractor) {
         this.params = params;
         this.collocatedGrpBy = collocatedGrpBy;
-
-        extractor = new PartitionExtractor(idx);
+        this.distributedJoins = distributedJoins;
+        this.extractor = extractor;
     }
 
     /**
@@ -207,7 +212,8 @@ public class GridSqlQuerySplitter {
 
         qry.explain(false);
 
-        GridSqlQuerySplitter splitter = new GridSqlQuerySplitter(params, collocatedGrpBy, h2);
+        GridSqlQuerySplitter splitter = new GridSqlQuerySplitter(params, collocatedGrpBy, distributedJoins,
+            h2.partitionExtractor());
 
         // Normalization will generate unique aliases for all the table filters in FROM.
         // Also it will collect all tables and schemas from the query.
@@ -262,7 +268,7 @@ public class GridSqlQuerySplitter {
         twoStepQry.distributedJoins(distributedJoins);
 
         // all map queries must have non-empty derivedPartitions to use this feature.
-        twoStepQry.derivedPartitions(splitter.extractor.merge(twoStepQry.mapQueries()));
+        twoStepQry.derivedPartitions(splitter.extractor.mergeMapQueries(twoStepQry.mapQueries()));
 
         twoStepQry.forUpdate(forUpdate);
 
@@ -1549,7 +1555,7 @@ public class GridSqlQuerySplitter {
         map.partitioned(hasPartitionedTables(mapQry));
         map.hasSubQueries(hasSubQueries);
 
-        if (map.isPartitioned())
+        if (map.isPartitioned() && !distributedJoins)
             map.derivedPartitions(extractor.extract(mapQry));
 
         mapSqlQrys.add(map);
index 62953ec..15788f2 100644 (file)
@@ -393,9 +393,12 @@ public class GridReduceQueryExecutor {
         int timeoutMillis,
         GridQueryCancel cancel,
         Object[] params,
-        final int[] parts,
+        int[] parts,
         boolean lazy,
         MvccQueryTracker mvccTracker) {
+        if (qry.isLocal() && parts != null)
+            parts = null;
+
         assert !qry.mvccEnabled() || mvccTracker != null;
 
         if (F.isEmpty(params))
index fbdbfb0..f6e73bc 100644 (file)
@@ -27,7 +27,6 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteException;
-import org.apache.ignite.IgniteSystemProperties;
 import org.apache.ignite.cache.CacheMode;
 import org.apache.ignite.cache.QueryEntity;
 import org.apache.ignite.cache.query.FieldsQueryCursor;
@@ -41,7 +40,6 @@ import org.apache.ignite.lang.IgniteInClosure;
 import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.spi.IgniteSpiException;
 import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
-import org.apache.ignite.testframework.GridTestUtils;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -481,22 +479,6 @@ public class BetweenOperationExtractPartitionSelfTest extends GridCommonAbstract
     }
 
     /**
-     * Check custom partitions limit exceeding.
-     */
-    @Test
-    public void testBetweenPartitionsCustomLimitExceeding() {
-        try (GridTestUtils.SystemProperty ignored = new GridTestUtils.
-            SystemProperty(IgniteSystemProperties.IGNITE_SQL_MAX_EXTRACTED_PARTS_FROM_BETWEEN, "4")){
-
-            // Default limit (16) not exceeded.
-            testBetweenConstOperator(BETWEEN_QRY, 1, 4, 4);
-
-            // Default limit (16) exceeded.
-            testBetweenConstOperator(BETWEEN_QRY, 1, 5, 5, EMPTY_PARTITIONS_ARRAY);
-        }
-    }
-
-    /**
      * Check range expression with constant values.
      */
     @Test
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/JoinPartitionPruningSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/JoinPartitionPruningSelfTest.java
new file mode 100644 (file)
index 0000000..1429f3f
--- /dev/null
@@ -0,0 +1,1303 @@
+/*
+ * 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.h2.twostep;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.CacheKeyConfiguration;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.cache.query.annotations.QuerySqlField;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+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.atomic.AtomicInteger;
+import java.util.function.Consumer;
+
+/**
+ * Tests for join partition pruning.
+ */
+@SuppressWarnings("deprecation")
+@RunWith(JUnit4.class)
+public class JoinPartitionPruningSelfTest extends GridCommonAbstractTest {
+    /** Number of intercepted requests. */
+    private static final AtomicInteger INTERCEPTED_REQS = new AtomicInteger();
+
+    /** Parititions tracked during query execution. */
+    private static final ConcurrentSkipListSet<Integer> INTERCEPTED_PARTS = new ConcurrentSkipListSet<>();
+
+    /** IP finder. */
+    private static final TcpDiscoveryVmIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder().setShared(true);
+
+    /** Client node name. */
+    private static final String CLI_NAME = "cli";
+
+    /** Memory. */
+    private static final String REGION_MEM = "mem";
+
+    /** Disk. */
+    private static final String REGION_DISK = "disk";
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTestsStarted() throws Exception {
+        cleanPersistenceDir();
+
+        startGrid(getConfiguration("srv1"));
+        startGrid(getConfiguration("srv2"));
+        startGrid(getConfiguration("srv3"));
+
+        startGrid(getConfiguration(CLI_NAME).setClientMode(true));
+
+        client().cluster().active(true);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        clearIoState();
+
+        Ignite cli = client();
+
+        cli.destroyCaches(cli.cacheNames());
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTestsStopped() throws Exception {
+        stopAllGrids();
+
+        cleanPersistenceDir();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String name) throws Exception {
+        IgniteConfiguration res = super.getConfiguration(name);
+
+        res.setDiscoverySpi(new TcpDiscoverySpi().setIpFinder(IP_FINDER));
+        res.setCommunicationSpi(new TrackingTcpCommunicationSpi());
+
+        res.setLocalHost("127.0.0.1");
+
+        DataRegionConfiguration memRegion =
+            new DataRegionConfiguration().setName(REGION_MEM).setPersistenceEnabled(false);
+
+        DataRegionConfiguration diskRegion =
+            new DataRegionConfiguration().setName(REGION_DISK).setPersistenceEnabled(true);
+
+        res.setDataStorageConfiguration(new DataStorageConfiguration().setDataRegionConfigurations(diskRegion)
+            .setDefaultDataRegionConfiguration(memRegion));
+
+        return res;
+    }
+
+    /**
+     * Test simple join.
+     */
+    @Test
+    public void testSimpleJoin() {
+        createPartitionedTable("t1",
+            pkColumn("k1"),
+            "v2");
+
+        createPartitionedTable("t2",
+            pkColumn("k1"),
+            affinityColumn("ak2"),
+            "v3");
+
+        executeSingle("INSERT INTO t1 VALUES ('1', '1')");
+        executeSingle("INSERT INTO t2 VALUES ('1', '1', '1')");
+
+        executeSingle("INSERT INTO t1 VALUES ('2', '2')");
+        executeSingle("INSERT INTO t2 VALUES ('2', '2', '2')");
+
+        executeSingle("INSERT INTO t1 VALUES ('3', '3')");
+        executeSingle("INSERT INTO t2 VALUES ('3', '3', '3')");
+
+        executeSingle("INSERT INTO t1 VALUES ('4', '4')");
+        executeSingle("INSERT INTO t2 VALUES ('4', '4', '4')");
+
+        executeSingle("INSERT INTO t1 VALUES ('5', '5')");
+        executeSingle("INSERT INTO t2 VALUES ('5', '5', '5')");
+
+        // Key (not alias).
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ?",
+            (res) -> {
+                assertPartitions(
+                    partition("t1", "1")
+                );
+                assertEquals(1, res.size());
+                assertEquals("1", res.get(0).get(0));
+            },
+            "1"
+        );
+
+        // Key (alias).
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1._KEY = ?",
+            (res) -> {
+                assertPartitions(
+                    partition("t1", "2")
+                );
+                assertEquals(1, res.size());
+                assertEquals("2", res.get(0).get(0));
+            },
+            "2"
+        );
+
+        // Non-affinity key.
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t2.k1 = ?",
+            (res) -> {
+                assertNoPartitions();
+                assertEquals(1, res.size());
+                assertEquals("3", res.get(0).get(0));
+            },
+            "3"
+        );
+
+        // Affinity key.
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t2.ak2 = ?",
+            (res) -> {
+                assertPartitions(
+                    partition("t2", "4")
+                );
+                assertEquals(1, res.size());
+                assertEquals("4", res.get(0).get(0));
+            },
+            "4"
+        );
+
+        // Complex key.
+        BinaryObject key = client().binary().builder("t2_key").setField("k1", "5").setField("ak2", "5").build();
+
+        List<List<?>> res = executeSingle("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t2._KEY = ?", key);
+        assertPartitions(
+            partition("t2", "5")
+        );
+        assertEquals(1, res.size());
+        assertEquals("5", res.get(0).get(0));
+    }
+
+    /**
+     * Test how partition ownership is transferred in various cases.
+     */
+    @Test
+    public void testPartitionTransfer() {
+        // First co-located table.
+        createPartitionedTable("t1",
+            pkColumn("k1"),
+            "v2"
+        );
+
+        // Second co-located table.
+        createPartitionedTable("t2",
+            pkColumn("k1"),
+            affinityColumn("ak2"),
+            "v3"
+        );
+
+        // Third co-located table.
+        createPartitionedTable("t3",
+            pkColumn("k1"),
+            affinityColumn("ak2"),
+            "v3",
+            "v4"
+        );
+
+        // Transfer through "AND".
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ? AND t2.ak2 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1", "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ? AND t2.ak2 = ?",
+            (res) -> assertNoRequests(),
+            "1", "2"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ? AND t2.ak2 IN (?, ?)",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1", "1", "2"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ? AND t2.ak2 IN (?, ?)",
+            (res) -> assertNoRequests(),
+            "1", "2", "3"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 IN (?, ?) AND t2.ak2 IN (?, ?)",
+            (res) -> assertPartitions(
+                partition("t1", "2")
+            ),
+            "1", "2", "2", "3"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 IN (?, ?) AND t2.ak2 IN (?, ?)",
+            (res) -> assertNoRequests(),
+            "1", "2", "3", "4"
+        );
+
+        // Transfer through "OR".
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ? OR t2.ak2 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1", "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ? OR t2.ak2 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1"),
+                partition("t2", "2")
+            ),
+            "1", "2"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ? OR t2.ak2 IN (?, ?)",
+            (res) -> assertPartitions(
+                partition("t1", "1"),
+                partition("t2", "2")
+            ),
+            "1", "1", "2"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ? OR t2.ak2 IN (?, ?)",
+            (res) -> assertPartitions(
+                partition("t1", "1"),
+                partition("t2", "2"),
+                partition("t2", "3")
+            ),
+            "1", "2", "3"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 IN (?, ?) OR t2.ak2 IN (?, ?)",
+            (res) -> assertPartitions(
+                partition("t1", "1"),
+                partition("t1", "2"),
+                partition("t2", "3")
+            ),
+            "1", "2", "2", "3"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 IN (?, ?) OR t2.ak2 IN (?, ?)",
+            (res) -> assertPartitions(
+                partition("t1", "1"),
+                partition("t1", "2"),
+                partition("t2", "3"),
+                partition("t2", "4")
+            ),
+            "1", "2", "3", "4"
+        );
+
+        // Multi-way co-located JOIN.
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 INNER JOIN t3 ON t1.k1 = t3.ak2 " +
+                "WHERE t1.k1 = ? AND t2.ak2 = ? AND t3.ak2 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1", "1", "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 INNER JOIN t3 ON t1.k1 = t3.ak2 " +
+                "WHERE t1.k1 = ? AND t2.ak2 = ? AND t3.ak2 = ?",
+            (res) -> assertNoRequests(),
+            "1", "2", "3"
+        );
+
+        // No transfer through intermediate table.
+        execute("SELECT * FROM t1 INNER JOIN t3 ON t1.k1 = t3.v3 INNER JOIN t2 ON t3.v4 = t2.ak2 " +
+                "WHERE t1.k1 = ? AND t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "1", "1"
+        );
+
+        // No transfer through disjunction.
+        execute("SELECT * FROM t1 INNER JOIN t2 ON 1=1 WHERE t1.k1 = ? OR t1.k1 = t2.ak2",
+            (res) -> assertNoPartitions(),
+            "1"
+        );
+    }
+
+    /**
+     * Test cross-joins. They cannot "transfer" partitions between joined tables.
+     */
+    @Test
+    public void testCrossJoin() {
+        createPartitionedTable("t1",
+            pkColumn("k1"),
+            "v2");
+
+        createPartitionedTable("t2",
+            pkColumn("k1"),
+            affinityColumn("ak2"),
+            "v3");
+
+        executeSingle("INSERT INTO t1 VALUES ('1', '1')");
+        executeSingle("INSERT INTO t2 VALUES ('1', '1', '1')");
+
+        executeSingle("INSERT INTO t1 VALUES ('2', '2')");
+        executeSingle("INSERT INTO t2 VALUES ('2', '2', '2')");
+
+        executeSingle("INSERT INTO t1 VALUES ('3', '3')");
+        executeSingle("INSERT INTO t2 VALUES ('3', '3', '3')");
+
+        // Left table, should work.
+        execute("SELECT * FROM t1, t2 WHERE t1.k1 = ?",
+            (res) -> {
+                assertPartitions(
+                    partition("t1", "1")
+                );
+                assertEquals(1, res.size());
+                assertEquals("1", res.get(0).get(0));
+            },
+            "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON 1=1 WHERE t1.k1 = ?",
+            (res) -> {
+                assertPartitions(
+                    partition("t1", "1")
+                );
+                assertEquals(1, res.size());
+                assertEquals("1", res.get(0).get(0));
+            },
+            "1"
+        );
+
+        // Right table, should work.
+        execute("SELECT * FROM t1, t2 WHERE t2.ak2 = ?",
+            (res) -> {
+                assertPartitions(
+                    partition("t2", "2")
+                );
+                assertEquals(1, res.size());
+                assertEquals("2", res.get(0).get(0));
+            },
+            "2"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON 1=1 WHERE t2.ak2 = ?",
+            (res) -> {
+                assertPartitions(
+                    partition("t2", "2")
+                );
+                assertEquals(1, res.size());
+                assertEquals("2", res.get(0).get(0));
+            },
+            "2"
+        );
+
+        execute("SELECT * FROM t1, t2 WHERE t1.k1=? AND t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "3", "3"
+        );
+
+        // Two tables, should not work.
+        execute("SELECT * FROM t1, t2 WHERE t1.k1=? AND t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "3", "3"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON 1=1 WHERE t1.k1=? AND t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "3", "3"
+        );
+    }
+
+    /**
+     * Test non-equijoins.
+     */
+    @Test
+    public void testThetaJoin() {
+        createPartitionedTable("t1",
+            pkColumn("k1"),
+            "v2");
+
+        createPartitionedTable("t2",
+            pkColumn("k1"),
+            affinityColumn("ak2"),
+            "v3");
+
+        // Greater than.
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 > t2.ak2 WHERE t1.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 > t2.ak2 WHERE t2.ak2 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 > t2.ak2 WHERE t1.k1 = ? AND t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "1", "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 > t2.ak2 WHERE t1.k1 = ? OR t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "1", "2"
+        );
+
+        // Less than.
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 < t2.ak2 WHERE t1.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 < t2.ak2 WHERE t2.ak2 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 < t2.ak2 WHERE t1.k1 = ? AND t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "1", "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 < t2.ak2 WHERE t1.k1 = ? OR t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "1", "2"
+        );
+
+        // Non-equal.
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 <> t2.ak2 WHERE t1.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 <> t2.ak2 WHERE t2.ak2 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 <> t2.ak2 WHERE t1.k1 = ? AND t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "1", "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 <> t2.ak2 WHERE t1.k1 = ? OR t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "1", "2"
+        );
+    }
+
+    /**
+     * Test joins with REPLICATED cache.
+     */
+    @Test
+    public void testJoinWithReplicated() {
+        // First co-located table.
+        createPartitionedTable("t1",
+            pkColumn("k1"),
+            "v2"
+        );
+
+        // Replicated table.
+        createReplicatedTable("t2",
+            pkColumn("k1"),
+            "v2",
+            "v3"
+        );
+
+        // Only partition from PARTITIONED cache should be used.
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.k1 WHERE t1.k1 = ? AND t2.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1", "2"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.k1 WHERE t1.k1 IN (?, ?) AND t2.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1"),
+                partition("t1", "2")
+            ),
+            "1", "2", "3"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.k1 WHERE t1.k1 = ? OR t2.k1 = ?",
+            (res) -> assertNoPartitions(),
+            "1", "2"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.k1 WHERE t2.k1 = ?",
+            (res) -> assertNoPartitions(),
+            "1"
+        );
+    }
+
+    /**
+     * Test joins with different affinity functions.
+     */
+    @Test
+    public void testJoinWithDifferentAffinityFunctions() {
+        // Partition count.
+        checkAffinityFunctions(
+            cacheConfiguration(256, 1, false, false, false),
+            cacheConfiguration(256, 1, false, false, false),
+            true
+        );
+
+        checkAffinityFunctions(
+            cacheConfiguration(1024, 1, false, false, false),
+            cacheConfiguration(256, 1, false, false, false),
+            false
+        );
+
+        checkAffinityFunctions(
+            cacheConfiguration(256, 1, false, false, false),
+            cacheConfiguration(1024, 1, false, false, false),
+            false
+        );
+
+        // Backups.
+        checkAffinityFunctions(
+            cacheConfiguration(256, 1, false, false, false),
+            cacheConfiguration(256, 2, false, false, false),
+            true
+        );
+
+        // Different affinity functions.
+        checkAffinityFunctions(
+            cacheConfiguration(256, 2, true, false, false),
+            cacheConfiguration(256, 2, false, false, false),
+            false
+        );
+
+        checkAffinityFunctions(
+            cacheConfiguration(256, 2, false, false, false),
+            cacheConfiguration(256, 2, true, false, false),
+            false
+        );
+
+        checkAffinityFunctions(
+            cacheConfiguration(256, 2, true, false, false),
+            cacheConfiguration(256, 2, true, false, false),
+            false
+        );
+
+        // Node filters.
+        checkAffinityFunctions(
+            cacheConfiguration(256, 2, false, true, false),
+            cacheConfiguration(256, 2, false, false, false),
+            false
+        );
+
+        checkAffinityFunctions(
+            cacheConfiguration(256, 2, false, false, false),
+            cacheConfiguration(256, 2, false, true, false),
+            false
+        );
+
+        checkAffinityFunctions(
+            cacheConfiguration(256, 2, false, true, false),
+            cacheConfiguration(256, 2, false, true, false),
+            false
+        );
+
+        // With and without persistence.
+        checkAffinityFunctions(
+            cacheConfiguration(256, 2, false, false, true),
+            cacheConfiguration(256, 2, false, false, false),
+            false
+        );
+
+        checkAffinityFunctions(
+            cacheConfiguration(256, 2, false, false, false),
+            cacheConfiguration(256, 2, false, false, true),
+            false
+        );
+
+        checkAffinityFunctions(
+            cacheConfiguration(256, 2, false, false, true),
+            cacheConfiguration(256, 2, false, false, true),
+            true
+        );
+    }
+
+    @SuppressWarnings("unchecked")
+    private void checkAffinityFunctions(CacheConfiguration ccfg1, CacheConfiguration ccfg2, boolean compatible) {
+        // Destroy old caches.
+        Ignite cli = client();
+
+        cli.destroyCaches(cli.cacheNames());
+
+        // Start new caches.
+        ccfg1.setName("t1");
+        ccfg2.setName("t2");
+
+        QueryEntity entity1 = new QueryEntity(KeyClass1.class, ValueClass.class).setTableName("t1");
+        QueryEntity entity2 = new QueryEntity(KeyClass2.class, ValueClass.class).setTableName("t2");
+
+        ccfg1.setQueryEntities(Collections.singletonList(entity1));
+        ccfg2.setQueryEntities(Collections.singletonList(entity2));
+
+        ccfg1.setKeyConfiguration(new CacheKeyConfiguration(entity1.getKeyType(), "k1"));
+        ccfg2.setKeyConfiguration(new CacheKeyConfiguration(entity2.getKeyType(), "ak2"));
+
+        ccfg1.setSqlSchema(QueryUtils.DFLT_SCHEMA);
+        ccfg2.setSqlSchema(QueryUtils.DFLT_SCHEMA);
+
+        client().createCache(ccfg1);
+        client().createCache(ccfg2);
+
+        // Conduct tests.
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t2.ak2 = ?",
+            (res) -> assertPartitions(
+                partition("t2", "2")
+            ),
+            "2"
+        );
+
+        if (compatible) {
+            execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ? OR t2.ak2 = ?",
+                (res) -> assertPartitions(
+                    partition("t1", "1"),
+                    partition("t2", "2")
+                ),
+                "1", "2"
+            );
+        }
+        else {
+            execute("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ? OR t2.ak2 = ?",
+                (res) -> assertNoPartitions(),
+                "1", "2"
+            );
+        }
+    }
+
+    /**
+     * Create custom cache configuration.
+     *
+     * @param parts Partitions.
+     * @param backups Backups.
+     * @param customAffinity Custom affinity function flag.
+     * @param nodeFilter Whether to set node filter.
+     * @param persistent Whether to enable persistence.
+     * @return Cache configuration.
+     */
+    private static CacheConfiguration cacheConfiguration(
+        int parts,
+        int backups,
+        boolean customAffinity,
+        boolean nodeFilter,
+        boolean persistent
+    ) {
+        CacheConfiguration ccfg = new CacheConfiguration();
+
+        ccfg.setCacheMode(CacheMode.PARTITIONED);
+        ccfg.setBackups(backups);
+
+        RendezvousAffinityFunction affFunc;
+
+        if (customAffinity)
+            affFunc = new CustomRendezvousAffinityFunction();
+        else
+            affFunc = new RendezvousAffinityFunction();
+
+        affFunc.setPartitions(parts);
+
+        ccfg.setAffinity(affFunc);
+
+        if (nodeFilter)
+            ccfg.setNodeFilter(new CustomNodeFilter());
+
+        if (persistent)
+            ccfg.setDataRegionName(REGION_DISK);
+
+        return ccfg;
+    }
+
+    /**
+     * Test joins with subqueries.
+     */
+    @Test
+    public void testJoinWithSubquery() {
+        createPartitionedTable("t1",
+            pkColumn("k1"),
+            "v2");
+
+        createPartitionedTable("t2",
+            pkColumn("k1"),
+            affinityColumn("ak2"),
+            "v3");
+
+        execute("SELECT * FROM t1 INNER JOIN (SELECT * FROM t2) T2_SUB ON t1.k1 = T2_SUB.ak2 WHERE t1.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 INNER JOIN (SELECT * FROM t2) T2_SUB ON t1.k1 = T2_SUB.ak2 WHERE T2_SUB.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "1"
+        );
+    }
+
+    /**
+     * Test joins when explicit partitions are set.
+     */
+    @Test
+    public void testExplicitPartitions() {
+        createPartitionedTable("t1",
+            pkColumn("k1"),
+            "v2");
+
+        createPartitionedTable("t2",
+            pkColumn("k1"),
+            affinityColumn("ak2"),
+            "v3");
+
+        executeSqlFieldsQuery(new SqlFieldsQuery("SELECT * FROM t1 INNER JOIN t2 ON t1.k1 = t2.ak2 " +
+            "WHERE t1.k1=? OR t2.ak2=?").setArgs("1", "2").setPartitions(1));
+
+        assertPartitions(1);
+    }
+
+    /**
+     * Test outer joins.
+     */
+    @Test
+    public void testOuterJoin() {
+        createPartitionedTable("t1",
+            pkColumn("k1"),
+            "v2");
+
+        createPartitionedTable("t2",
+            pkColumn("k1"),
+            affinityColumn("ak2"),
+            "v3");
+
+        execute("SELECT * FROM t1 LEFT OUTER JOIN t2 ON t1.k1 = t2.ak2 WHERE t1.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 LEFT OUTER JOIN t2 ON t1.k1 = t2.ak2 WHERE t2.ak2 = ?",
+            (res) -> assertNoPartitions(),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 LEFT OUTER JOIN t2 T2_1 ON t1.k1 = T2_1.ak2 INNER JOIN t2 T2_2 ON T2_1.k1 = T2_2.k1 " +
+                "WHERE T2_2.ak2 = ?",
+            (res) -> assertPartitions(
+                partition("t2", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 LEFT OUTER JOIN t2 T2_1 ON t1.k1 = T2_1.ak2 INNER JOIN t2 T2_2 ON t1.k1 = T2_2.ak2 " +
+                "WHERE T2_1.ak2 = ? AND T2_2.ak2=?",
+            (res) -> assertPartitions(
+                partition("t2", "2")
+            ),
+            "1", "2"
+        );
+    }
+
+    /**
+     * Test JOINs on a single table.
+     */
+    @Test
+    public void testSelfJoin() {
+        createPartitionedTable("t1",
+            pkColumn("k1"),
+            "v2");
+
+        execute("SELECT * FROM t1 A INNER JOIN t1 B ON A.k1 = B.k1 WHERE A.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1"
+        );
+
+        execute("SELECT * FROM t1 A INNER JOIN t1 B ON A.k1 = B.k1 WHERE A.k1 = ? AND B.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1")
+            ),
+            "1", "1"
+        );
+
+        execute("SELECT * FROM t1 A INNER JOIN t1 B ON A.k1 = B.k1 WHERE A.k1 = ? AND B.k1 = ?",
+            (res) -> assertNoRequests(),
+            "1", "2"
+        );
+
+        execute("SELECT * FROM t1 A INNER JOIN t1 B ON A.k1 = B.k1 WHERE A.k1 = ? OR B.k1 = ?",
+            (res) -> assertPartitions(
+                partition("t1", "1"),
+                partition("t1", "2")
+            ),
+            "1", "2"
+        );
+    }
+
+    /**
+     * Create PARTITIONED table.
+     *
+     * @param name Name.
+     * @param cols Columns.
+     */
+    private void createPartitionedTable(String name, Object... cols) {
+        createTable0(name, false, cols);
+    }
+
+    /**
+     * Create REPLICATED table.
+     *
+     * @param name Name.
+     * @param cols Columns.
+     */
+    @SuppressWarnings("SameParameterValue")
+    private void createReplicatedTable(String name, Object... cols) {
+        createTable0(name, true, cols);
+    }
+
+    /**
+     * Internal CREATE TABLE routine.
+     *
+     * @param name Name.
+     * @param replicated Replicated table flag.
+     * @param cols Columns.
+     */
+    @SuppressWarnings("StringConcatenationInsideStringBufferAppend")
+    private void createTable0(String name, boolean replicated, Object... cols) {
+        List<String> pkCols = new ArrayList<>();
+
+        String affCol = null;
+
+        StringBuilder sql = new StringBuilder("CREATE TABLE ").append(name).append("(");
+        for (Object col : cols) {
+            Column col0 = col instanceof Column ? (Column)col : new Column((String)col, false, false);
+
+            sql.append(col0.name()).append(" VARCHAR, ");
+
+            if (col0.pk())
+                pkCols.add(col0.name());
+
+            if (col0.affinity()) {
+                if (affCol != null)
+                    throw new IllegalStateException("Only one affinity column is allowed: " + col0.name());
+
+                affCol = col0.name();
+            }
+        }
+
+        if (pkCols.isEmpty())
+            throw new IllegalStateException("No PKs!");
+
+        sql.append("PRIMARY KEY (");
+
+        boolean firstPkCol = true;
+
+        for (String pkCol : pkCols) {
+            if (firstPkCol)
+                firstPkCol = false;
+            else
+                sql.append(", ");
+
+            sql.append(pkCol);
+        }
+
+        sql.append(")");
+
+        sql.append(") WITH \"template=" + (replicated ? "replicated" : "partitioned"));
+        sql.append(", CACHE_NAME=" + name);
+
+        if (affCol != null) {
+            sql.append(", AFFINITY_KEY=" + affCol);
+            sql.append(", KEY_TYPE=" + name + "_key");
+        }
+
+        sql.append("\"");
+
+        executeSingle(sql.toString());
+    }
+
+    /**
+     * Execute query with all possible combinations of argument placeholders.
+     *
+     * @param sql SQL.
+     * @param resConsumer Result consumer.
+     * @param args Arguments.
+     */
+    public void execute(String sql, Consumer<List<List<?>>> resConsumer, Object... args) {
+        System.out.println(">>> TEST COMBINATION: " + sql);
+
+        // Execute query as is.
+        List<List<?>> res = executeSingle(sql, args);
+
+        resConsumer.accept(res);
+
+        // Start filling arguments recursively.
+        if (args != null && args.length > 0)
+            executeCombinations0(sql, resConsumer, new HashSet<>(), args);
+
+        System.out.println();
+    }
+
+    /**
+     * Execute query with all possible combinations of argument placeholders.
+     *
+     * @param sql SQL.
+     * @param resConsumer Result consumer.
+     * @param executedSqls Already executed SQLs.
+     * @param args Arguments.
+     */
+    public 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] + 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)) {
+                List<List<?>> res = executeSingle(newSql, newArgs);
+
+                resConsumer.accept(res);
+            }
+
+            // Continue recursively.
+            if (newArgs.length > 0)
+                executeCombinations0(newSql, resConsumer, executedSqls, newArgs);
+        }
+    }
+
+    /**
+     * Execute SQL query.
+     *
+     * @param sql SQL.
+     */
+    private List<List<?>> executeSingle(String sql, Object... args) {
+        clearIoState();
+
+        if (args == null || args.length == 0)
+            System.out.println(">>> " + sql);
+        else
+            System.out.println(">>> " + sql + " " + Arrays.toString(args));
+
+        SqlFieldsQuery qry = new SqlFieldsQuery(sql);
+
+        if (args != null && args.length > 0)
+            qry.setArgs(args);
+
+        return executeSqlFieldsQuery(qry);
+    }
+
+    /**
+     * Execute prepared SQL fields query.
+     *
+     * @param qry Query.
+     * @return Result.
+     */
+    private List<List<?>> executeSqlFieldsQuery(SqlFieldsQuery qry) {
+        return client().context().query().querySqlFields(qry, false).getAll();
+    }
+
+    /**
+     * @return Client node.
+     */
+    private IgniteEx client() {
+        return grid(CLI_NAME);
+    }
+
+    /**
+     * Clear partitions.
+     */
+    private static void clearIoState() {
+        INTERCEPTED_REQS.set(0);
+        INTERCEPTED_PARTS.clear();
+    }
+
+    /**
+     * Make sure that expected partitions are logged.
+     *
+     * @param expParts Expected partitions.
+     */
+    private static void assertPartitions(int... expParts) {
+        Collection<Integer> expParts0 = new TreeSet<>();
+
+        for (int expPart : expParts)
+            expParts0.add(expPart);
+
+        assertPartitions(expParts0);
+    }
+
+    /**
+     * Make sure that expected partitions are logged.
+     *
+     * @param expParts Expected partitions.
+     */
+    private static void assertPartitions(Collection<Integer> expParts) {
+        TreeSet<Integer> expParts0 = new TreeSet<>(expParts);
+        TreeSet<Integer> actualParts = new TreeSet<>(INTERCEPTED_PARTS);
+
+        assertEquals("Unexpected partitions [exp=" + expParts + ", actual=" + actualParts + ']',
+            expParts0, actualParts);
+    }
+
+    /**
+     * Make sure that no partitions were extracted.
+     */
+    private static void assertNoPartitions() {
+        assertTrue("No requests were sent.", INTERCEPTED_REQS.get() > 0);
+        assertTrue("Partitions are not empty: " + INTERCEPTED_PARTS, INTERCEPTED_PARTS.isEmpty());
+    }
+
+    /**
+     * Make sure there were no requests sent because we determined empty partition set.
+     */
+    private static void assertNoRequests() {
+        assertEquals("Requests were sent: " + INTERCEPTED_REQS.get(), 0, INTERCEPTED_REQS.get());
+    }
+
+    /**
+     * @param cacheName Cache name.
+     * @param key Key.
+     * @return Partition.
+     */
+    private int partition(String cacheName, Object key) {
+        return client().affinity(cacheName).partition(key);
+    }
+
+    /**
+     * TCP communication SPI which will track outgoing query requests.
+     */
+    private static class TrackingTcpCommunicationSpi extends TcpCommunicationSpi {
+        @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC) {
+            if (msg instanceof GridIoMessage) {
+                GridIoMessage msg0 = (GridIoMessage)msg;
+
+                if (msg0.message() instanceof GridH2QueryRequest) {
+                    INTERCEPTED_REQS.incrementAndGet();
+
+                    GridH2QueryRequest req = (GridH2QueryRequest)msg0.message();
+
+                    int[] parts = req.queryPartitions();
+
+                    if (!F.isEmpty(parts)) {
+                        for (int part : parts)
+                            INTERCEPTED_PARTS.add(part);
+                    }
+                }
+            }
+
+            super.sendMessage(node, msg, ackC);
+        }
+    }
+
+    /**
+     * @param name Name.
+     * @return PK column.
+     */
+    public Column pkColumn(String name) {
+        return new Column(name, true, false);
+    }
+
+    /**
+     * @param name Name.
+     * @return Affintiy column.
+     */
+    public Column affinityColumn(String name) {
+        return new Column(name, true, true);
+    }
+
+    /**
+     * Column.
+     */
+    private static class Column {
+        /** Name. */
+        private final String name;
+
+        /** PK. */
+        private final boolean pk;
+
+        /** Affinity key. */
+        private final boolean aff;
+
+        /**
+         * Constructor.
+         *
+         * @param name Name.
+         * @param pk PK flag.
+         * @param aff Affinity flag.
+         */
+        public Column(String name, boolean pk, boolean aff) {
+            this.name = name;
+            this.pk = pk;
+            this.aff = aff;
+        }
+
+        /**
+         * @return Name.
+         */
+        public String name() {
+            return name;
+        }
+
+        /**
+         * @return PK flag.
+         */
+        public boolean pk() {
+            return pk;
+        }
+
+        /**
+         * @return Affintiy flag.
+         */
+        public boolean affinity() {
+            return aff;
+        }
+    }
+
+    /**
+     * Custom affinity function.
+     */
+    private static class CustomRendezvousAffinityFunction extends RendezvousAffinityFunction {
+        // No-op.
+    }
+
+    /**
+     * Custom node filter.
+     */
+    private static class CustomNodeFilter implements IgnitePredicate<ClusterNode> {
+        @Override public boolean apply(ClusterNode clusterNode) {
+            return true;
+        }
+    }
+
+    /**
+     * Key class 1.
+     */
+    @SuppressWarnings("unused")
+    private static class KeyClass1 {
+        /** Key. */
+        @QuerySqlField
+        private String k1;
+    }
+
+    /**
+     * Key class 2.
+     */
+    @SuppressWarnings("unused")
+    private static class KeyClass2 {
+        /** Key. */
+        @QuerySqlField
+        private String k1;
+
+        /** Affinity key. */
+        @QuerySqlField
+        private String ak2;
+    }
+
+    /**
+     * Value class.
+     */
+    @SuppressWarnings("unused")
+    private static class ValueClass {
+        /** Value. */
+        @QuerySqlField
+        private String v;
+    }
+}
index ce8f6cc..5ffd7fa 100644 (file)
@@ -215,6 +215,7 @@ import org.apache.ignite.internal.processors.query.h2.sql.H2CompareBigQueryTest;
 import org.apache.ignite.internal.processors.query.h2.twostep.AndOperationExtractPartitionSelfTest;
 import org.apache.ignite.internal.processors.query.h2.twostep.BetweenOperationExtractPartitionSelfTest;
 import org.apache.ignite.internal.processors.query.h2.twostep.InOperationExtractPartitionSelfTest;
+import org.apache.ignite.internal.processors.query.h2.twostep.JoinPartitionPruningSelfTest;
 import org.apache.ignite.internal.processors.sql.IgniteCachePartitionedAtomicColumnConstraintsTest;
 import org.apache.ignite.internal.processors.sql.IgniteCachePartitionedTransactionalColumnConstraintsTest;
 import org.apache.ignite.internal.processors.sql.IgniteCachePartitionedTransactionalSnapshotColumnConstraintTest;
@@ -527,6 +528,7 @@ import org.junit.runners.Suite;
     InOperationExtractPartitionSelfTest.class,
     AndOperationExtractPartitionSelfTest.class,
     BetweenOperationExtractPartitionSelfTest.class,
+    JoinPartitionPruningSelfTest.class,
 
     GridCacheDynamicLoadOnClientTest.class,
     GridCacheDynamicLoadOnClientPersistentTest.class,