This is an automated email from the ASF dual-hosted git repository.

tledkov pushed a commit to branch ignite-12248
in repository https://gitbox.apache.org/repos/asf/ignite.git


The following commit(s) were added to refs/heads/ignite-12248 by this push:
     new ae31145  IGNITE-13717 additional test coverage for calcite engine, fix 
distribution trait (closes #8475, #8481)
ae31145 is described below

commit ae31145fa3af01b31960f60439deeda9f64bb389
Author: tledkov <tled...@gridgain.com>
AuthorDate: Mon Nov 23 11:22:36 2020 +0300

    IGNITE-13717 additional test coverage for calcite engine, fix distribution 
trait (closes #8475, #8481)
---
 .../processors/query/calcite/prepare/Fragment.java |   5 +
 .../query/calcite/prepare/QueryPlanCache.java      |   6 +-
 .../query/calcite/rel/IgniteExchange.java          |   1 +
 .../query/calcite/schema/SchemaHolderImpl.java     |   1 +
 .../query/calcite/trait/DistributionFunction.java  |  35 ++-
 .../query/calcite/trait/DistributionTrait.java     |   2 +-
 .../query/calcite/CalciteQueryProcessorTest.java   | 298 ++++++++++++++++++---
 7 files changed, 294 insertions(+), 54 deletions(-)

diff --git 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Fragment.java
 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Fragment.java
index d832b68..5285df3 100644
--- 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Fragment.java
+++ 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Fragment.java
@@ -181,4 +181,9 @@ public class Fragment {
         return root instanceof IgniteSender
             && 
((IgniteSender)root).sourceDistribution().satisfies(IgniteDistributions.single());
     }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return rootSer;
+    }
 }
diff --git 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/QueryPlanCache.java
 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/QueryPlanCache.java
index 732830f..60c2fdd 100644
--- 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/QueryPlanCache.java
+++ 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/QueryPlanCache.java
@@ -18,7 +18,6 @@
 package org.apache.ignite.internal.processors.query.calcite.prepare;
 
 import java.util.List;
-
 import org.apache.ignite.internal.processors.query.calcite.util.Service;
 
 /**
@@ -32,4 +31,9 @@ public interface QueryPlanCache extends Service {
      * @return Query plan.
      */
     List<QueryPlan> queryPlan(PlanningContext ctx, CacheKey key, 
QueryPlanFactory factory);
+
+    /**
+     * Clear cache.
+     */
+    void clear();
 }
diff --git 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteExchange.java
 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteExchange.java
index 2ee7643..75dad60 100644
--- 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteExchange.java
+++ 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteExchange.java
@@ -48,6 +48,7 @@ public class IgniteExchange extends Exchange implements 
IgniteRel {
         super(cluster, traitSet, input, distribution);
     }
 
+    /** */
     public IgniteExchange(RelInput input) {
         super(changeTraits(input, IgniteConvention.INSTANCE));
     }
diff --git 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/SchemaHolderImpl.java
 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/SchemaHolderImpl.java
index 8fe9c50..bac2f37 100644
--- 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/SchemaHolderImpl.java
+++ 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/SchemaHolderImpl.java
@@ -74,6 +74,7 @@ public class SchemaHolderImpl extends AbstractService 
implements SchemaHolder, S
         /** */
         private final int hash;
 
