timeseries.h

fastdb-2.92的源码

//-< TIMESERIES.H >--------------------------------------------------*--------*
// FastDB                    Version 1.0         (c) 1999  GARRET    *     ?  *
// (Post Relational Database Management System)                      *   /\|  *
//                                                                   *  /  \  *
//                          Created:     22-Nov-2002  K.A. Knizhnik  * / [] \ *
//                          Last update: 22-Nov-2002  K.A. Knizhnik  * GARRET *
//-------------------------------------------------------------------*--------*
// Container for time serires data
//-------------------------------------------------------------------*--------*

#ifndef __TIMESERIES_H__
#define __TIMESERIES_H__

#include <time.h>
#include "fastdb.h"

#define INFINITE_TIME 0x7fffffff

/**
 * Time series block contaning array of elements. Grouping several elements in one block (record)
 * reduce space overhead and increase processing speed.<BR>
 * <B>Attention!</B> This class is not serialized, so it is can be accessed only by one thread<P>
 * <I>You are defining your own time series class, for example:</I>
 * <PRE>
 * class Stock {
 *   public:
 *     char const* name;
 *     TYPE_DESCRIPTOR((KEY(name, INDEXED)));
 * };
 * 
 * 
 * class Quote {
 *   public:
 *     int4        tickerDate;
 *     real4       bid;
 *     int4        bidSize;
 *     real4       ask;
 *     int4        askSize;
 * 
 *     time_t time() const { return tickerDate; } // this method should be defined 
 * 
 *     TYPE_DESCRIPTOR((FIELD(tickerDate), FIELD(bid), FIELD(bidSize), FIELD(ask), FIELD(askSize)));
 * };
 * typedef dbTimeSeriesBlock<Daily>  DailyBlock;
 * REGISTER(DailyBlock);
 * </PRE>    
 * <I>Now you can work with time series objects in the followin way:</I>
 * <PRE>
 * dbDatabase db;
 * if (db.open("mydatabase.dbs")) {
 *     dbTimeSeriesProcessor&lt;Quote&gt; proc(db, MIN_ELEMENTS_IN_BLOCK,MAX_ELEMENTS_IN_BLOCK);
 *     Quote quote;
 *     // initialize quote
 *     Stock stock;
 *     stock.name = "AAD";
 *     oid_t stockId = insert(oid).getOid();
 *     proc.add(stockId, quote); // add new element in time series
 * 
 *     Quote quoteBuf[MAX_QUOTES];
 *     // select quotes for the specified interval
 *     int n = proc.getInterval(stockId, fromDate, tillDate, quoteBuf, MAX_QUOTES);
 *     for (int i = 0; i < n; i++) {
 *         printf("bid=d ask=%d\n", quoteBuf[i].bid, quoteBuf[i].ask);
 *     }
 * }  
 * </PRE>
 */
template<class T>
class dbTimeSeriesBlock { 
  public:
    db_int8 blockId;
    db_int4 used;
    dbArray<T> elements;
  
    TYPE_DESCRIPTOR((KEY(blockId, INDEXED), FIELD(used), FIELD(elements)));
};


/**
 * Time series processor.<BR>
 * Element of time series can be arbitrary type with declared TYPE_DESCRIPTOR and defined
 * <code>time_t time()</code> method
 */
template<class T>
class dbTimeSeriesProcessor { 
    struct Interval { 
        db_int8 from;
        db_int8 till;
    };

  public:
    /**
     * Virtual method for processing elements, Should be redefinedin derived class.
     * @param data reference to the processed data element
     */
    virtual void process(T const&) {}
    
    /**
     * Add new element
     * @param oid time series identifer (OID of the object associated with this time series)
     * @param data reference to the inserted element
     */
    void add(oid_t oid, T const& data) 
    { 
        Interval interval;
        interval.from = generateBlockId(oid, data.time() - maxBlockTimeInterval);
        interval.till = generateBlockId(oid, data.time());
        dbCursor< dbTimeSeriesBlock<T> > blocks;
        blocks.select(selectBlock, dbCursorForUpdate, &interval);
        if (blocks.last()) { 
            insertInBlock(oid, blocks, data);
        } else { 
            addNewBlock(oid, data);
        }
    }

    /**
     * Process elements in the block belonging to the specified range
     * @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 select(oid_t oid, time_t from, time_t till) 
    { 
        Interval interval;
        interval.from = generateBlockId(oid, from - maxBlockTimeInterval);
        interval.till = generateBlockId(oid, till);
        dbCursor< dbTimeSeriesBlock<T> > blocks;
        if (blocks.select(selectBlock, dbCursorViewOnly, &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 (l < n && e[l].time() <= till) {
                    process(e[l++]);
                }
            } while (blocks.next());
        }
    }
    
    /**
     * Get the time of the first element in time series
     * @param oid time series identifer (OID of the object associated with this time series)
     * @return earliest time in times series or -1 if there are no elements in time series
     */
    time_t getFirstTime(oid_t oid) 
    {
        Interval interval;
        interval.from = generateBlockId(oid, 0);
        interval.till = generateBlockId(oid, INFINITE_TIME);
        dbCursor< dbTimeSeriesBlock<T> > blocks;
        blocks.setSelectionLimit(1);
        if (blocks.select(selectBlock, dbCursorViewOnly, &interval)) { 
            return blocks->elements[0].time();
        }
        return (time_t)-1;
    }
    
    /**
     * Get the time of the last element in time series
     * @param oid time series identifer (OID of the object associated with this time series)
     * @return latest time in times series or -1 if there are no elements in time series
     */
    time_t getLastTime(oid_t oid) 
    {
        Interval interval;
        interval.from = generateBlockId(oid, 0);
        interval.till = generateBlockId(oid, INFINITE_TIME);
        dbCursor< dbTimeSeriesBlock<T> > blocks;
        blocks.setSelectionLimit(1);
        if (blocks.select(selectBlockReverse, dbCursorViewOnly, &interval)) { 
            return blocks->elements[blocks->used-1].time();
        }
        return (time_t)-1;
    }
    
