Repository: incubator-ignite Updated Branches: refs/heads/ignite-sql-tests b689abdd3 -> 6942a85f1
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheQueryMultiThreadedSelfTest.java ---------------------------------------------------------------------- diff --cc modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheQueryMultiThreadedSelfTest.java index 88b5221,0000000..30ff110 mode 100644,000000..100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheQueryMultiThreadedSelfTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheQueryMultiThreadedSelfTest.java @@@ -1,754 -1,0 +1,755 @@@ +/* + * 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.cache; + +import org.apache.ignite.*; +import org.apache.ignite.cache.*; +import org.apache.ignite.cache.eviction.lru.*; +import org.apache.ignite.cache.query.*; +import org.apache.ignite.cache.query.annotations.*; ++import org.apache.ignite.cache.query.annotations.*; +import org.apache.ignite.configuration.*; +import org.apache.ignite.internal.*; +import org.apache.ignite.internal.processors.cache.query.*; +import org.apache.ignite.internal.processors.query.*; +import org.apache.ignite.internal.processors.query.h2.*; +import org.apache.ignite.internal.util.typedef.*; +import org.apache.ignite.marshaller.optimized.*; +import org.apache.ignite.spi.discovery.tcp.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*; +import org.apache.ignite.spi.swapspace.file.*; +import org.apache.ignite.testframework.junits.common.*; +import org.jetbrains.annotations.*; + +import javax.cache.*; +import java.io.*; +import java.util.*; +import java.util.concurrent.atomic.*; + +import static org.apache.ignite.cache.CacheAtomicityMode.*; +import static org.apache.ignite.cache.CacheMode.*; + +/** + * Multi-threaded tests for cache queries. + */ +@SuppressWarnings("StatementWithEmptyBody") +public class IgniteCacheQueryMultiThreadedSelfTest extends GridCommonAbstractTest { + /** */ + private static final boolean TEST_INFO = true; + + /** Number of test grids (nodes). Should not be less than 2. */ + private static final int GRID_CNT = 2; + + /** */ + private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); + + /** */ + private static AtomicInteger idxSwapCnt = new AtomicInteger(); + + /** */ + private static AtomicInteger idxUnswapCnt = new AtomicInteger(); + + /** */ + private static final long DURATION = 30 * 1000; + + /** Don't start grid by default. */ + public IgniteCacheQueryMultiThreadedSelfTest() { + super(false); + } + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(gridName); + + TcpDiscoverySpi disco = new TcpDiscoverySpi(); + + disco.setIpFinder(ipFinder); + + cfg.setDiscoverySpi(disco); + + cfg.setSwapSpaceSpi(new FileSwapSpaceSpi()); + cfg.setMarshaller(new OptimizedMarshaller(false)); + + CacheConfiguration cacheCfg = defaultCacheConfiguration(); + + cacheCfg.setCacheMode(PARTITIONED); + cacheCfg.setAtomicityMode(TRANSACTIONAL); + cacheCfg.setDistributionMode(CacheDistributionMode.NEAR_PARTITIONED); + cacheCfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC); + cacheCfg.setSwapEnabled(true); + cacheCfg.setBackups(1); + cacheCfg.setEvictionPolicy(evictsEnabled() ? new CacheLruEvictionPolicy(100) : null); + + CacheQueryConfiguration qcfg = new CacheQueryConfiguration(); + + qcfg.setIndexPrimitiveKey(true); + + cacheCfg.setQueryConfiguration(qcfg); + + if (offheapEnabled() && evictsEnabled()) + cacheCfg.setOffHeapMaxMemory(1000); // Small offheap for evictions. + + cfg.setCacheConfiguration(cacheCfg); + + QueryConfiguration indexing = new QueryConfiguration(); + + indexing.setMaxOffheapRowsCacheSize(128); + + if (offheapEnabled()) + indexing.setMaxOffHeapMemory(0); + + cfg.setQueryConfiguration(indexing); + + GridQueryProcessor.idxCls = FakeIndexing.class; + + return cfg; + } + + /** + * + */ + private static class FakeIndexing extends IgniteH2Indexing { + @Override public void onSwap(@Nullable String spaceName, Object key) throws IgniteCheckedException { + super.onSwap(spaceName, key); + + idxSwapCnt.incrementAndGet(); + } + + @Override public void onUnswap(@Nullable String spaceName, Object key, Object val, byte[] valBytes) + throws IgniteCheckedException { + super.onUnswap(spaceName, key, val, valBytes); + + idxUnswapCnt.incrementAndGet(); + } + } + + /** @return {@code true} If offheap enabled. */ + protected boolean offheapEnabled() { + return false; + } + + /** @return {@code true} If evictions enabled. */ + protected boolean evictsEnabled() { + return true; + } + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + super.beforeTest(); + + // Clean up all caches. + for (int i = 0; i < GRID_CNT; i++) { - GridCache<Object, Object> c = grid(i).cache(null); ++ GridCache<Object, Object> c = ((IgniteKernal)grid(i)).cache(null); + + assertEquals(0, c.size()); + } + } + + /** {@inheritDoc} */ + @Override protected void beforeTestsStarted() throws Exception { + assert GRID_CNT >= 2 : "Constant GRID_CNT must be greater than or equal to 2."; + + startGridsMultiThreaded(GRID_CNT); + } + + /** {@inheritDoc} */ + @Override protected void afterTestsStopped() throws Exception { + stopAllGrids(); + + if (evictsEnabled()) { + assertTrue(idxSwapCnt.get() > 0); + assertTrue(idxUnswapCnt.get() > 0); + } + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + // Clean up all caches. + for (int i = 0; i < GRID_CNT; i++) { - GridCache<Object, Object> c = grid(i).cache(null); ++ GridCache<Object, Object> c = ((IgniteKernal)grid(i)).cache(null); + + c.removeAll(); + + // Fix for tests where mapping was removed at primary node + // but was not removed at others. + // removeAll() removes mapping only when it presents at a primary node. + // To remove all mappings used force remove by key. + if (c.size() > 0) { + for (Object k : c.keySet()) { + c.remove(k); + } + } + + Iterator<Map.Entry<Object, Object>> it = c.swapIterator(); + + while (it.hasNext()) { + it.next(); + + it.remove(); + } + + it = c.offHeapIterator(); + + while (it.hasNext()) { + it.next(); + + it.remove(); + } + + assertEquals("Swap keys: " + c.swapKeys(), 0, c.swapKeys()); + assertEquals(0, c.offHeapEntriesCount()); + assertEquals(0, c.size()); + } + } + + /** {@inheritDoc} */ + @Override protected void info(String msg) { + if (TEST_INFO) + super.info(msg); + } + + /** + * @param entries Entries. + * @param g Grid. + * @return Affinity nodes. + */ + private Set<UUID> affinityNodes(Iterable<Cache.Entry<Integer, Integer>> entries, Ignite g) { + Set<UUID> nodes = new HashSet<>(); + + for (Cache.Entry<Integer, Integer> entry : entries) - nodes.add(g.cache(null).affinity().mapKeyToPrimaryAndBackups(entry.getKey()).iterator().next().id()); ++ nodes.add(((IgniteKernal)g).cache(null).affinity().mapKeyToPrimaryAndBackups(entry.getKey()).iterator().next().id()); + + return nodes; + } + + /** + * JUnit. + * + * @throws Exception If failed. + */ + @SuppressWarnings({"TooBroadScope"}) + public void testMultiThreadedSwapUnswapString() throws Exception { + int threadCnt = 150; + final int keyCnt = 2000; + final int valCnt = 10000; + + final Ignite g = grid(0); + + // Put test values into cache. + final IgniteCache<Integer, String> c = g.jcache(null); + final IgniteCache<Integer, Long> cl = g.jcache(null); + - assertEquals(0, g.cache(null).size()); ++ assertEquals(0, ((IgniteKernal)g).cache(null).size()); + assertEquals(0, c.query(new SqlQuery(String.class, "1 = 1")).getAll().size()); + assertEquals(0, cl.query(new SqlQuery(Long.class, "1 = 1")).getAll().size()); + + Random rnd = new Random(); + + for (int i = 0; i < keyCnt; i += 1 + rnd.nextInt(3)) { + c.put(i, String.valueOf(rnd.nextInt(valCnt))); + + if (evictsEnabled() && rnd.nextBoolean()) + c.localEvict(Arrays.asList(i)); + } + + final AtomicBoolean done = new AtomicBoolean(); + + IgniteInternalFuture<?> fut = multithreadedAsync(new CAX() { + @Override public void applyx() throws IgniteCheckedException { + Random rnd = new Random(); + + while (!done.get()) { + switch (rnd.nextInt(5)) { + case 0: + c.put(rnd.nextInt(keyCnt), String.valueOf(rnd.nextInt(valCnt))); + + break; + case 1: + if (evictsEnabled()) + c.localEvict(Arrays.asList(rnd.nextInt(keyCnt))); + + break; + case 2: + c.remove(rnd.nextInt(keyCnt)); + + break; + case 3: + c.get(rnd.nextInt(keyCnt)); + + break; + case 4: + int from = rnd.nextInt(valCnt); + + QueryCursor<Cache.Entry<Integer, String>> qry = c.query( - new SqlQuery(String.class, "_val between ? and ?").setArgs( - String.valueOf(from), String.valueOf(from + 250))); ++ new SqlQuery(String.class, "_val between ? and ?").setArgs( ++ String.valueOf(from), String.valueOf(from + 250))); + + Collection<Cache.Entry<Integer, String>> res = qry.getAll(); + + for (Cache.Entry<Integer, String> ignored : res) { + //No-op. + } + } + } + } + }, threadCnt); + + Thread.sleep(DURATION); + + done.set(true); + + fut.get(); + } + + /** + * JUnit. + * + * @throws Exception If failed. + */ + @SuppressWarnings({"TooBroadScope"}) + public void testMultiThreadedSwapUnswapLong() throws Exception { + int threadCnt = 150; + final int keyCnt = 2000; + final int valCnt = 10000; + + final Ignite g = grid(0); + + // Put test values into cache. + final IgniteCache<Integer, Long> c = g.jcache(null); + final IgniteCache<Integer, String> c1 = g.jcache(null); + - assertEquals(0, g.cache(null).size()); ++ assertEquals(0, ((IgniteKernal)g).cache(null).size()); + assertEquals(0, c1.query(new SqlQuery(String.class, "1 = 1")).getAll().size()); + assertEquals(0, c.query(new SqlQuery(Long.class, "1 = 1")).getAll().size()); + + Random rnd = new Random(); + + for (int i = 0; i < keyCnt; i += 1 + rnd.nextInt(3)) { - c.put(i, (long) rnd.nextInt(valCnt)); ++ c.put(i, (long)rnd.nextInt(valCnt)); + + if (evictsEnabled() && rnd.nextBoolean()) + c.localEvict(Arrays.asList(i)); + } + + final AtomicBoolean done = new AtomicBoolean(); + + IgniteInternalFuture<?> fut = multithreadedAsync(new CAX() { + @Override public void applyx() throws IgniteCheckedException { + Random rnd = new Random(); + + while (!done.get()) { + int key = rnd.nextInt(keyCnt); + + switch (rnd.nextInt(5)) { + case 0: - c.put(key, (long) rnd.nextInt(valCnt)); ++ c.put(key, (long)rnd.nextInt(valCnt)); + + break; + case 1: + if (evictsEnabled()) + c.localEvict(Arrays.asList(key)); + + break; + case 2: + c.remove(key); + + break; + case 3: + c.get(key); + + break; + case 4: + int from = rnd.nextInt(valCnt); + + Collection<Cache.Entry<Integer, Long>> res = c.query(new SqlQuery(Long.class, + "_val between ? and ?").setArgs(from, from + 250)).getAll(); + + for (Cache.Entry<Integer, Long> ignored : res) { + //No-op. + } + } + } + } + }, threadCnt); + + Thread.sleep(DURATION); + + done.set(true); + + fut.get(); + } + + /** + * JUnit. + * + * @throws Exception If failed. + */ + @SuppressWarnings({"TooBroadScope"}) + public void testMultiThreadedSwapUnswapLongString() throws Exception { + int threadCnt = 150; + final int keyCnt = 2000; + final int valCnt = 10000; + + final Ignite g = grid(0); + + // Put test values into cache. + final IgniteCache<Integer, Object> c = g.jcache(null); + + assertEquals(0, g.jcache(null).size()); + assertEquals(0, c.query(new SqlQuery(Object.class, "1 = 1")).getAll().size()); + + Random rnd = new Random(); + + for (int i = 0; i < keyCnt; i += 1 + rnd.nextInt(3)) { - c.put(i, rnd.nextBoolean() ? (long) rnd.nextInt(valCnt) : String.valueOf(rnd.nextInt(valCnt))); ++ c.put(i, rnd.nextBoolean() ? (long)rnd.nextInt(valCnt) : String.valueOf(rnd.nextInt(valCnt))); + + if (evictsEnabled() && rnd.nextBoolean()) + c.localEvict(Arrays.asList(i)); + } + + final AtomicBoolean done = new AtomicBoolean(); + + IgniteInternalFuture<?> fut = multithreadedAsync(new CAX() { + @Override public void applyx() throws IgniteCheckedException { + Random rnd = new Random(); + + while (!done.get()) { + int key = rnd.nextInt(keyCnt); + + switch (rnd.nextInt(5)) { + case 0: - c.put(key, rnd.nextBoolean() ? (long) rnd.nextInt(valCnt) : - String.valueOf(rnd.nextInt(valCnt))); ++ c.put(key, rnd.nextBoolean() ? (long)rnd.nextInt(valCnt) : ++ String.valueOf(rnd.nextInt(valCnt))); + + break; + case 1: + if (evictsEnabled()) + c.localEvict(Arrays.asList(key)); + + break; + case 2: + c.remove(key); + + break; + case 3: + c.get(key); + + break; + case 4: + int from = rnd.nextInt(valCnt); + + Collection<Cache.Entry<Integer, Object>> res = c.query( + new SqlQuery(Object.class, "_val between ? and ?").setArgs(from, from + 250)) + .getAll(); + + for (Cache.Entry<Integer, Object> ignored : res) { + //No-op. + } + } + } + } + }, threadCnt); + + Thread.sleep(DURATION); + + done.set(true); + + fut.get(); + } + + /** + * @throws Exception If failed. + */ + @SuppressWarnings({"TooBroadScope"}) + public void testMultiThreadedSwapUnswapObject() throws Exception { + int threadCnt = 50; + final int keyCnt = 4000; + final int valCnt = 10000; + + final Ignite g = grid(0); + + // Put test values into cache. + final IgniteCache<Integer, TestValue> c = g.jcache(null); + - assertEquals(0, g.cache(null).size()); ++ assertEquals(0, ((IgniteKernal)g).cache(null).size()); + assertEquals(0, c.query(new SqlQuery(TestValue.class, "1 = 1")).getAll().size()); + + Random rnd = new Random(); + + for (int i = 0; i < keyCnt; i += 1 + rnd.nextInt(3)) { + c.put(i, new TestValue(rnd.nextInt(valCnt))); + + if (evictsEnabled() && rnd.nextBoolean()) + c.localEvict(Arrays.asList(i)); + } + + final AtomicBoolean done = new AtomicBoolean(); + + IgniteInternalFuture<?> fut = multithreadedAsync(new CAX() { + @Override public void applyx() throws IgniteCheckedException { + Random rnd = new Random(); + + while (!done.get()) { + int key = rnd.nextInt(keyCnt); + + switch (rnd.nextInt(5)) { + case 0: + c.put(key, new TestValue(rnd.nextInt(valCnt))); + + break; + case 1: + if (evictsEnabled()) + c.localEvict(Arrays.asList(key)); + + break; + case 2: + c.remove(key); + + break; + case 3: + c.get(key); + + break; + case 4: + int from = rnd.nextInt(valCnt); + + Collection<Cache.Entry<Integer, TestValue>> res = + c.query(new SqlQuery(TestValue.class, "TestValue.val between ? and ?").setArgs( + from, from + 250)).getAll(); + + for (Cache.Entry<Integer, TestValue> ignored : res) { + //No-op. + } + } + } + } + }, threadCnt); + + Thread.sleep(DURATION); + + done.set(true); + + fut.get(); + } + + /** + * JUnit. + * + * @throws Exception If failed. + */ + @SuppressWarnings({"TooBroadScope"}) + public void testMultiThreadedSameQuery() throws Exception { + int threadCnt = 50; + final int keyCnt = 10; + final int logMod = 5000; + + final Ignite g = grid(0); + + // Put test values into cache. + final IgniteCache<Integer, Integer> c = g.jcache(null); + + for (int i = 0; i < keyCnt; i++) { + c.put(i, i); + + c.localEvict(Arrays.asList(i)); + } + + final AtomicInteger cnt = new AtomicInteger(); + + final AtomicBoolean done = new AtomicBoolean(); + + IgniteInternalFuture<?> fut = multithreadedAsync( + new CAX() { + @Override public void applyx() throws IgniteCheckedException { + int iter = 0; + + while (!done.get() && !Thread.currentThread().isInterrupted()) { + iter++; + + Collection<Cache.Entry<Integer, Integer>> entries = + c.query(new SqlQuery(Integer.class, "_val >= 0")).getAll(); + + assert entries != null; + + assertEquals("Query results [entries=" + entries + ", aff=" + affinityNodes(entries, g) + + ", iteration=" + iter + ']', keyCnt, entries.size()); + + if (cnt.incrementAndGet() % logMod == 0) { + GridCacheQueryManager<Object, Object> qryMgr = + ((IgniteKernal)g).internalCache().context().queries(); + + assert qryMgr != null; + + qryMgr.printMemoryStats(); + } + } + } + }, threadCnt); + + Thread.sleep(DURATION); + + info("Finishing test..."); + + done.set(true); + + fut.get(); + } + + /** + * JUnit. + * + * @throws Exception If failed. + */ + @SuppressWarnings({"TooBroadScope"}) + public void testMultiThreadedNewQueries() throws Exception { + int threadCnt = 50; + final int keyCnt = 10; + final int logMod = 5000; + + final Ignite g = grid(0); + + // Put test values into cache. + final IgniteCache<Integer, Integer> c = g.jcache(null); + + for (int i = 0; i < keyCnt; i++) { + c.put(i, i); + + c.localEvict(Arrays.asList(i)); + } + + final AtomicInteger cnt = new AtomicInteger(); + + final AtomicBoolean done = new AtomicBoolean(); + + IgniteInternalFuture<?> fut = multithreadedAsync(new CAX() { + @Override public void applyx() throws IgniteCheckedException { + int iter = 0; + + while (!done.get() && !Thread.currentThread().isInterrupted()) { + iter++; + + Collection<Cache.Entry<Integer, Integer>> entries = + c.query(new SqlQuery(Integer.class, "_val >= 0")).getAll(); + + assert entries != null; + + assertEquals("Entries count is not as expected on iteration: " + iter, keyCnt, entries.size()); + + if (cnt.incrementAndGet() % logMod == 0) { + GridCacheQueryManager<Object, Object> qryMgr = + ((IgniteKernal)g).internalCache().context().queries(); + + assert qryMgr != null; + + qryMgr.printMemoryStats(); + } + } + } + }, threadCnt); + + Thread.sleep(DURATION); + + done.set(true); + + fut.get(); + } + + /** + * JUnit. + * + * @throws Exception If failed. + */ + @SuppressWarnings({"TooBroadScope"}) + public void testMultiThreadedScanQuery() throws Exception { + int threadCnt = 50; + final int keyCnt = 500; + final int logMod = 5000; + + final Ignite g = grid(0); + + // Put test values into cache. + final IgniteCache<Integer, Integer> c = g.jcache(null); + + for (int i = 0; i < keyCnt; i++) + c.put(i, i); + + final AtomicInteger cnt = new AtomicInteger(); + + final AtomicBoolean done = new AtomicBoolean(); + + IgniteInternalFuture<?> fut = multithreadedAsync( + new CAX() { + @Override public void applyx() throws IgniteCheckedException { + int iter = 0; + + while (!done.get() && !Thread.currentThread().isInterrupted()) { + iter++; + + // Scan query. + Collection<Cache.Entry<Integer, Integer>> entries = + c.query(new ScanQuery<Integer, Integer>()).getAll(); + + assert entries != null; + + assertEquals("Entries count is not as expected on iteration: " + iter, keyCnt, entries.size()); + + if (cnt.incrementAndGet() % logMod == 0) { + GridCacheQueryManager<Object, Object> qryMgr = + ((IgniteKernal)g).internalCache().context().queries(); + + assert qryMgr != null; + + qryMgr.printMemoryStats(); + } + } + } + }, threadCnt); + + Thread.sleep(DURATION); + + done.set(true); + + fut.get(); + } + + /** + * Test value. + */ + private static class TestValue implements Serializable { + /** Value. */ + @QuerySqlField + private int val; + + /** + * @param val Value. + */ + private TestValue(int val) { + this.val = val; + } + + /** + * @return Value. + */ + public int value() { + return val; + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheSqlQueryMultiThreadedSelfTest.java ---------------------------------------------------------------------- diff --cc modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheSqlQueryMultiThreadedSelfTest.java index 844826d,0000000..14bf9cc mode 100644,000000..100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheSqlQueryMultiThreadedSelfTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheSqlQueryMultiThreadedSelfTest.java @@@ -1,135 -1,0 +1,139 @@@ +/* + * 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.cache; + +import org.apache.ignite.*; +import org.apache.ignite.cache.query.*; +import org.apache.ignite.cache.query.annotations.*; ++import org.apache.ignite.cache.*; ++import org.apache.ignite.cache.query.annotations.*; +import org.apache.ignite.configuration.*; ++import org.apache.ignite.internal.processors.cache.query.*; ++import org.apache.ignite.internal.*; +import org.apache.ignite.spi.discovery.tcp.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*; +import org.apache.ignite.testframework.*; +import org.apache.ignite.testframework.junits.common.*; + +import javax.cache.*; +import java.io.*; +import java.util.*; +import java.util.concurrent.*; + +import static org.apache.ignite.cache.CacheAtomicityMode.*; +import static org.apache.ignite.cache.CacheDistributionMode.*; +import static org.apache.ignite.cache.CacheMode.*; + +/** + * + */ +public class IgniteCacheSqlQueryMultiThreadedSelfTest extends GridCommonAbstractTest { + /** */ + private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration c = super.getConfiguration(gridName); + + TcpDiscoverySpi disco = new TcpDiscoverySpi(); + + disco.setIpFinder(ipFinder); + + c.setDiscoverySpi(disco); + + CacheConfiguration ccfg = new CacheConfiguration(); + + ccfg.setCacheMode(PARTITIONED); + ccfg.setDistributionMode(PARTITIONED_ONLY); + ccfg.setQueryIndexEnabled(true); + ccfg.setBackups(1); + ccfg.setAtomicityMode(TRANSACTIONAL); + + c.setCacheConfiguration(ccfg); + + return c; + } + + /** {@inheritDoc} */ + @Override protected void beforeTestsStarted() throws Exception { + super.beforeTestsStarted(); + + startGrids(2); + + awaitPartitionMapExchange(); + } + + /** {@inheritDoc} */ + @Override protected void afterTestsStopped() throws Exception { + super.afterTestsStopped(); + + stopAllGrids(); + } + + /** + * @throws Exception If failed. + */ + public void testQuery() throws Exception { + final IgniteCache<Integer, Person> cache = grid(0).jcache(null); + + for (int i = 0; i < 2000; i++) + cache.put(i, new Person(i)); + + GridTestUtils.runMultiThreaded(new Callable<Void>() { + @Override public Void call() throws Exception { + for (int i = 0; i < 100; i++) { + QueryCursor<Cache.Entry<Integer, Person>> qry = + cache.query(new SqlQuery("age >= 0").setType("Person")); + + int cnt = 0; + + for (Cache.Entry<Integer, Person> e : qry) + cnt++; + + assertEquals(2000, cnt); + } + + return null; + } + }, 16, "test"); + } + + /** + * + */ + private static class Person implements Serializable { + /** */ + @QuerySqlField + private int age; + + /** + * @param age Age. + */ + Person(int age) { + this.age = age; + } + + /** + * @return Age/ + */ + public int age() { + return age; + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheAtomicFieldsQuerySelfTest.java ---------------------------------------------------------------------- diff --cc modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheAtomicFieldsQuerySelfTest.java index 7174b84,0000000..0d920f8 mode 100644,000000..100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheAtomicFieldsQuerySelfTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheAtomicFieldsQuerySelfTest.java @@@ -1,59 -1,0 +1,61 @@@ +/* + * 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.cache.distributed.near; + - import org.apache.ignite.cache.*; +import org.apache.ignite.cache.query.*; ++import org.apache.ignite.cache.*; ++import org.apache.ignite.internal.processors.cache.query.*; ++import org.apache.ignite.internal.*; +import org.apache.ignite.internal.util.typedef.*; + +import java.util.*; + +import static org.apache.ignite.cache.CacheAtomicityMode.*; +import static org.apache.ignite.cache.CacheDistributionMode.*; + +/** + * Tests for fields queries. + */ +public class IgniteCacheAtomicFieldsQuerySelfTest extends IgniteCachePartitionedFieldsQuerySelfTest { + /** {@inheritDoc} */ + @Override protected CacheAtomicityMode atomicityMode() { + return ATOMIC; + } + + /** {@inheritDoc} */ + @Override protected CacheDistributionMode distributionMode() { + return PARTITIONED_ONLY; + } + + /** + * + */ + public void testUnsupportedOperations() { + try { + QueryCursor<List<?>> qry = grid(0).jcache(null).queryFields(Query.sql( + "update Person set name = ?").setArgs("Mary Poppins")); + + qry.getAll(); + + fail("We don't support updates."); + } + catch (Exception e) { + X.println("___ " + e.getMessage()); + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryNodeRestartSelfTest.java ---------------------------------------------------------------------- diff --cc modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryNodeRestartSelfTest.java index 11ada41,0000000..a2afa48 mode 100644,000000..100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryNodeRestartSelfTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryNodeRestartSelfTest.java @@@ -1,223 -1,0 +1,224 @@@ +/* + * 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.cache.distributed.near; + ++import org.apache.ignite.cache.query.*; +import org.apache.ignite.*; +import org.apache.ignite.cache.*; - import org.apache.ignite.cache.query.*; +import org.apache.ignite.configuration.*; +import org.apache.ignite.events.*; +import org.apache.ignite.internal.*; +import org.apache.ignite.internal.processors.cache.*; ++import org.apache.ignite.internal.processors.cache.query.*; +import org.apache.ignite.internal.util.typedef.*; +import org.apache.ignite.internal.util.typedef.internal.*; +import org.apache.ignite.lang.*; +import org.apache.ignite.spi.discovery.tcp.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*; + +import javax.cache.*; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +import static org.apache.ignite.cache.CacheAtomicityMode.*; +import static org.apache.ignite.cache.CacheDistributionMode.*; +import static org.apache.ignite.cache.CacheMode.*; + +/** + * Test for distributed queries with node restarts. + */ +public class IgniteCacheQueryNodeRestartSelfTest extends GridCacheAbstractSelfTest { + /** */ + private static final int GRID_CNT = 3; + + /** */ + private static final int KEY_CNT = 1000; + + /** */ + private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); + + /** {@inheritDoc} */ + @Override protected int gridCount() { + return GRID_CNT; + } + + /** {@inheritDoc} */ + @Override protected long getTestTimeout() { + return 90 * 1000; + } + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration c = super.getConfiguration(gridName); + + TcpDiscoverySpi disco = new TcpDiscoverySpi(); + + disco.setIpFinder(ipFinder); + + c.setDiscoverySpi(disco); + + CacheConfiguration cc = defaultCacheConfiguration(); + + cc.setCacheMode(PARTITIONED); + cc.setBackups(1); + cc.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC); + cc.setAtomicityMode(TRANSACTIONAL); + cc.setDistributionMode(NEAR_PARTITIONED); + + CacheQueryConfiguration qcfg = new CacheQueryConfiguration(); + + qcfg.setIndexPrimitiveKey(true); + + cc.setQueryConfiguration(qcfg); + + c.setCacheConfiguration(cc); + + return c; + } + + /** + * JUnit. + * + * @throws Exception If failed. + */ + @SuppressWarnings({"TooBroadScope"}) + public void testRestarts() throws Exception { + int duration = 60 * 1000; + int qryThreadNum = 10; + final long nodeLifeTime = 2 * 1000; + final int logFreq = 20; + + final IgniteCache<Integer, Integer> cache = grid(0).jcache(null); + + assert cache != null; + + for (int i = 0; i < KEY_CNT; i++) + cache.put(i, i); + + assertEquals(KEY_CNT, cache.localSize()); + + final AtomicInteger qryCnt = new AtomicInteger(); + + final AtomicBoolean done = new AtomicBoolean(); + + IgniteInternalFuture<?> fut1 = multithreadedAsync(new CAX() { + @Override public void applyx() throws IgniteCheckedException { + while (!done.get()) { + Collection<Cache.Entry<Integer, Integer>> res = + cache.query(new SqlQuery(Integer.class, "_val >= 0")).getAll(); + + assertFalse(res.isEmpty()); + + int c = qryCnt.incrementAndGet(); + + if (c % logFreq == 0) + info("Executed queries: " + c); + } + } + }, qryThreadNum); + + final AtomicInteger restartCnt = new AtomicInteger(); + + CollectingEventListener lsnr = new CollectingEventListener(); + + for (int i = 0; i < GRID_CNT; i++) + grid(i).events().localListen(lsnr, EventType.EVT_CACHE_PRELOAD_STOPPED); + + IgniteInternalFuture<?> fut2 = multithreadedAsync(new Callable<Object>() { + @SuppressWarnings({"BusyWait"}) + @Override public Object call() throws Exception { + while (!done.get()) { + int idx = GRID_CNT; + + startGrid(idx); + + Thread.sleep(nodeLifeTime); + + stopGrid(idx); + + int c = restartCnt.incrementAndGet(); + + if (c % logFreq == 0) + info("Node restarts: " + c); + } + + return true; + } + }, 1); + + Thread.sleep(duration); + + done.set(true); + + fut1.get(); + fut2.get(); + + info("Awaiting preload events [restartCnt=" + restartCnt.get() + ']'); + + boolean success = lsnr.awaitEvents(GRID_CNT * 2 * restartCnt.get(), 15000); + + for (int i = 0; i < GRID_CNT; i++) + grid(i).events().stopLocalListen(lsnr, EventType.EVT_CACHE_PRELOAD_STOPPED); + + assert success; + } + + /** Listener that will wait for specified number of events received. */ + private class CollectingEventListener implements IgnitePredicate<Event> { + /** Registered events count. */ + private int evtCnt; + + /** {@inheritDoc} */ + @Override public synchronized boolean apply(Event evt) { + evtCnt++; + + info("Processed event [evt=" + evt + ", evtCnt=" + evtCnt + ']'); + + notifyAll(); + + return true; + } + + /** + * Waits until total number of events processed is equal or greater then argument passed. + * + * @param cnt Number of events to wait. + * @param timeout Timeout to wait. + * @return {@code True} if successfully waited, {@code false} if timeout happened. + * @throws InterruptedException If thread is interrupted. + */ + public synchronized boolean awaitEvents(int cnt, long timeout) throws InterruptedException { + long start = U.currentTimeMillis(); + + long now = start; + + while (start + timeout > now) { + if (evtCnt >= cnt) + return true; + + wait(start + timeout - now); + + now = U.currentTimeMillis(); + } + + return false; + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/IgniteCacheReplicatedQuerySelfTest.java ---------------------------------------------------------------------- diff --cc modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/IgniteCacheReplicatedQuerySelfTest.java index 422f95b,0000000..25ab2ac mode 100644,000000..100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/IgniteCacheReplicatedQuerySelfTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/IgniteCacheReplicatedQuerySelfTest.java @@@ -1,577 -1,0 +1,579 @@@ +/* + * 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.cache.distributed.replicated; + +import org.apache.ignite.*; +import org.apache.ignite.cache.*; +import org.apache.ignite.cache.query.*; +import org.apache.ignite.cache.query.annotations.*; +import org.apache.ignite.configuration.*; ++import org.apache.ignite.cache.query.annotations.*; ++import org.apache.ignite.cluster.*; +import org.apache.ignite.events.*; +import org.apache.ignite.internal.*; +import org.apache.ignite.internal.processors.cache.*; +import org.apache.ignite.internal.processors.cache.query.*; +import org.apache.ignite.internal.util.future.*; +import org.apache.ignite.internal.util.lang.*; +import org.apache.ignite.internal.util.typedef.*; +import org.apache.ignite.internal.util.typedef.internal.*; +import org.apache.ignite.lang.*; +import org.apache.ignite.testframework.*; +import org.apache.ignite.transactions.*; +import org.springframework.util.*; + +import javax.cache.*; +import java.io.*; +import java.lang.reflect.*; +import java.sql.*; +import java.util.*; +import java.util.concurrent.*; + +import static org.apache.ignite.cache.CacheMode.*; +import static org.apache.ignite.events.EventType.*; + +/** + * Tests replicated query. + */ +public class IgniteCacheReplicatedQuerySelfTest extends IgniteCacheAbstractQuerySelfTest { + /** */ + private static final boolean TEST_DEBUG = false; + + /** Grid1. */ + private static Ignite ignite1; + + /** Grid2. */ + private static Ignite ignite2; + + /** Grid3. */ + private static Ignite ignite3; + + /** Cache1. */ + private static IgniteCache<CacheKey, CacheValue> cache1; + + /** Cache2. */ + private static IgniteCache<CacheKey, CacheValue> cache2; + + /** Cache3. */ + private static IgniteCache<CacheKey, CacheValue> cache3; + + /** Key serialization cnt. */ + private static volatile int keySerCnt; + + /** Key deserialization count. */ + private static volatile int keyDesCnt; + + /** Value serialization count. */ + private static volatile int valSerCnt; + + /** Value deserialization count. */ + private static volatile int valDesCnt; + + /** {@inheritDoc} */ + @Override protected int gridCount() { + return 3; + } + + /** {@inheritDoc} */ + @Override protected CacheMode cacheMode() { + return REPLICATED; + } + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + super.beforeTest(); + + ignite1 = grid(0); + ignite2 = grid(1); + ignite3 = grid(2); + + cache1 = ignite1.jcache(null); + cache2 = ignite2.jcache(null); + cache3 = ignite3.jcache(null); + } + + /** + * @throws Exception If failed. + */ + public void testClientOnlyNode() throws Exception { + try { + Ignite g = startGrid("client"); + + IgniteCache<Integer, Integer> c = g.jcache(null); + + for (int i = 0; i < 10; i++) + c.put(i, i); + + // Client cache should be empty. + assertEquals(0, c.localSize()); + + Collection<Cache.Entry<Integer, Integer>> res = + c.query(new SqlQuery(Integer.class, "_key >= 5 order by _key")).getAll(); + + assertEquals(5, res.size()); + + int i = 5; + + for (Cache.Entry<Integer, Integer> e : res) { + assertEquals(i, e.getKey().intValue()); + assertEquals(i, e.getValue().intValue()); + + i++; + } + } + finally { + stopGrid("client"); + } + } + + /** + * JUnit. + * + * @throws Exception If failed. + */ + public void testIterator() throws Exception { + int keyCnt = 100; + + for (int i = 0; i < keyCnt; i++) + cache1.put(new CacheKey(i), new CacheValue("val" + i)); + + assertEquals(keyCnt, cache1.localSize()); + assertEquals(keyCnt, cache2.localSize()); + assertEquals(keyCnt, cache3.localSize()); + + QueryCursor<Cache.Entry<CacheKey, CacheValue>> qry = + cache1.query(new SqlQuery(CacheValue.class, "true")); + + Iterator<Cache.Entry<CacheKey, CacheValue>> iter = qry.iterator(); + + assert iter.hasNext(); + + int cnt = 0; + + while (iter.hasNext()) { + iter.next(); + + cnt++; + } + + assertEquals(keyCnt, cnt); + } + + /** + * @throws Exception If test failed. + */ + public void testLocalQuery() throws Exception { + cache1.clear(); + + IgniteTx tx = ignite1.transactions().txStart(); + + try { + cache1.put(new CacheKey(1), new CacheValue("1")); + cache1.put(new CacheKey(2), new CacheValue("2")); + cache1.put(new CacheKey(3), new CacheValue("3")); + cache1.put(new CacheKey(4), new CacheValue("4")); + + tx.commit(); + + info("Committed transaction: " + tx); + } + catch (IgniteException e) { + tx.rollback(); + + throw e; + } + + checkQueryResults(cache1); + checkQueryResults(cache2); + checkQueryResults(cache3); + } + + /** + * @throws Exception If test failed. + */ + public void testDistributedQuery() throws Exception { + int keyCnt = 4; + + final CountDownLatch latch = new CountDownLatch(keyCnt * 2); + + IgnitePredicate<Event> lsnr = new IgnitePredicate<Event>() { + @Override public boolean apply(Event evt) { + latch.countDown(); + + return true; + } + }; + + ignite2.events().localListen(lsnr, EventType.EVT_CACHE_OBJECT_PUT); + ignite3.events().localListen(lsnr, EventType.EVT_CACHE_OBJECT_PUT); + + IgniteTx tx = ignite1.transactions().txStart(); + + try { + for (int i = 1; i <= keyCnt; i++) + cache1.put(new CacheKey(i), new CacheValue(String.valueOf(i))); + + tx.commit(); + + info("Committed transaction: " + tx); + } + catch (IgniteException e) { + tx.rollback(); + + throw e; + } + + latch.await(); + + QueryCursor<Cache.Entry<CacheKey, CacheValue>> qry = + cache1.query(new SqlQuery(CacheValue.class, "val > 1 and val < 4")); + + // Distributed query. + assertEquals(2, qry.getAll().size()); + + // Create new query, old query cannot be modified after it has been executed. + qry = cache3.localQuery(new SqlQuery(CacheValue.class, "val > 1 and val < 4")); + + // Tests execute on node. + Iterator<Cache.Entry<CacheKey, CacheValue>> iter = qry.iterator(); + + assert iter != null; + assert iter.hasNext(); + + iter.next(); + + assert iter.hasNext(); + + iter.next(); + + assert !iter.hasNext(); + } + + /** + * Returns private field {@code qryIters} of {@link GridCacheQueryManager} for the given grid. + * + * @param g Grid which {@link GridCacheQueryManager} should be observed. + * @return {@code qryIters} of {@link GridCacheQueryManager}. + */ + private ConcurrentMap<UUID, + Map<Long, GridFutureAdapter<GridCloseableIterator<IgniteBiTuple<CacheKey, CacheValue>>>>> + distributedQueryManagerQueryItersMap(Ignite g) { + GridCacheContext ctx = ((IgniteKernal)g).internalCache().context(); + + Field qryItersField = ReflectionUtils.findField(ctx.queries().getClass(), "qryIters"); + + qryItersField.setAccessible(true); + + return (ConcurrentMap<UUID, + Map<Long, GridFutureAdapter<GridCloseableIterator<IgniteBiTuple<CacheKey, CacheValue>>>>>) + ReflectionUtils.getField(qryItersField, ctx.queries()); + } + + /** + * @throws Exception If test failed. + */ + public void testToString() throws Exception { + int keyCnt = 4; + + for (int i = 1; i <= keyCnt; i++) + cache1.put(new CacheKey(i), new CacheValue(String.valueOf(i))); + + // Create query with key filter. + + QueryCursor<Cache.Entry<CacheKey, CacheValue>> qry = + cache1.query(new SqlQuery(CacheValue.class, "val > 0")); + + assertEquals(keyCnt, qry.getAll().size()); + } + + /** + * TODO + * + * @throws Exception If failed. + */ + public void _testLostIterator() throws Exception { + IgniteCache<Integer, Integer> cache = ignite.jcache(null); + + for (int i = 0; i < 1000; i++) + cache.put(i, i); + + QueryCursor<Cache.Entry<Integer, Integer>> fut = null; + + for (int i = 0; i < cache.getConfiguration(CacheConfiguration.class).getMaximumQueryIteratorCount() + 1; i++) { + QueryCursor<Cache.Entry<Integer, Integer>> q = + cache.query(new SqlQuery(Integer.class, "_key >= 0 order by _key")); + + assertEquals(0, (int)q.iterator().next().getKey()); + + if (fut == null) + fut = q; + } + + final QueryCursor<Cache.Entry<Integer, Integer>> fut0 = fut; + + GridTestUtils.assertThrows(log, new Callable<Object>() { + @Override public Object call() throws Exception { + int i = 0; + + Cache.Entry<Integer, Integer> e; + + while ((e = fut0.iterator().next()) != null) + assertEquals(++i, (int)e.getKey()); + + return null; + } + }, IgniteException.class, null); + } + + /** + * TODO enable + * + * @throws Exception If failed. + */ + public void _testNodeLeft() throws Exception { + try { + Ignite g = startGrid(); + + IgniteCache<Integer, Integer> cache = g.jcache(null); + + for (int i = 0; i < 1000; i++) + cache.put(i, i); + + QueryCursor<Cache.Entry<Integer, Integer>> q = + cache.query(new SqlQuery(Integer.class, "_key >= 0 order by _key")); + + assertEquals(0, (int) q.iterator().next().getKey()); + + final ConcurrentMap<UUID, Map<Long, GridFutureAdapter<GridCloseableIterator< + IgniteBiTuple<Integer, Integer>>>>> map = + U.field(((IgniteKernal)grid(0)).internalCache().context().queries(), "qryIters"); + + // fut.nextX() does not guarantee the request has completed on remote node + // (we could receive page from local one), so we need to wait. + assertTrue(GridTestUtils.waitForCondition(new PA() { + @Override public boolean apply() { + return map.size() == 1; + } + }, getTestTimeout())); + + Map<Long, GridFutureAdapter<GridCloseableIterator<IgniteBiTuple<Integer, Integer>>>> futs = + map.get(g.cluster().localNode().id()); + + assertEquals(1, futs.size()); + + GridCloseableIterator<IgniteBiTuple<Integer, Integer>> iter = + (GridCloseableIterator<IgniteBiTuple<Integer, Integer>>)((IgniteInternalFuture)F.first(futs.values()).get()).get(); + + ResultSet rs = U.field(iter, "data"); + + assertFalse(rs.isClosed()); + + final UUID nodeId = g.cluster().localNode().id(); + final CountDownLatch latch = new CountDownLatch(1); + + grid(0).events().localListen(new IgnitePredicate<Event>() { + @Override public boolean apply(Event evt) { + if (((DiscoveryEvent)evt).eventNode().id().equals(nodeId)) + latch.countDown(); + + return true; + } + }, EVT_NODE_LEFT); + + stopGrid(); + + latch.await(); + + assertEquals(0, map.size()); + assertTrue(rs.isClosed()); + } + finally { + // Ensure that additional node is stopped. + stopGrid(); + } + } + + /** + * @param cache Cache. + * @throws Exception If check failed. + */ + private void checkQueryResults(IgniteCache<CacheKey, CacheValue> cache) throws Exception { + QueryCursor<Cache.Entry<CacheKey, CacheValue>> qry = + cache.localQuery(new SqlQuery(CacheValue.class, "val > 1 and val < 4")); + + Iterator<Cache.Entry<CacheKey, CacheValue>> iter = qry.iterator(); + + assert iter != null; + assert iter.hasNext(); + + Cache.Entry<CacheKey, CacheValue> entry = iter.next(); + + assert entry.getKey().equals(new CacheKey(2)) || entry.getKey().equals(new CacheKey(3)); + + assert iter.hasNext(); + + entry = iter.next(); + + assert entry.getKey().equals(new CacheKey(2)) || entry.getKey().equals(new CacheKey(3)); + assert !iter.hasNext(); + } + + /** + * Cache key. + */ + private static class CacheKey implements Externalizable { + /** Key. */ + private int key; + + /** + * @param key Key. + */ + CacheKey(int key) { + this.key = key; + } + + /** + * + */ + public CacheKey() { + /* No-op. */ + } + + /** + * @return Key. + */ + public int getKey() { + return key; + } + + /** {@inheritDoc} */ + @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + key = in.readInt(); + + keyDesCnt++; + + if (TEST_DEBUG) + X.println("Deserialized demo key [keyDesCnt=" + keyDesCnt + ", key=" + this + ']'); + } + + /** {@inheritDoc} */ + @Override public void writeExternal(ObjectOutput out) throws IOException { + out.writeInt(key); + + keySerCnt++; + + if (TEST_DEBUG) + X.println("Serialized demo key [serCnt=" + keySerCnt + ", key=" + this + ']'); + } + + /** {@inheritDoc} */ + @Override public boolean equals(Object o) { + CacheKey cacheKey; + + if (o instanceof CacheKey) + cacheKey = (CacheKey)o; + else + return false; + + return key == cacheKey.key; + } + + /** {@inheritDoc} */ + @Override public int hashCode() { + return key; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(CacheKey.class, this); + } + } + + /** + * Cache value.. + */ + private static class CacheValue implements Externalizable { + /** Value. */ + @QuerySqlField + private String val; + + /** + * @param val Value. + */ + CacheValue(String val) { + this.val = val; + } + + /** + * + */ + public CacheValue() { + /* No-op. */ + } + + /** + * @return Value. + */ + public String getValue() { + return val; + } + + /** {@inheritDoc} */ + @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + val = U.readString(in); + + valDesCnt++; + + if (TEST_DEBUG) + X.println("Deserialized demo value [valDesCnt=" + valDesCnt + ", val=" + this + ']'); + } + + /** {@inheritDoc} */ + @Override public void writeExternal(ObjectOutput out) throws IOException { + U.writeString(out, val); + + valSerCnt++; + + if (TEST_DEBUG) + X.println("Serialized demo value [serCnt=" + valSerCnt + ", val=" + this + ']'); + } + + /** {@inheritDoc} */ + @Override public boolean equals(Object o) { + if (this == o) + return true; + + if (o == null || getClass() != o.getClass()) + return false; + + CacheValue val = (CacheValue)o; + + return !(this.val != null ? !this.val.equals(val.val) : val.val != null); + } + + /** {@inheritDoc} */ + @Override public int hashCode() { + return val != null ? val.hashCode() : 0; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(CacheValue.class, this); + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/local/IgniteCacheLocalQuerySelfTest.java ---------------------------------------------------------------------- diff --cc modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/local/IgniteCacheLocalQuerySelfTest.java index 8a92553,0000000..5e9cf62 mode 100644,000000..100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/local/IgniteCacheLocalQuerySelfTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/local/IgniteCacheLocalQuerySelfTest.java @@@ -1,82 -1,0 +1,85 @@@ +/* + * 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.cache.local; + +import org.apache.ignite.*; +import org.apache.ignite.cache.*; ++import org.apache.ignite.internal.*; +import org.apache.ignite.cache.query.*; +import org.apache.ignite.internal.processors.cache.*; ++import org.apache.ignite.internal.processors.cache.query.*; ++import org.apache.ignite.lang.*; + +import javax.cache.*; +import java.util.*; + +import static org.apache.ignite.cache.CacheMode.*; + +/** + * Tests local query. + */ +public class IgniteCacheLocalQuerySelfTest extends IgniteCacheAbstractQuerySelfTest { + /** {@inheritDoc} */ + @Override protected int gridCount() { + return 1; + } + + /** {@inheritDoc} */ + @Override protected CacheMode cacheMode() { + return LOCAL; + } + + /** - * @throws IgniteCheckedException If test failed. ++ * @throws Exception If test failed. + */ + public void testQueryLocal() throws Exception { + IgniteCache<Integer, String> cache = ignite.jcache(null); + + cache.put(1, "value1"); + cache.put(2, "value2"); + cache.put(3, "value3"); + cache.put(4, "value4"); + cache.put(5, "value5"); + + // Tests equals query. + QueryCursor<Cache.Entry<Integer, String>> qry = + cache.localQuery(new SqlQuery(String.class, "_val='value1'")); + + Iterator<Cache.Entry<Integer, String>> iter = qry.iterator(); + + Cache.Entry<Integer, String> entry = iter.next(); + + assert !iter.hasNext(); + + assert entry != null; + assert entry.getKey() == 1; + assert "value1".equals(entry.getValue()); + + // Tests like query. + qry = cache.localQuery(new SqlQuery(String.class, "_val like 'value%'")); + + iter = qry.iterator(); + + assert iter.next() != null; + assert iter.next() != null; + assert iter.next() != null; + assert iter.next() != null; + assert iter.next() != null; + assert !iter.hasNext(); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/GridH2IndexRebuildTest.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/sql/GridQueryParsingTest.java ---------------------------------------------------------------------- diff --cc modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/sql/GridQueryParsingTest.java index bb7550b,12e5bf8..5f97c86 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/sql/GridQueryParsingTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/sql/GridQueryParsingTest.java @@@ -19,7 -19,8 +19,9 @@@ package org.apache.ignite.internal.proc import org.apache.ignite.*; import org.apache.ignite.cache.*; +import org.apache.ignite.cache.query.annotations.*; + import org.apache.ignite.cache.query.*; + import org.apache.ignite.cache.query.annotations.*; import org.apache.ignite.configuration.*; import org.apache.ignite.internal.*; import org.apache.ignite.internal.processors.query.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java ---------------------------------------------------------------------- diff --cc modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java index 8edce6c,bf71d62..83b63d8 --- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java +++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java @@@ -43,37 -43,38 +43,39 @@@ public class IgniteCacheQuerySelfTestSu suite.addTestSuite(GridQueryParsingTest.class); // Queries tests. - suite.addTestSuite(GridCacheQueryLoadSelfTest.class); - suite.addTestSuite(GridCacheQueryMetricsSelfTest.class); - suite.addTestSuite(GridCacheLocalQuerySelfTest.class); - suite.addTestSuite(GridCacheLocalAtomicQuerySelfTest.class); - suite.addTestSuite(GridCacheReplicatedQuerySelfTest.class); - suite.addTestSuite(GridCacheReplicatedQueryP2PEnabledSelfTest.class); - suite.addTestSuite(GridCachePartitionedQuerySelfTest.class); - suite.addTestSuite(GridCacheAtomicQuerySelfTest.class); - suite.addTestSuite(GridCacheAtomicNearEnabledQuerySelfTest.class); - suite.addTestSuite(GridCachePartitionedQueryP2PEnabledSelfTest.class); - suite.addTestSuite(GridCachePartitionedQueryMultiThreadedSelfTest.class); - suite.addTestSuite(GridCacheQueryIndexSelfTest.class); + suite.addTestSuite(GridCacheQueryIndexDisabledSelfTest.class); + suite.addTestSuite(IgniteCacheQueryLoadSelfTest.class); + suite.addTestSuite(IgniteCacheLocalQuerySelfTest.class); + suite.addTestSuite(IgniteCacheLocalAtomicQuerySelfTest.class); + suite.addTestSuite(IgniteCacheReplicatedQuerySelfTest.class); + suite.addTestSuite(IgniteCacheReplicatedQueryP2PDisabledSelfTest.class); + suite.addTestSuite(IgniteCachePartitionedQuerySelfTest.class); + suite.addTestSuite(IgniteCacheAtomicQuerySelfTest.class); + suite.addTestSuite(IgniteCacheAtomicNearEnabledQuerySelfTest.class); + suite.addTestSuite(IgniteCachePartitionedQueryP2PDisabledSelfTest.class); + suite.addTestSuite(IgniteCachePartitionedQueryMultiThreadedSelfTest.class); + suite.addTestSuite(IgniteCacheQueryIndexSelfTest.class); suite.addTestSuite(GridCacheQueryInternalKeysSelfTest.class); - suite.addTestSuite(GridCacheQueryMultiThreadedSelfTest.class); - suite.addTestSuite(GridCacheQueryEvictsMultiThreadedSelfTest.class); - suite.addTestSuite(GridCacheQueryOffheapMultiThreadedSelfTest.class); - suite.addTestSuite(GridCacheQueryOffheapEvictsMultiThreadedSelfTest.class); - suite.addTestSuite(GridCacheQueryNodeRestartSelfTest.class); + suite.addTestSuite(IgniteCacheQueryMultiThreadedSelfTest.class); + //suite.addTestSuite(IgniteCacheQueryEvictsMultiThreadedSelfTest.class); + suite.addTestSuite(IgniteCacheQueryOffheapMultiThreadedSelfTest.class); + suite.addTestSuite(IgniteCacheQueryOffheapEvictsMultiThreadedSelfTest.class); + suite.addTestSuite(IgniteCacheSqlQueryMultiThreadedSelfTest.class); + suite.addTestSuite(IgniteCacheQueryNodeRestartSelfTest.class); suite.addTestSuite(GridCacheReduceQueryMultithreadedSelfTest.class); suite.addTestSuite(GridCacheCrossCacheQuerySelfTest.class); + suite.addTestSuite(GridCacheCrossCacheQuerySelfTestNewApi.class); + suite.addTestSuite(GridCacheSqlQueryMultiThreadedSelfTest.class); // Fields queries. - suite.addTestSuite(GridCacheLocalFieldsQuerySelfTest.class); - suite.addTestSuite(GridCacheReplicatedFieldsQuerySelfTest.class); - suite.addTestSuite(GridCacheReplicatedFieldsQueryP2PEnableSelfTest.class); - suite.addTestSuite(GridCachePartitionedFieldsQuerySelfTest.class); - suite.addTestSuite(GridCacheAtomicFieldsQuerySelfTest.class); - suite.addTestSuite(GridCacheAtomicNearEnabledFieldsQuerySelfTest.class); - suite.addTestSuite(GridCachePartitionedFieldsQueryP2PEnabledSelfTest.class); - suite.addTestSuite(GridCacheFieldsQueryNoDataSelfTest.class); + suite.addTestSuite(IgniteCacheLocalFieldsQuerySelfTest.class); + suite.addTestSuite(IgniteCacheReplicatedFieldsQuerySelfTest.class); + suite.addTestSuite(IgniteCacheReplicatedFieldsQueryP2PDisabledSelfTest.class); + suite.addTestSuite(IgniteCachePartitionedFieldsQuerySelfTest.class); + suite.addTestSuite(IgniteCacheAtomicFieldsQuerySelfTest.class); + suite.addTestSuite(IgniteCacheAtomicNearEnabledFieldsQuerySelfTest.class); + suite.addTestSuite(IgniteCachePartitionedFieldsQueryP2PDisabledSelfTest.class); + suite.addTestSuite(IgniteCacheFieldsQueryNoDataSelfTest.class); // Continuous queries. suite.addTestSuite(GridCacheContinuousQueryLocalSelfTest.class); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/scalar/src/main/scala/org/apache/ignite/scalar/scalar.scala ---------------------------------------------------------------------- diff --cc modules/scalar/src/main/scala/org/apache/ignite/scalar/scalar.scala index 3aa3ecb,6f5553b..86a6b81 --- a/modules/scalar/src/main/scala/org/apache/ignite/scalar/scalar.scala +++ b/modules/scalar/src/main/scala/org/apache/ignite/scalar/scalar.scala @@@ -17,18 -17,16 +17,18 @@@ package org.apache.ignite.scalar +import org.apache.ignite.cache.GridCache +import org.apache.ignite.cache.query.annotations.{QuerySqlField, QueryTextField} + import java.net.URL + import java.util.UUID + + import org.apache.ignite._ + import org.apache.ignite.cache.query.annotations.{QuerySqlField, QueryTextField} import org.apache.ignite.cluster.ClusterNode import org.apache.ignite.configuration.IgniteConfiguration - import org.apache.ignite.internal.IgniteVersionUtils - import IgniteVersionUtils._ - import org.apache.ignite.{Ignite, IgniteDataLoader, IgniteState, Ignition} + import org.apache.ignite.internal.IgniteVersionUtils._ import org.jetbrains.annotations.Nullable - import java.net.URL - import java.util.UUID - import scala.annotation.meta.field /** http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/cache/VisorCacheCommand.scala ---------------------------------------------------------------------- diff --cc modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/cache/VisorCacheCommand.scala index 40337e2,3c1aa01..99232ec --- a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/cache/VisorCacheCommand.scala +++ b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/cache/VisorCacheCommand.scala @@@ -837,12 -833,9 +833,8 @@@ object VisorCacheCommand cacheT += ("Default Query Timeout", defaultCfg.queryTimeout()) cacheT += ("Query Indexing Enabled", cfg.queryIndexEnabled()) cacheT += ("Query Iterators Number", cfg.maxQueryIteratorCount()) - cacheT += ("Indexing SPI Name", cfg.indexingSpiName()) cacheT += ("Cache Interceptor", cfg.interceptor()) - cacheT += ("DGC Frequency", dgcCfg.frequency()) - cacheT += ("DGC Remove Locks Flag", dgcCfg.removedLocks()) - cacheT += ("DGC Suspect Lock Timeout", dgcCfg.suspectLockTimeout()) - cacheT += ("Store Enabled", storeCfg.enabled()) cacheT += ("Store", storeCfg.store()) cacheT += ("Store Values In Bytes", storeCfg.valueBytes()) http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/41bc7728/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/model/Person.java ---------------------------------------------------------------------- diff --cc modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/model/Person.java index d809bd1,df8d58b..28d47f6 --- a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/model/Person.java +++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/model/Person.java @@@ -17,9 -17,8 +17,10 @@@ package org.apache.ignite.yardstick.cache.model; + import org.apache.ignite.cache.query.annotations.*; +import org.apache.ignite.cache.query.annotations.*; + import java.io.*; /**
