improve scalability of sql queries

Author: yuriy-glotanov

src/main/java/su/interference/sql/FrameApiJoin.java

@@ -0,0 +1,113 @@
+/**
+ The MIT License (MIT)
+
+ Copyright (c) 2010-2019 head systems, ltd
+
+ Permission is hereby granted, free of charge, to any person obtaining a copy of
+ this software and associated documentation files (the "Software"), to deal in
+ the Software without restriction, including without limitation the rights to
+ use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ the Software, and to permit persons to whom the Software is furnished to do so,
+ subject to the following conditions:
+
+ The above copyright notice and this permission notice shall be included in all
+ copies or substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+ */
+
+package su.interference.sql;
+
+import su.interference.core.Config;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+/**
+ * @author Yuriy Glotanov
+ * @since 1.0
+ */
+
+public class FrameApiJoin implements Serializable, Callable<FrameApiJoin> {
+    private int nodeId;
+    private final transient SQLCursor cur;
+    private final transient FrameApi bd1;
+    private final transient FrameApi bd2;
+    private final transient FrameJoinTask frameJoinTask;
+    private final long leftAllocId;
+    private final long rightAllocId;
+    private List<Object> result;
+
+    public FrameApiJoin(int nodeId, SQLCursor cur, FrameApi bd1, FrameApi bd2, Map<Integer, Map<String, FrameApiJoin>> joins) {
+        this.nodeId = nodeId;
+        this.cur = cur;
+        this.bd1 = bd1;
+        this.bd2 = bd2;
+        this.leftAllocId = bd1.getAllocId();
+        this.rightAllocId = bd2 == null ? 0 : bd2.getAllocId();
+        if (nodeId == Config.getConfig().LOCAL_NODE_ID) {
+            frameJoinTask = cur.buildFrameJoinTask(nodeId, bd1, bd2);
+        } else {
+            frameJoinTask = null;
+            joins.get(nodeId).put(this.getKey(), this);
+        }
+    }
+
+    public FrameApiJoin call() throws Exception {
+        if (nodeId == Config.getConfig().LOCAL_NODE_ID) {
+            final FrameJoinTask frameJoinTask = cur.buildFrameJoinTask(nodeId, bd1, bd2);
+            result = frameJoinTask.call();
+        }
+        return this;
+    }
+
+    protected String getKey() {
+        return this.leftAllocId + "-" + this.rightAllocId;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        return this.getKey().equals(((FrameApiJoin)obj).getKey());
+    }
+
+    @Override
+    public int hashCode() {
+        return this.getKey().hashCode();
+    }
+
+    public int getNodeId() {
+        return nodeId;
+    }
+
+    public void setNodeId(int nodeId) {
+        this.nodeId = nodeId;
+    }
+
+    public FrameJoinTask getFrameJoinTask() {
+        return frameJoinTask;
+    }
+
+    public long getLeftAllocId() {
+        return leftAllocId;
+    }
+
+    public long getRightAllocId() {
+        return rightAllocId;
+    }
+
+    public List<Object> getResult() {
+        return result;
+    }
+
+    public void setResult(List<Object> result) {
+        this.result = result;
+    }
+}

src/main/java/su/interference/sql/FrameJoinTask.java