+        /** */
         public AffinityIdentity(AffinityFunction aff, int backups, 
IgnitePredicate<ClusterNode> nodeFilter) {
             affFuncCls = aff.getClass();
 
diff --git 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionFunction.java
 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionFunction.java
index 52cf5e9..5c3183f 100644
--- 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionFunction.java
+++ 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionFunction.java
@@ -39,7 +39,9 @@ public abstract class DistributionFunction {
     private String name;
 
     /** */
-    private DistributionFunction(){}
+    private DistributionFunction() {
+        // No-op.
+    }
 
     /**
      * @return Distribution function type.
@@ -130,8 +132,20 @@ public abstract class DistributionFunction {
     }
 
     /** */
-    public static DistributionFunction affinity(int cacheId, Object key) {
-        return new AffinityDistribution(cacheId, key);
+    public static DistributionFunction affinity(int cacheId, Object identity) {
+        return new AffinityDistribution(cacheId, identity);
+    }
+
+    /** */
+    public static boolean satisfy(DistributionFunction f0, 
DistributionFunction f1) {
+        if (f0 == f1 || f0.name() == f1.name())
+            return true;
+
+        if (f0 instanceof AffinityDistribution && f1 instanceof 
AffinityDistribution &&
+            Objects.equals(((AffinityDistribution)f0).identity(), 
((AffinityDistribution)f1).identity()))
+            return true;
+
+        return false;
     }
 
     /** */
@@ -245,15 +259,15 @@ public abstract class DistributionFunction {
         private final int cacheId;
 
         /** */
-        private final Object key;
+        private final Object identity;
 
         /**
          * @param cacheId Cache ID.
-         * @param key Affinity identity key.
+         * @param identity Affinity identity key.
          */
-        public AffinityDistribution(int cacheId, Object key) {
+        public AffinityDistribution(int cacheId, Object identity) {
             this.cacheId = cacheId;
-            this.key = key;
+            this.identity = identity;
         }
 
         /** {@inheritDoc} */
@@ -288,9 +302,14 @@ public abstract class DistributionFunction {
             return new Partitioned<>(assignments, affinity);
         }
 
+        /** */
+        public Object identity() {
+            return identity;
+        }
+
         /** {@inheritDoc} */
         @Override protected String name0() {
-            return "affinity[" + key + "]";
+            return "affinity[identity=" + identity + ", cacheId=" + cacheId + 
']';
         }
     }
 }
diff --git 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionTrait.java
 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionTrait.java
index e8cf58e..0238848 100644
--- 
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionTrait.java
+++ 
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionTrait.java
@@ -137,7 +137,7 @@ public final class DistributionTrait implements 
IgniteDistribution {
         if (getType() == other.getType())
             return getType() != HASH_DISTRIBUTED
                 || (Objects.equals(keys, other.keys)
-                    && Objects.equals(function, other.function));
+                    && DistributionFunction.satisfy(function, other.function));
 
         if (other.getType() == RANDOM_DISTRIBUTED)
             return getType() == HASH_DISTRIBUTED;
diff --git 
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java
 
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java
index 04b854c..62f7490 100644
--- 
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java
+++ 
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.processors.query.calcite;
 import java.util.Arrays;
 import java.util.List;
 import com.google.common.collect.ImmutableMap;
+import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.cache.CacheMode;
 import org.apache.ignite.cache.QueryEntity;
@@ -29,14 +30,13 @@ import org.apache.ignite.cache.query.SqlFieldsQuery;
 import org.apache.ignite.cache.query.annotations.QuerySqlField;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.internal.IgniteEx;
-import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.query.QueryEngine;
 import org.apache.ignite.internal.processors.query.calcite.util.Commons;
 import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
 import org.apache.ignite.testframework.junits.WithSystemProperty;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.junit.After;
-import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 /**
@@ -45,20 +45,29 @@ import org.junit.Test;
 @WithSystemProperty(key = "calcite.debug", value = "false")
 public class CalciteQueryProcessorTest extends GridCommonAbstractTest {
     /** */
-    private IgniteEx ignite;
+    private static IgniteEx ignite;
 
-    @Before
-    public void setup() throws Exception {
+    /** {@inheritDoc} */
+    @Override protected void beforeTestsStarted() throws Exception {
         startGrids(5);
 
         ignite = startClientGrid();
     }
 
-    @After
-    public void tearDown() throws Exception {
-        stopAllGrids();
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        for (Ignite ign : G.allGrids()) {
+            for (String cacheName : ign.cacheNames())
+                ign.destroyCache(cacheName);
+
+            CalciteQueryProcessor qryProc = 
(CalciteQueryProcessor)Commons.lookupComponent(
+                ((IgniteEx)ign).context(), QueryEngine.class);
+
+            qryProc.queryPlanCache().clear();
+        }
     }
 
+    /** */
     @Test
     public void unionAll() throws Exception {
         IgniteCache<Integer, Employer> employer1 = ignite.getOrCreateCache(new 
CacheConfiguration<Integer, Employer>()
@@ -89,7 +98,7 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
         employer2.put(2, new Employer("Roman", 15d));
         employer3.put(3, new Employer("Nikolay", 20d));
 
-        waitForReadyTopology(internalCache(employer1).context().topology(), 
new AffinityTopologyVersion(5, 4));
+        awaitPartitionMapExchange(true, true, null);
 
         QueryEngine engine = Commons.lookupComponent(grid(1).context(), 
QueryEngine.class);
 
@@ -106,6 +115,7 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
         assertEquals(6, rows.size());
     }
 
+    /** */
     @Test
     public void union() throws Exception {
         IgniteCache<Integer, Employer> employer1 = ignite.getOrCreateCache(new 
CacheConfiguration<Integer, Employer>()
@@ -136,7 +146,7 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
         employer2.put(2, new Employer("Roman", 15d));
         employer3.put(3, new Employer("Nikolay", 20d));
 
-        waitForReadyTopology(internalCache(employer1).context().topology(), 
new AffinityTopologyVersion(5, 4));
+        awaitPartitionMapExchange(true, true, null);
 
         QueryEngine engine = Commons.lookupComponent(grid(1).context(), 
QueryEngine.class);
 
@@ -153,6 +163,206 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
         assertEquals(3, rows.size());
     }
 
+    /** */
+    private void populateTables() throws InterruptedException {
+        IgniteCache<Integer, Employer> orders = ignite.getOrCreateCache(new 
CacheConfiguration<Integer, Employer>()
+            .setName("orders")
+            .setSqlSchema("PUBLIC")
+            .setQueryEntities(F.asList(new QueryEntity(Integer.class, 
Employer.class).setTableName("orders")))
+            .setBackups(2)
+        );
+
+        IgniteCache<Integer, Employer> account = ignite.getOrCreateCache(new 
CacheConfiguration<Integer, Employer>()
+            .setName("account")
+            .setSqlSchema("PUBLIC")
+            .setQueryEntities(F.asList(new QueryEntity(Integer.class, 
Employer.class).setTableName("account")))
+            .setBackups(1)
+        );
+
+        orders.put(1, new Employer("Igor", 10d));
+        orders.put(2, new Employer("Igor", 11d));
+        orders.put(3, new Employer("Igor", 12d));
+        orders.put(4, new Employer("Igor1", 13d));
+        orders.put(5, new Employer("Igor1", 13d));
+        orders.put(6, new Employer("Igor1", 13d));
+        orders.put(7, new Employer("Roman", 14d));
+
+        account.put(1, new Employer("Roman", 10d));
+        account.put(2, new Employer("Roman", 11d));
+        account.put(3, new Employer("Roman", 12d));
+        account.put(4, new Employer("Roman", 13d));
+        account.put(5, new Employer("Igor1", 13d));
+        account.put(6, new Employer("Igor1", 13d));
+
+        /*
+        select * from orders;
+        +----+-------+-------+
+        | ID | NAME  | SALARY|
+        +----+-------+-------+
+        |  1 | igor  |   10  |
+        |  2 | igor  |   11  |
+        |  3 | igor  |   12  |
+        |  4 | igor1 |   13  |
+        |  5 | igor1 |   13  |
+        |  6 | igor1 |   13  |
+        |  7 | roman |   14  |
+        +----+-------+-------+
+
+        select * from account;
+        +----+-------+-------+
+        | ID | NAME  | SALARY|
+        +----+-------+-------+
+        |  1 | Roman |   10  |
+        |  2 | Roman |   11  |
+        |  3 | Roman |   12  |
+        |  4 | Roman |   13  |
+        |  5 | igor1 |   13  |
+        |  6 | igor1 |   13  |
+        +----+-------+-------+
+         */
+
+        awaitPartitionMapExchange(true, true, null);
+    }
+
+    /** */
+    @Test
+    public void testEqConditionWithDistinctSubquery() throws Exception {
+        populateTables();
+
+        List<List<?>> rows = sql(
+            "SELECT name FROM Orders WHERE salary = (SELECT DISTINCT(salary) 
from Account WHERE name='Igor1')");
+
+        assertEquals(3, rows.size());
+    }
+
+    /** */
+    @Test
+    public void testEqConditionWithAggregateSubqueryMax() throws Exception {
+        populateTables();
+
+        List<List<?>> rows = sql(
+            "SELECT name FROM Orders WHERE salary = (SELECT MAX(salary) from 
Account WHERE name='Roman')");
+
+        assertEquals(3, rows.size());
+    }
+
+    /** */
+    @Test
+    public void testEqConditionWithAggregateSubqueryMin() throws Exception {
+        populateTables();
+
+        List<List<?>> rows = sql(
+            "SELECT name FROM Orders WHERE salary = (SELECT MIN(salary) from 
Account WHERE name='Roman')");
+
+        assertEquals(1, rows.size());
+    }
+
+    /** */
+    @Test
+    public void testInConditionWithSubquery() throws Exception {
+        populateTables();
+
+        List<List<?>> rows = sql(
+            "SELECT name FROM Orders WHERE name IN (SELECT name from 
Account)");
+
+        assertEquals(4, rows.size());
+    }
+
+    /** */
+    @Test
+    public void testDistinctQueryWithInConditionWithSubquery() throws 
Exception {
+        populateTables();
+
+        List<List<?>> rows = sql(
+                "SELECT distinct(name) FROM Orders o WHERE name IN (" +
+                "   SELECT name" +
+                "   FROM Account)");
+
+        assertEquals(2, rows.size());
+    }
+
+
+    /** */
+    @Test
+    public void testUnion() throws Exception {
+        populateTables();
+
+        List<List<?>> rows = sql(
+            "SELECT name FROM Orders UNION SELECT name from Account");
+
+        assertEquals(3, rows.size());
+    }
+
+    /** */
+    @Test
+    @Ignore("https://issues.apache.org/jira/browse/IGNITE-13727";)
+    public void testUnionWithDistinct() throws Exception {
+        populateTables();
+
+        List<List<?>> rows = sql(
+            "SELECT distinct(name) FROM Orders UNION SELECT name from 
Account");
+
+        assertEquals(3, rows.size());
+    }
+
+    /** */
+    @Test
+    @Ignore("https://issues.apache.org/jira/browse/IGNITE-13721";)
+    public void testNotInConditionWithSubquery() throws Exception {
+        populateTables();
+
+        List<List<?>> rows = sql(
+            "SELECT name FROM Orders WHERE name NOT IN (SELECT name from 
Account)");
+
+        assertEquals(3, rows.size());
+    }
+
+    /** */
+    @Test
+    public void testExistsConditionWithSubquery() throws Exception {
+        populateTables();
+
+        List<List<?>> rows = sql(
+                "SELECT name FROM Orders o WHERE EXISTS (" +
+                "   SELECT 1" +
+                "   FROM Account a" +
+                "   WHERE o.name = a.name)");
+
+        assertEquals(4, rows.size());
+    }
+
+    /** */
+    @Test
+    @Ignore("https://issues.apache.org/jira/browse/IGNITE-13729";)
+    public void testNotExistsConditionWithSubquery() throws Exception {
+        populateTables();
+
+        List<List<?>> rows = sql(
+                "EXPLAIN PLAN FOR SELECT name FROM Orders o WHERE NOT EXISTS 
(" +
+                "   SELECT 1" +
+                "   FROM Account a" +
+                "   WHERE o.name = a.name)");
+
+        assertEquals(3, rows.size());
+
+        rows = sql(
+            "SELECT name FROM Orders o WHERE NOT EXISTS (" +
+                "   SELECT name" +
+                "   FROM Account a" +
+                "   WHERE o.name = a.name)");
+
+        assertEquals(3, rows.size());
+
+        rows = sql(
+                "SELECT distinct(name) FROM Orders o WHERE NOT EXISTS (" +
+                "   SELECT name" +
+                "   FROM Account a" +
+                "   WHERE o.name = a.name)");
+
+        assertEquals(1, rows.size());
+    }
+
+    /** */
     @Test
     public void aggregate() throws Exception {
         IgniteCache<Integer, Employer> employer = ignite.getOrCreateCache(new 
CacheConfiguration<Integer, Employer>()
@@ -162,14 +372,14 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
             .setBackups(2)
         );
 
-        waitForReadyTopology(internalCache(employer).context().topology(), new 
AffinityTopologyVersion(5, 2));
-
         employer.putAll(ImmutableMap.of(
             0, new Employer("Igor", 10d),
             1, new Employer("Roman", 15d),
             2, new Employer("Nikolay", 20d)
         ));
 
+        awaitPartitionMapExchange(true, true, null);
+
         QueryEngine engine = Commons.lookupComponent(grid(1).context(), 
QueryEngine.class);
 
         List<FieldsQueryCursor<List<?>>> query = engine.query(null, "PUBLIC",
@@ -182,33 +392,30 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
         assertEquals(Arrays.asList("Roman", 15d), F.first(rows));
     }
 
+    /** */
     @Test
     public void query() throws Exception {
-        IgniteCache<Integer, Developer> developer = 
ignite.getOrCreateCache(new CacheConfiguration<Integer, Developer>()
+        IgniteCache<Integer, Developer> developer = grid(1).createCache(new 
CacheConfiguration<Integer, Developer>()
             .setName("developer")
             .setSqlSchema("PUBLIC")
             .setIndexedTypes(Integer.class, Developer.class)
             .setBackups(2)
         );
 
-        IgniteCache<Integer, Project> project = ignite.getOrCreateCache(new 
CacheConfiguration<Integer, Project>()
+        IgniteCache<Integer, Project> project = grid(1).createCache(new 
CacheConfiguration<Integer, Project>()
             .setName("project")
             .setSqlSchema("PUBLIC")
             .setIndexedTypes(Integer.class, Project.class)
             .setBackups(2)
         );
 
-        waitForReadyTopology(internalCache(project).context().topology(), new 
AffinityTopologyVersion(5, 3));
+        project.put(0, new Project("Ignite"));
+        project.put(1, new Project("Calcite"));
 
-        project.putAll(ImmutableMap.of(
-            0, new Project("Ignite"),
-            1, new Project("Calcite")
-        ));
+        developer.put(0, new Developer("Igor", 1));
+        developer.put(1, new Developer("Roman", 0));
 
-        developer.putAll(ImmutableMap.of(
-            0, new Developer("Igor", 1),
-            1, new Developer("Roman", 0)
-        ));
+        awaitPartitionMapExchange(true, true, null);
 
         QueryEngine engine = Commons.lookupComponent(grid(1).context(), 
QueryEngine.class);
 
@@ -220,33 +427,28 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
         assertEqualsCollections(Arrays.asList("Igor", 1, "Calcite"), 
F.first(query.get(0).getAll()));
     }
 
+    /** */
     @Test
     public void query2() throws Exception {
-        IgniteCache<Integer, Developer> developer = 
ignite.getOrCreateCache(new CacheConfiguration<Integer, Developer>()
+        IgniteCache<Integer, Developer> developer = 
grid(1).getOrCreateCache(new CacheConfiguration<Integer, Developer>()
             .setName("developer")
             .setSqlSchema("PUBLIC")
             .setIndexedTypes(Integer.class, Developer.class)
             .setCacheMode(CacheMode.REPLICATED)
         );
 
-        IgniteCache<Integer, Project> project = ignite.getOrCreateCache(new 
CacheConfiguration<Integer, Project>()
+        IgniteCache<Integer, Project> project = grid(1).getOrCreateCache(new 
CacheConfiguration<Integer, Project>()
             .setName("project")
             .setSqlSchema("PUBLIC")
             .setIndexedTypes(Integer.class, Project.class)
             .setBackups(2)
         );
 
-        waitForReadyTopology(internalCache(project).context().topology(), new 
AffinityTopologyVersion(5, 3));
+        project.put(0, new Project("Ignite"));
+        project.put(1, new Project("Calcite"));
 
-        project.putAll(ImmutableMap.of(
-            0, new Project("Ignite"),
-            1, new Project("Calcite")
-        ));
-
-        developer.putAll(ImmutableMap.of(
-            0, new Developer("Igor", 1),
-            1, new Developer("Roman", 0)
-        ));
+        developer.put(0, new Developer("Igor", 1));
+        developer.put(1, new Developer("Roman", 0));
 
         QueryEngine engine = Commons.lookupComponent(grid(1).context(), 
QueryEngine.class);
 
@@ -258,24 +460,23 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
         assertEqualsCollections(Arrays.asList("Igor", 1, "Calcite"), 
F.first(query.get(0).getAll()));
     }
 
+    /** */
     @Test
     public void queryMultiStatement() throws Exception {
-        IgniteCache<Integer, Developer> developer = 
ignite.getOrCreateCache(new CacheConfiguration<Integer, Developer>()
+        IgniteCache<Integer, Developer> developer = 
grid(1).getOrCreateCache(new CacheConfiguration<Integer, Developer>()
             .setName("developer")
             .setSqlSchema("PUBLIC")
             .setIndexedTypes(Integer.class, Developer.class)
             .setBackups(2)
         );
 
-        IgniteCache<Integer, Project> project = ignite.getOrCreateCache(new 
CacheConfiguration<Integer, Project>()
+        IgniteCache<Integer, Project> project = grid(1).getOrCreateCache(new 
CacheConfiguration<Integer, Project>()
             .setName("project")
             .setSqlSchema("PUBLIC")
             .setIndexedTypes(Integer.class, Project.class)
             .setBackups(2)
         );
 
-        waitForReadyTopology(internalCache(project).context().topology(), new 
AffinityTopologyVersion(5, 3));
-
         project.putAll(ImmutableMap.of(
             0, new Project("Ignite"),
             1, new Project("Calcite")
@@ -299,17 +500,16 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
         assertEqualsCollections(Arrays.asList("Roman", 0, "Ignite"), 
F.first(query.get(1).getAll()));
     }
 
+    /** */
     @Test
     public void testInsertPrimitiveKey() throws Exception {
-        IgniteCache<Integer, Developer> developer = 
ignite.getOrCreateCache(new CacheConfiguration<Integer, Developer>()
+        IgniteCache<Integer, Developer> developer = 
grid(1).getOrCreateCache(new CacheConfiguration<Integer, Developer>()
             .setName("developer")
             .setSqlSchema("PUBLIC")
             .setIndexedTypes(Integer.class, Developer.class)
             .setBackups(2)
         );
 
-        waitForReadyTopology(internalCache(developer).context().topology(), 
new AffinityTopologyVersion(5, 3));
-
         QueryEngine engine = Commons.lookupComponent(grid(1).context(), 
QueryEngine.class);
 
         List<FieldsQueryCursor<List<?>>> query = engine.query(null, "PUBLIC", 
"INSERT INTO DEVELOPER VALUES (?, ?, ?)", 0, "Igor", 1);
@@ -337,6 +537,7 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
         assertEqualsCollections(Arrays.asList(0, "Igor", 1), row);
     }
 
+    /** */
     @Test
     public void testInsertUpdateDeleteNonPrimitiveKey() throws Exception {
         IgniteCache<Key, Developer> developer = ignite.getOrCreateCache(new 
CacheConfiguration<Key, Developer>()
@@ -346,7 +547,7 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
             .setBackups(2)
         );
 
-        waitForReadyTopology(internalCache(developer).context().topology(), 
new AffinityTopologyVersion(5, 2));
+        awaitPartitionMapExchange(true, true, null);
 
         QueryEngine engine = Commons.lookupComponent(grid(1).context(), 
QueryEngine.class);
 
@@ -460,6 +661,15 @@ public class CalciteQueryProcessorTest extends 
GridCommonAbstractTest {
     }
 
     /** */
+    private List<List<?>> sql(String sql) {
+        QueryEngine engine = Commons.lookupComponent(grid(1).context(), 
QueryEngine.class);
+
+        List<FieldsQueryCursor<List<?>>> cursors = engine.query(null, 
"PUBLIC", sql);
+
+        return cursors.get(0).getAll();
+    }
+
+    /** */
     public static class Key {
         /** */
         @QuerySqlField

Reply via email to