stream prototype

Author: yuriy-glotanov

src/main/java/su/interference/core/SyncQueue.java

@@ -31,6 +31,7 @@ this software and associated documentation files (the "Software"), to deal in
 import org.slf4j.LoggerFactory;
 import su.interference.persistent.*;
 import su.interference.exception.*;
+import su.interference.sql.SQLCursor;
 import su.interference.transport.TransportSyncTask;
 
 /**
@@ -66,6 +67,7 @@ private synchronized boolean syncFramesFromQueue() throws Exception {
             FreeFrame fb = null;
             try {
                 frames.add(new SyncFrame((Frame) entry.getValue(), s, fb));
+                SQLCursor.addStreamFrame(((Frame) entry.getValue()).getFrameData());
             } catch (MissingSyncFrameException e) {
                 logger.debug("Unable to sync frame "+((Frame) entry.getValue()).getPtr()+" because removed by freeing");
             }

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

@@ -48,6 +48,8 @@ public class Cursor implements Serializable {
     @Transient
     public static final int SLAVE_TYPE      = 2;
     @Transient
+    public static final int STREAM_TYPE     = 3;
+    @Transient
     public static final int STATE_IDLE      = 1;
     @Transient
     public static final int STATE_PREPARED  = 2;
@@ -211,4 +213,9 @@ public void setState(int state) {
     public void startStatement(long tranId) throws Exception {
         this.session.startStatement(tranId);
     }
+
+    public boolean isStream() {
+        return this.type == STREAM_TYPE ? true : false;
+    }
+
 }

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

@@ -1164,6 +1164,10 @@ protected DataChunk persist (final Object o, final Session s, final LLT extllt)
         }
     }
 
+    public Object poll() {
+        return null;
+    }
+
     private void persistIndexes(DataChunk c, Session s, LLT llt) throws Exception {
         for (IndexDescript ids : this.getIndexNames()) {
             final Table ixt = Instance.getInstance().getTableByName("su.interference.persistent."+ids.getName());

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

@@ -53,6 +53,10 @@ public DataChunk persist(Object o, Session s) {
         return null;
     }
 
+    public Object poll() {
+        return null;
+    }
+
     public List<Chunk> getAll(Session s) {
         return null;
     }

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

@@ -41,6 +41,7 @@ this software and associated documentation files (the "Software"), to deal in
 
 public interface ResultSet {
     DataChunk persist(Object o, Session s) throws Exception;
+    Object poll();
     List<Chunk> getAll(Session s) throws Exception;
     ArrayList<Object> getAll(Session s, int ptr) throws Exception;
     int getObjectId();

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

@@ -56,6 +56,10 @@ public DataChunk persist(Object o, Session s) throws Exception {
         return null;
     }
 
+    public Object poll() {
+        return null;
+    }
+
     public List<Chunk> getAll(Session s) throws Exception {
         return null;
     }

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

@@ -34,11 +34,10 @@ this software and associated documentation files (the "Software"), to deal in
 
 import java.lang.reflect.Field;
 import java.net.MalformedURLException;
-import java.util.List;
-import java.util.UUID;
+import java.util.*;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
-import java.util.ArrayList;
 import java.lang.reflect.InvocationTargetException;
 import java.io.IOException;
 
@@ -72,6 +71,7 @@ public class SQLCursor implements FrameIterator {
     private final SQLColumn joinedCC;
     private SQLColumn extJoinedCC;
 
+    private static Map<Integer, ConcurrentLinkedQueue<FrameApi>> sfmap = new HashMap();
     private static final int BATCH_SIZE = 4;
     private final static Logger logger = LoggerFactory.getLogger(SQLCursor.class);
 
@@ -88,13 +88,19 @@ public SQLCursor (int id, FrameIterator lbi, FrameIterator rbi, NestedCondition
         this.last = last;
         this.peristent = !cur.getSqlStmt().isEntityResult();
 
+        if (cur.isStream()) {
+            sfmap.put(lbi.getObjectId(), new ConcurrentLinkedQueue<FrameApi>());
+        }
+
         //todo wrong case - column set must be rebuilded for prevent bad indexes intersect cases
         //todo need to refactor SQLJoin - extJoinedCC must be set on construct time
 
         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 {
+        } 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();
         }
         current = new FrameHolder(target);
 
@@ -192,6 +198,37 @@ public void build() throws InternalException, IOException, ClassNotFoundExceptio
         logger.debug("SQL cursor is build: tasks amount = "+tasks.size()+", use NC check = "+last);
     }
 
+    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 (q.peek() == null) {
+                Thread.sleep(1000);
+            }
+        }
+    }
+
+    public static void addStreamFrame(FrameApi f) {
+        final Queue<FrameApi> q = sfmap.get(f.getObjectId());
+        if (q != null) {
+            q.add(f);
+        }
+    }
+
     // execute single task (on local node only)
     public List<Object> execute(FrameApi bd1, FrameApi bd2) throws Exception {
         final FrameJoinTask task = new FrameJoinTask(cur, bd1, bd2, target, rscols, nc, id, Config.getConfig().LOCAL_NODE_ID, last, lbi.isLeftfs(), hmap, s);

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

@@ -124,7 +124,11 @@ public ResultSet executeJoin(Session s, int mode, ArrayList<SQLSetValue> uset) t
             SQLCursor last = null;
             for (SQLCursor sqlc : preparedCursors) {
                 last = sqlc;
-                sqlc.build();
+                if (cur.isStream()) {
+                    sqlc.stream();
+                } else {
+                    sqlc.build();
+                }
             }
 
             temp = last.flushTarget();

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(8);
+    private static ExecutorService exec = Executors.newFixedThreadPool(4);
 
     public static ExecutorService getThreadPool() {
         return exec;

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

@@ -148,16 +148,16 @@ private final void parseSQL (String s, Cursor cur, Session sn) throws Exception
         final String sql = s.trim();
         final String SQL = s.toUpperCase().trim();
 
-        cursor = cur==null?new Cursor(sql, Cursor.MASTER_TYPE):cur;
+        cursor = cur==null?new Cursor(sql, this.stream ? Cursor.STREAM_TYPE : Cursor.MASTER_TYPE):cur;
         cursor.setSqlStmt(this);
 
         if (!SQL.startsWith(SELECT_CLAUSE)) {
             throw new InvalidSQLStatement();
         }
-        if (SQL.indexOf(STREAM_CLAUSE) >6 ) {
+        if (SQL.indexOf(STREAM_CLAUSE) == SELECT_CLAUSE.length()) {
             this.stream = true;
         }
-        if (SQL.indexOf(DISTINCT_CLAUSE) >6 ) {
+        if (SQL.indexOf(DISTINCT_CLAUSE) == SELECT_CLAUSE.length()) {
             this.distinct = true;
         }
         if (SQL.indexOf(FROM_CLAUSE) < SELECT_CLAUSE.length() + 1) {
@@ -186,7 +186,10 @@ private final void parseSQL (String s, Cursor cur, Session sn) throws Exception
             }
         }
 
-        final String[] clds = sql.substring(SELECT_CLAUSE.length(), SQL.indexOf(FROM_CLAUSE)).trim().split(",");
+        final int cldsStart = stream ? SELECT_CLAUSE.length() + STREAM_CLAUSE.length() :
+                              distinct ? SELECT_CLAUSE.length() + DISTINCT_CLAUSE.length() :
+                              SELECT_CLAUSE.length();
+        final String[] clds = sql.substring(cldsStart, SQL.indexOf(FROM_CLAUSE)).trim().split(",");
         final String[] tbls = sql.substring(SQL.indexOf(FROM_CLAUSE)+6, baselen).trim().split(",");
 
         final int wpos = SQL.indexOf(WHERE_CLAUSE);