/**********************************************************************
// @@@ 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 @@@
**********************************************************************/
#include "GroupAttr.h"
#include "AllRelExpr.h"
#include "RelSequence.h"
#include "Generator.h"
#include "GenExpGenerator.h"
#include "ExpCriDesc.h"
#include "ComTdbSequence.h"
#include "NumericType.h"
#include "ItmFlowControlFunction.h"
#include "ExplainTupleMaster.h"
#include "ComDefs.h"

#include "HashBufferHeader.h"

ItemExpr *
addConvNode(ItemExpr *childExpr,
            ValueIdMap *mapping,
            CollHeap *wHeap)

{
  if(childExpr->getOperatorType() != ITM_CONVERT &&
     !childExpr->isASequenceFunction()) {

    ValueId topValue;
    mapping->mapValueIdUp(topValue,
                          childExpr->getValueId());
    if(topValue == childExpr->getValueId()) {

      // add the convert node
      ItemExpr *newChild = new(wHeap) Convert (childExpr);
      newChild->synthTypeAndValueId(TRUE);
      mapping->addMapEntry(newChild->getValueId(),
                           childExpr->getValueId());
      return newChild;
    } else {
      return topValue.getItemExpr();
    }
  }
  return childExpr;
}

// getHistoryAttributes
//
// Helper function that traverses the set of root sequence functions
// supplied by the compiler and constructs the set of all of the
// attributes that must be materialized in the history row.
// 
void PhysSequence::getHistoryAttributes(const ValueIdSet &sequenceFunctions,
                                        const ValueIdSet &outputFromChild,
                                        ValueIdSet &historyAttributes,
                                        NABoolean addConvNodes,
                                        CollHeap *wHeap,
                                        ValueIdMap *origAttributes) const
{
  if(addConvNodes && !origAttributes) {
    origAttributes = new (wHeap) ValueIdMap();
  }

  ValueIdSet children;
  for(ValueId valId = sequenceFunctions.init();
      sequenceFunctions.next(valId);
      sequenceFunctions.advance(valId)) {

    if(valId.getItemExpr()->isASequenceFunction()) {
      ItemExpr *itmExpr = valId.getItemExpr();

      switch(itmExpr->getOperatorType())
        {
          // The child needs to be in the history row.
          //
        case ITM_OLAP_LEAD:
        case ITM_OLAP_LAG:
        case ITM_OFFSET:
        case ITM_ROWS_SINCE:
        case ITM_THIS:
        case ITM_NOT_THIS:

          // If the child needs to be in the history buffer, then
          // add a Convert node to force the value to be moved to the
          // history buffer.
          if (addConvNodes)
            {
              itmExpr->child(0) = 
                addConvNode(itmExpr->child(0), origAttributes, wHeap);
            }
          historyAttributes += itmExpr->child(0)->getValueId();
          break;

          // The sequence function needs to be in the history row.
          //
        case ITM_RUNNING_SUM:
        case ITM_RUNNING_COUNT:
        case ITM_RUNNING_MIN:
        case ITM_RUNNING_MAX:
        case ITM_LAST_NOT_NULL:
          historyAttributes += itmExpr->getValueId();
          break;
/*
        // after PhysSequence precode gen OLAP sum and count are already transform,ed into running
        // this is used during optimization phase-- 
        case ITM_OLAP_SUM:
        case ITM_OLAP_COUNT:
        case ITM_OLAP_RANK:
        case ITM_OLAP_DRANK:
          if (addConvNodes)
            {
              itmExpr->child(0) = 
                addConvNode(itmExpr->child(0), origAttributes, wHeap);
            }

          historyAttributes += itmExpr->child(0)->getValueId();
          //historyAttributes += itmExpr->getValueId();	  
          break;
*/
          // The child and sequence function need to be in the history row.
          //
        case ITM_OLAP_MIN:
        case ITM_OLAP_MAX:
        case ITM_MOVING_MIN:
        case ITM_MOVING_MAX:

          // If the child needs to be in the history buffer, then
          // add a Convert node to force the value to be moved to the
          // history buffer.
          if (addConvNodes)
            {
              itmExpr->child(0) = 
                addConvNode(itmExpr->child(0), origAttributes, wHeap);
            }

          historyAttributes += itmExpr->child(0)->getValueId();
          historyAttributes += itmExpr->getValueId();	  
          break;

        case ITM_RUNNING_CHANGE:
          if (itmExpr->child(0)->getOperatorType() == ITM_ITEM_LIST)
            {
              // child is a multi-valued expression
              // 
              ExprValueId treePtr = itmExpr->child(0);

              ItemExprTreeAsList changeValues(&treePtr,
                                              ITM_ITEM_LIST,
                                              RIGHT_LINEAR_TREE);

              CollIndex nc = changeValues.entries();
              
              ItemExpr *newChild = NULL;
              if(addConvNodes) {
                newChild = addConvNode(changeValues[nc-1], origAttributes, wHeap);
                historyAttributes += newChild->getValueId();
              } else {
                historyAttributes += changeValues[nc-1]->getValueId();
              }

              // add each item in the list
              // 
              for (CollIndex i = nc; i > 0; i--)
                {
                  if(addConvNodes) {
                    ItemExpr *conv
                      = addConvNode(changeValues[i-1], origAttributes, wHeap);

                    newChild = new(wHeap) ItemList(conv, newChild);
                    newChild->synthTypeAndValueId(TRUE);
                    historyAttributes += conv->getValueId();
                  } else {
                    historyAttributes += changeValues[i-1]->getValueId();
                  }
                }

              if(addConvNodes) {
                itmExpr->child(0) = newChild;
              }
            }
          else
            {

              // If the child needs to be in the history buffer, then
              // add a Convert node to force the value to be moved to the
              // history buffer.
              if (addConvNodes)
                {
                  itmExpr->child(0) = 
                    addConvNode(itmExpr->child(0), origAttributes, wHeap);
                }

              historyAttributes += itmExpr->child(0)->getValueId();
            }

          historyAttributes += itmExpr->getValueId();  
          break;

        default:
          CMPASSERT(0);
        }
    }

    // Gather all the children, and if not empty, recurse down to the
    // next level of the tree.
    //
    for(Lng32 i = 0; i < valId.getItemExpr()->getArity(); i++) 
    {
      if (!outputFromChild.contains(valId.getItemExpr()->child(i)->getValueId()))
        //!valId.getItemExpr()->child(i)->nodeIsPreCodeGenned()) 
      {
        children += valId.getItemExpr()->child(i)->getValueId();
      }
    }
  }
  
  if (NOT children.isEmpty())
  {
    getHistoryAttributes( children,
                          outputFromChild,
                          historyAttributes, 
                          addConvNodes, 
                          wHeap, 
                          origAttributes);
  }

} // PhysSequence::getHistoryAttributes

