JoinQBox.h

Go to the documentation of this file.

#ifndef JOIN_QBOX_H
#define JOIN_QBOX_H

#include "QBox.h"
#include "Tuple.h"
#include <list>

BOREALIS_NAMESPACE_BEGIN;

class JoinQBox : public QBox
{

    friend class JoinBoxState;

    // SJoinQBox is a kind of JoinBox. It manipulates the _buffer directly 
    friend class SJoinQBox; 

  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);

  private:
    // TODO: Consider tpeseafe enumeration.
    static const int LEFT;
    static const int RIGHT;

    string tupleToString(int stream_id, const void *data)
    {
        return getInputDescription( stream_id ).tupleToString( data );
    }

    void runStream( QBoxInvocation  &inv, int  stream_id, bool  &outputted );

    void emitTuple( EvalContext  &ctxt );

    void produceDontEmitTuple( EvalContext  &ctxt );

    bool       _timing_out;
    Timestamp  _timeout;
    
    // Structure for storing a tuple in the buffer.
    class BufferElement
    {
      public:
        const void *getData() const
        {
            //return _tuple.getData();
            return _data.data();
        }

        const Tuple &getTuple() const
        {   
            if (_tuple.getData() != _data.data())
            { FATAL << "Tuple points to wrong data";
            }
            return _tuple;
        }

        Timestamp getTimestamp() const
        {   return _tuple.getTimestamp();
        }

        int getFieldValue() const
        {   return _field_value;
        }

        BufferElement( const void  *data,
                           size_t   length,
                              int   field_value = 0 ) :

            _data( (const char *)data, length ),
            //_tuple(Tuple::factory( data )),  --> not data will go away!!! 
            _tuple(Tuple::factory( _data)),
            _field_value( field_value ) {}
            
      private:
        const string  _data;
        const Tuple   _tuple;
        const int     _field_value;
    };


    class JoinBuffer
    {
      public:
        JoinBuffer( int stream_id, JoinQBox *qbox ) :
            _stream_id( stream_id ),
            _qbox( qbox ),
            _tuple_size( _qbox->getInputDescription( _stream_id ).getSizeInBytes()),
            _ctxt( 2 )
        {}

        virtual ~JoinBuffer() {}

        virtual void insertTuple( const void  *tuple_data );

        void join( const void  *other_tuple,
                   Expression  *predicate,
                         bool  &outputted );

        void joinNoEmit( const void  *other_tuple,
                         Expression  *predicate,
                         bool  &outputted );

        void clear();

        int size() { return _buffer.size(); }

        int dump(dynbuf& dst); // Copy tuples to buffer and return nb tuples copied

        void init(const dynbuf& src, int nb_tuples); // Insert the tuples from this buffer into self

      protected:
        typedef list<BufferElement> BufferList;
        BufferList  _buffer;
        int         _stream_id;
        JoinQBox   *_qbox;

      private:
        size_t       _tuple_size;
        EvalContext  _ctxt;
    };

    friend class JoinBuffer;

    // Subclass for ORDER_BY_VALUES
    class TupleCountJoinBuffer : public JoinBuffer
    {
      public:
        TupleCountJoinBuffer(int stream_id, JoinQBox *qbox,
                             size_t buffer_size) :
            JoinBuffer(stream_id, qbox),
            _buffer_size(buffer_size)
        { }
        virtual void insertTuple(const void *tuple_data);
      private:
        size_t _buffer_size;
    };

    // Subclass for ORDER_BY_FIELD
    class FieldJoinBuffer : public JoinBuffer
    {
      public:
        FieldJoinBuffer(int stream_id, JoinQBox *qbox,
                        ptr<Expression> field,
                        size_t buffer_size) :
            JoinBuffer(stream_id, qbox), _field(field),
            _buffer_size(buffer_size)
        { 
            if (!_field->is<int32>())
                Throw (AuroraTypingException,
                       "Order by field value must be of type int.");
        }

        virtual void insertTuple(const void *tuple_data);
      private:
        int fieldValue(const void *tuple_data);
    
        ptr<Expression> _field;
        size_t _buffer_size;
        EvalContext _ctxt;
    };

    // Create "left" and "right" buffers.
    ptr<JoinBuffer> createBufferByParameter(int stream_id);

    // Left and right join buffers.
    ptr<JoinBuffer> _buffer[2];

    ptr<Expression> _predicate;

    // What kind of output.
    bool _just_concat;
    vector<ptr<Expression> > _output_expressions;

    // Cached information about tuples
    unsigned int _left_body_size, _right_body_size;

    AURORA_DECLARE_QBOX(JoinQBox, "join");
};



class JoinBoxState : public AbstractBoxState
{
 public:
  JoinBoxState() : _left_nb_tuples(0), _right_nb_tuples(0) {}
  JoinBoxState(const dynbuf& left, int nb_left, const dynbuf& right, int nb_right) :
      _left_buffer(left), _left_nb_tuples(nb_left), 
      _right_buffer(right), _right_nb_tuples(nb_right) 
      {
          // The copy constructor of a dynbuff makes another reference to
          // the other one's underlying storage. The basic dynbuf constructor
          // uses malloc so the dynbufs are always on the heap
          //memcpy(_left_buffer.data(),left->data(),left->max_length());
          //memcpy(_right_buffer.data(),right->data(),right->max_length());
      }
  ~JoinBoxState() {}

  NMSTL_SERIAL_SUBCLASS(JoinBoxState, AbstractBoxState, << _left_buffer << _left_nb_tuples 
                      << _right_buffer << _right_nb_tuples);

  const dynbuf& getLeft() const { return _left_buffer; }
  int getNbLeftTuples() const { return _left_nb_tuples; }

  const dynbuf& getRight() const { return _right_buffer; }
  int getNbRightTuples() const { return _right_nb_tuples; }
  

 private:
  dynbuf _left_buffer;
  int _left_nb_tuples;

  dynbuf _right_buffer;
  int _right_nb_tuples;

};


BOREALIS_NAMESPACE_END;

#endif // JOIN_QBOX_H

---