AggregateQBox.h

Go to the documentation of this file.

#ifndef SIMPLEAGG_QBOX_H
#define SIMPLEAGG_QBOX_H

#include <ext/hash_map>
#include <vector>
#include <list>

#include "QBox.h"
#include "Aggregate.h"
#include "Timestamp.h"

#include <NMSTL/serial>
#include <NMSTL/iserial>
#include <NMSTL/oserial>

BOREALIS_NAMESPACE_BEGIN;

using __gnu_cxx::hash_map;
using NMSTL::to_string;

class AggregateQBox : public QBox
{
  friend class AggregateBoxState;

  private:

    struct HashString
    {
        size_t operator()(const string s) const {
            // TODO: Find a better hash function or something.
            uint32  h = 0;
            for (string::const_iterator i = s.begin() ; i != s.end(); ++i)
                h = (5 * h) + *i;
            return size_t(h);
        }

        NMSTL_TO_STRING(HashString);
        NMSTL_SIMPLY_SERIALIZABLE(HashString, );
    };


    typedef vector<ptr<Aggregate> > AFVector;
    AFVector _afs;

    // Size of the input tuples
    size_t _in_tuple_size;

    // Size of the group_by values.
    size_t _group_by_value_size;
    
    // Window will contain the last window_size values of
    // window_field, advancing by advance_size values
    // when the window fills.
    enum WindowMethod {
        WINDOW_BY_TUPLES,
        WINDOW_BY_VALUES
    };
    WindowMethod _window_method;
    int _window_size;
    int _advance_size;
    // If WINDOW_BY_VALUES
    ptr<Expression> _window_field;

    // Keep track of when the next window starts for each group by
    // value.
    // TODO: Consider tossing this in with the WindowList.
    struct WindowStart {
        bool initialized;
        int next;
        WindowStart() : initialized(false) { }

        NMSTL_TO_STRING(WindowStart);
        NMSTL_SIMPLY_SERIALIZABLE(WindowStart, << initialized << next);

    };
    hash_map<string, WindowStart, HashString> _window_starts;

    enum OrderByMethod {
        ORDER_BY_FIELD,
        ORDER_BY_TUPLENUM
    };
    OrderByMethod _order_by_method;

    EvalContext _ctxt;

    // Group by a set of attributes.
    vector<ptr<Expression> > _group_by_fields;

    class Window;
    typedef list<ptr<Window> > WindowList;

    // Abstract class for structure that stores a Window
    // Manages the state of the Aggregate Functions.
    class Window {
      public:
        Window(const AFVector &af, const string &group_by) :
            _windows(),
            _group_by(group_by), _initialized(false)
        {
            // Open up windows.
            AFVector::const_iterator i;
            for (i = af.begin(); i != af.end(); ++i) {
                ptr<Aggregate::Window>w((*i)->openWindow());
                ptr<Aggregate>agg_ptr(*i);
                //w->setAggregate(**i);
                w->setAggregate(agg_ptr);
                w->init();
                _windows.push_back(w);
            }
        };
    
        Window() {}
        virtual ~Window() {};

        // Return true if this window should emit now.
        virtual bool insert_tuple(const EvalContext& ctxt);

        virtual const int getStartValue() const { return -999; };

        const Timestamp t0() const { return _t0; }
        const Timestamp lt0() const { return _lt0; }
        const int tid() const { return _tid; }

        const string &group_by_value() const { return _group_by; }

        size_t final_value(char *pos) const;

        virtual string as_string() const { return NULL; };

        void set_timeout_i(WindowList::iterator i) {_timeout_i = i; }
        void set_open_windows_i(WindowList::iterator i) {_ow_i = i; }
        WindowList::iterator get_timeout_i() { return _timeout_i; }
        WindowList::iterator get_open_windows_i() { return _ow_i; }

    // set _window_field for ValueWindow
    // here because of serialization problems with Expression
    virtual void set_window_field(ptr<Expression> window_field) {}

