ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [2/5] incubator-ignite git commit: ignite-484 - fixes + tests enabled
Date Tue, 19 May 2015 21:51:32 GMT
ignite-484 - fixes + tests enabled


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/5e877cc7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/5e877cc7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/5e877cc7

Branch: refs/heads/ignite-484
Commit: 5e877cc75d75c63240296e1e17ba517805a30827
Parents: f2b96e0
Author: S.Vladykin <svladykin@gridgain.com>
Authored: Tue May 19 22:43:43 2015 +0300
Committer: S.Vladykin <svladykin@gridgain.com>
Committed: Tue May 19 22:43:43 2015 +0300

----------------------------------------------------------------------
 .../h2/twostep/GridMergeIndexUnsorted.java      |   2 +-
 .../h2/twostep/GridReduceQueryExecutor.java     | 122 ++++++++++++++-----
 .../query/h2/twostep/GridResultPage.java        |   2 +-
 .../IgniteCacheQueryNodeRestartSelfTest.java    |  21 +++-
 .../IgniteCacheQuerySelfTestSuite.java          |   2 +-
 5 files changed, 111 insertions(+), 38 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5e877cc7/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndexUnsorted.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndexUnsorted.java
b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndexUnsorted.java
index 76a52e9..fdee17a 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndexUnsorted.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndexUnsorted.java
@@ -60,7 +60,7 @@ public class GridMergeIndexUnsorted extends GridMergeIndex {
             Iterator<Value[]> iter = Collections.emptyIterator();
 
             @Override public boolean hasNext() {
-                while (!iter.hasNext()){
+                while (!iter.hasNext()) {
                     GridResultPage page;
 
                     try {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5e877cc7/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
index 76de71b..bb6801c 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
@@ -145,7 +145,8 @@ public class GridReduceQueryExecutor {
                 for (QueryRun r : runs.values()) {
                     for (GridMergeTable tbl : r.tbls) {
                         if (tbl.getScanIndex(null).hasSource(nodeId)) {
-                            fail(r, nodeId, "Node left the topology.");
+                            // Will attempt to retry. If reduce query was started it will
fail on next page fetching.
+                            retry(r, topologyVersion(), nodeId);
 
                             break;
                         }
@@ -201,15 +202,8 @@ public class GridReduceQueryExecutor {
      * @param msg Error message.
      */
     private void fail(QueryRun r, UUID nodeId, String msg) {
-        if (r != null) {
-            r.rmtErr = new CacheException("Failed to execute map query on the node: " + nodeId
+ ", " + msg);
-
-            while(r.latch.getCount() != 0)
-                r.latch.countDown();
-
-            for (GridMergeTable tbl : r.tbls)
-                tbl.getScanIndex(null).fail(nodeId);
-        }
+        if (r != null)
+            r.state(new CacheException("Failed to execute map query on the node: " + nodeId
+ ", " + msg), nodeId);
     }
 
     /**
@@ -234,8 +228,16 @@ public class GridReduceQueryExecutor {
         try {
             page = new GridResultPage(ctx, node.id(), msg, false) {
                 @Override public void fetchNextPage() {
-                    if (r.rmtErr != null)
-                        throw new CacheException("Next page fetch failed.", r.rmtErr);
+                    Object errState = r.state.get();
+
+                    if (errState != null) {
+                        CacheException e = new CacheException("Failed to fetch data from
node: " + node.id());
+
+                        if (errState instanceof CacheException)
+                            e.addSuppressed((Throwable)errState);
+
+                        throw e;
+                    }
 
                     try {
                         GridQueryNextPageRequest msg0 = new GridQueryNextPageRequest(qryReqId,
qry, pageSize);
@@ -261,17 +263,29 @@ public class GridReduceQueryExecutor {
 
         idx.addPage(page);
 
-        if (msg.retry() != null) {
-            r.retry = msg.retry();
-
-            while (r.latch.getCount() != 0)
-                r.latch.countDown();
-        }
+        if (msg.retry() != null)
+            retry(r, msg.retry(), node.id());
         else if (msg.allRows() != -1) // Only the first page contains row count.
             r.latch.countDown();
     }
 
     /**
+     * @param r Query run.
+     * @param retryVer Retry version.
+     * @param nodeId Node ID.
+     */
+    private void retry(QueryRun r, AffinityTopologyVersion retryVer, UUID nodeId) {
+        r.state(retryVer, nodeId);
+    }
+
+    /**
+     * @return Current topology version.
+     */
+    private AffinityTopologyVersion topologyVersion() {
+        return ctx.discovery().topologyVersionEx();
+    }
+
+    /**
      * @param cctx Cache context.
      * @param qry Query.
      * @return Cursor.
@@ -290,10 +304,13 @@ public class GridReduceQueryExecutor {
 
             r.conn = (JdbcConnection)h2.connectionForSpace(space);
 
-            AffinityTopologyVersion topVer = ctx.discovery().topologyVersionEx();
+            AffinityTopologyVersion topVer = topologyVersion();
 
             Collection<ClusterNode> nodes = ctx.discovery().cacheAffinityNodes(space,
topVer);
 
+            if (F.isEmpty(nodes))
+                throw new CacheException("No data nodes found for cache: " + space);
+
             if (cctx.isReplicated() || qry.explain()) {
                 assert qry.explain() || !nodes.contains(ctx.cluster().get().localNode())
: "We must be on a client node.";
 
@@ -342,17 +359,37 @@ public class GridReduceQueryExecutor {
                         mapQry.marshallParams(m);
                 }
 
-                send(nodes, new GridQueryRequest(qryReqId, r.pageSize, space, mapQrys, topVer,
-                    extraSpaces(space, qry.spaces())));
+                boolean ok = false;
+
+                try {
+                    send(nodes, new GridQueryRequest(qryReqId, r.pageSize, space, mapQrys,
topVer,
+                        extraSpaces(space, qry.spaces())));
+
+                    ok = true;
+                }
+                catch (IgniteCheckedException e) {
+                    U.warn(log, "Failed to send query request to nodes: " + nodes);
+                }
+
+                AffinityTopologyVersion retry = null;
 
-                U.await(r.latch);
+                if (ok) { // Sent successfully.
+                    U.await(r.latch);
 
-                if (r.rmtErr != null)
-                    throw new CacheException("Failed to run map query remotely.", r.rmtErr);
+                    Object state = r.state.get();
 
-                ResultSet res = null;
+                    if (state != null) {
+                        if (state instanceof CacheException)
+                            throw new CacheException("Failed to run map query remotely.",
(CacheException)state);
 
-                AffinityTopologyVersion retry = r.retry;
+                        if (state instanceof AffinityTopologyVersion)
+                            retry = (AffinityTopologyVersion)state; // Remote nodes can ask
us to retry.
+                    }
+                }
+                else  // Send failed -> retry.
+                    retry = topologyVersion();
+
+                ResultSet res = null;
 
                 if (retry == null) {
                     if (qry.explain())
@@ -364,8 +401,14 @@ public class GridReduceQueryExecutor {
                 }
 
                 for (GridMergeTable tbl : r.tbls) {
-                    if (!tbl.getScanIndex(null).fetchedAll()) // We have to explicitly cancel
queries on remote nodes.
-                        send(nodes, new GridQueryCancelRequest(qryReqId));
+                    if (!tbl.getScanIndex(null).fetchedAll()) { // We have to explicitly
cancel queries on remote nodes.
+                        try {
+                            send(nodes, new GridQueryCancelRequest(qryReqId));
+                        }
+                        catch (IgniteCheckedException e) {
+                            U.warn(log, "Failed to send cancel request to nodes: " + nodes);
+                        }
+                    }
 
 //                dropTable(r.conn, tbl.getName()); TODO
                 }
@@ -697,11 +740,26 @@ public class GridReduceQueryExecutor {
         /** */
         private int pageSize;
 
-        /** */
-        private volatile CacheException rmtErr;
+        /** Can be either CacheException in case of error or AffinityTopologyVersion to retry
if needed. */
+        private final AtomicReference<Object> state = new AtomicReference<>();
 
-        /** */
-        private volatile AffinityTopologyVersion retry;
+        /**
+         * @param o Fail state object.
+         * @param nodeId Node ID.
+         */
+        void state(Object o, UUID nodeId) {
+            assert o != null;
+            assert o instanceof CacheException || o instanceof AffinityTopologyVersion :
o.getClass();
+
+            if (!state.compareAndSet(null, o))
+                return;
+
+            while (latch.getCount() != 0) // We don't need to wait for all nodes to reply.
+                latch.countDown();
+
+            for (GridMergeTable tbl : tbls) // Fail all merge indexes.
+                tbl.getScanIndex(null).fail(nodeId);
+        }
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5e877cc7/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridResultPage.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridResultPage.java
b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridResultPage.java
index 9392fd1..35bfab9 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridResultPage.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridResultPage.java
@@ -161,7 +161,7 @@ public class GridResultPage {
      * Request next page.
      */
     public void fetchNextPage() {
-        throw new UnsupportedOperationException();
+        throw new CacheException("Failed to fetch data from node: " + src);
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5e877cc7/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryNodeRestartSelfTest.java
----------------------------------------------------------------------
diff --git 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
index 5dce126..4edef55 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
@@ -78,6 +78,7 @@ public class IgniteCacheQueryNodeRestartSelfTest extends GridCacheAbstractSelfTe
         cc.setBackups(1);
         cc.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
         cc.setAtomicityMode(TRANSACTIONAL);
+        cc.setRebalanceMode(CacheRebalanceMode.SYNC);
         cc.setIndexedTypes(
             Integer.class, Integer.class
         );
@@ -106,7 +107,7 @@ public class IgniteCacheQueryNodeRestartSelfTest extends GridCacheAbstractSelfTe
         for (int i = 0; i < KEY_CNT; i++)
             cache.put(i, i);
 
-        assertEquals(KEY_CNT, cache.localSize());
+        assertEquals(KEY_CNT, cache.size());
 
         final AtomicInteger qryCnt = new AtomicInteger();
 
@@ -116,9 +117,23 @@ public class IgniteCacheQueryNodeRestartSelfTest extends GridCacheAbstractSelfTe
             @Override public void applyx() throws IgniteCheckedException {
                 while (!done.get()) {
                     Collection<Cache.Entry<Integer, Integer>> res =
-                        cache.query(new SqlQuery(Integer.class, "_val >= 0")).getAll();
+                        cache.query(new SqlQuery<Integer, Integer>(Integer.class, "true")).getAll();
 
-                    assertFalse(res.isEmpty());
+                    Set<Integer> keys = new HashSet<>();
+
+                    for (Cache.Entry<Integer,Integer> entry : res)
+                        keys.add(entry.getKey());
+
+                    if (KEY_CNT > keys.size()) {
+                        for (int i = 0; i < KEY_CNT; i++) {
+                            if (!keys.contains(i))
+                                assertEquals(Integer.valueOf(i), cache.get(i));
+                        }
+
+                        fail("res size: " + res.size());
+                    }
+
+                    assertEquals(KEY_CNT, keys.size());
 
                     int c = qryCnt.incrementAndGet();
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5e877cc7/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 ce05980..915b118 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
@@ -64,7 +64,7 @@ public class IgniteCacheQuerySelfTestSuite extends TestSuite {
         suite.addTestSuite(IgniteCacheQueryOffheapEvictsMultiThreadedSelfTest.class);
         suite.addTestSuite(IgniteCacheSqlQueryMultiThreadedSelfTest.class);
         suite.addTestSuite(IgniteCacheOffheapTieredMultithreadedSelfTest.class);
-//        suite.addTestSuite(IgniteCacheQueryNodeRestartSelfTest.class); TODO IGNITE-484
+        suite.addTestSuite(IgniteCacheQueryNodeRestartSelfTest.class);
         suite.addTestSuite(GridCacheReduceQueryMultithreadedSelfTest.class);
         suite.addTestSuite(GridCacheCrossCacheQuerySelfTest.class);
         suite.addTestSuite(GridCacheQuerySerializationSelfTest.class);


Mime
View raw message