// PhysSequence::computeHistoryAttributes
//
// Helper function to compute the attribute for the history buffer based 
// on the items projected from the child and the computed history items.
// Also, adds the attribute information the the map table.
//
void
PhysSequence::computeHistoryAttributes(Generator *generator,
                                       MapTable *localMapTable, 
                                       Attributes **attrs,
                                       const ValueIdSet &historyIds) const
{
  // Get a local handle on some of the generator objects.
  //
  CollHeap *wHeap = generator->wHeap();

  // Populate the attribute vector with the flattened list of sequence 
  // functions and/or sequence function arguments that must be in the
  // history row. Add convert nodes for the items that are not sequence
  // functions to force them to be moved into the history row.
  //
  if(NOT historyIds.isEmpty())
    {
      Int32 i = 0;
      ValueId valId;

      for (valId = historyIds.init();
           historyIds.next(valId);
           historyIds.advance(valId))
        {
          // If this is not a sequence function, then insert a convert
          // node.
          //
          if(!valId.getItemExpr()->isASequenceFunction())
             {
               // Get a handle on the original expression and erase
               // the value ID.
               //
               ItemExpr *origExpr = valId.getItemExpr();
               origExpr->setValueId(NULL_VALUE_ID);
               origExpr->markAsUnBound();

               // Construct the cast expression with the original expression
               // as the child -- must have undone the child value ID to
               // avoid recursion later.
               //
               ItemExpr *castExpr = new(wHeap) 
                 Cast(origExpr, &(valId.getType()));

               // Replace the expression for the original value ID and the
               // synthesize the types and value ID for the new expression.
               //
               valId.replaceItemExpr(castExpr);
               castExpr->synthTypeAndValueId(TRUE);
             }
          attrs[i++] = (generator->addMapInfoToThis(localMapTable, valId, 0))->getAttr();
        }
    }
} // PhysSequence::computeHistoryAttributes

// computeHistoryBuffer
//
// Helper function that traverses the set of root sequence functions
// supplied by the compiler and dynamically determines the size
// of the history buffer.
// 
void PhysSequence::computeHistoryRows(const ValueIdSet &sequenceFunctions,//historyIds
                                      Lng32 &computedHistoryRows,
                                      Lng32 &unableToCalculate,
                                      NABoolean &unboundedFollowing, 
                                      Lng32 &minFollowingRows,
                                      const ValueIdSet &outputFromChild) 
{
  ValueIdSet children;
  ValueIdSet historyAttributes;
  Lng32 value = 0;

  for(ValueId valId = sequenceFunctions.init();
      sequenceFunctions.next(valId);
      sequenceFunctions.advance(valId)) 
  {
    if(valId.getItemExpr()->isASequenceFunction()) 
    {
      ItemExpr *itmExpr = valId.getItemExpr();

      switch(itmExpr->getOperatorType())
        {

        // THIS and NOT THIS are not dynamically computed
        //
        case ITM_THIS:
        case ITM_NOT_THIS:
          break;

        // The RUNNING functions and LastNotNull all need to go back just one row.
        //
        case ITM_RUNNING_SUM:
        case ITM_RUNNING_COUNT:
        case ITM_RUNNING_MIN:
        case ITM_RUNNING_MAX:
        case ITM_RUNNING_CHANGE:   
        case ITM_LAST_NOT_NULL:
          computedHistoryRows = MAXOF(computedHistoryRows, 2);
          break;

        case ITM_OLAP_LEAD:
            value = ((ItmLeadOlapFunction*)itmExpr)->getOffset();

        ///set to unable to compute for now-- will change later to compte values from frameStart_ and frameEnd_
        case ITM_OLAP_SUM:
        case ITM_OLAP_COUNT:
        case ITM_OLAP_MIN:
        case ITM_OLAP_MAX:
        case ITM_OLAP_RANK:
        case ITM_OLAP_DRANK:
        {
          if ( !outputFromChild.contains(itmExpr->getValueId()))
          {
            ItmSeqOlapFunction * olap = (ItmSeqOlapFunction*)itmExpr;

            if (olap->isFrameStartUnboundedPreceding()) //(olap->getframeStart() == - INT_MAX)
            {
              computedHistoryRows = MAXOF(computedHistoryRows, 2);
            }
            else
            {
              computedHistoryRows = MAXOF(computedHistoryRows, ABS(olap->getframeStart()) + 2);
            }
            if (!olap->isFrameEndUnboundedFollowing()) //(olap->getframeEnd() != INT_MAX)
            {
              computedHistoryRows = MAXOF(computedHistoryRows, ABS(olap->getframeEnd()) + 1);
            }

            if (olap->isFrameEndUnboundedFollowing()) //(olap->getframeEnd() == INT_MAX)
            {
              unboundedFollowing = TRUE;
              if (olap->getframeStart() > 0) 
              {
                minFollowingRows = ((minFollowingRows > olap->getframeStart()) ?  
                                    minFollowingRows : olap->getframeStart());
              }
            } else  if (olap->getframeEnd() > 0)
            {
              minFollowingRows = ((minFollowingRows > olap->getframeEnd()) ?  
                                  minFollowingRows : olap->getframeEnd());
            }
          }
        }

        break;

        // If 'rows since', we cannot determine how much history is needed.  
        case ITM_ROWS_SINCE:
          unableToCalculate = 1;
          break;

        // The MOVING and OFFSET functions need to go back as far as the value
        // of their second child.
        //
        //  The second argument can be:
        //    Constant: for these, we can use the constant value to set the upper bound
        //              for the history buffer.
        //    ItmScalarMinMax(child0, child1) (with operType = ITM_SCALAR_MIN)  
        //      - if child0 or child1 is a constant, then we can use either one
        //        to set the upper bound.
        
        case ITM_MOVING_MIN:
        case ITM_MOVING_MAX:
        case ITM_OFFSET:
        case ITM_OLAP_LAG: 
          for(Lng32 i = 1; i < itmExpr->getArity(); i++)
          {
            if (itmExpr->child(i)->getOperatorType() != ITM_NOTCOVERED)
            {
              ItemExpr * exprPtr = itmExpr->child(i);
              NABoolean negate;
              ConstValue *cv = exprPtr->castToConstValue(negate);
              if (cv AND cv->canGetExactNumericValue())
                {
                  Lng32 scale;
                  Int64 value64 = cv->getExactNumericValue(scale);

                  if(scale == 0 && value64 >= 0 && value64 < INT_MAX) 
                    {
                      value64 = (negate ? -value64 : value64);
                      value = MAXOF((Lng32)value64, value);
                    }
                 }
              else
                {
                  if (exprPtr->getOperatorType() == ITM_SCALAR_MIN)
                    {
                      for(Lng32 j = 0; j < exprPtr->getArity(); j++)
                        {
                          if (exprPtr->child(j)->getOperatorType()
                            != ITM_NOTCOVERED)
                            {
                               ItemExpr * exprPtr1 = exprPtr->child(j);
                               NABoolean negate1;
                               ConstValue *cv1 = exprPtr1->castToConstValue(negate1);
                               if (cv1 AND cv1->canGetExactNumericValue())
                                 {
                                   Lng32 scale1;
                                   Int64 value64_1 = cv1->getExactNumericValue(scale1);

                                   if(scale1 == 0 && value64_1 >= 0 && value64_1 < INT_MAX) 
                                     {
                                       value64_1 = (negate1 ? -value64_1 : value64_1);
                                       value = MAXOF((Lng32)value64_1, value);
                                     }
                                  }
                              }
                          }   
                     }   
                }  // end of inner else
            }// end of if

          }// end of for

          // Check if the value is greater than zero.
          // If it is, then save the value, but first
          // increment the returned ConstValue by one.
          // Otherwise, the offset or moving value was unable
          // to be calculated.

          if (value > 0)
          {
            value++;
            computedHistoryRows = MAXOF(computedHistoryRows, value);
            value = 0;
          }
          else
            unableToCalculate = 1;

          break;

        default:
          CMPASSERT(0);
        }
    }
   
    // Gather all the children, and if not empty, recurse down to the
    // next level of the tree.
    //

    for(Lng32 i = 0; i < valId.getItemExpr()->getArity(); i++) {
      if (//valId.getItemExpr()->child(i)->getOperatorType() != ITM_NOTCOVERED //old stuff
          !outputFromChild.contains(valId.getItemExpr()->child(i)->getValueId()))
      {
        children += valId.getItemExpr()->child(i)->getValueId();
      }
    }
  }
  
  if (NOT children.isEmpty())
  {
    computeHistoryRows(children, 
                       computedHistoryRows, 
                       unableToCalculate, 
                       unboundedFollowing, 
                       minFollowingRows,
                       outputFromChild);  
  }
} // PhysSequence::computeHistoryRows