      private:
        vector<ptr<Aggregate::Window> > _windows;
        string _group_by;
        bool _initialized;
        Timestamp _t0;
        Timestamp _lt0;
        int _tid;
        WindowList::iterator _timeout_i, _ow_i;

    public:
        NMSTL_TO_STRING(Window);
        NMSTL_SERIAL_BASE(Window, int32, 
                          << _windows  << _group_by  
                          << _initialized << _t0
                          << _lt0 << _tid);


    };

    // Structure that stores a Window by values
    class ValueWindow : public Window {
      public:
        ValueWindow(const AFVector &af, const string &group_by,
                    ptr<Expression> window_field,
                    int window_start, int window_end,
                    int slack) :
            Window(af, group_by), _window_field(window_field),
            _first_val(window_start), _last_val(window_end),
            _slack_remaining(slack)
        {}
        
        ValueWindow() {};

        virtual bool insert_tuple(const EvalContext& ctxt);
        
        virtual const int getStartValue() const {
            return _first_val; 
        }
    
        void set_window_field(ptr<Expression> window_field) {
            _window_field = ptr<Expression>(window_field);
        }
    
        virtual string as_string() const {
            return ("ValueWindow{[" + to_string(_first_val) + ", " + 
                    to_string(_last_val) + ") }");
        }

      private:
        ptr<Expression> _window_field;
        // Window runs from _first_val (inclusive) to _last_val (exclusive).
        int _first_val, _last_val;
        int _slack_remaining;

    public:
        NMSTL_TO_STRING(ValueWindow);
        NMSTL_SERIAL_SUBCLASS(ValueWindow, Window,
                              << _first_val << _last_val << _slack_remaining);
    };

    // Structure that stores a Window by tuples
    class TupleWindow : public Window {
      public:
        TupleWindow(const AFVector &af, const string group_by,
                    int window_size, int start_value) :
            Window(af, group_by), _window_size(window_size),
            _count(0), _start_value(start_value) {};

    TupleWindow() {};

        virtual bool insert_tuple(const EvalContext& ctxt);
        virtual const int getStartValue() const { return _start_value; }

        int size() const { return _count; };

        virtual string as_string() const {
            return ("TupleWindow{[" + to_string(_count) + " out of "
                    + to_string(_window_size) + " tuples] }");
        }

      private:
        int _window_size;
        int _count;
        int _start_value;

    public:
        NMSTL_TO_STRING(TupleWindow);
        NMSTL_SERIAL_SUBCLASS(TupleWindow, Window, 
                              << _window_size << _count << _start_value);

    };

    // Map of group-by values to lists of open windows.
    typedef hash_map<string, WindowList, HashString> WindowHash;
    WindowHash _open_windows;

    // Slack stuff
    int _slack;

    // Timeout stuff.
    bool _timing_out;           // True if we are doing timeout.
    Timestamp _timeout;         // Timeout in seconds and useconds.
    WindowList _timeout_list;   // List of windows, in the order they
                                // will be timed-out.

    // Private functions.
    void setupParameters();
    TupleDescription createOutputDescription();
    void parseGroupBy(const string atts, const ExprContext& ctxt);
    void emitForWindow(const ptr<Window> w);
    string groupByForTuple(const EvalContext& ctxt) const;
    void insert_new_window(WindowList &wl, ptr<Window> w);


  protected:
    void setupImpl() throw (AuroraException);
    void initImpl() throw (AuroraException);
    void runImpl(QBoxInvocation&) throw (AuroraException);
    
    // packing and unpacking states
    void setPendingBoxState(ptr<AbstractBoxState> packed_box);
    ptr<AbstractBoxState> packState();
    void unpackState(ptr<AbstractBoxState> box_state);

    AURORA_DECLARE_QBOX(AggregateQBox, "simpleagg");

    friend string to_string(const WindowHash& m);
};

