RevisionAggregateQBox.h

Go to the documentation of this file.

#ifndef REVISION_AGG_QBOX_H
#define REVISION_AGG_QBOX_H

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

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

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

BOREALIS_NAMESPACE_BEGIN;

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

#define BEFORE_INDEX 0
#define AFTER_INDEX  1

class RevisionAggregateQBox : 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);
    // returns true if the tuple is to be put in this window
    virtual bool window_affected(const EvalContext& ctxt) { return false; }
    // removes the tuple from the window
    void remove_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;
    string final_value_string() 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 bool window_affected(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 bool window_affected(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);



  /****** Added on this box *******/
  int _revision_id_counter;
  map<int, const char*> _revision_tuple_store;
  // the very first window value, since this value determines values for all windows
  int _first_window_start;

  struct RevisionStates {
    bool initialized;
    int revision_id;
    bool revision_processed;
    WindowHash before_states;
    WindowHash after_states;
    hash_map<string, WindowStart, HashString> before_wstarts;
    hash_map<string, WindowStart, HashString> after_wstarts;
    hash_map<string, WindowStart, HashString> wstarts;

    RevisionStates() : initialized(false), revision_processed(false) {}
    
    string as_string() const {
      return string() + "RevisionStates{" + 
        "revision-id=" + revision_id 
        + "," + before_states.size() + " before_states" + 
        + "," + after_states.size() + " after_states" +
        + "," + before_wstarts.size() + " before_wstarts" +
        + "," + after_wstarts.size() + " after_wstarts" +
        "}";
    }
  };
      
  map<int, RevisionStates> _revision_states;

  // process revision tuple
  // returns true when revision tuple is outputted
  bool processRevision(Tuple revision_tuple, Tuple historical_tuple, int revision_id);
  // returns a pair tid->concatenated group-by, order-by, agg values
  pair<int, string> produceCoreTuple(const ptr<Window> w);

  bool checkAndEmitRevision(bool before_emitp, bool after_emitp, ptr<Window> before_w, ptr<Window> after_w);
  void emitRevision(const ptr<Window> w, TupleType revision_type);
  /*******************************/

 protected:
  void setupImpl() throw (AuroraException);
  void initImpl() throw (AuroraException);
  void runImpl(QBoxInvocation&) throw (AuroraException);
    
  AURORA_DECLARE_QBOX(RevisionAggregateQBox, "revisionagg");

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

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

  for (RevisionAggregateQBox::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;
}

NMSTL_SERIAL_DEFINE(RevisionAggregateQBox::Window,                1100);
NMSTL_SERIAL_DEFINE(RevisionAggregateQBox::ValueWindow,           1200);
NMSTL_SERIAL_DEFINE(RevisionAggregateQBox::TupleWindow,           1300);


BOREALIS_NAMESPACE_END;

#endif

---