core/sql/executor/ExSequence.h

/**********************************************************************
// @@@ START COPYRIGHT @@@
//
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.
//
// @@@ END COPYRIGHT @@@
**********************************************************************/
#ifndef ExSequence_h
#define ExSequence_h

/* -*-C++-*-
******************************************************************************
*
* File:         ExSequence.h
* Description:  Class declarations for ExSequence. ExSequence performs
*               sequence functions, like running sums, averages, etc.
* Created:      
* Language:     C++
*
*
*
*
******************************************************************************
*/

// External forward declarations
//
class ExSimpleSQLBuffer;

// Task Definition Block
//
#include "ComTdbSequence.h"

#include "NAMemory.h"
#include "NABasicObject.h"
#include "ExpError.h"
#include "cluster.h" 

// -----------------------------------------------------------------------
// Classes defined in this file
// -----------------------------------------------------------------------
class ExSequenceTdb;
class ExSequenceTcb;
class ExSequencePrivateState;

// -----------------------------------------------------------------------
// Classes referenced in this file
// -----------------------------------------------------------------------
class ex_tcb;
class ex_tdb;
//

// -----------------------------------------------------------------------
// ExSequenceTdb
// -----------------------------------------------------------------------
class ExSequenceTdb : public ComTdbSequence
{
public:

  // ---------------------------------------------------------------------
  // Constructor is only called to instantiate an object used for
  // retrieval of the virtual table function pointer of the class while
  // unpacking. An empty constructor is enough.
  // ---------------------------------------------------------------------
  ExSequenceTdb()
  {};

  virtual ~ExSequenceTdb()
  {};

  // ---------------------------------------------------------------------
  // Build a TCB for this TDB. Redefined in the Executor project.
  // ---------------------------------------------------------------------
  virtual ex_tcb *build(ex_globals *globals);

private:
  // ---------------------------------------------------------------------
  // !!!!!!! IMPORTANT -- NO DATA MEMBERS ALLOWED IN EXECUTOR TDB !!!!!!!!
  // *********************************************************************
  // The Executor TDB's are only used for the sole purpose of providing a
  // way to supplement the Compiler TDB's (in comexe) with methods whose
  // implementation depends on Executor objects. This is done so as to
  // decouple the Compiler from linking in Executor objects unnecessarily.
  //
  // When a Compiler generated TDB arrives at the Executor, the same data
  // image is "cast" as an Executor TDB after unpacking. Therefore, it is
  // a requirement that a Compiler TDB has the same object layout as its
  // corresponding Executor TDB. As a result of this, all Executor TDB's
  // must have absolutely NO data members, but only member functions. So,
  // if you reach here with an intention to add data members to a TDB, ask
  // yourself two questions:
  //
  // 1. Are those data members Compiler-generated?
  //    If yes, put them in the ComTdbSequence instead.
  //    If no, they should probably belong to someplace else (like TCB).
  // 
  // 2. Are the classes those data members belong defined in the executor
  //    project?
  //    If your answer to both questions is yes, you might need to move
  //    the classes to the comexe project.
  // ---------------------------------------------------------------------
};

//
// Task control block
//
class ExSequenceTcb : public ex_tcb
{
  friend class   ExSequenceTdb;
  friend class   ExSequencePrivateState;
  
public:
  enum RequestState {
    ExSeq_EMPTY,
    ExSeq_WORKING_READ,
    ExSeq_WORKING_RETURN,
    ExSeq_CANCELLED,
    ExSeq_ERROR,
    ExSeq_DONE,
    ExSeq_OVERFLOW_WRITE,
    ExSeq_OVERFLOW_READ,
    ExSeq_END_OF_PARTITION
  };
  
  ExSequenceTcb(const ExSequenceTdb & seq_tdb,    
		const ex_tcb &    child_tcb,
		ex_globals * glob);
  ~ExSequenceTcb();  

  virtual void registerSubtasks();
  void freeResources();
  short work();
  
  inline ExSequenceTdb & myTdb() const;

  ex_queue_pair getParentQueue() const
  {
  return (qparent_);
  }

  inline ex_expr * sequenceExpr() const;
  inline ex_expr * returnExpr() const;
  inline ex_expr * postPred() const;
  inline ex_expr * cancelExpr() const;
  inline ex_expr * checkPartitionChangeExpr() const;

  inline ex_expr * moveExpr() const;

  friend char *GetHistoryRow(void *data, Int32 n,NABoolean leading,Lng32 winSize, Int32&);
  friend char *GetHistoryRowOLAP(void *data, Int32 n,NABoolean leading,Lng32 winSize, Int32&);
  friend char *GetHistoryRowFollowingOLAP(void *data, Int32 n,NABoolean leading,Lng32 winSize, Int32&);

  //inline char * GetFirstHistoryRowPtr();
//
  NABoolean advanceHistoryRow(NABoolean checkMemoryPressure = FALSE);
  //void unAdvanceHistoryRow();
  inline NABoolean isHistoryFull() const;
  inline NABoolean isHistoryEmpty() const;//if the history buffer is empty (i.e. histRowsToReturn_ == 0)
  inline NABoolean canReturnRows() const;
  inline Lng32 numFollowingRows() const;//number of rows following the current row
  void advanceReturnHistoryRow();
  //inline char * getCurrentRetHistRowPtr() 
  inline void updateHistRowsToReturn() ;
  void initializeHistory();
  void createCluster();

  NABoolean removeOLAPBuffer();
  NABoolean shrinkOLAPBufferList();

  inline NABoolean isOverflowStarted();
  inline NABoolean isUnboundedFollowing();