// PhysSequence::generateChildProjectExpression
//
// Helper function to compute the child project expression. Also, sets up the
// local map table so that the moved items will be referenced from their new
// location. Returns the resulting project expression.
//
#ifdef OLD
ex_expr *
PhysSequence::generateChildProjectExpression(Generator *generator, 
                                             MapTable *mapTable, 
                                             MapTable *localMapTable,
                                             const ValueIdSet &childProjectIds) const
{
  ex_expr * projectExpr = NULL;
  if(NOT childProjectIds.isEmpty())
    {
      // Generate the clauses for the expression
      //
      generator->getExpGenerator()->generateSetExpr(childProjectIds,
                                                    ex_expr::exp_ARITH_EXPR,
                                                    &projectExpr);

      // Add the projected values to the local map table.
      //
      ValueId valId;
      for(valId = childProjectIds.init();
          childProjectIds.next(valId);
          childProjectIds.advance(valId))
        {
          // Get the attribute information from the convert destination.
          //
          Attributes *newAttr = mapTable->getMapInfo(valId)->getAttr();

          // Add the original value to the local map table with the
          // attribute information from the convert desination.
          //
          MapInfo *mapInfo = localMapTable->addMapInfoToThis
            (valId.getItemExpr()->child(0)->getValueId(), newAttr);

          // Nothing more needs to be done for this item.
          //
          mapInfo->codeGenerated();
        }
    }

  return projectExpr;
} // PhysSequence::generateChildProjectExpression
#endif


void
RelSequence::addCancelExpr(CollHeap *wHeap)
{
  ItemExpr *cPred = NULL;

  if (this->partition().entries() > 0)
  {
    return;
  }
  if(cancelExpr().entries() > 0) 
  {
    return;
  }

  for(ValueId valId = selectionPred().init();
      selectionPred().next(valId);
      selectionPred().advance(valId)) 
  {
    ItemExpr *pred = valId.getItemExpr();

    // Look for preds that select a prefix of the sequence.
    // Rank() < const; Rank <= const; const > Rank; const >= Rank
    ItemExpr *op1 = NULL;
    ItemExpr *op2 = NULL;

    if(pred->getOperatorType() == ITM_LESS ||
       pred->getOperatorType() == ITM_LESS_EQ) 
    {
      op1 = pred->child(0);
      op2 = pred->child(1);
    }
    else if (pred->getOperatorType() == ITM_GREATER ||
             pred->getOperatorType() == ITM_GREATER_EQ) 
    {
      op1 = pred->child(1);
      op2 = pred->child(0);
    }
    NABoolean negate;
    if (op1 && op2 &&
        (op2->getOperatorType() == ITM_CONSTANT || 
         op2->getOperatorType() == ITM_DYN_PARAM)  &&
         (op1->getOperatorType() == ITM_OLAP_RANK ||
          op1->getOperatorType() == ITM_OLAP_DRANK ||
          (op1->getOperatorType() == ITM_OLAP_COUNT &&
           op1->child(0)->getOperatorType() == ITM_CONSTANT &&
           !op1->child(0)->castToConstValue(negate)->isNull())))
    {
       cPred = new(wHeap) UnLogic(ITM_NOT, pred);
       //break at first occurence
       break;
    }
  }
  
  if(cPred) 
  {
    cPred->synthTypeAndValueId(TRUE);
    cancelExpr().insert(cPred->getValueId());
  }
}

void PhysSequence::addCheckPartitionChangeExpr( Generator *generator,
                                                NABoolean addConvNodes,
                                                ValueIdMap *origAttributes) 
{
  if(!(partition().entries() > 0)) 
  {
    return;
  }
  CollHeap * wHeap = generator->wHeap();
    
  if(addConvNodes && !origAttributes) 
  {
    origAttributes = new (wHeap) ValueIdMap();
  }

  ItemExpr * checkPartChng= NULL;
  for (CollIndex ix = 0; ix < partition().entries(); ix++)
  {
    ItemExpr *iePtr = partition().at(ix).getItemExpr();
    ItemExpr * convIePtr = addConvNode(iePtr, origAttributes,wHeap);

    movePartIdsExpr() += convIePtr->getValueId();
    ItemExpr * comp =  new (wHeap) BiRelat(ITM_EQUAL,
                                           iePtr,
                                           convIePtr,
                                           TRUE);
    if (!checkPartChng)
    {
      checkPartChng = comp;
    }
    else
    {
      checkPartChng = new (wHeap) BiLogic( ITM_AND,
                                           checkPartChng,
                                           comp);
    }
  } 

  checkPartChng->convertToValueIdSet(checkPartitionChangeExpr(),
                                      generator->getBindWA(),
                                      ITM_AND);
}


