timeseries.h

最新版本！fastdb是高效的内存数据库系统

     * to doubled number of seconds in day multipied on maxElementsInBlock
     */
    dbTimeSeriesProcessor(dbDatabase& database, int minElementsInBlock=100, int maxElementsInBlock=100, time_t maxBlockTimeInterval=0) :
        db(database) 
    {
        assert(minElementsInBlock > 0 && maxElementsInBlock >= minElementsInBlock);
        if (maxBlockTimeInterval == 0) { 
            maxBlockTimeInterval = 2*(maxElementsInBlock*24*60*60); // doubled interval in seconds, one element per day
        }        
        this->maxElementsInBlock = maxElementsInBlock;
        this->minElementsInBlock = minElementsInBlock;
        this->maxBlockTimeInterval = maxBlockTimeInterval;

        // correct instance of interval will be specified in select
        Interval* dummy = NULL;
        selectBlock = "blockId between",dummy->from,"and",dummy->till;
        selectBlockReverse = "blockId between",dummy->from,"and",dummy->till,"order by blockId desc";
    }

    /**
     * Remove elements for the sepcified period
     * @param oid time series identifer (OID of the object associated with this time series)
     * @param from inclusive low bound for element timestamp (set 0 to disable this criteria)
     * @param till inclusive high bound for element timestamp (set INFINITE_TIME to disable this criteria)
     * @return number of removed elements
     */
    int remove(oid_t oid, time_t from, time_t till)
    {
        Interval interval;
        interval.from = generateBlockId(oid, from == 0 ? 0 : from - maxBlockTimeInterval);
        interval.till = generateBlockId(oid, till);
        dbCursor< dbTimeSeriesBlock<T> > blocks;
        size_t nRemoved = 0;
        if (blocks.select(selectBlock, dbCursorForUpdate, &interval)) { 
            do { 
                int n = blocks->used;
                T const* e =  blocks->elements.get();
                int l = 0, r = n;
                while (l < r)  {
                    int i = (l+r) >> 1;
                    if (from > e[i].time()) { 
                        l = i+1;
                    } else { 
                        r = i;
                    }
                }
                assert(l == r && (l == n || e[l].time() >= from)); 
                while (r < n && e[r].time() <= till) {
                    r += 1;
                    nRemoved += 1;
                }
                if (l == 0 && r == n) { 
                    blocks.remove();
                } else if (l < n && l != r) { 
                    if (l == 0) { 
                        blocks->blockId = generateBlockId(oid, e[r].time());
                    }
                    T* ue = blocks->elements.update();
                    while (r < n) { 
                        ue[l++] = ue[r++];
                    }
                    blocks->used = l;
                    blocks.update();
                }
            } while (blocks.nextAvailable());
        }
        return nRemoved;
    }        
    
    virtual~dbTimeSeriesProcessor() {}

    /**
     * This method should be actually private but since there is no portable way of declaration 
     * of friend templates classes recognized by all C++ compiler, it is made public.
     * Do not use this method yourself.
     */
    int _openIteratorCursor(dbCursor< dbTimeSeriesBlock<T> >& cursor, oid_t oid, time_t from, time_t till) 
    { 
        Interval interval;
        interval.from = generateBlockId(oid, from == 0 ? 0 : from - maxBlockTimeInterval);
        interval.till = generateBlockId(oid, till);
        return cursor.select(selectBlock, dbCursorViewOnly, &interval);
    }

   private:
     db_int8 generateBlockId(oid_t oid, time_t date) 
     {
        return cons_int8(oid, date);
     }
     
     
     void addNewBlock(oid_t oid, T const& data)
     {
         dbTimeSeriesBlock<T> block;
         block.blockId = generateBlockId(oid, data.time());
         block.elements.resize(minElementsInBlock);
         block.used = 1;
         block.elements.putat(0, data);
         insert(block);
     }

     void insertInBlock(oid_t oid, dbCursor< dbTimeSeriesBlock<T> >& blocks, T const& data)
     {
         time_t t = data.time();
         int i, n = blocks->used;

         T const* e =  blocks->elements.get();
         int l = 0, r = n;
         while (l < r)  {
             i = (l+r) >> 1;
             if (t > e[i].time()) { 
                 l = i+1;
             } else { 
                 r = i;
             }
         }
         assert(l == r && (l == n || e[l].time() >= t));
         if (r == 0) { 
             if (e[n-1].time() - t > maxBlockTimeInterval || n == maxElementsInBlock) { 
                 addNewBlock(oid, data);
                 return;
             }
             blocks->blockId = generateBlockId(oid, t);
         } else if (r == n) {
             if (t - e[0].time() > maxBlockTimeInterval || n == maxElementsInBlock) { 
                 addNewBlock(oid, data);
                 return;
             } 
         }
         if ((size_t)n == blocks->elements.length()) { 
             if (n == maxElementsInBlock) { 
                 T* u = blocks->elements.update();
                 addNewBlock(oid, u[n-1]);
                 for (i = n; --i > r; ) { 
                     u[i] = u[i-1];
                 }
                 u[r] = data;
                 blocks.update();
                 return;
             }
             blocks->elements.resize(n + minElementsInBlock < maxElementsInBlock ? n + minElementsInBlock : maxElementsInBlock);
         }
         T* u = blocks->elements.update();
         for (i = n; i > r; i--) { 
             u[i] = u[i-1];
         }
         u[r] = data;
         blocks->used += 1;
         blocks.update();
     }

     dbDatabase& db;
     int         maxElementsInBlock;
     int         minElementsInBlock;
     time_t      maxBlockTimeInterval;     
     dbQuery     selectBlock;
     dbQuery     selectBlockReverse; 
};
    