    /**
     * Get number of elements in time series.
     * @param oid time series identifer (OID of the object associated with this time series)
     * @return number of elements in time series.
     */
    size_t getNumberOfElements(oid_t oid) 
    {
        Interval interval;
        interval.from = generateBlockId(oid, 0);
        interval.till = generateBlockId(oid, INFINITE_TIME);
        dbCursor< dbTimeSeriesBlock<T> > blocks;
        int n = 0;
        if (blocks.select(selectBlock, dbCursorViewOnly, &interval)) {
            do { 
                n += blocks->used;
            } while (blocks.next());
        }
        return n;
    }
        
    /**
     * Select elements belonging to the specified interval
     * @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)
     * @param buf destination buffer for selected elements
     * @param bufSize size of buffer: up to bufSize elements will be placed in buffer
     * @return number of elements belonging to the specified interval (can be greater than bufSize)
     */
    size_t getInterval(oid_t oid, time_t from, time_t till, T* buf, size_t bufSize) 
    { 
        Interval interval;
        interval.from = generateBlockId(oid, from == 0 ? 0 : from - maxBlockTimeInterval);
        interval.till = generateBlockId(oid, till);
        dbCursor< dbTimeSeriesBlock<T> > blocks;
        size_t nSelected = 0;
        if (blocks.select(selectBlock, dbCursorViewOnly, &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 (l < n && e[l].time() <= till) {
                    if (nSelected < bufSize) { 
                        buf[nSelected] = e[l];
                    }
                    l += 1;
                    nSelected += 1;
                }
            } while (blocks.next());
        }
        return nSelected;
    }        
        
    /**
     * Get time series element with specified time
     * @param oid time series identifer (OID of the object associated with this time series)
     * @param elem reference to the extracted element 
     * @param t timestamp of extracted element
     * @return <code>true</code> if element with specifed times exists in time series
     */
    bool getElement(oid_t oid, T& elem, time_t t) 
    { 
        return getInterval(oid, t, t, &elem, 1) == 1;
    }        
        
    /**
     * Select first N elements of times series with timestamp less than or equal to specified
     * @param oid time series identifer (OID of the object associated with this time series)
     * @param till inclusive high bound for element timestamp (set INFINITE_TIME to disable this criteria) 
     * @param buf destination buffer for selected elements
     * @param bufSize size of buffer: up to bufSize elements will be placed in buffer
     * @return number of selected elements (can be less than bufSize if there are less elements in time series
     * with timestamp less or equal than specified, but can not be greater than bufSize)
     */
    size_t getFirstInterval(oid_t oid, time_t till, T* buf, size_t bufSize) 
    {
        if (bufSize == 0) { 
            return 0;
        }
        Interval interval;
        interval.from = generateBlockId(oid, 0);
        interval.till = generateBlockId(oid, till);
        dbCursor< dbTimeSeriesBlock<T> > blocks;
        size_t nSelected = 0;
        if (blocks.select(selectBlock, dbCursorViewOnly, &interval)) { 
            do { 
                int n = blocks->used;
                T const* e =  blocks->elements.get();
                for (int i = 0; i < n && e[i].time() <= till; i++) { 
                    buf[nSelected++] = e[i];
                    if (nSelected == bufSize) { 
                        return nSelected;
                    }
                }
            } while (blocks.next());
        }
        return nSelected;
    }        


    /**
     * Select last N elements of times series with timestamp greater than or equal to specified
     * @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 buf destination buffer for selected elements
     * @param bufSize size of buffer: up to bufSize elements will be placed in buffer
     * @return number of selected elements (can be less than bufSize if there are less elements in time series
     * with timestamp greater or equal than specified, but can not be greater than bufSize)
     */
    size_t getLastInterval(oid_t oid, time_t from, T* buf, size_t bufSize) 
    {
        if (bufSize == 0) { 
            return 0;
        }
        Interval interval;
        interval.from = generateBlockId(oid, from == 0 ? 0 : from - maxBlockTimeInterval);
        interval.till = generateBlockId(oid, INFINITE_TIME);
        dbCursor< dbTimeSeriesBlock<T> > blocks;

        size_t nSelected = 0;
        blocks.select(selectBlock, dbCursorViewOnly, &interval);
        if (blocks.last()) { 
            do { 
                int n = blocks->used;
                T const* e =  blocks->elements.get();
                for (int i = n; --i >= 0 && e[i].time() >= from;) { 
                    buf[nSelected++] = e[i];
                    if (nSelected == bufSize) { 
                        return nSelected;
                    }
                }
            } while (blocks.prev());
        }
        return nSelected;
    }        



    /**
     * Check if there is element for specified data in time series
     * @param oid time series identifer (OID of the object associated with this time series)
     * @param t timestamp of checked element
     * @return <code>true</code> if element with specifed times exists in time series
     */
    bool hasElement(oid_t oid, time_t t) 
    { 
        T dummy;
        return getElement(oid, dummy, t);
    }        

    /**
     * TimeSeries processor constructor
     * @param database reference to the database
     * @param minElementsInBlock preallocated number of the elements in the block: 
     * array with specified number of elements will be allocated for new block
     * @param maxElementsInBlock maximal number of the elements in the block: block will be splitten if it has maxElementsInBlock
     * elements and new is added to the block
     * @param maxBlockTimeInterval maximal interval between first and last element in the block, new block will be created if 
     * adding new element to the block cause violation of this assumption. If maxBlockTimeInterval is 0, then it is assigned