// PhysSequence::preCodeGen() -------------------------------------------
// Perform local query rewrites such as for the creation and
// population of intermediate tables, for accessing partitioned
// data. Rewrite the value expressions after minimizing the dataflow
// using the transitive closure of equality predicates.
//
// PhysSequence::preCodeGen() - is basically the same as the RelExpr::
// preCodeGen() except that here we replace the VEG references in the
// sortKey() list, as well as the selectionPred().
//
// Parameters:
//
// Generator *generator
//    IN/OUT : A pointer to the generator object which contains the state,
//             and tools (e.g. expression generator) to generate code for
//             this node.
//
// ValueIdSet &externalInputs
//    IN    : The set of external Inputs available to this node.
//
//
RelExpr * PhysSequence::preCodeGen(Generator * generator, 
                                   const ValueIdSet & externalInputs,
                                   ValueIdSet &pulledNewInputs)
{
  if (nodeIsPreCodeGenned())
    return this;

  // Resolve the VEGReferences and VEGPredicates, if any, that appear
  // in the Characteristic Inputs, in terms of the externalInputs.
  //
  getGroupAttr()->resolveCharacteristicInputs(externalInputs);

  // My Characteristic Inputs become the external inputs for my children.
  //
  ValueIdSet childPulledInputs;

  child(0) = child(0)->preCodeGen(generator, externalInputs, pulledNewInputs);
  if (! child(0).getPtr())
    return NULL;

  // process additional input value ids the child wants
  getGroupAttr()->addCharacteristicInputs(childPulledInputs);
  pulledNewInputs += childPulledInputs;

  // The VEG expressions in the selection predicates and the characteristic
  // outputs can reference any expression that is either a potential output
  // or a characteristic input for this RelExpr. Supply these values for
  // rewriting the VEG expressions.
  //
  ValueIdSet availableValues;
  getInputValuesFromParentAndChildren(availableValues);

  // The sequenceFunctions() have access to only the Input
  // Values. These can come from the parent or be the outputs of the
  // child.
  //
  sequenceFunctions().
    replaceVEGExpressions(availableValues,
                          getGroupAttr()->getCharacteristicInputs());

  sequencedColumns().
    replaceVEGExpressions(availableValues,
                          getGroupAttr()->getCharacteristicInputs());

  requiredOrder().
    replaceVEGExpressions(availableValues,
                          getGroupAttr()->getCharacteristicInputs());

  cancelExpr().
    replaceVEGExpressions(availableValues,
                          getGroupAttr()->getCharacteristicInputs());

  partition().
    replaceVEGExpressions(availableValues,
                          getGroupAttr()->getCharacteristicInputs());

  //checkPartitionChangeExpr().
   // replaceVEGExpressions(availableValues,
   //                       getGroupAttr()->getCharacteristicInputs());

  // The selectionPred has access to only the output values generated by
  // Sequence and input values from the parent.
  //
  getInputAndPotentialOutputValues(availableValues);

  // Rewrite the selection predicates.
  //
  NABoolean replicatePredicates = TRUE;
  selectionPred().replaceVEGExpressions
    (availableValues,
     getGroupAttr()->getCharacteristicInputs(),
     FALSE, // no key predicates here
     0 /* no need for idempotence here */,
     replicatePredicates
     ); 

  // Replace VEG references in the outputs and remove redundant
  // outputs.
  //
  getGroupAttr()->resolveCharacteristicOutputs
    (availableValues,
     getGroupAttr()->getCharacteristicInputs());

  
  addCancelExpr(generator->wHeap());

  ///addCheckPartitionChangeExpr(generator->wHeap());

  transformOlapFunctions(generator->wHeap());
  if ( getUnboundedFollowing() ) {
    // Count this Seq as a BMO and add its needed memory to the total needed
    generator->incrNumBMOs();
    
    if ((ActiveSchemaDB()->getDefaults()).
	getAsDouble(BMO_MEMORY_LIMIT_PER_NODE_IN_MB) > 0)
      generator->incrBMOsMemory(getEstimatedRunTimeMemoryUsage(generator, TRUE));
  }
  else
    generator->incrNBMOsMemoryPerNode(getEstimatedRunTimeMemoryUsage(generator, TRUE));

  markAsPreCodeGenned();

  return this;
} // PhysSequence::preCodeGen


