Aggregate.h

Go to the documentation of this file.

#ifndef AGGREGATE_H
#define AGGREGATE_H

#include "TupleDescription.h"
#include "Registry.h"
#include "Expression.h"

using namespace Borealis;

class Aggregate {
  public:
    Aggregate() : _out_size(0) {}
    virtual ~Aggregate() {}
        
    class Window {
      public:
#ifdef ENABLE_SERIALIZATION
      Window() {}
#endif
        virtual ~Window() {}

        virtual void init() {};
        virtual void incr(const EvalContext& ctxt) throw (EvalException) {};
        virtual void final(char *out_data) {};

        virtual void remove(const EvalContext& ctxt) throw (EvalException) {};

        const ptr<Aggregate> getAggregate() { return _agg; }
        void setAggregate(ptr<Aggregate> agg) { 
      //      _agg = &agg; 
      _agg = agg;
    }

#ifdef ENABLE_SERIALIZATION
    NMSTL_SERIAL_BASE(Window, long, << _agg);
    /*
    void freeze(OSerial &out) const {
      out << _agg;
    }
    void unfreeze(ISerial &in) {
      in >> _agg;
    }
    */
#endif

      protected:
        unsigned int numArgs() { return _agg->numArgs(); }
        Expression& arg(unsigned int index) { return _agg->arg(index); }

      private:
        //Aggregate *_agg;
    ptr<Aggregate> _agg;
    };

    typedef ::Registry<Aggregate> Registry;

    // Create a new window
    virtual Window *openWindow() { return new Window(); };

    // Figure out the output TupleDescription based on args; call
    // setTupleDescription
    virtual void setup(const ExprContext& ctxt)
        throw (AuroraException) {};

    void setArgs(const vector<ptr<Expression> >& args) { _args = args; }

    int numOutputFields() const { return _out_fields.size(); }
    TupleDescription::Field getOutputField(unsigned int i) const {
        ASSERT(i < _out_fields.size());
        return _out_fields[i];
    }
    unsigned int getOutputSize() const { return _out_size; }
    TupleDescription getOutputTupleDescription() const {
        return TupleDescription(_out_fields.begin(), _out_fields.end());
    }

    virtual const string as_string() const {
        return "Aggregate(out:" + to_string(_out_fields)
            + " args:" + to_string(_args) + ")";
    }

  public:
    static ptr<Aggregate> parse(string expr, const ExprContext& ctxt)
        throw (ExprException);

  protected:
    // Description of aggregate outputs; to be called by subclass
    // constructor
    void addOutputField(TupleDescription::Field field) {
        _out_fields.push_back(field);
        _out_size += field.getSize();
    }

    void addInt32OutputField( string  name = string() )
    {   addOutputField( Field<int32>( name )); }

    void addInt64OutputField( string  name = string() )
    {   addOutputField( Field<int64>( name )); }

    void addSingleOutputField( string  name = string() )
    {   addOutputField( Field<single>( name )); }

    void addDoubleOutputField( string  name = string() )
    {   addOutputField( Field<double>( name )); }

    void addStringOutputField( unsigned int  length )
    {   addOutputField( Field<string>( length )); }

    void addStringOutputField( string  name, unsigned int  length )
    {   addOutputField( Field<string>( name, length )); }

    Expression  &arg( unsigned int  index ) const
    {
        ASSERT( index < _args.size() );
        return( *_args[ index ]);
    }

    unsigned int numArgs() const
    {
        return( _args.size() );
    }


  private:
    vector<TupleDescription::Field> _out_fields;
    unsigned int _out_size;

    vector<ptr<Expression> > _args;

 public:
#ifdef ENABLE_SERIALIZATION
    NMSTL_SERIAL_BASE(Aggregate, long, << _out_fields << _out_size); // << _args);
    /*
    void freeze(OSerial &out) const {
      out << _out_fields << _out_size;
    }
    void unfreeze(ISerial &in) {
      in >> _out_fields >> _out_size;
    }
    */
#endif

};

#define AURORA_DECLARE_AGGREGATE(ClassName) AURORA_DECLARE_REG_CLASS(Aggregate, ClassName)
#define AURORA_DEFINE_AGGREGATE(ClassName, FuncName) \
    AURORA_DEFINE_REG_CLASS_WITH_KEY(Aggregate, ClassName, FuncName)

#endif

---