inline string to_string(const AggregateQBox::WindowHash& m) {
    string out = "WindowHash{";

    for (AggregateQBox::WindowHash::const_iterator i = m.begin(); i != m.end(); ++i) {
        if (i != m.begin()) out.append("; ");
        out.append(to_escaped_string(i->first)).append("->").append(to_string(i->second));
    }

    out.append("}");
    return out;
}


class AggregateBoxState : public AbstractBoxState
{

 public:
  // Keep track of when the next window starts for each group by value.
  typedef hash_map<string, AggregateQBox::WindowStart, AggregateQBox::HashString> WindowStarts;
  // List of windows, in the order they will be timed-out.
  typedef AggregateQBox::WindowList WindowList;
  // hash_map of group-by values to lists of open windows
  typedef AggregateQBox::WindowHash WindowHash;


  AggregateBoxState(WindowStarts window_starts, WindowList timeout_list, WindowHash open_windows) 
    {
      /*
       NMSTL OSerial can easily serialize vector, map, set
       serializing and deserializing hash_maps and lists are difficult.
         - need to write to_string methods to serialize
         - DON'T KNOW WHAT METHODS TO WRITE TO DESERIALIZE
       So the solution I (Anurag) came up with 
         - convert hash_map to map to serialize
     - convert list to vector
     - vice-versa to deserialize
       IF SOMEONE KNOWS A BETTER WAY - LEMME KNOW
      */
       
      // convert hash_map window_starts to map
      map<string, AggregateQBox::WindowStart> window_starts_map;
      for (WindowStarts::iterator iter = window_starts.begin(); iter != window_starts.end(); ++iter) {
    window_starts_map[iter->first] = iter->second;
      }
      DEBUG << "Original WindowStarts hash_map size = " << window_starts.size() << " || "
        << "Copied WindowStarts map size = " << window_starts_map.size();
    
      // convert list timeout_list to vector
      vector<ptr<AggregateQBox::Window> > timeout_list_vector;
      for (WindowList::iterator iter = timeout_list.begin(); iter != timeout_list.end(); ++iter) {
    timeout_list_vector.push_back(*iter);
      }
      DEBUG << "Original WindowList list size = " << timeout_list.size() << " || "
        << "Copied WindowList vector size = " << timeout_list_vector.size();
      
      // convert hash_map open_windows to map
      map<string, vector<ptr<AggregateQBox::Window> > > open_windows_map;
      for (WindowHash::iterator iter = open_windows.begin(); iter != open_windows.end(); ++iter) {
    WindowList temp_list = iter->second;
    vector<ptr<AggregateQBox::Window> > temp_vector;
    // change the list to vector
    for (WindowList::iterator in_iter = temp_list.begin(); in_iter != temp_list.end(); ++in_iter) {
      temp_vector.push_back(*in_iter);
    }
    open_windows_map[iter->first] = temp_vector;
      }
      DEBUG << "Original WindowHash hash_map size = " << open_windows.size() << " || "
        << "Copied WindowHash map size = " << open_windows_map.size();

      OSerialString serial_box;
      serial_box << window_starts_map 
             << timeout_list_vector
         << open_windows_map;
      _serialized_state = serial_box.str();

      DEBUG << _serialized_state;
    }

  AggregateBoxState() {}
  ~AggregateBoxState() {}

  string as_string() const {
    string out("AggregateQBoxState{");
    out += _serialized_state;
    out += "}";
    return out;
  }
  
  NMSTL_SERIAL_SUBCLASS(AggregateBoxState, AbstractBoxState, );


};

//NMSTL_TO_STRING(AggregateBoxState);

NMSTL_SERIAL_DEFINE(AggregateQBox::Window,                1100);
NMSTL_SERIAL_DEFINE(AggregateQBox::ValueWindow,           1200);
NMSTL_SERIAL_DEFINE(AggregateQBox::TupleWindow,           1300);
NMSTL_SERIAL_DEFINE(AggregateBoxState,     1000);

BOREALIS_NAMESPACE_END;

#endif

---