short
PhysSequence::codeGen(Generator *generator) 
{
  // Get a local handle on some of the generator objects.
  //
  CollHeap *wHeap = generator->wHeap();
  Space *space = generator->getSpace();
  ExpGenerator *expGen = generator->getExpGenerator();
  MapTable *mapTable = generator->getMapTable();

  // Allocate a new map table for this node. This must be done
  // before generating the code for my child so that this local
  // map table will be sandwiched between the map tables already
  // generated and the map tables generated by my offspring.
  //
  // Only the items available as output from this node will
  // be put in the local map table. Before exiting this function, all of
  // my offsprings map tables will be removed. Thus, none of the outputs 
  // from nodes below this node will be visible to nodes above it except 
  // those placed in the local map table and those that already exist in
  // my ancestors map tables. This is the standard mechanism used in the
  // generator for managing the access to item expressions.
  //
  MapTable *localMapTable = generator->appendAtEnd();

  // Since this operation doesn't modify the row on the way down the tree,
  // go ahead and generate the child subtree. Capture the given composite row
  // descriptor and the child's returned TDB and composite row descriptor.
  //
  ex_cri_desc * givenCriDesc = generator->getCriDesc(Generator::DOWN);
  child(0)->codeGen(generator);
  ComTdb *childTdb = (ComTdb*)generator->getGenObj();
  ex_cri_desc * childCriDesc = generator->getCriDesc(Generator::UP);
  ExplainTuple *childExplainTuple = generator->getExplainTuple();

  // Make all of my child's outputs map to ATP 1. The child row is only 
  // accessed in the project expression and it will be the second ATP 
  // (ATP 1) passed to this expression.
  //
  localMapTable->setAllAtp(1);

  // My returned composite row has an additional tupp.
  //
  Int32 numberTuples = givenCriDesc->noTuples() + 1;
  ex_cri_desc * returnCriDesc 
    = new (space) ex_cri_desc(numberTuples, space);

  // For now, the history buffer row looks just the return row. Later,
  // it may be useful to add an additional tupp for sequence function
  // itermediates that are not needed above this node -- thus, this
  // ATP is kept separate from the returned ATP.
  //
  const Int32 historyAtp = 0;
  const Int32 historyAtpIndex = numberTuples-1;
  ex_cri_desc *historyCriDesc = new (space) ex_cri_desc(numberTuples, space);
  ExpTupleDesc *historyDesc = 0;

  //seperate the read and retur expressions
  seperateReadAndReturnItems(wHeap);

  // The history buffer consists of items projected directly from the
  // child, the root sequence functions, the value arguments of the 
  // offset functions, and running sequence functions. These elements must 
  // be materialized in the  history buffer in order to be able to compute 
  // the outputs of this node -- the items projected directly from the child 
  // (projectValues) and the root sequence functions (sequenceFunctions).
  //
  // Compute the set of sequence function items that must be materialized
  // int the history buffer. -- sequenceItems
  //
  // Compute the set of items in the history buffer: the union of the 
  // projected values and the value arguments. -- historyIds
  //
  // Compute the set of items in the history buffer that are computed:
  // the difference between all the elements in the history buffer
  // and the projected items. -- computedHistoryIds
  //

  // KB---will need to return atp with 3 tups only 0,1 and 2 
  // 2 -->values from history buffer after ther are moved to it

 
  addCheckPartitionChangeExpr(generator, TRUE);

  ValueIdSet historyIds;

  historyIds += movePartIdsExpr(); 
  historyIds += sequencedColumns();
  
  ValueIdSet outputFromChild = child(0)->getGroupAttr()->getCharacteristicOutputs();

  getHistoryAttributes(readSeqFunctions(),outputFromChild, historyIds, TRUE, wHeap);

  // remove LEAD from readSeqFunctions() and remeber it in leadFuncs
  // Its child has CONVERT added in getHistoryAttributes() call.
  ValueIdSet leadFuncs;
  ValueIdSet leadFuncChildren;

  readSeqFunctions().findAllOpType(ITM_OLAP_LEAD, leadFuncs);
  leadFuncs.findAllChildren(leadFuncChildren);
  readSeqFunctions() -= leadFuncs;
  readSeqFunctions() += leadFuncChildren;

  // Add in the top level sequence functions.
  historyIds += readSeqFunctions();

  // remove LEAD in old form from the return function
  ValueIdSet leadFuncsInReturn;
  returnSeqFunctions().findAllOpType(ITM_OLAP_LEAD, leadFuncsInReturn);
  returnSeqFunctions() -= leadFuncsInReturn;

  // add in the LEAD in new form
  returnSeqFunctions() += leadFuncs;

  getHistoryAttributes(returnSeqFunctions(),outputFromChild, historyIds, TRUE, wHeap);
  // Add in the top level functions.
  historyIds += returnSeqFunctions();
  
  // Layout the work tuple format which consists of the projected
  // columns and the computed sequence functions. First, compute
  // the number of attributes in the tuple.
  //
  ULng32 numberAttributes 
    = ((NOT historyIds.isEmpty()) ? historyIds.entries() : 0);

  // Allocate an attribute pointer vector from the working heap.
  //
  Attributes **attrs = new(wHeap) Attributes*[numberAttributes];

  // Fill in the attributes vector for the history buffer including
  // adding the entries to the map table. Also, compute the value ID
  // set for the elements to project from the child row.
  //
  //??????????re-visit this function??
  computeHistoryAttributes(generator, 
                           localMapTable,
                           attrs,
                           historyIds);

  // Create the tuple descriptor for the history buffer row and
  // assign the offsets to the attributes. For now, this layout is 
  // identical to the returned row. Set the tuple descriptors for
  // the return and history rows.
  //
  ULng32 historyRecLen;
  expGen->processAttributes(numberAttributes,
                            attrs,
                            ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
                            historyRecLen,
                            historyAtp,
                            historyAtpIndex,
                            &historyDesc,
                            ExpTupleDesc::SHORT_FORMAT);
  NADELETEBASIC(attrs, wHeap);
  returnCriDesc->setTupleDescriptor(historyAtpIndex, historyDesc);
  historyCriDesc->setTupleDescriptor(historyAtpIndex, historyDesc);

  // If there are any sequence function items, generate the sequence 
  // function expressions.
  //
  ex_expr * readSeqExpr = NULL;
  if(NOT readSeqFunctions().isEmpty())
    {
      ValueIdSet seqVals = readSeqFunctions();
      seqVals += sequencedColumns();
      seqVals += movePartIdsExpr(); 
      expGen->generateSequenceExpression(seqVals,
                                         readSeqExpr);
    }

  ex_expr *checkPartChangeExpr = NULL;
  if (!checkPartitionChangeExpr().isEmpty()) {
    ItemExpr * newCheckPartitionChangeTree= 
        checkPartitionChangeExpr().rebuildExprTree(ITM_AND,TRUE,TRUE);

    expGen->generateExpr(newCheckPartitionChangeTree->getValueId(), 
                         ex_expr::exp_SCAN_PRED,
                         &checkPartChangeExpr);
  }

  //unsigned long rowLength;
  ex_expr * returnExpr = NULL;
  if(NOT returnSeqFunctions().isEmpty())
  {
    expGen->generateSequenceExpression(returnSeqFunctions(),
                                         returnExpr);

  }

  // Generate expression to evaluate predicate on the output
  //
  ex_expr *postPred = 0;

  if (! selectionPred().isEmpty()) {
    ItemExpr * newPredTree = 
      selectionPred().rebuildExprTree(ITM_AND,TRUE,TRUE);

    expGen->generateExpr(newPredTree->getValueId(), ex_expr::exp_SCAN_PRED,
                         &postPred);
  }


  // Reset ATP's to zero for parent.
  //
  localMapTable->setAllAtp(0);


  // Generate expression to evaluate the cancel expression
  //
  ex_expr *cancelExpression = 0;

  if (! cancelExpr().isEmpty()) {
    ItemExpr * newCancelExprTree = 
      cancelExpr().rebuildExprTree(ITM_AND,TRUE,TRUE);

    expGen->generateExpr(newCancelExprTree->getValueId(), ex_expr::exp_SCAN_PRED,
                         &cancelExpression);
  }

  //
  //  For overflow
  //
  // ( The following are meaningless if ! unlimitedHistoryRows() ) 
  NABoolean noOverflow =  
    CmpCommon::getDefault(EXE_BMO_DISABLE_OVERFLOW) == DF_ON ;
  NABoolean logDiagnostics = 
    CmpCommon::getDefault(EXE_DIAGNOSTIC_EVENTS) == DF_ON ;
  NABoolean possibleMultipleCalls = generator->getRightSideOfFlow() ;
  short scratchTresholdPct = 
    (short) CmpCommon::getDefaultLong(SCRATCH_FREESPACE_THRESHOLD_PERCENT);
  // determione the memory usage (amount of memory as percentage from total
  // physical memory used to initialize data structures)
  unsigned short memUsagePercent =
    (unsigned short) getDefault(BMO_MEMORY_USAGE_PERCENT);
  short memPressurePct = (short)getDefault(GEN_MEM_PRESSURE_THRESHOLD);

  historyRecLen = ROUND8(historyRecLen);

  Lng32 maxNumberOfOLAPBuffers;
  Lng32 maxRowsInOLAPBuffer;
  Lng32 minNumberOfOLAPBuffers;
  Lng32 numberOfWinOLAPBuffers;
  Lng32 olapBufferSize;

  computeHistoryParams(historyRecLen,
                       maxRowsInOLAPBuffer,
                       minNumberOfOLAPBuffers,
                       numberOfWinOLAPBuffers,
                       maxNumberOfOLAPBuffers,
                       olapBufferSize);

  ComTdbSequence *sequenceTdb
    = new(space) ComTdbSequence(readSeqExpr,
                                returnExpr,
                                postPred,
                                cancelExpression,
                                getMinFollowingRows(),
                                historyRecLen,
                                historyAtpIndex,
                                childTdb,
                                givenCriDesc,
                                returnCriDesc,
                                (queue_index)getDefault(GEN_SEQFUNC_SIZE_DOWN),
                                (queue_index)getDefault(GEN_SEQFUNC_SIZE_UP),
                                getDefault(GEN_SEQFUNC_NUM_BUFFERS),
                                getDefault(GEN_SEQFUNC_BUFFER_SIZE),
				olapBufferSize,
                                maxNumberOfOLAPBuffers,
                                numHistoryRows(),
                                getUnboundedFollowing(),
				logDiagnostics,
				possibleMultipleCalls,
				scratchTresholdPct,
				memUsagePercent,
				memPressurePct,
                                maxRowsInOLAPBuffer,
                                minNumberOfOLAPBuffers,
                                numberOfWinOLAPBuffers,
                                noOverflow,
                                checkPartChangeExpr);
  generator->initTdbFields(sequenceTdb);

  // update the estimated value of HistoryRowLength with actual value
  //setEstHistoryRowLength(historyIds.getRowLength());

  double sequenceMemEst = generator->getEstMemPerInst(getKey());
  generator->addToTotalEstimatedMemory(sequenceMemEst);

  if(!generator->explainDisabled()) {
    generator->
      setExplainTuple(addExplainInfo(sequenceTdb,
                                     childExplainTuple,
                                     0,
                                     generator));
  }

  sequenceTdb->setScratchIOVectorSize((Int16)getDefault(SCRATCH_IO_VECTOR_SIZE_HASH));
  sequenceTdb->setOverflowMode(generator->getOverflowMode());

  sequenceTdb->setBmoMinMemBeforePressureCheck((Int16)getDefault(EXE_BMO_MIN_SIZE_BEFORE_PRESSURE_CHECK_IN_MB));
  
  if(generator->getOverflowMode() == ComTdb::OFM_SSD )
    sequenceTdb->setBMOMaxMemThresholdMB((UInt16)(ActiveSchemaDB()->
				   getDefaults()).
			  getAsLong(SSD_BMO_MAX_MEM_THRESHOLD_IN_MB));
  else
    sequenceTdb->setBMOMaxMemThresholdMB((UInt16)(ActiveSchemaDB()->
				   getDefaults()).
			  getAsLong(EXE_MEMORY_AVAILABLE_IN_MB));

  // The CQD EXE_MEM_LIMIT_PER_BMO_IN_MB has precedence over the mem quota sys
  NADefaults &defs = ActiveSchemaDB()->getDefaults();
  UInt16 mmu = (UInt16)(defs.getAsDouble(EXE_MEM_LIMIT_PER_BMO_IN_MB));
  UInt16 numBMOsInFrag = (UInt16)generator->getFragmentDir()->getNumBMOs();
  Lng32 numStreams;
  double bmoMemoryUsagePerNode = generator->getEstMemPerNode(getKey(), numStreams);
  double memQuota = 0;
  double memQuotaRatio;
  if (mmu != 0)
    sequenceTdb->setMemoryQuotaMB(mmu);
  else {
    // Apply quota system if either one the following two is true:
    //   1. the memory limit feature is turned off and more than one BMOs 
    //   2. the memory limit feature is turned on
    NABoolean mlimitPerNode = defs.getAsDouble(BMO_MEMORY_LIMIT_PER_NODE_IN_MB) > 0;

    if ( mlimitPerNode || numBMOsInFrag > 1 ) {
        double memQuota = 
           computeMemoryQuota(generator->getEspLevel() == 0,
                              mlimitPerNode,
                              generator->getBMOsMemoryLimitPerNode().value(),
                              generator->getTotalNumBMOs(),
                              generator->getTotalBMOsMemoryPerNode().value(),
                              numBMOsInFrag, 
                              bmoMemoryUsagePerNode,
                              numStreams,
                              memQuotaRatio
                             );
    }
    Lng32 seqMemoryLowbound = defs.getAsLong(EXE_MEMORY_LIMIT_LOWER_BOUND_SEQUENCE);
    Lng32 memoryUpperbound = defs.getAsLong(BMO_MEMORY_LIMIT_UPPER_BOUND);

    if ( memQuota < seqMemoryLowbound ) {
       memQuota = seqMemoryLowbound;
       memQuotaRatio = BMOQuotaRatio::MIN_QUOTA;
    }
    else if (memQuota >  memoryUpperbound)
       memQuota = memoryUpperbound;
           
    sequenceTdb->setMemoryQuotaMB( UInt16(memQuota) );
  }

  generator->setCriDesc(givenCriDesc, Generator::DOWN);
  generator->setCriDesc(returnCriDesc, Generator::UP);
  generator->setGenObj(this, sequenceTdb);

  return 0;

}


