fix wrong index scans within distibute queries

Author: yuriy-glotanov

src/main/java/su/interference/persistent/Session.java

@@ -111,6 +111,8 @@ public static int getCLASS_ID() {
 
     @Transient
     private final ExecutorService streampool = Executors.newCachedThreadPool();
+    @Transient
+    private volatile boolean stream;
 
     public static Session getContextSession() {
         return contextSession.get();
@@ -539,4 +541,11 @@ public static void setDntmSession(Session dntmSession) {
         Session.dntmSession = dntmSession;
     }
 
+    public boolean isStream() {
+        return stream;
+    }
+
+    public void setStream(boolean stream) {
+        this.stream = stream;
+    }
 }

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

@@ -25,46 +25,53 @@ this software and associated documentation files (the "Software"), to deal in
 package su.interference.sql;
 
 import su.interference.core.Config;
+import su.interference.metrics.Metrics;
 
 import java.io.Serializable;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
 
 /**
  * @author Yuriy Glotanov
  * @since 1.0
  */
 
 public class FrameApiJoin implements Serializable, Callable<FrameApiJoin> {
-    private int nodeId;
+    private final 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 final transient CountDownLatch latch = new CountDownLatch(1);
     private List<Object> result;
+    private boolean failed;
 
-    public FrameApiJoin(int nodeId, SQLCursor cur, FrameApi bd1, FrameApi bd2, Map<Integer, Map<String, FrameApiJoin>> joins) {
+    public FrameApiJoin(int nodeId, SQLCursor cur, FrameApi bd1, FrameApi bd2) {
         this.nodeId = nodeId;
         this.cur = cur;
         this.bd1 = bd1;
         this.bd2 = bd2;
         this.leftAllocId = bd1.getAllocId();
-        this.rightAllocId = bd2 == null ? 0 : bd2.getAllocId();
+        this.rightAllocId = bd2 == null ? 0 : bd2 instanceof SQLIndexFrame ? 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);
+            Metrics.get("localTask").start();
             result = frameJoinTask.call();
+            Metrics.get("localTask").stop();
+        } else {
+            Metrics.get("remoteTask").start();
+            latch.await();
+            Metrics.get("remoteTask").stop();
         }
         return this;
     }
@@ -87,14 +94,6 @@ public int getNodeId() {
         return nodeId;
     }
 
-    public void setNodeId(int nodeId) {
-        this.nodeId = nodeId;
-    }
-
-    public FrameJoinTask getFrameJoinTask() {
-        return frameJoinTask;
-    }
-
     public long getLeftAllocId() {
         return leftAllocId;
     }
@@ -103,11 +102,29 @@ public long getRightAllocId() {
         return rightAllocId;
     }
 
+    public FrameApi getBd1() {
+        return bd1;
+    }
+
+    public FrameApi getBd2() {
+        return bd2;
+    }
+
     public List<Object> getResult() {
         return result;
     }
 
     public void setResult(List<Object> result) {
         this.result = result;
+        latch.countDown();
+    }
+
+    public boolean isFailed() {
+        return failed;
+    }
+
+    public void setFailed(boolean failed) {
+        this.failed = failed;
+        latch.countDown();
     }
 }

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