/**
 * Time series forward iterator 
 */
template<class T>
class dbTimeSeriesIterator { 
  public:
    /**
     * Start iteration through elements belonging to the specified range.
     * @param processor pointer to time series processor
     * @param oid time series identifer (OID of the object associated with this time series)
     * @param from inclusive low bound for element timestamp (set 0 to disable this criteria)
     * @param till inclusive high bound for element timestamp (set INFINITE_TIME to disable this criteria)
     */
    void start(dbTimeSeriesProcessor<T>* processor, oid_t oid, time_t from, time_t till) { 
        first = pos = -1;
        this->till = till;
        if (processor->_openIteratorCursor(blocks, oid, from, till)) { 
            do { 
                int n = blocks->used;
                T const* e =  blocks->elements.get();
                int l = 0, r = n;
                while (l < r)  {
                    int i = (l+r) >> 1;
                    if (from > e[i].time()) { 
                        l = i+1;
                    } else { 
                        r = i;
                    }
                }
                assert(l == r && (l == n || e[l].time() >= from)); 
                if (l < n) { 
                    if (e[l].time() <= till) {
                        first = pos = l;
                    }
                    return;
                }
            } while (blocks.next());
        }        
    }
            
    /**
     * Get current iterator element
     * @return <code>true</code> if there is current element, <code>false</code> otherwise
     */
    bool current(T& elem) { 
        if (pos >= 0) { 
            elem = blocks->elements[pos];
            return true;
        }
        return false;
    }
    
    /**
     * Move iterator position to next element.
     * @return <code>true</code> if next element exists, <code>false</code> otherwise
     */
    bool next() { 
        if (pos >= 0) { 
            if (++pos == blocks->used) { 
                if (!blocks.next()) { 
                    pos = -1;
                    return false;
                }
                pos = 0;
            }
            if (blocks->elements[pos].time() <= till) {
                return true;
            }
            pos = -1;
        }
        return false;
    }

    /**
     * Reset iterator to the initial state
     */
    void reset() { 
        blocks.first();
        pos = first;
    }
    
    /**
     * Iterator costructor. If current() or next() method will always return false if
     * them are invoked prior to start()
     */
    dbTimeSeriesIterator() {
        first = pos = -1;
    }
  private:
    dbCursor< dbTimeSeriesBlock<T> > blocks;
    int                              pos;
    int                              first;
    time_t                           till;
};
    
/**
 * Time series reverse iterator 
 */
template<class T>
class dbTimeSeriesReverseIterator { 
  public:
    /**
     * Start iteration through elements belonging to the specified range.
     * @param processor pointer to time series processor
     * @param oid time series identifer (OID of the object associated with this time series)
     * @param from inclusive low bound for element timestamp (set 0 to disable this criteria)
     * @param till inclusive high bound for element timestamp (set INFINITE_TIME to disable this criteria)
     */
    void start(dbTimeSeriesProcessor<T>* processor, oid_t oid, time_t from, time_t till) { 
        last = pos = -1;
        this->from = from;
        if (processor->_openIteratorCursor(blocks, oid, from, till)) { 
            do { 
                int n = blocks->used;
                blocks.last();
                T const* e =  blocks->elements.get();
                int l = 0, r = n;
                while (l < r)  {
                    int i = (l+r) >> 1;
                    if (till >= e[i].time()) { 
                        l = i+1;
                    } else { 
                        r = i;
                    }
                }
                assert(l == r && (l == n || e[l].time() > till)); 
                if (l > 0) {
                    if (e[l-1].time() >= from) {
                        last = pos = l-1;
                    }
                    return;
                }
            } while (blocks.prev());
        }        
    }
            
    /**
     * Get current iterator element
     * @return <code>true</code> if there is current element, <code>false</code> otherwise
     */
    bool current(T& elem) { 
        if (pos >= 0) { 
            elem = blocks->elements[pos];
            return true;
        }
        return false;
    }
    
    /**
     * Move iterator position to next element.
     * @return <code>true</code> if next element exists, <code>false</code> otherwise
     */
    bool next() { 
        if (pos >= 0) { 
            if (--pos < 0) {
                if (!blocks.prev()) { 
                    return false;
                }
                pos = blocks->used-1;
            }
            if (blocks->elements[pos].time() >= from) {
                return true;
            }
            pos = -1;
        }
        return false;
    }

    /**
     * Reset iterator to the initial state
     */
    void reset() { 
        blocks.last();
        pos = last;
    }
    
    /**
     * Iterator costructor. If current() or next() method will always return false if
     * them are invoked prior to start()
     */
    dbTimeSeriesReverseIterator() {
        last = pos = -1;
    }
  private:
    dbCursor< dbTimeSeriesBlock<T> > blocks;
    int                              pos;
    int                              last;
    time_t                           from;
};
    
END_FASTDB_NAMESPACE

#endif