void PhysSequence::seperateReadAndReturnItems(
                                //ValueIdSet & readPhaseSet, 
                                //ValueIdSet & returnPhaseSet, 
                                CollHeap *wHeap)
{

  ValueIdSet outputFromChild = child(0)->getGroupAttr()->getCharacteristicOutputs();

  ValueIdSet seqFuncs = sequenceFunctions();
  
  for(ValueId valId = seqFuncs.init();
      seqFuncs.next(valId);
      seqFuncs.advance(valId)) 
  {
    computeReadNReturnItems(valId,
                            valId,
                            //sequenceFunctions(), 
                            //returnSeqFunctions(),
                            outputFromChild,
                            wHeap);
  }
}

void PhysSequence::transformOlapFunctions(CollHeap *wHeap)
{

 
  for(ValueId valId = sequenceFunctions().init();
      sequenceFunctions().next(valId);
      sequenceFunctions().advance(valId)) 
  {
    
    ItemExpr * itmExpr = valId.getItemExpr();

    //NAType *itmType = itmExpr->getValueId().getType().newCopy(wHeap);

    if (itmExpr->isOlapFunction())
    {
      NAType *itmType = itmExpr->getValueId().getType().newCopy(wHeap);

      itmExpr = ((ItmSeqOlapFunction*)itmExpr)->transformOlapFunction(wHeap);

      if ( itmExpr->getOperatorType() == ITM_OLAP_LEAD ) {
        computeAndSetMinFollowingRows(((ItmLeadOlapFunction*)itmExpr)->getOffset());
      }

      CMPASSERT(itmExpr);
      if(itmExpr->getValueId() != valId)
      {
	itmExpr = new (wHeap) Cast(itmExpr, itmType);
	itmExpr->synthTypeAndValueId(TRUE);
	valId.replaceItemExpr(itmExpr);
	itmExpr->getValueId().changeType(itmType);//????
      }
    }
      itmExpr->transformOlapFunctions(wHeap);
  }
}

