MeanAQPIAWJOperator.h

Go to the documentation of this file.

 
//
// Created by tony on 03/12/22.
//
 
#ifndef INTELLISTREAM_INCLUDE_OPERATOR_MEANAQPMeanAQPIAWJOPERATOR_H_
#define INTELLISTREAM_INCLUDE_OPERATOR_MEANAQPMeanAQPIAWJOPERATOR_H_
#include <Operator/AbstractOperator.h>
#include <Common/Window.h>
#include <atomic>
#include <WaterMarker/LatenessWM.h>
#include <Common/StateOfKey.h>
namespace OoOJoin {
 
class MeanAQPIAWJOperator : public AbstractOperator {
 protected:
  Window myWindow;
  size_t intermediateResult = 0;
  size_t confirmedResult = 0;
  uint64_t windowBound = 0;
  // double alphaArrivalRate=0.125;
  double alphaArrivalSkew = 0.125;
  double betaArrivalSkew = 0.25;
//  tsType lastTimeS = 0, lastTimeR = 0;
  double aqpScale = 0.1;
  void conductComputation();
  atomic_bool lockedByWaterMark = false;
  AbstractWaterMarkerPtr wmGen = nullptr;
  StateOfKeyHashTablePtr stateOfKeyTableR, stateOfKeyTableS;
  tsType lastTimeOfR = 0;
  tsType timeBreakDown_index;
  tsType timeBreakDown_prediction;
  tsType timeBreakDown_join;
  class MeanStateOfKey : public AbstractStateOfKey {
   public:
    size_t arrivedTupleCnt = 0;
    double arrivalSkew = 0, sigmaArrivalSkew = 0;
    TrackTuplePtr lastEventTuple = nullptr, lastArrivalTuple = nullptr;
    // tsType  lastSeenTime=0;
    MeanStateOfKey() {}
    ~MeanStateOfKey() {}
  };
  typedef std::shared_ptr<MeanStateOfKey> MeanStateOfKeyPtr;
#define newMeanStateOfKey std::make_shared<MeanStateOfKey>
  void updateStateOfKey(MeanStateOfKeyPtr sk, TrackTuplePtr tp);
  // void updateStateOfKeyR(MeanStateOfKeyPtr sk,TrackTuplePtr tp);
  void lazyComputeOfAQP();
  double predictUnarrivedTuples(MeanStateOfKeyPtr px);
 public:
  MeanAQPIAWJOperator() {}
  ~MeanAQPIAWJOperator() {}
  virtual bool setConfig(ConfigMapPtr cfg);
  virtual bool feedTupleS(TrackTuplePtr ts);
 
  virtual bool feedTupleR(TrackTuplePtr tr);
 
  virtual bool start();
 
  virtual bool stop();
 
  virtual size_t getResult();
 
  virtual size_t getAQPResult();
  virtual ConfigMapPtr getTimeBreakDown();
};
typedef std::shared_ptr<class MeanAQPIAWJOperator> MeanAQPIAWJOperatorPtr;
#define newMeanAQPIAWJOperator std::make_shared<OoOJoin::MeanAQPIAWJOperator>
}
#endif //INTELLISTREAM_INCLUDE_OPERATOR_MEANAQPMeanAQPIAWJOPERATOR_H_