@@ -44,7 +44,7 @@ public class FrameHolder {
     private final boolean persistent;
 
     public FrameHolder(ResultSet t) {
-        if (t.getClass().getName().equals("su.interference.persistent.Table")) {
+        if (t != null && t.getClass().getName().equals("su.interference.persistent.Table")) {
             Table tt = (Table)t;
             cframes = new AtomicReference[tt.getLbs().length];
             for (int i=0; i<tt.getLbs().length; i++) {

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

@@ -117,7 +117,6 @@ public List<Object> call() throws Exception {
         }
 */
 
-        Metrics.get("localTask").start();
         final ArrayList<Object> drs1 = bd1.getFrameEntities(s);
         final ArrayList<Object> drs2 = bd2==null?null:bd2.getFrameEntities(s);
 
@@ -221,7 +220,10 @@ public List<Object> call() throws Exception {
                     } else {
                         //one table loop
                         if (r == null) {
-                            res.add(o1); //target table is null -> result class is null -> returns generic entities
+                            //todo need to cast o1 to RS type
+                            //if (nc.checkNC(o1, sqlcid, last)) {
+                                res.add(o1); //target table is null -> result class is null -> returns generic entities
+                            //}
                         } else {
                             Object j = joinDataRecords(r, c1, c2, t1, t2, o1, null, cols, c1rs, s);
                             if (nc.checkNC(j, sqlcid, last)) {
@@ -240,7 +242,6 @@ public List<Object> call() throws Exception {
                 }
             }
         }
-        Metrics.get("localTask").stop();
         Metrics.get("recordLCount").put(res.size());
         // logger.info(res.size()+" records returns from local node " + nodeId);
         return res;

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

@@ -55,14 +55,15 @@ public RemoteTask(Cursor cur, int nodeId, Map<String, FrameApiJoin> joins, Strin
     }
 
     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()) {
+                for (Map.Entry<String, FrameApiJoin> entry : joins.entrySet()) {
+                    entry.getValue().setFailed(true);
+                }
                 throw new InternalException();
             }
             Metrics.get("recordRCount").put(rs.size());
@@ -71,7 +72,7 @@ public Boolean call() throws Exception {
             }
         } else {
             for (Map.Entry<String, FrameApiJoin> entry : joins.entrySet()) {
-                entry.getValue().setNodeId(Config.getConfig().LOCAL_NODE_ID);
+                entry.getValue().setFailed(true);
             }
         }
         return true;

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

@@ -37,6 +37,7 @@ this software and associated documentation files (the "Software"), to deal in
 import java.util.*;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.lang.reflect.InvocationTargetException;
 import java.io.IOException;
@@ -51,9 +52,11 @@ public class SQLCursor implements FrameIterator {
     private final int id;
     private ResultSet target;
     private ExecutorService exec = SQLJoinThreadPool.getThreadPool();
+    private ExecutorService exec2 = SQLJoinThreadPool.getThreadPool2();
+    private ExecutorService remotepool = Executors.newCachedThreadPool();
+    private ExecutorService streampool = Executors.newCachedThreadPool();
     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;
@@ -100,16 +103,16 @@ public SQLCursor (int id, FrameIterator lbi, FrameIterator rbi, NestedCondition
 
         if (cur.getType() == Cursor.SLAVE_TYPE && cur.getResultTargetName() != null && this.id == 1) {
             target = this.peristent ? s.registerTable(cur.getResultTargetName(), s, rscols, null, null, ixflag && last) : new ResultList(cur.getSqlStmt().getEntityTable());
-        } else if (cur.getType() == Cursor.MASTER_TYPE) {
-            target = this.peristent ? s.registerTable("su.interference.persistent.R$" + UUID.randomUUID().toString().replace('-', '$'), s, rscols, null, null, ixflag && last) : new ResultList(cur.getSqlStmt().getEntityTable());
         } else if (cur.getType() == Cursor.STREAM_TYPE) {
             target = new StreamQueue();
+        } else
+            {
+            target = this.peristent ? s.registerTable("su.interference.persistent.R$" + UUID.randomUUID().toString().replace('-', '$'), s, rscols, null, null, ixflag && last) : new ResultList(cur.getSqlStmt().getEntityTable());
         }
         current = new FrameHolder(target);
 
         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;
@@ -184,19 +187,14 @@ protected FrameJoinTask buildFrameJoinTask(int nodeId, FrameApi bd1, FrameApi bd
 
     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()) {
             final FrameApi bd1 = lbi.nextFrame();
 
             if (rbi == null) {
-                tasks.add(new FrameApiJoin(ns[i], this, bd1, null, joins));
+                tasks.add(new FrameApiJoin(ns[i], this, bd1, null));
                 i++;
                 if (i == ns.length) { i = 0; }
             } else {
@@ -205,45 +203,67 @@ public void build() throws InternalException, IOException, ClassNotFoundExceptio
                     if (rightType == null) {
                         rightType = bd2.getClass().getSimpleName();
                     }
-                    tasks.add(new FrameApiJoin(ns[i], this, bd1, bd2, joins));
+                    tasks.add(new FrameApiJoin(ns[i], this, bd1, bd2));
                     i++;
                     if (i == ns.length) { i = 0; }
                 }
                 rbi.resetIterator();
             }
         }
-        logger.debug("SQL cursor is build: local tasks amount = "+tasks.size()+", use NC check = "+last);
+        logger.debug("SQL cursor is build: 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);
+            for (Integer nodeId : ns) {
+                if (nodeId != Config.getConfig().LOCAL_NODE_ID) {
+                    final Map<String, FrameApiJoin> joins = new HashMap<>();
+                    for (FrameApiJoin j : tasks) {
+                        if (j.getNodeId() == nodeId) {
+                            joins.put(j.getKey(), j);
+                        }
+                    }
+                    final RemoteTask rt = new RemoteTask(cur, nodeId, joins, rightType);
+                    remotepool.submit(rt);
+                }
             }
             sent = true;
         }
     }
 
     public void stream() throws Exception {
         final Queue<FrameApi> q = sfmap.get(lbi.getObjectId());
-        boolean cnue = true;
         if (!cur.isStream()) {
             logger.error("wrong stream method call: SQL statement is not a stream");
         }
         if (q == null) {
             throw new RuntimeException("internal error: queue not exist for object id = "+lbi.getObjectId());
         }
-        while (cnue) {
-            FrameApi f = q.poll();
-            if (f != null) {
-                FrameJoinTask task = new FrameJoinTask(cur, f, null, target, rscols, nc, id, Config.getConfig().LOCAL_NODE_ID, last, lbi.isLeftfs(), null, s);
-                final Future<List<Object>> ft = exec.submit(task);
-                for (Object o : ft.get()) {
-                    target.persist(o, s);
+        if (!(target instanceof StreamQueue)) {
+            throw new RuntimeException("internal error: wrong target type for object id = "+lbi.getObjectId());
+        }
+        s.setStream(true);
+        Runnable r = new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    while (((StreamQueue) target).isRunning()) {
+                        FrameApi f = q.poll();
+                        if (f != null) {
+                            FrameJoinTask task = new FrameJoinTask(cur, f, null, target, rscols, nc, id, Config.getConfig().LOCAL_NODE_ID, last, lbi.isLeftfs(), null, s);
+                            final Future<List<Object>> ft = exec.submit(task);
+                            for (Object o : ft.get()) {
+                                target.persist(o, s);
+                            }
+                        }
+                        if (q.peek() == null) {
+                            Thread.sleep(100);
+                        }
+                    }
+                } catch (Exception e) {
+                    ((StreamQueue) target).stop(s);
+                    throw new RuntimeException(e);
                 }
             }
-            if (q.peek() == null) {
-                Thread.sleep(1000);
-            }
-        }
+        };
+        streampool.submit(r);
     }
 
     public static void addStreamFrame(FrameApi f) {
@@ -270,21 +290,33 @@ private synchronized FrameData nextFrame2() throws InternalException {
 
         while (!done && ret==null) {
             final ArrayList<Future<FrameApiJoin>> flist = new ArrayList<Future<FrameApiJoin>>();
+            final ArrayList<Future<FrameApiJoin>> flist2 = new ArrayList<Future<FrameApiJoin>>();
             try {
                 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));
+                        if (j.getNodeId() == Config.getConfig().LOCAL_NODE_ID) {
+                            flist.add(exec.submit(j));
+                        } else {
+                            flist2.add(exec2.submit(j));
+                        }
                     }
                 }
                 for (Future<FrameApiJoin> f : flist) {
                     final FrameApiJoin j = f.get();
-                    final List<Object> ol = j.getResult();
-                    if (ol == null) {
-                        tasks_.add(j);
+                    logger.debug("SQL cursor next frame: the jointask call returned " + j.getResult().size() + " records");
+                    for (Object o : j.getResult()) {
+                        target.persist(o, s);
+                    }
+                }
+                for (Future<FrameApiJoin> f : flist2) {
+                    final FrameApiJoin j = f.get();
+                    if (j.isFailed()) {
+                        FrameApiJoin j_ = new FrameApiJoin(Config.getConfig().LOCAL_NODE_ID, this, j.getBd1(), j.getBd2());
+                        tasks_.add(j_);
                     } else {
-                        logger.debug("SQL cursor next frame: the jointask call returned " + ol.size() + " records");
-                        for (Object o : ol) {
+                        logger.debug("SQL cursor next frame: the jointask call returned " + j.getResult().size() + " records");
+                        for (Object o : j.getResult()) {
                             target.persist(o, s);
                         }
                     }