void PhysSequence::computeHistoryParams(Lng32 histRecLength,
                                        Lng32 &maxRowsInOLAPBuffer,
                                        Lng32 &minNumberOfOLAPBuffers,
                                        Lng32 &numberOfWinOLAPBuffers,
                                        Lng32 &maxNumberOfOLAPBuffers,
                                        Lng32 &olapBufferSize)
{
  Lng32 maxFWAdditionalBuffers = getDefault(OLAP_MAX_FIXED_WINDOW_EXTRA_BUFFERS);
  maxNumberOfOLAPBuffers = getDefault(OLAP_MAX_NUMBER_OF_BUFFERS);
  // For testing we may force a smaller max # rows in a buffer
  Lng32 forceMaxRowsInOLAPBuffer =  getDefault(OLAP_MAX_ROWS_IN_OLAP_BUFFER);
  minNumberOfOLAPBuffers = 0;
  numberOfWinOLAPBuffers = 0;
  olapBufferSize = getDefault(OLAP_BUFFER_SIZE);
  Lng32 olapAvailableBufferSize = olapBufferSize
                                 - ROUND8(sizeof(HashBufferHeader)); // header
  // also consider the trailer reserved for DP2's checksum, for overflow only
  if ( getUnboundedFollowing() ) olapAvailableBufferSize -= 8 ; 

  if ( histRecLength > olapAvailableBufferSize)
  { // history row exceeds size limit fir the overflow
    if (getUnboundedFollowing())
    {
       *CmpCommon::diags() << DgSqlCode(-4390)
			    << DgInt0(olapAvailableBufferSize);
       GenExit();
       return ;
    }
    else
    {
      olapAvailableBufferSize = histRecLength;
      // Buffer needs to accomodate the header, but not the trailer (no O/F)
      olapBufferSize = histRecLength + ROUND8(sizeof(HashBufferHeader)) ;
    }
  }

  // Calculate the max # rows in a buffer
  maxRowsInOLAPBuffer = olapAvailableBufferSize / histRecLength ;

  // For testing - we may override the above max # rows
  if ( forceMaxRowsInOLAPBuffer > 0 && 
       forceMaxRowsInOLAPBuffer < maxRowsInOLAPBuffer )
    maxRowsInOLAPBuffer = forceMaxRowsInOLAPBuffer ;

  minNumberOfOLAPBuffers = numHistoryRows_ / maxRowsInOLAPBuffer;

  if ( numHistoryRows_ % maxRowsInOLAPBuffer >0)
  {
    minNumberOfOLAPBuffers++;
  }

  if (getUnboundedFollowing())
  {
    numberOfWinOLAPBuffers = minFollowingRows_/maxRowsInOLAPBuffer ;
    if (minFollowingRows_ % maxRowsInOLAPBuffer >0)
    {
      numberOfWinOLAPBuffers++;
    }

    if (numberOfWinOLAPBuffers +1 > minNumberOfOLAPBuffers)
    {
      minNumberOfOLAPBuffers = numberOfWinOLAPBuffers +1 ;
    }

    //produce an error here if maxNumberOfOLAPBuffers < minNumberOfOLAPBuffers 
  }
  else
  {
    maxNumberOfOLAPBuffers = minNumberOfOLAPBuffers + maxFWAdditionalBuffers;
  }
}


CostScalar PhysSequence::getEstimatedRunTimeMemoryUsage(Generator *generator, NABoolean perNode, Lng32 *numStreams)
{
  // input param is not used as this operator does not participate in the
  // quota system.

  ValueIdSet outputFromChild = child(0).getGroupAttr()->getCharacteristicOutputs();
  //TODO: Line below dumps core at times 
  //const CostScalar maxCard = child(0).getGroupAttr()->getResultMaxCardinalityForEmptyInput();
  const CostScalar maxCard = 0;
  const CostScalar rowCount = numHistoryRows();

  //ValueIdSet historyIds;
  //getHistoryAttributes(sequenceFunctions(),outputFromChild, historyIds);
  //historyIds += sequenceFunctions();
  const Lng32 historyBufferWidthInBytes = getEstHistoryRowLength(); //historyIds.getRowLength();
  const double historyBufferSizeInBytes = rowCount.value() * 
                                            historyBufferWidthInBytes;

  // totalMemory is per CPU at this point of time.
  double totalMemory = historyBufferSizeInBytes;
  CostScalar estMemPerNode;
  CostScalar estMemPerInst;

  const PhysicalProperty* const phyProp = getPhysicalProperty();
  Lng32 numOfStreams = 1;
  if (phyProp != NULL)
  {
     PartitioningFunction * partFunc = phyProp -> getPartitioningFunction() ;
     numOfStreams = partFunc->getCountOfPartitions();
     if (numOfStreams <= 0)
        numOfStreams = 1;
     // totalMemory is for all CPUs at this point of time.
     totalMemory *= numOfStreams;
  }
  if (numStreams != NULL)
     *numStreams = numOfStreams;
  estMemPerNode = totalMemory /= MINOF(MAXOF(gpClusterInfo->getTotalNumberOfCPUs(), 1), numOfStreams);
  estMemPerInst = totalMemory /= numOfStreams;
  OperBMOQuota *operBMOQuota = new (generator->wHeap()) OperBMOQuota(getKey(), numOfStreams,         
                                                  estMemPerNode, estMemPerInst, rowCount, maxCard);
  generator->getBMOQuotaMap()->insert(operBMOQuota);
  if (perNode)
     return estMemPerNode;
  else
     return estMemPerInst; 
}

