package org.yamcs.yarch.rocksdb;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.PriorityQueue;

import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yamcs.TimeInterval;
import org.yamcs.yarch.HistogramRecord;
import org.yamcs.yarch.HistogramSegment;
import org.yamcs.yarch.Partition;
import org.yamcs.yarch.PartitionManager;
import org.yamcs.yarch.TableDefinition;
import org.yamcs.yarch.YarchDatabase;

import static org.yamcs.yarch.rocksdb.CfTableWriter.zerobytes;

/**
 * 
 * @author nm
 *
 */
class RdbHistogramIterator implements Iterator<HistogramRecord> {

    private Iterator<List<Partition>> partitionIterator;
    private RocksIterator segmentIterator;

    private PriorityQueue<HistogramRecord> records = new PriorityQueue<HistogramRecord>();

    private final TimeInterval interval;
    private final long mergeTime;

    YarchDatabase ydb;
    TableDefinition tblDef;
    YRDB rdb;

    Logger log;
    String colName;


    //FIXME: mergeTime does not merge records across partitions or segments
    public RdbHistogramIterator(YarchDatabase ydb, TableDefinition tblDef, String colName, TimeInterval interval, long mergeTime) throws RocksDBException {
        this.interval = interval;
        this.mergeTime = mergeTime;
        this.ydb = ydb;
        this.tblDef = tblDef;
        this.colName = colName;

        PartitionManager partMgr = RdbStorageEngine.getInstance(ydb).getPartitionManager(tblDef);
        partitionIterator = partMgr.iterator(interval.getStart(), null);
        log = LoggerFactory.getLogger(this.getClass().getName()+"["+tblDef.getName()+"]");
        readNextPartition();
    }

    private void readNextPartition() throws RocksDBException {
        try {
            while(partitionIterator.hasNext()) {
                RdbPartition part  = (RdbPartition) partitionIterator.next().get(0);
                if(interval.hasStop() && part.getStart()>interval.getStop()) {
                    break; //finished
                }
                RDBFactory rdbf = RDBFactory.getInstance(ydb.getName());
                String dbDir = part.dir;

                log.debug("opening database "+ dbDir);
                if(rdb!=null) {
                    rdbf.dispose(rdb);
                }
                rdb = rdbf.getRdb(tblDef.getDataDir()+"/"+dbDir, false);
                String histoCfName = InKeyTableWriter.getHistogramColumnFamilyName(colName);
                ColumnFamilyHandle cfh = rdb.getColumnFamilyHandle(histoCfName);
                if(cfh!=null) {
                    segmentIterator = rdb.newIterator(cfh);
                    if(!interval.hasStart()) {
                        segmentIterator.seek(HistogramSegment.key(0, zerobytes));
                    } else {
                        int sstart=(int)(interval.getStart()/HistogramSegment.GROUPING_FACTOR);
                        segmentIterator.seek(HistogramSegment.key(sstart, zerobytes));
                    }

                    if(segmentIterator.isValid()) {
                        readNextSegments();
                        break;
                    }
                }
                rdbf.dispose(rdb);
            }
        } catch (IOException e) {
            log.error("Failed to open database", e);
        }
    }

    //reads all the segments with the same sstart time
    private void readNextSegments() throws RocksDBException {
        ByteBuffer bb = ByteBuffer.wrap(segmentIterator.key());
        int sstart = bb.getInt();
        if(sstart==Integer.MAX_VALUE) {
            readNextPartition();
            return;
        }
        
        while(true) {
            if(!addRecords(segmentIterator.key(), segmentIterator.value())){
                break;
            }
            segmentIterator.next();
            if(!segmentIterator.isValid()) {
                readNextPartition();
                break;
            }
            bb = ByteBuffer.wrap(segmentIterator.key());
            int g = bb.getInt();
            if(g!=sstart) break;
        }
    }       

    public void close() {
        if(rdb!=null) {
            RDBFactory.getInstance(ydb.getName()).dispose(rdb);
            rdb = null;
        }
    }

    private boolean addRecords(byte[] key, byte[] val) {
        ByteBuffer kbb = ByteBuffer.wrap(key);
        int sstart = kbb.getInt();
        byte[] columnv = new byte[kbb.remaining()];
        kbb.get(columnv);
        ByteBuffer vbb=ByteBuffer.wrap(val);
        HistogramRecord r = null;
        while(vbb.hasRemaining()) {
            long start = sstart*HistogramSegment.GROUPING_FACTOR + vbb.getInt();
            long stop = sstart*HistogramSegment.GROUPING_FACTOR + vbb.getInt();              
            int num = vbb.getShort();
            if((interval.hasStart()) && (stop<interval.getStart())) continue;
            
            if((interval.hasStop()) && (start>interval.getStop())) {
                return false;
            }
            if(r==null) {
                r = new HistogramRecord(columnv, start, stop, num);
            } else {
                if(start-r.getStop()<mergeTime) {
                    r = new HistogramRecord(r.getColumnv(), r.getStart(), stop, r.getNumTuples()+num);
                } else {
                    records.add(r);
                    r = new HistogramRecord(columnv, start, stop, num);
                }
            }
        }
        records.add(r);
        return true;
    }

    public boolean hasNext() {
        return !records.isEmpty();
    }

    public HistogramRecord next() {
        if(records.isEmpty()) throw new NoSuchElementException();
        HistogramRecord r = records.poll();

        if(records.isEmpty()) {
            try {
                readNextSegments();
            } catch (RocksDBException e) {
               throw new RuntimeException(e);
            }
        }
        return r;
    }
}

@@ -1171,16 +1171,48 @@
 olName;%0A
+    boolean stopReached = false;
 %0A%0A    //
@@ -3881,20 +3881,37 @@
         
-if(!
+boolean beyondStop = 
 addRecor
@@ -3964,10 +3964,9 @@
 e())
-)%7B
+;
 %0A   
@@ -3978,32 +3978,43 @@
     
-    break;%0A            %7D
+if(beyondStop) stopReached = true;%0A
 %0A   
@@ -4459,32 +4459,168 @@
        %7D%0A    %7D%0A%0A
+    //add all records from this segment into the queue %0A    // if the stop has been reached add only partially the records, return true%0A
     private bool
@@ -4656,24 +4656,77 @@
 te%5B%5D val) %7B%0A
+    //    System.out.println(%22interval: %22+interval);%0A
         Byte
@@ -5279,29 +5279,16 @@
 ntinue;%0A
-            %0A
         
@@ -5344,24 +5344,68 @@
 tStop())) %7B%0A
+                if(r!=null) records.add(r);%0A
             
@@ -5415,20 +5415,19 @@
  return 
-fals
+tru
 e;%0A     
@@ -5913,27 +5913,28 @@
      return 
-tru
+fals
 e;%0A    %7D%0A%0A  
@@ -6147,24 +6147,24 @@
 ds.poll();%0A%0A
-
         if(r
@@ -6179,16 +6179,32 @@
 sEmpty()
+ && !stopReached
 ) %7B%0A    

fix the stop