ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [11/50] [abbrv] incubator-ignite git commit: ignite-sql-tests - direct marshallable
Date Wed, 18 Mar 2015 01:16:48 GMT
ignite-sql-tests - direct marshallable


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

Branch: refs/heads/sprint-2
Commit: cef213040fb2201e25fefbeedb5bf2ea8e18df78
Parents: c459f30
Author: S.Vladykin <svladykin@gridgain.com>
Authored: Thu Mar 12 04:44:05 2015 +0300
Committer: S.Vladykin <svladykin@gridgain.com>
Committed: Thu Mar 12 04:44:05 2015 +0300

----------------------------------------------------------------------
 .../ignite/codegen/MessageCodeGenerator.java    |   8 ++
 .../communication/GridIoMessageFactory.java     |  26 ++++
 .../messages/GridQueryCancelRequest.java        |  66 +++++++++-
 .../twostep/messages/GridQueryFailResponse.java |  80 +++++++++++-
 .../messages/GridQueryNextPageRequest.java      |  94 +++++++++++++-
 .../messages/GridQueryNextPageResponse.java     | 114 ++++++++++++++++-
 .../h2/twostep/messages/GridQueryRequest.java   | 126 ++++++++++++++++++-
 .../processors/query/h2/IgniteH2Indexing.java   |  14 +++
 .../query/h2/twostep/GridMapQueryExecutor.java  |  81 +++++++-----
 .../h2/twostep/GridReduceQueryExecutor.java     |  92 +++++++++-----
 10 files changed, 626 insertions(+), 75 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cef21304/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
----------------------------------------------------------------------
diff --git a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
index 7962a4b..76fed7e 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
@@ -18,6 +18,7 @@
 package org.apache.ignite.codegen;
 
 import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.processors.query.h2.twostep.messages.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.lang.*;
 import org.apache.ignite.plugin.extensions.communication.*;