ExplainTuple*
PhysSequence::addSpecificExplainInfo(ExplainTupleMaster *explainTuple, 
                                              ComTdb * tdb, 
                                              Generator *generator)
{

 NAString buffer = "num_history_rows: ";

 char buf[20];
 sprintf(buf, "%d ", numHistoryRows());
 buffer += buf;

 const Lng32 bufferWidth = ((ComTdbSequence *)tdb)->getRecLength();

 buffer += " history_row_size: ";
 sprintf(buf, "%d ", bufferWidth);
 buffer += buf;

 buffer += " est_history_row_size: ";
 sprintf(buf, "%d ", getEstHistoryRowLength());
 buffer += buf;

 //Int16 int16Val = ((ComTdbSequence *)tdb)->getUnboundedFollowing();
 Int16 int16Val = ((ComTdbSequence *)tdb)->isUnboundedFollowing();
 buffer += " unbounded_following: ";
 sprintf(buf, "%d ", int16Val);
 buffer += buf;

 Int32 int32Val;
 int32Val = ((ComTdbSequence *)tdb)->getMaxRowsInOLAPBuffer();
 buffer += " max_rows_in_olap_buffer: ";
 sprintf(buf, "%d ", int32Val);
 buffer += buf;

 int32Val = ((ComTdbSequence *)tdb)->getMinNumberOfOLAPBuffers();
 buffer += " min_number_of_olap_buffer: ";
 sprintf(buf, "%d ", int32Val);
 buffer += buf;

 int32Val = ((ComTdbSequence *)tdb)->getMaxNumberOfOLAPBuffers();
 buffer += " max_number_of_olap_buffer: ";
 sprintf(buf, "%d ", int32Val);
 buffer += buf;

 int32Val = ((ComTdbSequence *)tdb)->getOLAPBufferSize();
 buffer += " olap_buffer_size: ";
 sprintf(buf, "%d ", int32Val);
 buffer += buf;

 int32Val = ((ComTdbSequence *)tdb)->getNumberOfWinOLAPBuffers();
 buffer += " number_of_win_olap_buffers: ";
 sprintf(buf, "%d ", int32Val);
 buffer += buf;

 int16Val = ! ((ComTdbSequence *)tdb)->isNoOverflow();
 buffer += " overflow_enabled: ";
 sprintf(buf, "%d ", int16Val);
 buffer += buf;

 
 int32Val = ((ComTdbSequence *)tdb)->getMinFollowing();
 buffer += " minimum_following: ";
 sprintf(buf, "%d ", int32Val);
 buffer += buf;


 explainTuple->setDescription(buffer);

 return(explainTuple);
}        


void PhysSequence::computeReadNReturnItems( ValueId topSeqVid,
                                            ValueId vid,
                                            const ValueIdSet &outputFromChild,
                                            CollHeap *wHeap)
{
  ItemExpr * itmExpr = vid.getItemExpr();


  if (outputFromChild.contains(vid)) 
  {
    return;
  }

  // Make sure both the return and returning function
  // contain the same LEAD function. later on in
  // Sequence::codeGen() near seperateReadAndReturnItems(),
  // the function is separated as follows.
  //
  // LEAD in old form: LEAD(c)
  // LEAD in new form: LEAD(CONVERT(c))
  //
  // After separation:
  //
  //    readFunction: COVNERT(C)
  //    returnFunction: LEAD(CONVERT(C))
  //
  //  In particular, both functions refer to CONVERT(C)
  //  with the same valueId.
  //
  if ( itmExpr->getOperatorType() == ITM_OLAP_LEAD )
  {
    readSeqFunctions() += topSeqVid;
    returnSeqFunctions() += topSeqVid;
    return;
  }

  //test if itm_minus and then if negative offset ....
  if ( itmExpr->getOperatorType() == ITM_OFFSET &&
      ((ItmSeqOffset *)itmExpr)->getOffsetConstantValue() < 0)
  {
    readSeqFunctions() -= topSeqVid;
    returnSeqFunctions() += topSeqVid;

    readSeqFunctions() += itmExpr->child(0)->castToItemExpr()->getValueId();
    return;
  }
  
  if (itmExpr->getOperatorType() == ITM_MINUS)
  {
    ItemExpr * chld0  = itmExpr->child(0)->castToItemExpr();
    if ( chld0->getOperatorType() == ITM_OFFSET &&
        ((ItmSeqOffset *)chld0)->getOffsetConstantValue() <0)
    {
      readSeqFunctions() -= topSeqVid;
      returnSeqFunctions() += topSeqVid;

      readSeqFunctions() += chld0->child(0)->castToItemExpr()->getValueId();

      ItemExpr * chld1  = itmExpr->child(1)->castToItemExpr();
      if (chld1->getOperatorType() == ITM_OFFSET &&
          ((ItmSeqOffset *)chld1)->getOffsetConstantValue() < 0)
      {
        readSeqFunctions() += chld1->child(0)->castToItemExpr()->getValueId();
      }
      else
      {
        readSeqFunctions() += chld1->getValueId();
      }
      return;
    }
    
  }
  
  
  if (itmExpr->getOperatorType() == ITM_OLAP_MIN || 
           itmExpr->getOperatorType() == ITM_OLAP_MAX) 
  { 
    ItmSeqOlapFunction * olap = (ItmSeqOlapFunction *)itmExpr;
    if (olap->getframeEnd()>0)
    {
      readSeqFunctions() -= topSeqVid;
      returnSeqFunctions() += topSeqVid;

      ItemExpr *newChild = new(wHeap) Convert (itmExpr->child(0)->castToItemExpr());
      newChild->synthTypeAndValueId(TRUE);

      itmExpr->child(0) = newChild;

      readSeqFunctions() += newChild->getValueId();
      return;
    }
  }
  
  if (itmExpr->getOperatorType() == ITM_SCALAR_MIN || 
           itmExpr->getOperatorType() == ITM_SCALAR_MAX) 
  {
    ItemExpr * chld0  = itmExpr->child(0)->castToItemExpr();
    ItemExpr * chld1  = itmExpr->child(1)->castToItemExpr();
    if ((chld0->getOperatorType() == ITM_OLAP_MIN && chld1->getOperatorType() == ITM_OLAP_MIN )|| 
        (chld0->getOperatorType() == ITM_OLAP_MAX && chld1->getOperatorType() == ITM_OLAP_MAX ))
    {
      ItmSeqOlapFunction * olap0 = (ItmSeqOlapFunction *)chld0;
      ItmSeqOlapFunction * olap1 = (ItmSeqOlapFunction *)chld1;
      if ( olap1->getframeEnd()>0)
      { 
        CMPASSERT(olap0->getframeEnd()==0);

        readSeqFunctions() -= topSeqVid;
        returnSeqFunctions() += topSeqVid;
        readSeqFunctions() += olap0->getValueId();
        
        ItemExpr *newChild = new(wHeap) Convert (olap1->child(0)->castToItemExpr());
        newChild->synthTypeAndValueId(TRUE);

        olap1->child(0) = newChild;

        readSeqFunctions() += newChild->getValueId();
      }
      else
      {
        CMPASSERT(olap1->getframeEnd()==0);

        readSeqFunctions() -= topSeqVid;
        returnSeqFunctions() += topSeqVid;
        readSeqFunctions() += olap1->getValueId();

        ItemExpr *newChild = new(wHeap) Convert (olap0->child(0)->castToItemExpr());
        newChild->synthTypeAndValueId(TRUE);

        olap0->child(0) = newChild;

        readSeqFunctions() += newChild->getValueId();
      }
      return;
    }
  }

  for (Int32 i= 0 ; i < itmExpr->getArity(); i++)
  {
    ItemExpr * chld= itmExpr->child(i);
    computeReadNReturnItems(topSeqVid,
                            chld->getValueId(),
                            outputFromChild,
                            wHeap);
  }
}//void PhysSequence::computeReadNReturnItems(ItemExpr * other)