@@ -101,6 +101,7 @@ public List<Object> call() throws Exception {
         ResultSetEntity rsa = (ResultSetEntity)c1.getAnnotation(ResultSetEntity.class);
         final boolean c1rs = rsa!=null?true:false;
 
+/*
         if (cur.getType() == Cursor.MASTER_TYPE && nodeId != Config.getConfig().LOCAL_NODE_ID && this.leftfs) {
             Metrics.get("remoteTask").start();
             final TransportEvent transportEvent = new SQLEvent(nodeId, cur.getCursorId(), bd1.getAllocId(), bd2==null?0:bd2.getAllocId(), bd2==null?null:bd2.getClass().getSimpleName(), 0, null, null, 0, true);
@@ -114,6 +115,7 @@ public List<Object> call() throws Exception {
                 return rs;
             }
         }
+*/
 
         Metrics.get("localTask").start();
         final ArrayList<Object> drs1 = bd1.getFrameEntities(s);

src/main/java/su/interference/sql/RemoteTask.java

@@ -0,0 +1,79 @@
+/**
+ The MIT License (MIT)
+
+ Copyright (c) 2010-2019 head systems, ltd
+
+ Permission is hereby granted, free of charge, to any person obtaining a copy of
+ this software and associated documentation files (the "Software"), to deal in
+ the Software without restriction, including without limitation the rights to
+ use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ the Software, and to permit persons to whom the Software is furnished to do so,
+ subject to the following conditions:
+
+ The above copyright notice and this permission notice shall be included in all
+ copies or substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+ */
+
+package su.interference.sql;
+
+import su.interference.core.Config;
+import su.interference.exception.InternalException;
+import su.interference.metrics.Metrics;
+import su.interference.persistent.Cursor;
+import su.interference.transport.SQLEvent;
+import su.interference.transport.TransportContext;
+import su.interference.transport.TransportEvent;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+/**
+ * @author Yuriy Glotanov
+ * @since 1.0
+ */
+
+public class RemoteTask implements Callable<Boolean> {
+    private final Cursor cur;
+    private final int nodeId;
+    private final Map<String, FrameApiJoin> joins;
+    private final String rightType;
+
+    public RemoteTask(Cursor cur, int nodeId, Map<String, FrameApiJoin> joins, String rightType) {
+        this.cur = cur;
+        this.nodeId = nodeId;
+        this.joins = joins;
+        this.rightType = rightType;
+    }
+
+    public Boolean call() throws Exception {
+        Metrics.get("remoteTask").start();
+        final TransportEvent transportEvent = new SQLEvent(nodeId, cur.getCursorId(), joins, rightType, 0, null, null, 0, true);
+        TransportContext.getInstance().send(transportEvent);
+        transportEvent.getLatch().await();
+        Metrics.get("remoteTask").stop();
+        if (!transportEvent.isFail()) {
+            final List<FrameApiJoin> rs = ((SQLEvent) transportEvent).getCallback().getResult().getResultSet();
+            if (rs.size() != joins.size()) {
+                throw new InternalException();
+            }
+            Metrics.get("recordRCount").put(rs.size());
+            for (FrameApiJoin j : rs) {
+                joins.get(j.getKey()).setResult(j.getResult());
+            }
+        } else {
+            for (Map.Entry<String, FrameApiJoin> entry : joins.entrySet()) {
+                entry.getValue().setNodeId(Config.getConfig().LOCAL_NODE_ID);
+            }
+        }
+        return true;
+    }
+}

src/main/java/su/interference/sql/SQLCursor.java

@@ -50,12 +50,14 @@ public class SQLCursor implements FrameIterator {
 
     private final int id;
     private ResultSet target;
-    private boolean done;
     private ExecutorService exec = SQLJoinThreadPool.getThreadPool();
-    private ArrayList<FrameJoinTask> tasks;
+    private List<FrameApiJoin> tasks;
+    private List<FrameApiJoin> tasks_;
+    private Map<Integer, Map<String, FrameApiJoin>> joins;
     private FrameData bdnext;
     private FrameHolder current;
     private int ptr = 0;
+    private boolean sent;
     private Cursor cur;
     private Session s;
     private ArrayList<SQLColumn> rscols;
@@ -66,6 +68,7 @@ public class SQLCursor implements FrameIterator {
     private final FrameIterator lbi;
     private final FrameIterator rbi;
     private final List<Integer> objectIds;
+    private String rightType;
     private boolean leftFS;
     private final boolean furtherUseUC;
     private final SQLColumn joinedCC;
@@ -104,7 +107,9 @@ public SQLCursor (int id, FrameIterator lbi, FrameIterator rbi, NestedCondition
         }
         current = new FrameHolder(target);
 
-        tasks = new ArrayList<FrameJoinTask>();
+        tasks = new ArrayList<FrameApiJoin>();
+        tasks_ = new ArrayList<FrameApiJoin>();
+        joins = new HashMap<>();
 
         //rebuild column set for sqlcursor iterator
         final SQLCursor cursor_ = lbi.getType() == FrameIterator.TYPE_CURSOR ? (SQLCursor) lbi : rbi != null && rbi.getType() == FrameIterator.TYPE_CURSOR ? (SQLCursor) rbi : null;
@@ -173,29 +178,48 @@ public int getObjectId() {
         return target.getObjectId();
     }
 
+    protected FrameJoinTask buildFrameJoinTask(int nodeId, FrameApi bd1, FrameApi bd2) {
+        return new FrameJoinTask(cur, bd1, bd2, target, rscols, nc, id, nodeId, last, lbi.isLeftfs(), hmap, s);
+    }
+
     public void build() throws InternalException, IOException, ClassNotFoundException, InvocationTargetException, NoSuchMethodException, InstantiationException, IllegalAccessException {
         final Integer[] ns = TransportContext.getInstance().getOnlineNodesWithLocal();
+        for (Integer nodeId : ns) {
+            if (nodeId != Config.getConfig().LOCAL_NODE_ID) {
+                joins.put(nodeId, new HashMap<>());
+            }
+        }
         int i = 0;
         //int tnode = 0;
 
         while (lbi.hasNextFrame()) {
-            FrameApi bd1 = lbi.nextFrame();
+            final FrameApi bd1 = lbi.nextFrame();
 
             if (rbi == null) {
-                tasks.add(new FrameJoinTask(cur, bd1, null, target, rscols, nc, id, ns[i], last, lbi.isLeftfs(), null, s));
+                tasks.add(new FrameApiJoin(ns[i], this, bd1, null, joins));
                 i++;
                 if (i == ns.length) { i = 0; }
             } else {
                 while (rbi.hasNextFrame()) {
-                    FrameApi bd2 = rbi.nextFrame();
-                    tasks.add(new FrameJoinTask(cur, bd1, bd2, target, rscols, nc, id, ns[i], last, lbi.isLeftfs(), hmap, s));
+                    final FrameApi bd2 = rbi.nextFrame();
+                    if (rightType == null) {
+                        rightType = bd2.getClass().getSimpleName();
+                    }
+                    tasks.add(new FrameApiJoin(ns[i], this, bd1, bd2, joins));
                     i++;
                     if (i == ns.length) { i = 0; }
                 }
                 rbi.resetIterator();
             }
         }
-        logger.debug("SQL cursor is build: tasks amount = "+tasks.size()+", use NC check = "+last);
+        logger.debug("SQL cursor is build: local tasks amount = "+tasks.size()+", use NC check = "+last);
+        if (!sent) {
+            for (Map.Entry<Integer, Map<String, FrameApiJoin>> entry : joins.entrySet()) {
+                final RemoteTask rt = new RemoteTask(cur, entry.getKey(), entry.getValue(), rightType);
+                exec.submit(rt);
+            }
+            sent = true;
+        }
     }
 
     public void stream() throws Exception {
@@ -241,27 +265,38 @@ public synchronized FrameData nextFrame() {
     }
 
     private synchronized FrameData nextFrame2() throws InternalException {
-        boolean done = !(ptr<tasks.size());
+        boolean done = !(ptr < tasks.size());
         FrameData ret = current.getFrame(done);
 
-        while (!done&&ret==null) {
+        while (!done && ret==null) {
+            final ArrayList<Future<FrameApiJoin>> flist = new ArrayList<Future<FrameApiJoin>>();
             try {
-                ArrayList<Future<List<Object>>> flist = new ArrayList<Future<List<Object>>>();
-                for (int i=0; i<BATCH_SIZE; i++) {
-                    if ((ptr + i)<tasks.size()) {
-                        FrameJoinTask jt = tasks.get(ptr + i);
-                        flist.add(exec.submit(jt));
+                for (int i=0; i < BATCH_SIZE; i++) {
+                    if ((ptr + i) < tasks.size()) {
+                        final FrameApiJoin j = tasks.get(ptr + i);
+                        flist.add(exec.submit(j));
                     }
                 }
-                for (Future<List<Object>> f : flist) {
-                    List<Object> ol = f.get();
-                    logger.debug("SQL cursor next frame: the jointask call returned " + ol.size() + " records");
-                    for (Object o : ol) {
-                        target.persist(o, s);
+                for (Future<FrameApiJoin> f : flist) {
+                    final FrameApiJoin j = f.get();
+                    final List<Object> ol = j.getResult();
+                    if (ol == null) {
+                        tasks_.add(j);
+                    } else {
+                        logger.debug("SQL cursor next frame: the jointask call returned " + ol.size() + " records");
+                        for (Object o : ol) {
+                            target.persist(o, s);
+                        }
                     }
                 }
                 ptr = ptr + BATCH_SIZE;
-                done = !(ptr<tasks.size());
+                done = !(ptr < tasks.size());
+                if (done && tasks_.size() > 0) {
+                    done = false;
+                    tasks = tasks_;
+                    tasks_ = new ArrayList<>();
+                    ptr = 0;
+                }
                 ret = current.getFrame(done);
             } catch (Exception e) {
                 e.printStackTrace();

src/main/java/su/interference/sql/SQLJoinThreadPool.java

@@ -34,7 +34,7 @@ this software and associated documentation files (the "Software"), to deal in
 
 public class SQLJoinThreadPool {
 
-    private static ExecutorService exec = Executors.newFixedThreadPool(4);
+    private static ExecutorService exec = Executors.newFixedThreadPool(8);
 
     public static ExecutorService getThreadPool() {
         return exec;

src/main/java/su/interference/sql/SQLSelect.java

@@ -328,7 +328,7 @@ private final void parseSQL (String s, Cursor cur, Session sn) throws Exception
 
             final Integer[] ns = TransportContext.getInstance().getOnlineNodes();
             for (Integer nodeId : TransportContext.getInstance().getOnlineNodes()) {
-                final TransportEvent transportEvent = new SQLEvent(nodeId, cursor.getCursorId(), 0, 0, null,
+                final TransportEvent transportEvent = new SQLEvent(nodeId, cursor.getCursorId(), null, null,
                         this.join.getTargetId(), cursor.getSql(), this.join.getResultTargetName(), sn.getTransaction().getTransId(), false);
                 TransportContext.getInstance().send(transportEvent);
                 logger.debug("create remote cursor on node " + nodeId);