CPView.h

Go to the documentation of this file.

#ifndef CPVIEW_H
#define CPVIEW_H

#include "TupleQueue.h"
#include <Stream.h>
#include <Expression.h>
#include "CPViewDescription.h"

BOREALIS_NAMESPACE_BEGIN;

class CPView {

  friend class CP;

 public:

  //Constructors
  CPView(ptr<CPView> base_view, ptr<CP> base_cp, string new_view_id);
  CPView(ptr<CPViewDescription> desc, ptr<CP> base_cp, string new_view_id, TupleDescription tuple_desc);
     
  //Destructor
  ~CPView();

  //Getter for unique id
  string getID();

  //Getter for description
  ptr<CPViewDescription> getDescription();

  //returns all unseen tuples packaged into a StreamEvent object
  vector<char*> read(); 

  //returns all tuples available on that CPView
  vector<char*> replayAll();  

  //returns all tuples of this CPView starting from t 
  vector<char*> replaySince(char* t);

  //moves man_start_ptr of this CPView to point to tuple t  
  void trimStorage(char* t);  


  // ------------- DIFFERENT WAYS TO REPLAY FROM CPVIEW ----------------

  //places on the corresponding queue tuples between t1 and t2, appended with id
  void replay(char* t1, char* t2, bool includeRevisions, int id);    

  // replays tuples between t1 and t2 with key for tuples given
  void replay(char* t1, char* t2, vector<ptr<Expression> > keys);  

  // replays tuple with the given tuple_key
  void replay(string tuple_key, vector<ptr<Expression> > keys, int revision_id);   

  // replay with given tuple key and change revision_id in replayed tuples
  void replay(int tuple_id, int revision_id);
  
  // replays tuples from start_order_by to end_order_by
  void replay(int start_order_by, int end_order_by, int revision_id);

  // replays tuples from start_order_by to end_order_by 
  // includes a tuple with fake_flag = true to signal end of replay
  void replayWithEndMarker(int start_order_by, int end_order_by, int revision_id);

  // make a fake (fake_flag=true) HISTORICAL tuple with revision_id
  char* makeFakeTuple(int revision_id); 
  
  // ------------------------------------


  //places on the corresponding queue deletion tuples for tuples between t1 and t2
  void undo(char* t1, char* t2); 

  //when partitioning happens on an arc that this view is on, this command initializes 
  //manual start/end pointers and makes this view manula type
  void makeBoundary();

  // is called by CP on all its views every time CP recieves new tuples to update 
  //start/end pointers according to spec and current state of stream/CP
  void updatePtrs(char* last_tuple);

  void updateAutoStartPtr(char* last_tuple); 
  void updateAutoEndPtr(char* last_tuple);


  void initAutoStartPtr();
  void initAutoEndPtr();
 
  //checks if tuple pointed to by p is within scope of this CPView (>= first CPView's tuple)
  bool checkFirst(char* ptr);
  //checks if tuple pointed to by p is within scope of this CPView (<= first CPView's tuple)
  bool checkLast(char* ptr);
  //returns true if order_by value of tuple pointed to by p1is >= to value of p2's tuple  
  bool isLater(char* p1, char* p2);

  void setTupleQueue(ptr<TupleQueue> tq);
  ptr<TupleQueue> getTupleQueue() { return _tq; }

  ptr<Expression> getOrderByField() {
    return _order_by_field;
  } 

 private:

  // alter the tuple : change type to historical and change revision_id to given
  vector<char*> modifyMultipleTuples(vector<char*> tuples, int tuple_size, TupleType type, int revision_id, bool revisions);
  //  char* modifyTuple(char *tuple, int tuple_size, int type, int revision_id);


  enum ViewType {AUTO, MANUAL, BOTH};
 
  ViewType _view_type;
  char* _man_start_ptr;
  char* _man_end_ptr;
  char* _auto_start_ptr;
  char* _auto_end_ptr;

  ptr<CP> _cp;
  ptr<TupleQueue> _tq;
  string _view_id;
  CPViewDescription _view_desc;
  ptr<Expression> _order_by_field;

  TupleDescription _tuple_desc;   // tuple description of the tuple on cpview's stream

  CPViewDescription::CPViewType _cp_view_type;
  CPViewDescription::OrderByMethod _order_by_method;

};

BOREALIS_NAMESPACE_END;

#endif

---