sprint-3 - Renaming colocated -> collocated. Word-count example fixes.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/da0f1933 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/da0f1933 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/da0f1933 Branch: refs/heads/ignite-683-2 Commit: da0f193301260c6b550de4f33dd6b6a42a675b98 Parents: b39fd2a Author: Dmitiry Setrakyan <dsetrak...@gridgain.com> Authored: Sun Apr 12 18:19:53 2015 -0700 Committer: Dmitiry Setrakyan <dsetrak...@gridgain.com> Committed: Sun Apr 12 18:19:53 2015 -0700 ---------------------------------------------------------------------- .../streaming/wordcount/CacheConfig.java | 5 + .../streaming/wordcount/QueryWords.java | 10 +- .../ignite/cache/query/SqlFieldsQuery.java | 45 +++- .../processors/query/h2/IgniteH2Indexing.java | 2 +- .../IgniteCacheCollocatedQuerySelfTest.java | 209 +++++++++++++++++++ .../IgniteCacheColocatedQuerySelfTest.java | 209 ------------------- .../IgniteCacheQuerySelfTestSuite.java | 2 +- 7 files changed, 259 insertions(+), 223 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da0f1933/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/CacheConfig.java ---------------------------------------------------------------------- diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/CacheConfig.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/CacheConfig.java index 58704ca..bb2a18e 100644 --- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/CacheConfig.java +++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/CacheConfig.java @@ -18,6 +18,7 @@ package org.apache.ignite.examples.streaming.wordcount; import org.apache.ignite.cache.affinity.*; +import org.apache.ignite.cache.eviction.fifo.*; import org.apache.ignite.configuration.*; import javax.cache.configuration.*; @@ -43,6 +44,10 @@ public class CacheConfig { // Sliding window of 1 seconds. cfg.setExpiryPolicyFactory(FactoryBuilder.factoryOf(new CreatedExpiryPolicy(new Duration(SECONDS, 1)))); + // Do not allow more than 1 million entries. + // Allows to run this example with smaller available memory. + cfg.setEvictionPolicy(new FifoEvictionPolicy<>(1_000_000)); + return cfg; } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da0f1933/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java ---------------------------------------------------------------------- diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java index c32d8e8..8f4c6d4 100644 --- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java +++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java @@ -49,7 +49,9 @@ public class QueryWords { // Select top 10 words. SqlFieldsQuery top10Qry = new SqlFieldsQuery( - "select _val, count(_val) as cnt from String group by _val order by cnt desc limit 10"); + "select _val, count(_val) as cnt from String group by _val order by cnt desc limit 10", + true /*collocated*/ + ); // Select average, min, and max counts among all the words. SqlFieldsQuery statsQry = new SqlFieldsQuery( @@ -64,8 +66,12 @@ public class QueryWords { // Print average count. List<?> row = stats.get(0); + int size = stmCache.size(); + if (row.get(0) != null) - System.out.printf("Query results [avg=%.2f, min=%d, max=%d]%n", row.get(0), row.get(1), row.get(2)); + + System.out.printf("Query results [total=%d, avg=%.2f, min=%d, max=%d]%n", + size, row.get(0), row.get(1), row.get(2)); // Print top 10 words. ExamplesUtils.printQueryResults(top10); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da0f1933/modules/core/src/main/java/org/apache/ignite/cache/query/SqlFieldsQuery.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/query/SqlFieldsQuery.java b/modules/core/src/main/java/org/apache/ignite/cache/query/SqlFieldsQuery.java index 63b9b6e..1ffd14f 100644 --- a/modules/core/src/main/java/org/apache/ignite/cache/query/SqlFieldsQuery.java +++ b/modules/core/src/main/java/org/apache/ignite/cache/query/SqlFieldsQuery.java @@ -24,7 +24,21 @@ import org.apache.ignite.internal.util.typedef.internal.*; import java.util.*; /** - * SQL Fields query. + * SQL Fields query. This query can return specific fields of data based + * on SQL {@code 'select'} clause, as opposed to {@link SqlQuery}, which always returns + * the whole key and value objects back. + * <h1 class="header">Collocated Flag</h1> + * Collocation flag si used for optimization purposes. Whenever Ignite executes + * a distributed query, it sends sub-queries to individual cluster members. + * If you know in advance that the elements of your query selection are collocated + * together on the same node, usually based on some <b>affinity-key</b>, Ignite + * can make significant performance and network optimizations. + * <p> + * For example, in case of Word-Count example, we know that all identical words + * are processed on the same cluster member, because we use the {@code word} itself + * as affinity key. This allows Ignite to execute the {@code 'limit'} clause on + * the remote nodes and bring back only the small data set specified within the 'limit' clause, + * instead of the whole query result as would happen in a non-collocated execution. * * @see IgniteCache#query(Query) */ @@ -39,19 +53,30 @@ public final class SqlFieldsQuery extends Query<List<?>> { @GridToStringInclude private Object[] args; - /** */ - private boolean colocated; + /** Collocation flag. */ + private boolean collocated; /** - * Constructs sql fields query. + * Constructs SQL fields query. * - * @param sql SQL Query. + * @param sql SQL query. */ public SqlFieldsQuery(String sql) { setSql(sql); } /** + * Constructs SQL fields query. + * + * @param sql SQL query. + * @param collocated Collocated flag. + */ + public SqlFieldsQuery(String sql, boolean collocated) { + this.sql = sql; + this.collocated = collocated; + } + + /** * Gets SQL clause. * * @return SQL clause. @@ -100,18 +125,18 @@ public final class SqlFieldsQuery extends Query<List<?>> { * * @return {@code true} If the query is colocated. */ - public boolean isColocated() { - return colocated; + public boolean isCollocated() { + return collocated; } /** * Sets flag defining if this query colocated. * - * @param colocated Flag value. + * @param collocated Flag value. * @return {@code this} For chaining. */ - public SqlFieldsQuery setColocated(boolean colocated) { - this.colocated = colocated; + public SqlFieldsQuery setCollocated(boolean collocated) { + this.collocated = collocated; return this; } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da0f1933/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java ---------------------------------------------------------------------- diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java index f1c10bc..9372901 100644 --- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java +++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java @@ -786,7 +786,7 @@ public class IgniteH2Indexing implements GridQueryIndexing { Collection<GridQueryFieldMetadata> meta; try { - twoStepQry = GridSqlQuerySplitter.split((JdbcPreparedStatement)stmt, qry.getArgs(), qry.isColocated()); + twoStepQry = GridSqlQuerySplitter.split((JdbcPreparedStatement)stmt, qry.getArgs(), qry.isCollocated()); meta = meta(stmt.getMetaData()); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da0f1933/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheCollocatedQuerySelfTest.java ---------------------------------------------------------------------- diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheCollocatedQuerySelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheCollocatedQuerySelfTest.java new file mode 100644 index 0000000..2c4a05c --- /dev/null +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheCollocatedQuerySelfTest.java @@ -0,0 +1,209 @@ +/* + * 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.affinity.*; +import org.apache.ignite.cache.query.*; +import org.apache.ignite.cache.query.annotations.*; +import org.apache.ignite.configuration.*; +import org.apache.ignite.internal.util.*; +import org.apache.ignite.internal.util.typedef.*; +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.junits.common.*; + +import java.io.*; +import java.util.*; +import java.util.concurrent.*; + +import static org.apache.ignite.cache.CacheAtomicityMode.*; +import static org.apache.ignite.cache.CacheMode.*; + +/** + */ +public class IgniteCacheCollocatedQuerySelfTest extends GridCommonAbstractTest { + /** */ + private static final String QRY = + "select productId, sum(price) s, count(1) c " + + "from Purchase " + + "group by productId " + + "having c > ? " + + "order by s desc, productId limit ? "; + + /** */ + private static final int PURCHASES = 1000; + + /** */ + private static final int PRODUCTS = 10; + + /** */ + private static final int MAX_PRICE = 5; + + /** */ + private static final long SEED = ThreadLocalRandom.current().nextLong(); + + /** */ + private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(gridName); + + TcpDiscoverySpi disco = new TcpDiscoverySpi(); + + disco.setIpFinder(ipFinder); + + cfg.setDiscoverySpi(disco); + + CacheConfiguration<?,?> cacheCfg = defaultCacheConfiguration(); + + cacheCfg.setCacheMode(PARTITIONED); + cacheCfg.setAtomicityMode(TRANSACTIONAL); + cacheCfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC); + cacheCfg.setSwapEnabled(false); + cacheCfg.setBackups(1); + cacheCfg.setIndexedTypes( + AffinityUuid.class, Purchase.class + ); + + cfg.setCacheConfiguration(cacheCfg); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void beforeTestsStarted() throws Exception { + startGridsMultiThreaded(3); + + X.println("--> seed: " + SEED); + } + + /** {@inheritDoc} */ + @Override protected void afterTestsStopped() throws Exception { + stopAllGrids(); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + ignite(0).cache(null).removeAll(); + } + + /** + * @param c Cache. + * @param collocated Colocated. + * @return Result. + */ + private static List<List<?>> query(IgniteCache<AffinityUuid,Purchase> c, boolean collocated) { + return c.query(new SqlFieldsQuery(QRY).setArgs(30, 5).setCollocated(collocated)).getAll(); + } + + /** + * Correct affinity. + */ + public void testColocatedQueryRight() { + IgniteCache<AffinityUuid,Purchase> c = ignite(0).cache(null); + + Random rnd = new GridRandom(SEED); + + for (int i = 0; i < PURCHASES; i++) { + Purchase p = new Purchase(); + + p.productId = rnd.nextInt(PRODUCTS); + p.price = rnd.nextInt(MAX_PRICE); + + c.put(new AffinityUuid(p.productId), p); // Correct affinity. + } + + List<List<?>> res1 = query(c, false); + List<List<?>> res2 = query(c, true); + + X.println("res1: " + res1); + X.println("res2: " + res2); + + assertFalse(res1.isEmpty()); + assertEquals(res1.toString(), res2.toString()); // TODO fix type conversion issue + } + + /** + * Correct affinity. + */ + public void testColocatedQueryWrong() { + IgniteCache<AffinityUuid,Purchase> c = ignite(0).cache(null); + + Random rnd = new GridRandom(SEED); + + for (int i = 0; i < PURCHASES; i++) { + Purchase p = new Purchase(); + + p.productId = rnd.nextInt(PRODUCTS); + p.price = rnd.nextInt(MAX_PRICE); + + c.put(new AffinityUuid(rnd.nextInt(PRODUCTS)), p); // Random affinity. + } + + List<List<?>> res1 = query(c, false); + List<List<?>> res2 = query(c, true); + + X.println("res1: " + res1); + X.println("res2: " + res2); + + assertFalse(res1.isEmpty()); + assertFalse(res1.equals(res2)); + } + + /** + * + */ + private static class Purchase implements Serializable { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + @QuerySqlField + int productId; + + /** */ + @QuerySqlField + int price; + + /** {@inheritDoc} */ + @Override public boolean equals(Object o) { + if (this == o) return true; + + if (o == null || getClass() != o.getClass()) return false; + + Purchase purchase = (Purchase)o; + + return productId == purchase.productId && price == purchase.price; + + } + + /** {@inheritDoc} */ + @Override public int hashCode() { + int result = productId; + + result = 31 * result + price; + + return result; + } + } +} + http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da0f1933/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheColocatedQuerySelfTest.java ---------------------------------------------------------------------- diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheColocatedQuerySelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheColocatedQuerySelfTest.java deleted file mode 100644 index b6051b7..0000000 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheColocatedQuerySelfTest.java +++ /dev/null @@ -1,209 +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.cache; - -import org.apache.ignite.*; -import org.apache.ignite.cache.*; -import org.apache.ignite.cache.affinity.*; -import org.apache.ignite.cache.query.*; -import org.apache.ignite.cache.query.annotations.*; -import org.apache.ignite.configuration.*; -import org.apache.ignite.internal.util.*; -import org.apache.ignite.internal.util.typedef.*; -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.junits.common.*; - -import java.io.*; -import java.util.*; -import java.util.concurrent.*; - -import static org.apache.ignite.cache.CacheAtomicityMode.*; -import static org.apache.ignite.cache.CacheMode.*; - -/** - */ -public class IgniteCacheColocatedQuerySelfTest extends GridCommonAbstractTest { - /** */ - private static final String QRY = - "select productId, sum(price) s, count(1) c " + - "from Purchase " + - "group by productId " + - "having c > ? " + - "order by s desc, productId limit ? "; - - /** */ - private static final int PURCHASES = 1000; - - /** */ - private static final int PRODUCTS = 10; - - /** */ - private static final int MAX_PRICE = 5; - - /** */ - private static final long SEED = ThreadLocalRandom.current().nextLong(); - - /** */ - private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); - - /** {@inheritDoc} */ - @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { - IgniteConfiguration cfg = super.getConfiguration(gridName); - - TcpDiscoverySpi disco = new TcpDiscoverySpi(); - - disco.setIpFinder(ipFinder); - - cfg.setDiscoverySpi(disco); - - CacheConfiguration<?,?> cacheCfg = defaultCacheConfiguration(); - - cacheCfg.setCacheMode(PARTITIONED); - cacheCfg.setAtomicityMode(TRANSACTIONAL); - cacheCfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC); - cacheCfg.setSwapEnabled(false); - cacheCfg.setBackups(1); - cacheCfg.setIndexedTypes( - AffinityUuid.class, Purchase.class - ); - - cfg.setCacheConfiguration(cacheCfg); - - return cfg; - } - - /** {@inheritDoc} */ - @Override protected void beforeTestsStarted() throws Exception { - startGridsMultiThreaded(3); - - X.println("--> seed: " + SEED); - } - - /** {@inheritDoc} */ - @Override protected void afterTestsStopped() throws Exception { - stopAllGrids(); - } - - /** {@inheritDoc} */ - @Override protected void afterTest() throws Exception { - ignite(0).cache(null).removeAll(); - } - - /** - * @param c Cache. - * @param colocated Colocated. - * @return Result. - */ - private static List<List<?>> query(IgniteCache<AffinityUuid,Purchase> c, boolean colocated) { - return c.query(new SqlFieldsQuery(QRY).setArgs(30, 5).setColocated(colocated)).getAll(); - } - - /** - * Correct affinity. - */ - public void testColocatedQueryRight() { - IgniteCache<AffinityUuid,Purchase> c = ignite(0).cache(null); - - Random rnd = new GridRandom(SEED); - - for (int i = 0; i < PURCHASES; i++) { - Purchase p = new Purchase(); - - p.productId = rnd.nextInt(PRODUCTS); - p.price = rnd.nextInt(MAX_PRICE); - - c.put(new AffinityUuid(p.productId), p); // Correct affinity. - } - - List<List<?>> res1 = query(c, false); - List<List<?>> res2 = query(c, true); - - X.println("res1: " + res1); - X.println("res2: " + res2); - - assertFalse(res1.isEmpty()); - assertEquals(res1.toString(), res2.toString()); // TODO fix type conversion issue - } - - /** - * Correct affinity. - */ - public void testColocatedQueryWrong() { - IgniteCache<AffinityUuid,Purchase> c = ignite(0).cache(null); - - Random rnd = new GridRandom(SEED); - - for (int i = 0; i < PURCHASES; i++) { - Purchase p = new Purchase(); - - p.productId = rnd.nextInt(PRODUCTS); - p.price = rnd.nextInt(MAX_PRICE); - - c.put(new AffinityUuid(rnd.nextInt(PRODUCTS)), p); // Random affinity. - } - - List<List<?>> res1 = query(c, false); - List<List<?>> res2 = query(c, true); - - X.println("res1: " + res1); - X.println("res2: " + res2); - - assertFalse(res1.isEmpty()); - assertFalse(res1.equals(res2)); - } - - /** - * - */ - private static class Purchase implements Serializable { - /** */ - private static final long serialVersionUID = 0L; - - /** */ - @QuerySqlField - int productId; - - /** */ - @QuerySqlField - int price; - - /** {@inheritDoc} */ - @Override public boolean equals(Object o) { - if (this == o) return true; - - if (o == null || getClass() != o.getClass()) return false; - - Purchase purchase = (Purchase)o; - - return productId == purchase.productId && price == purchase.price; - - } - - /** {@inheritDoc} */ - @Override public int hashCode() { - int result = productId; - - result = 31 * result + price; - - return result; - } - } -} - http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da0f1933/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java ---------------------------------------------------------------------- diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java index 08f6ed2..fe70c12 100644 --- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java +++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java @@ -55,7 +55,7 @@ public class IgniteCacheQuerySelfTestSuite extends TestSuite { suite.addTestSuite(IgniteCachePartitionedQueryP2PDisabledSelfTest.class); suite.addTestSuite(IgniteCachePartitionedQueryMultiThreadedSelfTest.class); suite.addTestSuite(IgniteCacheQueryIndexSelfTest.class); - suite.addTestSuite(IgniteCacheColocatedQuerySelfTest.class); + suite.addTestSuite(IgniteCacheCollocatedQuerySelfTest.class); suite.addTestSuite(IgniteCacheLargeResultSelfTest.class); suite.addTestSuite(GridCacheQueryInternalKeysSelfTest.class); suite.addTestSuite(IgniteCacheQueryMultiThreadedSelfTest.class);
