stream prototype (further impl)

Author: yuriy-glotanov

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

@@ -529,12 +529,20 @@ public synchronized ArrayList<Chunk> getFrameChunks (Session s) throws IOExcepti
         }
         if (dataObject.isNoTran() || s.isStream()) {
 
+            final int streamptr = s.isStream() ? s.streamFramePtr(this) : 0;
+
             for (Chunk c : data.getChunks()) {
                 if (c.getHeader().getState()==Header.RECORD_NORMAL_STATE) {
-                    res.add(c);
+                    if (c.getHeader().getPtr() >= streamptr) {
+                        res.add(c);
+                    }
                 }
             }
 
+            if (s.isStream()) {
+                s.streamFramePtr(this, rowCntr);
+            }
+
         } else { //if (local()) {
 
             final long tr = s.getTransaction().getTransId();

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

@@ -42,6 +42,7 @@ this software and associated documentation files (the "Software"), to deal in
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Modifier;
 import java.net.MalformedURLException;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import javax.persistence.*;
@@ -112,6 +113,8 @@ public static int getCLASS_ID() {
     @Transient
     private final ExecutorService streampool = Executors.newCachedThreadPool();
     @Transient
+    private final Map<Long, Integer> streammap = new ConcurrentHashMap<>();
+    @Transient
     private volatile boolean stream;
 
     public static Session getContextSession() {
@@ -541,6 +544,14 @@ public static void setDntmSession(Session dntmSession) {
         Session.dntmSession = dntmSession;
     }
 
+    public void streamFramePtr(Frame f, int ptr) {
+        streammap.put(f.getAllocFile()+f.getAllocPointer(), ptr);
+    }
+
+    public int streamFramePtr(Frame f) {
+        return streammap.get(f.getAllocFile()+f.getAllocPointer()) == null ? 0 : streammap.get(f.getAllocFile()+f.getAllocPointer());
+    }
+
     public boolean isStream() {
         return stream;
     }

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

@@ -0,0 +1,68 @@
+package su.interference.sql;
+
+import su.interference.core.DataChunk;
+import su.interference.core.EntityContainer;
+import su.interference.persistent.Cursor;
+import su.interference.persistent.Session;
+import su.interference.persistent.Table;
+
+import java.util.Queue;
+
+public class FrameGroupTask implements Runnable {
+
+    private final Cursor cur;
+    private final Queue<Object> q;
+    private final ResultSet target;
+    private final Table gtable;
+    private final Session s;
+
+    public FrameGroupTask(Cursor cur, Queue<Object> q, ResultSet target, Table gtable, Session s) {
+        this.cur = cur;
+        this.q = q;
+        this.target = target;
+        this.gtable = gtable;
+        this.s = s;
+    }
+
+    @Override
+    public void run() {
+
+        final boolean ixflag = cur.getSqlStmt().getCols().getOrderColumns().size() > 0;
+
+        try {
+            DataChunk cdc = null;
+            SQLGroup sqlg = null;
+            while (((StreamQueue) target).isRunning()) {
+                EntityContainer o = (EntityContainer) q.poll();
+                if (o != null) {
+                    if (cdc != null) {
+                        if (o.getDataChunk().compare(cdc, cur.getSqlStmt().getCols().getGroupColumns().size()) == 0) { //cdc & c chunks grouped
+                            sqlg.add(o.getDataChunk());
+                        } else { // start next group
+                            DataChunk gdc = sqlg.getDC();
+                            Object oo = gdc.getEntity(gtable);
+                            target.persist(oo, s);
+                            sqlg = new SQLGroup(o.getDataChunk(), cur.getSqlStmt().getCols().getColumns());
+                            sqlg.add(o.getDataChunk());
+                        }
+                    } else {
+                        sqlg = new SQLGroup(o.getDataChunk(), cur.getSqlStmt().getCols().getColumns());
+                        sqlg.add(o.getDataChunk());
+                    }
+                    cdc = o.getDataChunk();
+
+                    if (q.peek() == null) {
+                        Thread.sleep(100);
+                    }
+                }
+            }
+            DataChunk gdc = sqlg.getDC();
+            Object oo = gdc.getEntity(gtable);
+            target.persist(oo, s);
+        } catch (Exception e) {
+            ((StreamQueue) target).stop(s);
+            throw new RuntimeException(e);
+        }
+    }
+
+}

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

@@ -55,6 +55,7 @@ public class SQLCursor implements FrameIterator {
     private ExecutorService exec2 = SQLJoinThreadPool.getThreadPool2();
     private ExecutorService remotepool = Executors.newCachedThreadPool();
     private ExecutorService streampool = Executors.newCachedThreadPool();
+    private ExecutorService groupspool = Executors.newCachedThreadPool();
     private List<FrameApiJoin> tasks;
     private List<FrameApiJoin> tasks_;
     private FrameData bdnext;
@@ -229,6 +230,7 @@ public void build() throws InternalException, IOException, ClassNotFoundExceptio
 
     public void stream() throws Exception {
         final Queue<FrameApi> q = sfmap.get(lbi.getObjectId());
+        final Table gtable = Instance.getInstance().getTableById(lbi.getObjectId());
         if (!cur.isStream()) {
             logger.error("wrong stream method call: SQL statement is not a stream");
         }
@@ -242,14 +244,27 @@ public void stream() throws Exception {
         Runnable r = new Runnable() {
             @Override
             public void run() {
+                Thread.currentThread().setName("Stream SQL thread "+Thread.currentThread().getId());
                 try {
+                    final ConcurrentLinkedQueue<Object> q_in = new ConcurrentLinkedQueue<>();
+                    FrameGroupTask group = null;
                     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 (cur.getSqlStmt().isGroupedResult()) {
+                                if (group == null) {
+                                    group = new FrameGroupTask(cur, q_in, target, gtable, s);
+                                    groupspool.submit(group);
+                                }
+                                for (Object o : ft.get()) {
+                                    q_in.add(o);
+                                }
+                            } else {
+                                for (Object o : ft.get()) {
+                                    target.persist(o, s);
+                                }
                             }
                         }
                         if (q.peek() == null) {

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

@@ -383,6 +383,13 @@ public void setEntityResult(boolean entityResult) {
         this.entityResult = entityResult;
     }
 
+    public boolean isGroupedResult() {
+        if (cols.getGroupColumns().size() > 0 || cols.getFResultColumns().size() > 0) {
+            return true;
+        }
+        return false;
+    }
+
     public Table getEntityTable() {
         return entityTable;
     }