@@ -164,6 +165,13 @@ public class MessageCodeGenerator {
 //
 //        gen.generateAndWrite(GridCacheOptimisticCheckPreparedTxRequest.class);
 //        gen.generateAndWrite(GridCacheOptimisticCheckPreparedTxResponse.class);
+
+//        gen.generateAndWrite(GridQueryCancelRequest.class);
+//        gen.generateAndWrite(GridQueryFailResponse.class);
+//        gen.generateAndWrite(GridQueryNextPageRequest.class);
+//        gen.generateAndWrite(GridQueryNextPageResponse.class);
+//        gen.generateAndWrite(GridQueryRequest.class);
+//        gen.generateAndWrite(GridCacheSqlQuery.class);
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cef21304/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index 6109d74..4eb0c1d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -34,6 +34,7 @@ import org.apache.ignite.internal.processors.clock.*;
 import org.apache.ignite.internal.processors.continuous.*;
 import org.apache.ignite.internal.processors.dataload.*;
 import org.apache.ignite.internal.processors.igfs.*;
+import org.apache.ignite.internal.processors.query.h2.twostep.messages.*;
 import org.apache.ignite.internal.processors.rest.handlers.task.*;
 import org.apache.ignite.internal.processors.streamer.*;
 import org.apache.ignite.internal.util.*;
@@ -492,6 +493,31 @@ public class GridIoMessageFactory implements MessageFactory {
 
                 break;
 
+            case 90:
+                msg = new GridQueryCancelRequest();
+
+                break;
+
+            case 91:
+                msg = new GridQueryFailResponse();
+
+                break;
+
+            case 92:
+                msg = new GridQueryNextPageRequest();
+
+                break;
+
+            case 93:
+                msg = new GridQueryNextPageResponse();
+
+                break;
+
+            case 94:
+                msg = new GridQueryRequest();
+
+                break;
+
             default:
                 if (ext != null) {
                     for (MessageFactory factory : ext) {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cef21304/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
index 9bcb410..85371b8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
@@ -18,17 +18,25 @@
 package org.apache.ignite.internal.processors.query.h2.twostep.messages;
 
 import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.plugin.extensions.communication.*;
 
-import java.io.*;
+import java.nio.*;
 
 /**
  * Cancel request.
  */
-public class GridQueryCancelRequest implements Serializable {
+public class GridQueryCancelRequest implements Message {
     /** */
     private long qryReqId;
 
     /**
+     * Default constructor.
+     */
+    public GridQueryCancelRequest() {
+        // No-op.
+    }
+
+    /**
      * @param qryReqId Query request ID.
      */
     public GridQueryCancelRequest(long qryReqId) {
@@ -46,4 +54,58 @@ public class GridQueryCancelRequest implements Serializable {
     @Override public String toString() {
         return S.toString(GridQueryCancelRequest.class, this);
     }
+
+    /** {@inheritDoc} */
+    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+        writer.setBuffer(buf);
+
+        if (!writer.isHeaderWritten()) {
+            if (!writer.writeHeader(directType(), fieldsCount()))
+                return false;
+
+            writer.onHeaderWritten();
+        }
+
+        switch (writer.state()) {
+            case 0:
+                if (!writer.writeLong("qryReqId", qryReqId))
+                    return false;
+
+                writer.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+        reader.setBuffer(buf);
+
+        if (!reader.beforeMessageRead())
+            return false;
+
+        switch (reader.state()) {
+            case 0:
+                qryReqId = reader.readLong("qryReqId");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 90;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte fieldsCount() {
+        return 1;
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cef21304/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryFailResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryFailResponse.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryFailResponse.java
index f551ab8..7a2f22e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryFailResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryFailResponse.java
@@ -18,13 +18,14 @@
 package org.apache.ignite.internal.processors.query.h2.twostep.messages;
 
 import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.plugin.extensions.communication.*;
 
-import java.io.*;
+import java.nio.*;
 
 /**
  * Error message.
  */
-public class GridQueryFailResponse implements Serializable {
+public class GridQueryFailResponse implements Message {
     /** */
     private static final long serialVersionUID = 0L;
 
@@ -35,6 +36,13 @@ public class GridQueryFailResponse implements Serializable {
     private String errMsg;
 
     /**
+     * Default constructor.
+     */
+    public GridQueryFailResponse() {
+        // No-op.
+    }
+
+    /**
      * @param qryReqId Query request ID.
      * @param err Error.
      */
@@ -61,4 +69,72 @@ public class GridQueryFailResponse implements Serializable {
     @Override public String toString() {
         return S.toString(GridQueryFailResponse.class, this);
     }
+
+    /** {@inheritDoc} */
+    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+        writer.setBuffer(buf);
+
+        if (!writer.isHeaderWritten()) {
+            if (!writer.writeHeader(directType(), fieldsCount()))
+                return false;
+
+            writer.onHeaderWritten();
+        }
+
+        switch (writer.state()) {
+            case 0:
+                if (!writer.writeString("errMsg", errMsg))
+                    return false;
+
+                writer.incrementState();
+
+            case 1:
+                if (!writer.writeLong("qryReqId", qryReqId))
+                    return false;
+
+                writer.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+        reader.setBuffer(buf);
+
+        if (!reader.beforeMessageRead())
+            return false;
+
+        switch (reader.state()) {
+            case 0:
+                errMsg = reader.readString("errMsg");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 1:
+                qryReqId = reader.readLong("qryReqId");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 91;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte fieldsCount() {
+        return 2;
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cef21304/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryNextPageRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryNextPageRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryNextPageRequest.java
index ae7e1e3..2e49683 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryNextPageRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryNextPageRequest.java
@@ -19,13 +19,14 @@ package org.apache.ignite.internal.processors.query.h2.twostep.messages;
 
 
 import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.plugin.extensions.communication.*;
 
-import java.io.*;
+import java.nio.*;
 
 /**
  * Request to fetch next page.
  */
-public class GridQueryNextPageRequest implements Serializable {
+public class GridQueryNextPageRequest implements Message {
     /** */
     private static final long serialVersionUID = 0L;
 
@@ -39,6 +40,13 @@ public class GridQueryNextPageRequest implements Serializable {
     private int pageSize;
 
     /**
+     * Default constructor.
+     */
+    public GridQueryNextPageRequest() {
+        // No-op.
+    }
+
+    /**
      * @param qryReqId Query request ID.
      * @param qry Query.
      * @param pageSize Page size.
@@ -74,4 +82,86 @@ public class GridQueryNextPageRequest implements Serializable {
     @Override public String toString() {
         return S.toString(GridQueryNextPageRequest.class, this);
     }
+
+    /** {@inheritDoc} */
+    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+        writer.setBuffer(buf);
+
+        if (!writer.isHeaderWritten()) {
+            if (!writer.writeHeader(directType(), fieldsCount()))
+                return false;
+
+            writer.onHeaderWritten();
+        }
+
+        switch (writer.state()) {
+            case 0:
+                if (!writer.writeInt("pageSize", pageSize))
+                    return false;
+
+                writer.incrementState();
+
+            case 1:
+                if (!writer.writeInt("qry", qry))
+                    return false;
+
+                writer.incrementState();
+
+            case 2:
+                if (!writer.writeLong("qryReqId", qryReqId))
+                    return false;
+
+                writer.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+        reader.setBuffer(buf);
+
+        if (!reader.beforeMessageRead())
+            return false;
+
+        switch (reader.state()) {
+            case 0:
+                pageSize = reader.readInt("pageSize");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 1:
+                qry = reader.readInt("qry");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 2:
+                qryReqId = reader.readLong("qryReqId");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 92;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte fieldsCount() {
+        return 3;
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cef21304/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryNextPageResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryNextPageResponse.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryNextPageResponse.java
index 3c4cc94..b2cdea6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryNextPageResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryNextPageResponse.java
@@ -18,13 +18,15 @@
 package org.apache.ignite.internal.processors.query.h2.twostep.messages;
 
 import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.plugin.extensions.communication.*;
 
 import java.io.*;
+import java.nio.*;
 
 /**
  * Next page response.
  */
-public class GridQueryNextPageResponse implements Externalizable {
+public class GridQueryNextPageResponse implements Externalizable, Message {
     /** */
     private static final long serialVersionUID = 0L;
 
@@ -125,4 +127,114 @@ public class GridQueryNextPageResponse implements Externalizable {
     @Override public String toString() {
         return S.toString(GridQueryNextPageResponse.class, this);
     }
+
+    /** {@inheritDoc} */
+    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+        writer.setBuffer(buf);
+
+        if (!writer.isHeaderWritten()) {
+            if (!writer.writeHeader(directType(), fieldsCount()))
+                return false;
+
+            writer.onHeaderWritten();
+        }
+
+        switch (writer.state()) {
+            case 0:
+                if (!writer.writeInt("allRows", allRows))
+                    return false;
+
+                writer.incrementState();
+
+            case 1:
+                if (!writer.writeInt("page", page))
+                    return false;
+
+                writer.incrementState();
+
+            case 2:
+                if (!writer.writeInt("qry", qry))
+                    return false;
+
+                writer.incrementState();
+
+            case 3:
+                if (!writer.writeLong("qryReqId", qryReqId))
+                    return false;
+
+                writer.incrementState();
+
+            case 4:
+                if (!writer.writeByteArray("rows", rows))
+                    return false;
+
+                writer.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+        reader.setBuffer(buf);
+
+        if (!reader.beforeMessageRead())
+            return false;
+
+        switch (reader.state()) {
+            case 0:
+                allRows = reader.readInt("allRows");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 1:
+                page = reader.readInt("page");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 2:
+                qry = reader.readInt("qry");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 3:
+                qryReqId = reader.readLong("qryReqId");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 4:
+                rows = reader.readByteArray("rows");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 93;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte fieldsCount() {
+        return 5;
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cef21304/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryRequest.java
index 2834b84..bc929dd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryRequest.java
@@ -17,17 +17,21 @@
 
 package org.apache.ignite.internal.processors.query.h2.twostep.messages;
 
+import org.apache.ignite.*;
+import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.processors.cache.query.*;
 import org.apache.ignite.internal.util.tostring.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.marshaller.*;
+import org.apache.ignite.plugin.extensions.communication.*;
 
-import java.io.*;
+import java.nio.*;
 import java.util.*;
 
 /**
  * Query request.
  */
-public class GridQueryRequest implements Serializable {
+public class GridQueryRequest implements Message {
     /** */
     private static final long serialVersionUID = 0L;
 
@@ -42,22 +46,35 @@ public class GridQueryRequest implements Serializable {
 
     /** */
     @GridToStringInclude
+    @GridDirectTransient
     private Collection<GridCacheSqlQuery> qrys;
 
+    /** */
+    private byte[] qrysBytes;
+
+    /**
+     * Default constructor.
+     */
+    public GridQueryRequest() {
+        // No-op.
+    }
+
     /**
      * @param reqId Request ID.
      * @param pageSize Page size.
      * @param space Space.
      * @param qrys Queries.
+     * @param qrysBytes Marshalled queries.
      */
-    public GridQueryRequest(long reqId, int pageSize, String space, Collection<GridCacheSqlQuery>
qrys) {
+    public GridQueryRequest(long reqId, int pageSize, String space, Collection<GridCacheSqlQuery>
qrys, byte[] qrysBytes) {
         this.reqId = reqId;
         this.pageSize = pageSize;
         this.space = space;
 
-        assert qrys instanceof Serializable;
+        assert qrysBytes != null;
 
         this.qrys = qrys;
+        this.qrysBytes = qrysBytes;
     }
 
     /**
@@ -84,7 +101,10 @@ public class GridQueryRequest implements Serializable {
     /**
      * @return Queries.
      */
-    public Collection<GridCacheSqlQuery> queries() {
+    public Collection<GridCacheSqlQuery> queries(Marshaller m) throws IgniteCheckedException
{
+        if (qrys == null && qrysBytes != null)
+            qrys = m.unmarshal(qrysBytes, null);
+
         return qrys;
     }
 
@@ -92,4 +112,100 @@ public class GridQueryRequest implements Serializable {
     @Override public String toString() {
         return S.toString(GridQueryRequest.class, this);
     }
+
+    /** {@inheritDoc} */
+    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+        writer.setBuffer(buf);
+
+        if (!writer.isHeaderWritten()) {
+            if (!writer.writeHeader(directType(), fieldsCount()))
+                return false;
+
+            writer.onHeaderWritten();
+        }
+
+        switch (writer.state()) {
+            case 0:
+                if (!writer.writeInt("pageSize", pageSize))
+                    return false;
+
+                writer.incrementState();
+
+            case 1:
+                if (!writer.writeByteArray("qrysBytes", qrysBytes))
+                    return false;
+
+                writer.incrementState();
+
+            case 2:
+                if (!writer.writeLong("reqId", reqId))
+                    return false;
+
+                writer.incrementState();
+
+            case 3:
+                if (!writer.writeString("space", space))
+                    return false;
+
+                writer.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+        reader.setBuffer(buf);
+
+        if (!reader.beforeMessageRead())
+            return false;
+
+        switch (reader.state()) {
+            case 0:
+                pageSize = reader.readInt("pageSize");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 1:
+                qrysBytes = reader.readByteArray("qrysBytes");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 2:
+                reqId = reader.readLong("reqId");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 3:
+                space = reader.readString("space");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 94;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte fieldsCount() {
+        return 4;
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cef21304/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 5fd837e..b210f4d 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
@@ -1017,6 +1017,20 @@ public class IgniteH2Indexing implements GridQueryIndexing {
         return ((Number)iter.next().get(0)).longValue();
     }
 
+    /**
+     * @return Map query executor.
+     */
+    public GridMapQueryExecutor mapQueryExecutor() {
+        return mapQryExec;
+    }
+
+    /**
+     * @return Reduce query executor.
+     */
+    public GridReduceQueryExecutor reduceQueryExecutor() {
+        return rdcQryExec;
+    }
+
     /** {@inheritDoc} */
     @SuppressWarnings("NonThreadSafeLazyInitialization")
     @Override public void start(GridKernalContext ctx) throws IgniteCheckedException {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cef21304/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
index 5b2a0ab..adade06 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
@@ -21,12 +21,12 @@ import org.apache.ignite.*;
 import org.apache.ignite.cluster.*;
 import org.apache.ignite.events.*;
 import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.managers.communication.*;
 import org.apache.ignite.internal.processors.cache.query.*;
 import org.apache.ignite.internal.processors.query.h2.*;
 import org.apache.ignite.internal.processors.query.h2.twostep.messages.*;
 import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.lang.*;
 import org.h2.jdbc.*;
 import org.h2.result.*;
 import org.h2.store.*;
@@ -45,7 +45,7 @@ import static org.apache.ignite.events.EventType.*;
 /**
  * Map query executor.
  */
-public class GridMapQueryExecutor {
+public class GridMapQueryExecutor implements GridMessageListener {
     /** */
     private static final Field RESULT_FIELD;
 
@@ -88,34 +88,33 @@ public class GridMapQueryExecutor {
 
         // TODO handle node failures.
 
-        ctx.io().addUserMessageListener(GridTopic.TOPIC_QUERY, new IgniteBiPredicate<UUID,
Object>() {
-            @Override public boolean apply(UUID nodeId, Object msg) {
-                try {
-                    assert msg != null;
+        ctx.io().addMessageListener(GridTopic.TOPIC_QUERY, this);
+    }
 
-                    ClusterNode node = ctx.discovery().node(nodeId);
+    /** {@inheritDoc} */
+    @Override public void onMessage(UUID nodeId, Object msg) {
+        try {
+            assert msg != null;
 
-                    boolean processed = true;
+            ClusterNode node = ctx.discovery().node(nodeId);
 
-                    if (msg instanceof GridQueryRequest)
-                        onQueryRequest(node, (GridQueryRequest)msg);
-                    else if (msg instanceof GridQueryNextPageRequest)
-                        onNextPageRequest(node, (GridQueryNextPageRequest)msg);
-                    else if (msg instanceof GridQueryCancelRequest)
-                        onCancel(node, (GridQueryCancelRequest)msg);
-                    else
-                        processed = false;
+            boolean processed = true;
 
-                    if (processed && log.isDebugEnabled())
-                        log.debug("Processed request: " + nodeId + "->" + ctx.localNodeId()
+ " " + msg);
-                }
-                catch(Throwable th) {
-                    U.error(log, "Failed to process message: " + msg, th);
-                }
+            if (msg instanceof GridQueryRequest)
+                onQueryRequest(node, (GridQueryRequest)msg);
+            else if (msg instanceof GridQueryNextPageRequest)
+                onNextPageRequest(node, (GridQueryNextPageRequest)msg);
+            else if (msg instanceof GridQueryCancelRequest)
+                onCancel(node, (GridQueryCancelRequest)msg);
+            else
+                processed = false;
 
-                return true;
-            }
-        });
+            if (processed && log.isDebugEnabled())
+                log.debug("Processed request: " + nodeId + "->" + ctx.localNodeId() +
" " + msg);
+        }
+        catch(Throwable th) {
+            U.error(log, "Failed to process message: " + msg, th);
+        }
     }
 
     /**
@@ -161,7 +160,16 @@ public class GridMapQueryExecutor {
     private void onQueryRequest(ClusterNode node, GridQueryRequest req) {
         ConcurrentMap<Long,QueryResults> nodeRess = resultsForNode(node.id());
 
-        QueryResults qr = new QueryResults(req.requestId(), req.queries().size());
+        Collection<GridCacheSqlQuery> qrys;
+
+        try {
+            qrys = req.queries(ctx.config().getMarshaller());
+        }
+        catch (IgniteCheckedException e) {
+            throw new IgniteException(e);
+        }
+
+        QueryResults qr = new QueryResults(req.requestId(), qrys.size());
 
         if (nodeRess.putIfAbsent(req.requestId(), qr) != null)
             throw new IllegalStateException();
@@ -176,7 +184,7 @@ public class GridMapQueryExecutor {
 
             String space = req.space();
 
-            for (GridCacheSqlQuery qry : req.queries()) {
+            for (GridCacheSqlQuery qry : qrys) {
                 ResultSet rs = h2.executeSqlQueryWithTimer(space, h2.connectionForSpace(space),
qry.query(),
                     F.asList(qry.parameters()));
 
@@ -233,8 +241,12 @@ public class GridMapQueryExecutor {
      */
     private void sendError(ClusterNode node, long qryReqId, Throwable err) {
         try {
-            ctx.io().sendUserMessage(F.asList(node), new GridQueryFailResponse(qryReqId,
err),
-                GridTopic.TOPIC_QUERY, false, 0);
+            GridQueryFailResponse msg = new GridQueryFailResponse(qryReqId, err);
+
+            if (node.isLocal())
+                h2.reduceQueryExecutor().onMessage(ctx.localNodeId(), msg);
+            else
+                ctx.io().send(node, GridTopic.TOPIC_QUERY, msg, GridIoPolicy.PUBLIC_POOL);
         }
         catch (Exception e) {
             e.addSuppressed(err);
@@ -284,10 +296,13 @@ public class GridMapQueryExecutor {
         }
 
         try {
-            ctx.io().sendUserMessage(F.asList(node),
-                new GridQueryNextPageResponse(qr.qryReqId, qry, page, page == 0 ? res.rowCount
: -1,
-                    marshallRows(rows)),
-                GridTopic.TOPIC_QUERY, false, 0);
+            GridQueryNextPageResponse msg = new GridQueryNextPageResponse(qr.qryReqId, qry,
page,
+                page == 0 ? res.rowCount : -1, marshallRows(rows));
+
+            if (node.isLocal())
+                h2.reduceQueryExecutor().onMessage(ctx.localNodeId(), msg);
+            else
+                ctx.io().send(node, GridTopic.TOPIC_QUERY, msg, GridIoPolicy.PUBLIC_POOL);
         }
         catch (IgniteCheckedException e) {
             log.error("Failed to send message.", e);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cef21304/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 4dc9e59..642a5ba 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
@@ -21,6 +21,7 @@ import org.apache.ignite.*;
 import org.apache.ignite.cache.query.*;
 import org.apache.ignite.cluster.*;
 import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.managers.communication.*;
 import org.apache.ignite.internal.processors.cache.*;
 import org.apache.ignite.internal.processors.cache.query.*;
 import org.apache.ignite.internal.processors.query.h2.*;
@@ -28,7 +29,7 @@ import org.apache.ignite.internal.processors.query.h2.sql.*;
 import org.apache.ignite.internal.processors.query.h2.twostep.messages.*;
 import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.lang.*;
+import org.apache.ignite.plugin.extensions.communication.*;
 import org.h2.command.ddl.*;
 import org.h2.command.dml.Query;
 import org.h2.engine.*;
@@ -51,7 +52,7 @@ import java.util.concurrent.atomic.*;
 /**
  * Reduce query executor.
  */
-public class GridReduceQueryExecutor {
+public class GridReduceQueryExecutor implements GridMessageListener {
     /** */
     private GridKernalContext ctx;
 
@@ -108,35 +109,34 @@ public class GridReduceQueryExecutor {
 
         // TODO handle node failure.
 
-        ctx.io().addUserMessageListener(GridTopic.TOPIC_QUERY, new IgniteBiPredicate<UUID,
Object>() {
-            @Override public boolean apply(UUID nodeId, Object msg) {
-                try {
-                    assert msg != null;
+        ctx.io().addMessageListener(GridTopic.TOPIC_QUERY, this);
 
-                    ClusterNode node = ctx.discovery().node(nodeId);
+        h2.executeStatement("PUBLIC", "CREATE ALIAS " + GridSqlQuerySplitter.TABLE_FUNC_NAME
+
+            " FOR \"" + GridReduceQueryExecutor.class.getName() + ".mergeTableFunction\"");
+    }
 
-                    boolean processed = true;
+    /** {@inheritDoc} */
+    @Override public void onMessage(UUID nodeId, Object msg) {
+        try {
+            assert msg != null;
 
-                    if (msg instanceof GridQueryNextPageResponse)
-                        onNextPage(node, (GridQueryNextPageResponse)msg);
-                    else if (msg instanceof GridQueryFailResponse)
-                        onFail(node, (GridQueryFailResponse)msg);
-                    else
-                        processed = false;
+            ClusterNode node = ctx.discovery().node(nodeId);
 
-                    if (processed && log.isDebugEnabled())
-                        log.debug("Processed response: " + nodeId + "->" + ctx.localNodeId()
+ " " + msg);
-                }
-                catch(Throwable th) {
-                    U.error(log, "Failed to process message: " + msg, th);
-                }
+            boolean processed = true;
 
-                return true;
-            }
-        });
+            if (msg instanceof GridQueryNextPageResponse)
+                onNextPage(node, (GridQueryNextPageResponse)msg);
+            else if (msg instanceof GridQueryFailResponse)
+                onFail(node, (GridQueryFailResponse)msg);
+            else
+                processed = false;
 
-        h2.executeStatement("PUBLIC", "CREATE ALIAS " + GridSqlQuerySplitter.TABLE_FUNC_NAME
+
-            " FOR \"" + GridReduceQueryExecutor.class.getName() + ".mergeTableFunction\"");
+            if (processed && log.isDebugEnabled())
+                log.debug("Processed response: " + nodeId + "->" + ctx.localNodeId() +
" " + msg);
+        }
+        catch(Throwable th) {
+            U.error(log, "Failed to process message: " + msg, th);
+        }
     }
 
     /**
@@ -174,8 +174,12 @@ public class GridReduceQueryExecutor {
         idx.addPage(new GridResultPage(node.id(), msg, false) {
             @Override public void fetchNextPage() {
                 try {
-                    ctx.io().sendUserMessage(F.asList(node), new GridQueryNextPageRequest(qryReqId,
qry, pageSize),
-                        GridTopic.TOPIC_QUERY, false, 0);
+                    GridQueryNextPageRequest msg0 = new GridQueryNextPageRequest(qryReqId,
qry, pageSize);
+
+                    if (node.isLocal())
+                        h2.mapQueryExecutor().onMessage(ctx.localNodeId(), msg0);
+                    else
+                        ctx.io().send(node, GridTopic.TOPIC_QUERY, msg0, GridIoPolicy.PUBLIC_POOL);
                 }
                 catch (IgniteCheckedException e) {
                     throw new IgniteException(e);
@@ -231,8 +235,8 @@ public class GridReduceQueryExecutor {
         runs.put(qryReqId, r);
 
         try {
-            ctx.io().sendUserMessage(nodes, new GridQueryRequest(qryReqId, r.pageSize, space,
qry.mapQueries()),
-                GridTopic.TOPIC_QUERY, false, 0);
+            send(nodes, new GridQueryRequest(qryReqId, r.pageSize, space, qry.mapQueries(),
+                ctx.config().getMarshaller().marshal(qry.mapQueries())));
 
             r.latch.await();
 
@@ -245,7 +249,7 @@ public class GridReduceQueryExecutor {
 
             for (GridMergeTable tbl : r.tbls) {
                 if (!tbl.getScanIndex(null).fetchedAll()) // We have to explicitly cancel
queries on remote nodes.
-                    ctx.io().sendUserMessage(nodes, new GridQueryCancelRequest(qryReqId),
GridTopic.TOPIC_QUERY, false, 0);
+                    send(nodes, new GridQueryCancelRequest(qryReqId));
 
 //                dropTable(r.conn, tbl.getName()); TODO
             }
@@ -269,6 +273,34 @@ public class GridReduceQueryExecutor {
     }
 
     /**
+     * @param nodes Nodes.
+     * @param msg Message.
+     * @throws IgniteCheckedException If failed.
+     */
+    private void send(Collection<ClusterNode> nodes, Message msg) throws IgniteCheckedException
{
+        for (ClusterNode node : nodes) {
+            if (node.isLocal()) {
+                ArrayList<ClusterNode> remotes = new ArrayList<>(nodes.size()
- 1);
+
+                for (ClusterNode node0 : nodes) {
+                    if (node0 != node)
+                        remotes.add(node0);
+                }
+
+                ctx.io().send(remotes, GridTopic.TOPIC_QUERY, msg, GridIoPolicy.PUBLIC_POOL);
+
+                // Local node goes the last to allow parallel execution.
+                h2.mapQueryExecutor().onMessage(ctx.localNodeId(), msg);
+
+                return;
+            }
+        }
+
+        // All the given nodes are remotes.
+        ctx.io().send(nodes, GridTopic.TOPIC_QUERY, msg, GridIoPolicy.PUBLIC_POOL);
+    }
+
+    /**
      * @param conn Connection.
      * @param tblName Table name.
      * @throws SQLException If failed.


Mime
View raw message