  inline NABoolean canAllocateOLAPBuffer();
  NABoolean addNewOLAPBuffer(NABoolean checkMemoryPressure = TRUE);

  void updateDiagsArea(ex_queue_entry * centry);
  void updateDiagsArea( ExeErrorCode rc_);

  NABoolean getPartitionEnd() const
  {
    return partitionEnd_;
  }
  void setPartitionEnd(NABoolean v) 
  {
    partitionEnd_ = v;
  }
  inline Lng32 recLen();
  
  virtual Int32 numChildren() const;
  virtual const ex_tcb* getChild(Int32 pos) const;

private:
  const ex_tcb * childTcb_;
  
  ex_queue_pair  qparent_;
  ex_queue_pair  qchild_;  

  atp_struct * workAtp_;
  ExSubtask  * ioEventHandler_;
  HashBuffer * firstOLAPBuffer_;
  HashBuffer * lastOLAPBuffer_;

  HashBuffer * currentOLAPBuffer_ ;
  HashBuffer * currentRetOLAPBuffer_;

  Lng32 currentHistRowInOLAPBuffer_;
  Lng32 currentRetHistRowInOLAPBuffer_;

  char * currentHistRowPtr_ ;
  char * currentRetHistRowPtr_;
  char *lastRow_;

  Lng32 minFollowing_;
  Lng32 numberHistoryRows_ ;
  Lng32 maxNumberHistoryRows_;
  Lng32 histRowsToReturn_ ;
  
  NABoolean partitionEnd_;
  NABoolean unboundedFollowing_;

  Lng32 allocRowLength_;  // allocated size of a row (original rounded up to 8)
  Lng32 maxRowsInOLAPBuffer_ ;
  Lng32 olapBufferSize_;

  Lng32 maxNumberOfOLAPBuffers_;
  Lng32 numberOfOLAPBuffers_;
  Lng32 minNumberOfOLAPBuffers_;

  ClusterDB * clusterDb_; // used to call ::enoughMemory(), etc.
  Cluster * cluster_; // used for overflow calls: flush(), read(), etc
  NABoolean OLAPBuffersFlushed_;
  NABoolean memoryPressureDetected_;
  HashBuffer *  firstOLAPBufferFromOF_;
  Lng32 numberOfOLAPBuffersFromOF_;
  ExeErrorCode rc_;

  Lng32 numberOfWinOLAPBuffers_;
  Lng32 maxNumberOfRowsReturnedBeforeReadOF_ ;
  Lng32 numberOfRowsReturnedBeforeReadOF_;

  NABoolean overflowEnabled_;

  queue_index    processedInputs_;

  CollHeap * heap_;
}; // class ExSequenceTcb

// ExSequenceTcb inline functions
//
inline ex_expr * ExSequenceTcb::sequenceExpr() const 
{ 
  return myTdb().sequenceExpr_; 
};

inline ex_expr * ExSequenceTcb::returnExpr() const 
{ 
  return myTdb().returnExpr_; 
};

inline ex_expr * ExSequenceTcb::postPred() const 
{ 
  return myTdb().postPred_; 
};

inline ex_expr * ExSequenceTcb::cancelExpr() const 
{ 
  return myTdb().cancelExpr_; 
};

inline ex_expr * ExSequenceTcb::checkPartitionChangeExpr() const 
{ 
  return myTdb().checkPartitionChangeExpr_; 
};

inline NABoolean ExSequenceTcb::isHistoryFull() const
{
  return (histRowsToReturn_ == maxNumberHistoryRows_);
};

inline NABoolean ExSequenceTcb::isHistoryEmpty() const
{
  return (histRowsToReturn_ == 0);
};

inline NABoolean ExSequenceTcb::canReturnRows() const
{
  return (numFollowingRows() >= minFollowing_);

};

inline Lng32 ExSequenceTcb::numFollowingRows() const
{
  return histRowsToReturn_ -1;
};

inline void ExSequenceTcb::updateHistRowsToReturn() 
{
  histRowsToReturn_--;
}

inline NABoolean ExSequenceTcb::isOverflowStarted()
{
  return cluster_ && cluster_->getState() == Cluster::FLUSHED ;
}

inline NABoolean ExSequenceTcb::canAllocateOLAPBuffer()
{
  return (numberOfOLAPBuffers_ < maxNumberOfOLAPBuffers_);

}

inline NABoolean ExSequenceTcb::isUnboundedFollowing()
{
  return  unboundedFollowing_;
}
inline Lng32 ExSequenceTcb::recLen() 
{ 
  //return myTdb().recLen_; 
  return allocRowLength_;
};

inline Int32 ExSequenceTcb::numChildren() const 
{ 
  return 1; 
}   

inline const ex_tcb* ExSequenceTcb::getChild(Int32 pos) const
{
   ex_assert((pos >= 0), ""); 
   if (pos == 0)
      return childTcb_;
   else
      return NULL;
}

inline ExSequenceTdb & ExSequenceTcb::myTdb() const
{
        return (ExSequenceTdb &) tdb;
};

// class ExSequencePrivateState
//
class ExSequencePrivateState : public ex_tcb_private_state
{
  friend class ExSequenceTcb;

public:
  ExSequencePrivateState(const ExSequenceTcb * tcb);
  ex_tcb_private_state * allocate_new(const ex_tcb * tcb);
  ~ExSequencePrivateState();

private:
  ExSequenceTcb::RequestState step_;
  queue_index index_;  
  Int64 matchCount_;
}; // class ExSequencePrivateState

#endif