text: /**********************************************************************
  // @@@ 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 @@@
  **********************************************************************/
  /* -*-C++-*-
  ******************************************************************************
  *
  * File:         GenRelEnforcer.C
  * Description:  Generating executor object for enforcer nodes
  *
  *
  * Created:      5/17/94
  * Language:     C++
  *
  *
  *
  *
  ******************************************************************************
  */
  
  // -----------------------------------------------------------------------
  
  #include "ComOptIncludes.h"
  #include "GroupAttr.h"
  #include "RelEnforcer.h"
  #include "RelMisc.h"
  #include "RelUpdate.h"
  #include "RelRoutine.h"
  #include "Generator.h"
  #include "GenExpGenerator.h"
  //#include "ex_stdh.h"
  #include "ExpCriDesc.h"
  #include "ComTdb.h"
  //#include "ex_tcb.h"
  #include "ComTdbSplitTop.h"
  #include "ComTdbSendTop.h"
  #include "ComTdbSplitBottom.h"
  #include "ComTdbSendBottom.h"
  #include "DefaultConstants.h"
  #include <ComCextdecs.h>
  
  /////////////////////////////////////////////////////////////////////
  //
  // Contents:
  //   
  //   Exchange::codeGen()
  //   Exchange::codeGenForSplitTop()
  //   Exchange::codeGenForESP()
  //
  //////////////////////////////////////////////////////////////////////
  
  
  //////////////////////////////////////////////////////////////////////
  //
  // Exchange::codeGen()
  //
  //////////////////////////////////////////////////////////////////////
  short Exchange::codeGen(Generator * generator)
  {
    ////////////////////////////////////////////////////////////////////////
    // In the case where there is no ESP involved, when we simply want to
    // parallelize the data stream that comes back from DP2 without having
    // to repartition, then produce a split top node only. In all other
    // cases, produce four nodes: split top, send top, send bottom, and
    // split bottom. If we create a split top node by itself, then the split
    // top node is responsible for producing partition input values,
    // otherwise the split bottom node expects to receive the partition
    // input values in a message.
    ////////////////////////////////////////////////////////////////////////
    if (isDP2Exchange())
      {
        GenAssert(isAPAPA(),"PA should have eliminated its exchange");
        return codeGenForSplitTop(generator);
      }
    else
      {
        return codeGenForESP(generator);
      }
  }
  
  short Exchange::codeGenForSplitTop(Generator * generator)
  {
    GenAssert(FALSE,"Split top PA node not supported");
    return 0;
  }
  
  short Exchange::codeGenForESP(Generator * generator)
  {
    ////////////////////////////////////////////////////////////////////////////
    // 
    // Case of generating a fragment to be downloaded into an ESP:
    // 
    //         split top     |
    //             |         |- top ESP(s) or master executor (current fragment)
    //         send top      |
    // 
    // 
    //        send bottom    |
    //             |         |
    //       split bottom    |- bottom ESP(s) (new fragment)
    //             |         |
    //        child tree     |
    // 
    ////////////////////////////////////////////////////////////////////////////
    ExpGenerator        * expGen = generator->getExpGenerator();
    MapTable            * mapTable = generator->getMapTable();
    Space               * space = generator->getSpace();
    FragmentDir         * fragmentDir = generator->getFragmentDir();
    MapTable            * parentsSavedMapTable;
  
    ////////////////////////////////////////////////////////////////////////////
    //
    // Returned atp layout (to parent of split top node):
    //
    // |-------------|   |------------------------------|
    // | input data  |   | input data  |   output row   |
    // | ( I tupps ) |   | ( I tupps ) |   ( 1 tupp )   |
    // |-------------|   |------------------------------|
    // <- input ATP ->   <-- returned ATP to parent ---->
    //
    // input data:        the atp input to this node by its parent. 
    // output row:        tupp where the row read from ESP is moved.
    //
    // The ATP layout between split top and send top nodes is the same as
    // the format between the split top node and its parent (the split top
    // node does not change the ATP format).
    //
    // ATP layout between send bottom and split bottom node:
    //
    // |---------------------|   |-----------------------------------|
    // |     input data      |   |   input data    |   output row    |
    // |     ( 1 tupp )      |   |   ( 2 tupps )   |   ( J tupps )   |
    // |---------------------|   |-----------------------------------|
    // <-- ATP sent down
    //     to split bottom -->   <-- returned ATP to send bottom ---->
    //
    // In other words, the split bottom node adds the partition input tuple
    // to the requests that it gets from the send bottom nodes. Note that
    // it does not strip the returned partition input values away before
    // passing the output rows back to the send bottom node.
    //
    // ATP layout of queues between split bottom node and its child
    // (not including the mandatory consts and temps entries)
    //
    // |---------------------------------|   |------------------------------|
    // | input data  |  part. input data |   | input data  |  output row    |
    // | ( 1 tupp )  |    ( 1 tupp )     |   | ( 2 tupps ) |  ( J tupps )   |
    // |---------------------------------|   |------------------------------|
    // <----- ATP sent down to child ---->   <-- returned ATP to parent ---->
    //
    // 2 input tupps (one is received from the parent, another one is received
    //                from the ESP manager when a partition got assigned to
    //                this ESP)
    // J output tupps (whatever the split bottom's child created)
    //
    // The messages that are exchanged between split top and split bottom
    // nodes contain 1 input tupp and 1 output tupp. The input contains
    // the characteristic inputs of the exchange node, the output contains
    // the characteristic outputs of the child of the exchange node.
    //
    // The split top node uses a work atp to encode a merge key from a child
    // row. It returns child rows in binary ascending order of the encoded keys.
    // Another tupp in its work atp is used to hold the result of the input
    // partitioning function:
    //
    // |---------------------------------------------|
    // | consts | temps  | merge key | input part no |
    // | 1 tupp | 1 tupp |  1 tupp   |    1 tupp     |
    // |---------------------------------------------|
    // <------------- split top work ATP ------------>
    //
    // The send top node uses a work atp to store the one input tupp that
    // goes down to the child (note that there is no down queue in the send
    // top node).
    //
    // |-----------------------------|
    // | consts | temps  | down tupp |
    // | 1 tupp | 1 tupp |   1 tupp  |
    // |-----------------------------|
    // <----- send top work ATP ----->
    //
    // The send bottom node uses a work ATP, too. Its work ATP contains
    // 1 tupp, describing the information sent up in the message (sql table row)
    // in addition to the mandatory consts and temps entries.
    //
    // |---------------------------------|
    // | consts | temps  |  return tupp  |
    // | 1 tupp | 1 tupp |     1 tupp    |
    // |---------------------------------|
    // <------ send bottom work ATP ----->
    //
    // Finally, the split bottom node also has a work ATP. It holds the
    // partition input data and a tupp that is used to calculate the output
    // partition number of rows that come up from the children:
    //
    // |-------------------------------------------------------------------|
    // | consts | temps  | part. input data | output part no | conv error  |
    // | 1 tupp | 1 tupp |      1 tupp      |     1 tupp     |   1 tupp    |
    // |-------------------------------------------------------------------|
    // <---------------------- split bottom work ATP ---------------------->
    //
    // Both input and output partition number are signed 32 bit quantities
    // (C++ "long" datatype) that cannot be NULL.
    //
    ////////////////////////////////////////////////////////////////////////////
  
    // work atp of split top node
    const Int32           mergeTuppAtp            = 1; // work atp
    const Int32           mergeTuppAtpIndex       = 2; // after consts and temps
    const Int32           inputPartNoAtp          = 1; // work ATP
    const Int32           inputPartNoAtpIndex     = mergeTuppAtpIndex+1;
    const Int32           splitTopWorkTupps       = inputPartNoAtpIndex+1;
  
    // work atp of send top node
    const Int32           downTuppAtp             = 1; // work atp
    const Int32           downTuppAtpIndex        = 2; // after consts and temps
    const Int32           sendTopWorkTupps        = downTuppAtpIndex+1;
  
    // work atp of send bottom node
    const Int32           returnTuppAtp           = 1; // work atp
    const Int32           returnTuppAtpIndex      = 2; // after consts and temps
    const Int32           sendBottomWorkTupps     = returnTuppAtpIndex+1;
  
    // work atp of split bottom node
    const Int32           partInputTuppAtpIndex   = 2; // after consts and temps
    const Int32           outputPartNoAtp         = 1; // work ATP
    const Int32           outputPartNoAtpIndex    = partInputTuppAtpIndex+1;
    const Int32           conversionErrorAtp      = 1; // work ATP
    const Int32           conversionErrorAtpIndex = outputPartNoAtpIndex+1;
    const Int32           splitBottomWorkTupps    = conversionErrorAtpIndex+1;
  
    // child of split bottom node
    const Int32           espChildAtp             = 0; // main ATP
    const Int32           espChildInputAtpIndex   = 2; // after const and temp
    const Int32           espChildPartInpAtpIndex = espChildInputAtpIndex+1;
    const Int32           espChildWorkTupps       = espChildPartInpAtpIndex+1;
  
    ex_cri_desc         *given_cri_desc 
      = generator->getCriDesc(Generator::DOWN);
  
    ex_cri_desc         *returned_cri_desc 
  #pragma nowarn(1506)   // warning elimination 
      = new(space) ex_cri_desc(given_cri_desc->noTuples() + 1, space);
  #pragma warn(1506)  // warning elimination 
  
    ex_cri_desc         *splitTopWorkCriDesc  = new(space) ex_cri_desc(splitTopWorkTupps,space);
    ex_cri_desc         *sendTopWorkCriDesc   = new(space) ex_cri_desc(sendTopWorkTupps,space);
    ex_cri_desc         *sendBottomWorkCriDesc;
    ex_cri_desc         *splitBottomWorkCriDesc;
    ex_cri_desc         *sendBottomDownCriDesc;
    ex_cri_desc         *sendBottomUpCriDesc;
    ex_cri_desc         *child_down_cri_desc;
    ex_cri_desc         *child_up_cri_desc;
    
    // two value id lists describing the columns that are sent up and down
    // via IPC messages, the map tables describing the data sent, and the
    // length of the sent records
    ValueIdList         sentInputValues;
    MapTable            *initialESPMapTable = NULL;
    ULng32       downRecordLength;
    ULng32       downSqlBufferLength;
    ULng32       numDownBuffers = getDefault(GEN_SNDT_NUM_BUFFERS);
    CostScalar          numRowsDown = (CostScalar) 1; // $$$$ get from input log props
  
    ValueIdList         returnedOutputValues;
    MapTable            *returnedValuesMapTable = NULL;
    ULng32       upRecordLength = 0;
    ULng32       upSqlBufferLength = 0;
    ULng32       numUpBuffers = getDefault(GEN_SNDB_NUM_BUFFERS);
    CostScalar          numRowsUp = getEstRowsUsed().value();
    ULng32       mergeKeyLength = 0;
  
    const PartitioningFunction *topPartFunc = getTopPartitioningFunction();
    const PartitioningFunction *bottomPartFunc =getBottomPartitioningFunction();
    Lng32                numTopPartitions = topPartFunc->getCountOfPartitions();
    Lng32                numBottomPartitions = bottomPartFunc->getCountOfPartitions();
    Lng32                numTopEsps = numTopPartitions;
    Lng32                numBottomEsps = numBottomPartitions;
    NABoolean           possiblePartNoConversionError = FALSE;
  
    // a value id list describing the values that the ESP needs to determine
    // the partition it is working on (part. number or key range) and its
    // corresponding map table
    const ValueIdList   &partitionInputValues = getBottomPartitionInputValues();
    Lng32                partitionInputDataLength = 0;
  
    // the four TDBs that will be generated
    ComTdbSplitTop    *splitTop = NULL;
    ComTdbSendTop     *sendTop = NULL;
    ComTdbSplitBottom *splitBottom = NULL;
    ComTdbSendBottom  *sendBottom = NULL;
  
    // expressions to be generated:
    //
    // - calculate input partition number (split top)
    // - encode merge key from a child into a key buffer so that
    //   a binary comparison can be done to merge rows
    // - move parent's input values into a contiguous buffer (send top)
    // - calculate output partition number (split bottom)
    // - move output values into contiguous buffer (send bottom)
    //
    ex_expr             *calcInputPartNoExpr    = NULL;
    ex_expr             *mergeKeyExpr           = NULL;
    ex_expr             *inputMoveExpr          = NULL;
    ex_expr             *calcOutputPartNoExpr   = NULL;
    ex_expr             *outputMoveExpr         = NULL;
  
    // helpers for skew buster
    //
    NABoolean useSkewBuster       = FALSE;
    NABoolean broadcastSkew       = FALSE;
    NABoolean broadcastOneRow     = FALSE;
    Lng32 numSkewHashValues        = 0;
    Int64 *skewHashValues         = NULL;
    SplitBottomSkewInfo *skewInfo = NULL;
    Int32 initialRoundRobin         = 0;
    Int32 finalRoundRobin           = numTopEsps - 1;
  
    generator->incrEspLevel();
  
    // If this is an extract producer query then manufacture a security key
    char extractSecurityKey[100];
    if (isExtractProducer_)
    {
      ComUID uid;
      uid.make_UID();
      Int64 i64 = uid.get_value();
      str_sprintf(extractSecurityKey, "%Ld", i64);
    }
    else
    {
      str_sprintf(extractSecurityKey, "0");
    }
  
    // Raise an error if this is a parallel extract consumer and the
    // child has a selection predicate.
    PhysicalExtractSource *extractConsumerChild = NULL;
    if (isExtractConsumer_)
    {
      GenAssert(child(0)->getOperatorType() == REL_EXTRACT_SOURCE,
                "Child of root must be REL_EXTRACT_SOURCE");
      extractConsumerChild = (PhysicalExtractSource *) (child(0).getPtr());
      if (extractConsumerChild->getSelectionPred().entries() != 0)
      {
        *CmpCommon::diags() << DgSqlCode(-7005);
        GenExit();
        return NULL;
      }
    }
  
    // ---------------------------------------------------------------------
    // make lists of values that go over the wire: input, output, and
    // partition input data
    // ---------------------------------------------------------------------
  
    // Take the characteristic inputs of the exchange node: these are the
    // input values that will travel down the wire. Decide on some sequence
    // in which to put them (choose value id order). Both split top and
    // split bottom node have to use the same sequence, of course.
    // Note that some additional input values that identify the actual
    // partition assigned to the split bottom node may exist and come from
    // another source. Those values are not added to the list.
    // This code decides the sequence of the values in the sent records.
    // Constants are not sent up or down in messages.
    for (ValueId x = getGroupAttr()->getCharacteristicInputs().init();
         getGroupAttr()->getCharacteristicInputs().next(x);
         getGroupAttr()->getCharacteristicInputs().advance(x))
      {
        if (x.getItemExpr()->getOperatorType() != ITM_CONSTANT AND
            NOT partitionInputValues.contains(x))
          sentInputValues.insert(x);
      }
  
    // Normally the outputs we send up are the non-constant values in
    // the characteristic outputs of the child.
    // 
    // Exceptions to this rule:
    // - A parallel extract consumer returns all of the child's
    //   output columns.
    // - A parallel extract producer returns everything in the root's
    //   select list.
    // 
    // Note that this code decides the sequence of the values in the
    // sent records.
    // 
    if (isExtractConsumer_)
    {
      returnedOutputValues =
        extractConsumerChild->getTableDesc()->getColumnList();
    }
    else if (isExtractProducer_)
    {
      GenAssert(extractSelectList_,
                "Select list should not be NULL for a producer query");
      returnedOutputValues = *extractSelectList_;
  
      RETDesc *retDesc = generator->getBindWA()->getTopRoot()->getRETDesc();
  
      for (CollIndex i = 0; i < returnedOutputValues.entries(); i++)
      {
        ValueId val_id = returnedOutputValues[i];
        ItemExpr * item_expr = val_id.getItemExpr();
        if (! (val_id.getType() == retDesc->getType(i)))
        {
          item_expr = new(generator->wHeap()) Cast(item_expr,
                                      &(retDesc->getType(i)));
          item_expr->bindNode(generator->getBindWA());
  
          returnedOutputValues[i] = item_expr->getValueId();
        }
      }
    }
    else
    {
      for (ValueId y =
             child(0)->getGroupAttr()->getCharacteristicOutputs().init();
           child(0)->getGroupAttr()->getCharacteristicOutputs().next(y);
           child(0)->getGroupAttr()->getCharacteristicOutputs().advance(y))
      {
        if (y.getItemExpr()->getOperatorType() != ITM_CONSTANT)
          returnedOutputValues.insert(y);
      }
    }
  
    // ---------------------------------------------------------------------
    // generate expressions to be evaluated in the client
    // ---------------------------------------------------------------------
  
    // generate expr to calculate input partition number
    if (FALSE /*  enable this later $$$$ */)
      {
        ItemExpr *bottomPartExpr = ((PartitioningFunction *)
          bottomPartFunc)->createPartitioningExpression();
        ValueIdList bottomPartList;
        ULng32 dummyLength;
  
        if (bottomPartExpr)
          {
            bottomPartList.insert(bottomPartExpr->getValueId());
            expGen->generateContiguousMoveExpr(
                       bottomPartList,
                       -1,
                       inputPartNoAtp,
                       inputPartNoAtpIndex,
                       ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
                       dummyLength,
                       &calcInputPartNoExpr);
          }
      }
  
    // expression to move input values into a single tuple
    expGen->generateContiguousMoveExpr(sentInputValues,
                                       -1, // add convert nodes
                                       downTuppAtp,
                                       downTuppAtpIndex,
                                       ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
                                       downRecordLength,
                                       &inputMoveExpr,
                                       0, 
                                       ExpTupleDesc::SHORT_FORMAT,
                                       &initialESPMapTable);
  
    // we MUST be able to fit at least one row into a buffer
  #pragma nowarn(1506)   // warning elimination
    downSqlBufferLength = 
      MAXOF((ULng32) ComTdbSendTop::minSendBufferSize(downRecordLength),
            (ULng32) (downMessageBufferLength_.getValue() * 1024));
  #pragma warn(1506)  // warning elimination 
  
    CollIndex childFragmentId = 0;
  
    NABoolean thisExchangeUsesSM = FALSE;
    if (thisExchangeCanUseSM(generator->getBindWA()))
      thisExchangeUsesSM = TRUE;
   
    if (isExtractConsumer_)
    {
      // Each extract consumer can change the format back to exploded if need be
      // to get the work distributed.
      //generator->setExplodedInternalFormat();
        ExpTupleDesc::TupleDataFormat tupleFormat = ExpTupleDesc::SQLARK_EXPLODED_FORMAT;
  
      // For an extract consumer we are going to bypass everything
      // related to the bottom fragment. All we need to do before
      // generating the top fragment (the split top and send top TDBs)
      // is add reply buffer output values to the map table.
      ULng32 dummyRecLen = 0;
      expGen->processValIdList(
        returnedOutputValues,                  // [IN] ValueIdList
        tupleFormat,        // [IN] tuple data format
        dummyRecLen,                           // [OUT] tuple length 
        0,                                     // [IN] atp number
        given_cri_desc->noTuples(),            // [IN] index into atp
        NULL,                                  // [optional OUT] tuple desc
        ExpTupleDesc::SHORT_FORMAT             // [optional IN] tuple desc format
        );
    }
    else
    {
      // ---------------------------------------------------------------------
      // Generate expressions and TDBs to be executed in the ESP
      // ---------------------------------------------------------------------
      childFragmentId = generator->getFragmentDir()->
        pushFragment(FragmentDir::ESP,
                     numBottomEsps,
                     getBottomPartitioningFunction());
      
      if (CmpCommon::getDefault(ODBC_PROCESS) == DF_ON)
      {
        generator->getFragmentDir()->setAllEspFragmentsNeedTransaction();
      }
      
      space = generator->getSpace();
      
      // generate a small piece of space at the beginning of this fragment,
      // to get rid of the useless object that has offset 0 (an offset
      // of 0 is identified as a NULL object)
      space->allocateAlignedSpace(1);
      
      // save my own stack of map tables and start over from scratch
      // with a single map table for the child that describes the received
      // input values (the split bottom node will pass this tupp down
      // to its child)
      parentsSavedMapTable = generator->getMapTable();
      generator->setMapTable(initialESPMapTable);
      mapTable = generator->getMapTable();
      
      // the sent down values' attributes need to be changed now from
      // the work atp from which they are referenced to the actual atp
      // in which they are passed to the child of the split bottom node
      // which is the first tupp in the main atp (atp = 0, atpindex = 2)
      CollIndex oi = 0;
      for (oi = 0; oi < sentInputValues.entries(); oi++)
      {
        Attributes * attr =
          generator->getMapInfo(sentInputValues[oi])->getAttr();
  
        // the row coming up from the connection is returned as the tupp
        // following the input tupps
        attr->setAtp(espChildAtp);
        attr->setAtpIndex(espChildInputAtpIndex);
      }
      
      // ---------------------------------------------------------------------
      // set down cri desc for child
      // ---------------------------------------------------------------------
      sendBottomWorkCriDesc = new(space) ex_cri_desc(sendBottomWorkTupps,space);
      splitBottomWorkCriDesc = new(space)
        ex_cri_desc(splitBottomWorkTupps,space);
      child_down_cri_desc = new(space) ex_cri_desc(espChildWorkTupps,space);
      generator->setCriDesc(child_down_cri_desc, Generator::DOWN);
      
      // allocate message data CRI descriptors for the server side
      sendBottomDownCriDesc   = new(space) ex_cri_desc(3,space);
      
      // Assign attributes to the partition input values
      // such that they appear in a contiguous tuple
      if (NOT partitionInputValues.isEmpty())
      {
        ((PartitioningFunction *) bottomPartFunc)->
          generatePivLayout(generator,
                            partitionInputDataLength,
                            espChildAtp,
                            espChildPartInpAtpIndex,
                            NULL);
        
        generator->getFragmentDir()->
          setPartInputDataLength(childFragmentId,
                                 (ULng32) partitionInputDataLength);
      }
      
      // Copy #BMOs value from Exchange node into the fragment
      generator->getFragmentDir()->setNumBMOs(childFragmentId, numBMOs_);
      generator->getFragmentDir()->setBMOsMemoryUsage(childFragmentId, 
                                                      BMOsMemoryUsage_.value());
      
      //store the numBottomEsps from this node for access by child sort operator.
      generator->setNumESPs(numBottomEsps);
  
      generator->getFragmentDir()->setEspLevel(childFragmentId, 
  					     generator->getEspLevel());
  
      // ---------------------------------------------------------------------
      // generate child tree
      // ---------------------------------------------------------------------
      child(0)->codeGen(generator);
      
      ComTdb * child_tdb  = (ComTdb *)(generator->getGenObj());
      ExplainTuple *childExplainTuple = generator->getExplainTuple();
      child_up_cri_desc   = generator->getCriDesc(Generator::UP);
      
      // split bottom passes up queue entries unchanged to send bottom node
      // (never mind the partition input values that travel with it, they
      // don't get moved into the message buffers)
      sendBottomUpCriDesc = child_up_cri_desc;
      
      // ---------------------------------------------------------------------
      // generate expressions for split bottom node
      // ---------------------------------------------------------------------
      
      // generate expr to calculate output partition number
      if ((numTopEsps > 1) && (NOT isAnESPAccess()))
      {
        UInt32 expectedPartInfoLength = sizeof(Lng32);
  
        ItemExpr *topPartExpr = topPartFunc->getPartitioningExpression();
  
        // A ReplicationPartitioningFunction has no partitioning expression.
        if (topPartExpr)
        {
          ValueIdList topPartExprAsList;
          ULng32 partNoValLength;
          ItemExpr *convErrExpr = topPartFunc->getConvErrorExpr();
          
          if (convErrExpr)
          {
            // There is a "Narrow" operator in the partitioning expression
            // and the split bottom TCB will have to handle cases of
            // errors during a data conversion. Add an entry to the
            // map table that assigns convErrExpr the right location
            // (ATP=conversionErrorAtp, ATPIndex=conversionErrorAtpIndex,
            // Offset=0)
            possiblePartNoConversionError = TRUE;
            
            ValueId convErrorValId = convErrExpr->getValueId();
            // should this use a separate map table?
            Attributes *convAttr = generator->addMapInfo(
                     convErrorValId, NULL)->getAttr();
            convAttr->setAtp(conversionErrorAtp);
            convAttr->setAtpIndex(conversionErrorAtpIndex);
            convAttr->setOffset(0);
            convAttr->setTupleFormat(ExpTupleDesc::SQLARK_EXPLODED_FORMAT);
          }
          
          // executor has hard-coded assumption that the result is an unsigned
          // long.
          // Add a Cast node to convert result to an unsigned long.
          
          ItemExpr * cast = new (generator->wHeap()) 
            Cast(topPartExpr, 
                 new (generator->wHeap()) 
                 SQLInt(FALSE,
                        topPartExpr->getValueId().getType().
                        supportsSQLnullLogical()));
          
          cast->bindNode(generator->getBindWA());
          
          topPartExprAsList.insert(cast->getValueId());
        
            const SkewedDataPartitioningFunction* skpf =
             topPartFunc->castToSkewedDataPartitioningFunction();
   
            // Process getSkewProperty.
            skewProperty::skewDataHandlingEnum skew = 
                (skpf) ? skpf->getSkewProperty().getIndicator() :
                         skewProperty::ANY;
  
            if ( skpf AND 
                 (skew == skewProperty::BROADCAST || 
                 skew == skewProperty::UNIFORM_DISTRIBUTE) )
              {
  
                // 1. Add the hashing expression (to return hash value
                //    only) to the topPartExprAsList.
               
                ItemExpr * hashExpr = topPartFunc->getHashingExpression();
  
                GenAssert( hashExpr != NULL, 
                    "getHashingExpression returned NULL");
  
                ItemExpr * hashExprResult = new (generator->wHeap()) 
                  Cast( hashExpr, 
                       new (generator->wHeap()) 
                       SQLLargeInt(TRUE, // allow negative values.
                                   FALSE // no nulls.
                            ));
  
                hashExprResult->bindNode(generator->getBindWA());
                topPartExprAsList.insert(hashExprResult ->getValueId());
  
                expectedPartInfoLength += sizeof(Int64) // hash value 
                                           + 4;          // alignment
  
                // 2. Prepare the array of hash values which 
                //    indicate possible skewed partitioning keys.
  
  #if 0
                // Please note that I never reserved COMP_BOOL_154 
                // and just added this for my private build. 
                if (CmpCommon::getDefault(COMP_BOOL_154) == DF_ON)
                  {
                    // Test limit of 10000.  Don't care that they
                    // are actual skewed values.
  
                    numSkewHashValues = 10000;     
                    skewHashValues = new (space) Int64[numSkewHashValues];
                    for (Int32 sv = 0; sv < numSkewHashValues; sv++)
                      skewHashValues[sv] = (Int64) sv;
                    skewInfo = new (space) 
                        SplitBottomSkewInfo(numSkewHashValues, skewHashValues);
                  }
                else
  #else
                  {
                    Int64List *partFuncSkewedValues = 
                        ((SkewedDataPartitioningFunction*)topPartFunc)->
                            buildHashListForSkewedValues();
  
                    GenAssert(partFuncSkewedValues != NULL,
                          "NULL returned from buildHashListForSkewedValues");
  
                    numSkewHashValues = partFuncSkewedValues->entries();
  
                    GenAssert(numSkewHashValues > 0,
                    "buildHashListForSkewedValues returned zero or fewer values");
  
                    skewHashValues = new (space) Int64[numSkewHashValues];
  
                    for (Int32 sv = 0; sv < numSkewHashValues; sv++)
                      skewHashValues[sv] = (*partFuncSkewedValues)[sv];
  
                    skewInfo = new (space) 
                        SplitBottomSkewInfo(numSkewHashValues, skewHashValues);
                  }
  #endif
  
                // 3. Prepare the skew properties for ComTdbSplitBottom.
                
                useSkewBuster = TRUE;
                broadcastOneRow = FALSE;
                if (skew == skewProperty::BROADCAST)
                {
                  broadcastSkew = TRUE;
                  broadcastOneRow = skpf->getSkewProperty().getBroadcastOneRow();
                }
                else
                  broadcastSkew = FALSE;
                srand((UInt32) JULIANTIMESTAMP(3));
                initialRoundRobin = rand() % numTopEsps;
                Int32 antiSkewESPs = skpf->getSkewProperty().getAntiSkewESPs();
                if (antiSkewESPs <= 0)
                  {
                    // For hash join, make sure all consumer ESPs get the 
                    // the skewed value rows.
                    antiSkewESPs = numTopEsps;
                  }
                else
                  {
                    // For NJ OCR skewbuster, just use the CQD.
                  }
  
                finalRoundRobin = (initialRoundRobin + antiSkewESPs - 1)
                                 % numTopEsps;
              }
            
            expGen->generateContiguousMoveExpr(
                        topPartExprAsList,
                        0, // don't add convert node
                        outputPartNoAtp,
                        outputPartNoAtpIndex,
                        ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
                        partNoValLength,
                        &calcOutputPartNoExpr);
            GenAssert(partNoValLength == expectedPartInfoLength,
                      "Unexpected length of result of part. function.");
          }
        else
        {
          GenAssert(topPartFunc->isAReplicationPartitioningFunction(),
                    "Did not create part. expr. for repartitioning function");
        }
      }
  
      if(isAnESPAccess())
        {
        GenAssert((calcOutputPartNoExpr == NULL),
  		"ESP ACCESS node must not have calcOutputPartNoExpr");
        }
      
      // If this is a top level ESP then switch from internal format to
      // exploded format here in the ESP rather than having to do the switch
      // of data formats in the master for all ESPs.
       ExpTupleDesc::TupleDataFormat tupleFormat = generator->getInternalFormat();
       NABoolean resizeCifRecord = FALSE;
       NABoolean considerBufferDefrag = FALSE;
       NABoolean bmo_affinity = (CmpCommon::getDefault(COMPRESSED_INTERNAL_FORMAT_BMO_AFFINITY) == DF_ON);
       if (! bmo_affinity &&
           getCachedTupleFormat() != ExpTupleDesc::UNINITIALIZED_FORMAT &&
           CmpCommon::getDefault(COMPRESSED_INTERNAL_FORMAT) == DF_SYSTEM &&
           CmpCommon::getDefault(COMPRESSED_INTERNAL_FORMAT_BMO) == DF_SYSTEM)
       {
         resizeCifRecord = getCachedResizeCIFRecord();
         tupleFormat = getCachedTupleFormat();
         considerBufferDefrag = getCachedDefrag() && resizeCifRecord;
       }
       else
       {
         tupleFormat = determineInternalFormat( returnedOutputValues,
                                                this,
                                                resizeCifRecord,
                                                generator,
                                                bmo_affinity,
                                                considerBufferDefrag);
         considerBufferDefrag = considerBufferDefrag && resizeCifRecord;
       }
  
      if ((CmpCommon::getDefault(COMPRESSED_INTERNAL_FORMAT_ROOT_DOES_CONVERSION) == DF_OFF ||
           getExtractProducerFlag()) && // if extract producer convert to exploded format
      	generator->isCompressedInternalFormat() &&
      	isParentRoot())
  			{
           tupleFormat = ExpTupleDesc::SQLARK_EXPLODED_FORMAT; 
           resizeCifRecord = FALSE;
  			   considerBufferDefrag = FALSE;
  		   }
  
      // move J output tupps to a contiguous buffer (have to move even if
      // the output already has a single tupp, since we have to allocate the
      // target tupp inside a message buffer)
      expGen->generateContiguousMoveExpr(returnedOutputValues,
                                         -1, // add convert nodes
                                         returnTuppAtp,
                                         returnTuppAtpIndex,
                                         tupleFormat,
                                         upRecordLength,
                                         &outputMoveExpr,
                                         NULL,
                                         ExpTupleDesc::SHORT_FORMAT,
                                         &returnedValuesMapTable);
  
      // we MUST be able to fit at least one row into a buffer
  #pragma nowarn(1506)   // warning elimination 
      upSqlBufferLength =
        MAXOF((ULng32) ComTdbSendTop::minReceiveBufferSize(upRecordLength),
              (ULng32) (upMessageBufferLength_.getValue() * 1024));
  #pragma warn(1506)  // warning elimination 
  
      // ---------------------------------------------------------------------
      // generate send bottom tdb
      // ---------------------------------------------------------------------
   
  #pragma nowarn(1506)   // warning elimination 
      sendBottom = new(space) ComTdbSendBottom(
         outputMoveExpr,
         (queue_index)getDefault(GEN_SNDB_SIZE_DOWN), 
         (queue_index)getDefault(GEN_SNDB_SIZE_UP), 
         sendBottomDownCriDesc,
         sendBottomUpCriDesc,
         sendBottomWorkCriDesc,
         returnTuppAtpIndex,
         downRecordLength,
         upRecordLength,
         downSqlBufferLength,
         numDownBuffers, 
         upSqlBufferLength,
         numUpBuffers,
         (Cardinality) numRowsDown.value(),
         (Cardinality) numRowsUp.value());
  #pragma warn(1506)  // warning elimination 
   
      sendBottom->setConsiderBufferDefrag(considerBufferDefrag);
      sendBottom->setCIFON( (tupleFormat == ExpTupleDesc::SQLMX_ALIGNED_FORMAT));
      // ---------------------------------------------------------------------
      // generate split bottom tdb
      // ---------------------------------------------------------------------
      splitBottom = new(space) ComTdbSplitBottom(
         child_tdb,
         sendBottom,
         calcOutputPartNoExpr,
         outputPartNoAtpIndex,
         possiblePartNoConversionError,
         conversionErrorAtpIndex,
         partInputTuppAtpIndex,
         partitionInputDataLength,
         (Cardinality) getGroupAttr()->
         getOutputLogPropList()[0]->getResultCardinality().value(),
         sendBottomDownCriDesc,
         sendBottomUpCriDesc,
         splitBottomWorkCriDesc,
         TRUE,
         numTopEsps,
         numTopPartitions,
         numBottomEsps,
         numBottomPartitions,
         skewInfo);
      // It is important that we initialize splitBottom first
      // and then sendBottom to get the tdbId 
      generator->initTdbFields(splitBottom);
      generator->initTdbFields(sendBottom);
  
      splitBottom->setUseSkewBuster(useSkewBuster);
      splitBottom->setBroadcastSkew(broadcastSkew);
  
      splitBottom->setInitialRoundRobin(initialRoundRobin);
      splitBottom->setFinalRoundRobin(finalRoundRobin);
      splitBottom->setBroadcastOneRow(broadcastOneRow);
  
      splitBottom->setCIFON( (tupleFormat == ExpTupleDesc::SQLMX_ALIGNED_FORMAT));
  
      if (generator->processLOB())
        splitBottom->setProcessLOB(TRUE);
  
      if (CmpCommon::getDefault(COMP_BOOL_153) == DF_ON)
        splitBottom->setForceSkewRoundRobin(TRUE);
      
      splitBottom->setAbendType(
        (Lng32) CmpCommon::getDefaultNumeric(COMP_INT_39) );
   double cpuLimitCheckFreq = CmpCommon::getDefaultNumeric(COMP_INT_48);
      if (cpuLimitCheckFreq > SHRT_MAX)
        cpuLimitCheckFreq = SHRT_MAX;
      splitBottom->setCpuLimitCheckFreq((Int32) cpuLimitCheckFreq);
  
     
      // tell the tdb whether we collect statistics or not
      if (generator->computeStats())
      {
        splitBottom->setCollectStats(generator->computeStats());
        splitBottom->setCollectStatsType(generator->collectStatsType());
        splitBottom->setCollectRtsStats(generator->collectRtsStats());
      }
  
      //Set overflow mode. Needed for accumulated stats.
      splitBottom->setOverflowMode(generator->getOverflowMode());
  
      // Config query execution limits.
      Lng32 cpuLimit = (Lng32) CmpCommon::getDefaultNumeric(QUERY_LIMIT_SQL_PROCESS_CPU);
      if (cpuLimit > 0)
        splitBottom->setCpuLimit(cpuLimit);  
  
      if (CmpCommon::getDefault(QUERY_LIMIT_SQL_PROCESS_CPU_DEBUG) == DF_ON)
        splitBottom->setQueryLimitDebug();
  
      // There are two SeaMonster flags in split bottom
      // * Whether SM is used somewhere in the query
      // * Whether SM is used in this exchange
      // 
      // Send top, send bottom, and split top only carry the exchange
      // flag not the query-level flag
      if (generator->getQueryUsesSM())
        splitBottom->setQueryUsesSM();
      
      if (thisExchangeUsesSM)
      {
        splitBottom->setExchangeUsesSM();
        sendBottom->setExchangeUsesSM();
      }
  
      if (isExtractProducer_)
      {
        // Set a flag in the split bottom and send bottom TDBs
        splitBottom->setExtractProducerFlag();
        sendBottom->setExtractProducerFlag();
  
        // Create a copy of the security key and give the split bottom a
        // pointer to it
        char *keyCopy =
          space->allocateAndCopyToAlignedSpace(extractSecurityKey,
                                               str_len(extractSecurityKey),
                                               0);
  
        ComExtractProducerInfo *producerInfo = new (space)
          ComExtractProducerInfo();
  
        producerInfo->setSecurityKey(keyCopy);
        splitBottom->setExtractProducerInfo(producerInfo);
      }
      
      if (hash2RepartitioningWithSameKey())
        splitBottom->setMWayRepartitionFlag();
  
      if(isAnESPAccess()) {
        splitBottom->setMWayRepartitionFlag();
        splitBottom->setIsAnESPAccess();
        generator->getFragmentDir()->setSoloFragment(childFragmentId, TRUE);
      }
  
  // Compute memory estimate for SendBottom
  // -----------------------------------------------------------------------------------------
      double totalMemorySB = numRowsUp.value() * upRecordLength + 
                             numRowsDown.value() * downRecordLength;
      // divide by 2 to get an everage values since actual number of up buffers
      // varies at run time based on the rate at which rows are consumed and 
      // traffic at a given a SendBottom.  In bytes
      const double memoryLimitPerCpuSB = 
        (numUpBuffers * upSqlBufferLength)/2 +
         (numDownBuffers * downSqlBufferLength); 
  
      if ( bottomPartFunc -> isAReplicationPartitioningFunction() == TRUE ) 
      {
        totalMemorySB *= numBottomEsps;
      }
      double memoryPerCpuSB = totalMemorySB/numBottomEsps ;
      if ( memoryPerCpuSB > memoryLimitPerCpuSB ) 
          memoryPerCpuSB = memoryLimitPerCpuSB;
      totalMemorySB = memoryPerCpuSB * numBottomEsps ;
  
  // Compute memory estimate for SendTop
  // -----------------------------------------------------------------------------------------
      double totalMemoryST = numRowsUp.value() * upRecordLength + 
                             numRowsDown.value() * downRecordLength;
      // divide by 2 to get an everage values since actual number of up buffers
      // varies at run time based on the rate at which rows are consumed and 
      // traffic at a given a SendBottom.  In bytes
      const double memoryLimitPerCpuST = 
        (numUpBuffers * upSqlBufferLength)/2 + 
        (numDownBuffers * downSqlBufferLength);  
  
      if ( topPartFunc -> isAReplicationPartitioningFunction() == TRUE ) 
      {
        totalMemoryST *= numTopEsps;
      }
      double memoryPerCpuST = totalMemoryST/numTopEsps ;
      if ( memoryPerCpuST > memoryLimitPerCpuST ) 
          memoryPerCpuST = memoryLimitPerCpuST;
      totalMemoryST = memoryPerCpuST * numTopEsps ;
  
      generator->addToTotalEstimatedMemory(totalMemoryST + totalMemorySB);
  
      if(!generator->explainDisabled())
      {
        Lng32 sbMemEstInKBPerCPU = (Lng32) ((totalMemoryST + totalMemorySB) / 1024) ;
        sbMemEstInKBPerCPU = sbMemEstInKBPerCPU/
          (MAXOF(generator->compilerStatsInfo().dop(),1));
        generator->setOperEstimatedMemory(sbMemEstInKBPerCPU);
         
        generator->setExplainTuple(
  	  addExplainInfo(splitBottom, childExplainTuple, 0, generator));
        sendBottom->setExplainNodeId(generator->getExplainNodeId());
  
        generator->setOperEstimatedMemory(0);
      }
      
      // ExplainTuple *sendBotExplain = 
      //  addExplainInfo(sendBottom, splitBotExplain, 0, generator);
      
      generator->getFragmentDir()->setTopObj((char *) splitBottom);
      
      // ---------------------------------------------------------------------
      // return back to the original fragment, the server fragment has
      // been generated
      // ---------------------------------------------------------------------
      generator->getFragmentDir()->popFragment();
      space = generator->getSpace();
      
      // get rid of the child's map tables, restore the old view
      generator->removeAll();
      
      generator->setMapTable(parentsSavedMapTable);
      mapTable = generator->getMapTable();
      
      // the returned values are available to the parent node
      generator->appendAtEnd(returnedValuesMapTable);
      
      // the returned values' attributes need to be changed now from
      // the work atp from which they are referenced to the actual atp
      // in which they are returned to the parent of the send top node
      // (atp = 0, atpindex = # of input tupps)
      for (oi = 0; oi < returnedOutputValues.entries(); oi++)
      {
        Attributes * attr = generator->getMapInfoFromThis(
  	   generator->getLastMapTable(),
             returnedOutputValues[oi])->getAttr();
  
        // the row coming up from the connection is returned as the tupp
        // following the input tupps
        attr->setAtp(0);
  #pragma nowarn(1506)   // warning elimination 
        attr->setAtpIndex(given_cri_desc->noTuples());
  #pragma warn(1506)  // warning elimination 
      }
      
      // ---------------------------------------------------------------------
      // Generate comparison expression for merging sorted streams
      // ---------------------------------------------------------------------
      
      // generate merge key expression. If this is an extract
      // producer query, there is no need for merging because data rows
      // are going to be routed to alternate masters.
      // Also, if this esp change was added during PreCodeGen for halloween
      // protection, ignore the merge, because otherwise we get deadlocks
      // -- see solution 10-081023-6759.
      if (!isExtractProducer_ && 
          !forcedHalloweenProtection_ && 
          sortKeyForMyOutput_.entries() > 0)
      {
        ValueIdList encodeList;
        
        for (CollIndex i = 0; i < sortKeyForMyOutput_.entries(); i++)
        {
          ItemExpr *ix = sortKeyForMyOutput_[i].getItemExpr();
          
          if (ix->getOperatorType() == ITM_INVERSE)
            ix = new(generator->wHeap()) CompEncode(ix->child(0).getPtr(),TRUE);
          else
            ix = new(generator->wHeap()) CompEncode(ix,FALSE);
          
          ix->synthTypeAndValueId();
          encodeList.insert(ix->getValueId());
        }
        
        expGen->generateContiguousMoveExpr(
             encodeList, 
             0, // don't add convert nodes,
             mergeTuppAtp,
             mergeTuppAtpIndex,
             ExpTupleDesc::SQLMX_KEY_FORMAT,
             mergeKeyLength,
             &mergeKeyExpr);
      }
      
    } // if (isExtractConsumer) else ...
  
    // ---------------------------------------------------------------------
    // Generate send top tdb
    // ---------------------------------------------------------------------
  
  #pragma nowarn(1506)   // warning elimination 
    sendTop = new(space) ComTdbSendTop(
         childFragmentId,
         inputMoveExpr,
         given_cri_desc,
         returned_cri_desc,
         NULL,
         NULL, // get rid of these later
         sendTopWorkCriDesc,
         downTuppAtpIndex,
         (queue_index)getDefault(GEN_SNDT_SIZE_DOWN),
         (queue_index)getDefault(GEN_SNDT_SIZE_UP),
         downRecordLength,
         upRecordLength, 
         downSqlBufferLength,
         numDownBuffers,
         upSqlBufferLength,
         numUpBuffers,
         (Cardinality) numRowsDown.value(),
         (Cardinality) numRowsUp.value(),
         (CmpCommon::getDefault(EXE_DIAGNOSTIC_EVENTS) == DF_ON));
  #pragma warn(1506)  // warning elimination 
    generator->initTdbFields(sendTop);
  
    if (isExtractConsumer_)
    {
      // Set a flag in the send top TDB
      sendTop->setExtractConsumerFlag();
  
      // Create a copy of the phandle string and the security key and
      // give the TDB pointers to them
      const NAString &espForExtract = extractConsumerChild->getEspPhandle();
      const NAString &securityKey = extractConsumerChild->getSecurityKey();
      const char *espData = espForExtract.data();
      const char *keyData = securityKey.data();
      char *espCopy =
        space->allocateAndCopyToAlignedSpace(espData, str_len(espData), 0);
      char *keyCopy =
        space->allocateAndCopyToAlignedSpace(keyData, str_len(keyData), 0);
  
      ComExtractConsumerInfo *consumerInfo = new (space)
        ComExtractConsumerInfo();
  
      consumerInfo->setEspPhandle(espCopy);
      consumerInfo->setSecurityKey(keyCopy);
      sendTop->setExtractConsumerInfo(consumerInfo);
    }
  
    // Allow the fix for soln 10-100508-0135 to be undone
    if (CmpCommon::getDefault(COMP_BOOL_118) == DF_ON)
      sendTop->setUseOldStatsNoWaitDepth();
  
    // Set the flag that tells send top TCB whether to restrict the
    // number of send buffers to 1. By default the CQD is ON and the
    // flag is set.
    if (CmpCommon::getDefault(GEN_SNDT_RESTRICT_SEND_BUFFERS) == DF_ON)
      sendTop->setRestrictSendBuffers();
  
    // ---------------------------------------------------------------------
    // Create a split top node, to be executed in the current process
    // ---------------------------------------------------------------------
    splitTop = new(space) ComTdbSplitTop(
         sendTop,
         calcInputPartNoExpr,
         inputPartNoAtpIndex,
         mergeKeyExpr,
         mergeTuppAtpIndex,
         (Lng32) mergeKeyLength,
         NULL,
         -1,
         -1,
         given_cri_desc,
         returned_cri_desc,
         given_cri_desc,
         splitTopWorkCriDesc,
         FALSE,
         (queue_index) 2,
         (queue_index)getDefault(GEN_SPLT_SIZE_UP),
         (Cardinality) getGroupAttr()->
         getOutputLogPropList()[0]->getResultCardinality().value(),
         numBottomPartitions,
  #pragma warning (disable : 4244)  //warning elimination
  #pragma nowarn(1506)   // warning elimination 
         CmpCommon::getDefaultNumeric(STREAM_TIMEOUT),
         getDefault(GEN_SID_NUM_BUFFERS),
         getDefault(GEN_SID_BUFFER_SIZE)
         );
  #pragma warn(1506)  // warning elimination 
  #pragma warning (default : 4244)  //warning elimination
  
    generator->initTdbFields(splitTop);
  
    if (generator->isLRUOperation())
      splitTop->setLRUOperation();
  
    if (thisExchangeUsesSM)
    {
      splitTop->setExchangeUsesSM();
      sendTop->setExchangeUsesSM();
    }
  
    if (isExtractProducer_)
    {
      // Set a flag in the split top and send top TDBs
      splitTop->setExtractProducerFlag();
      sendTop->setExtractProducerFlag();
  
      // Create a copy of the security key and give the split top a
      // pointer to it
      char *keyCopy =
        space->allocateAndCopyToAlignedSpace(extractSecurityKey,
                                             str_len(extractSecurityKey),
                                             0);
      
      ComExtractProducerInfo *producerInfo = new (space)
        ComExtractProducerInfo();
      
      producerInfo->setSecurityKey(keyCopy);
      splitTop->setExtractProducerInfo(producerInfo);
    }
  
    if (isExtractConsumer_)
      splitTop->setExtractConsumerFlag();
  
    if (hash2RepartitioningWithSameKey())
      splitTop->setMWayRepartitionFlag();
  
    if(!generator->explainDisabled())
    {
      if (isExtractConsumer_)
        generator->setExplainTuple(addExplainInfo(sendTop, 0, 0, generator));
      else
        sendTop->setExplainNodeId(generator->getExplainNodeId());
      splitTop->setExplainNodeId(generator->getExplainNodeId());
    }
  
    // Assign a SeaMonster tag to send top and send bottom. Both TDBs
    // use the same tag. When EXPLAIN is enabled the tag can be the
    // EXPLAIN ID. Otherwise we use an integer unique within this plan.
    if (thisExchangeUsesSM)
    {
      Int32 smTag = generator->getExplainNodeId();
      if (generator->explainDisabled())
        smTag = generator->getNextSMTag();
      sendTop->setSMTag(smTag);
      sendBottom->setSMTag(smTag);
    }
  
    if(isAnESPAccess()) {
      splitTop->setMWayRepartitionFlag();
    }
  
    generator->decrEspLevel();
  
    // ---------------------------------------------------------------------
    // setup everything and leave
    // ---------------------------------------------------------------------
    generator->setCriDesc(given_cri_desc, Generator::DOWN);
    generator->setCriDesc(returned_cri_desc, Generator::UP);
    generator->setGenObj(this, splitTop);
  
    return 0;
  }
  
  ExpTupleDesc::TupleDataFormat Exchange::determineInternalFormat( const ValueIdList & valIdList,
                                                                     RelExpr * relExpr,
                                                                     NABoolean & resizeCifRecord,
                                                                     Generator * generator,
                                                                     NABoolean bmo_affinity,
                                                                     NABoolean & considerBufferDefrag)
  {
  
    RelExpr::CifUseOptions bmo_cif = RelExpr::CIF_SYSTEM;
  
  
    if (CmpCommon::getDefault(COMPRESSED_INTERNAL_FORMAT_BMO) == DF_OFF)
    {
      bmo_cif = RelExpr::CIF_OFF;
    }
    else
    if (CmpCommon::getDefault(COMPRESSED_INTERNAL_FORMAT_BMO) == DF_ON)
    {
      bmo_cif = RelExpr::CIF_ON;
    }
  
    //CIF_SYSTEM
  
  
    return generator->determineInternalFormat(valIdList, 
                                              relExpr, 
                                              resizeCifRecord, 
                                              bmo_cif,
                                              bmo_affinity,
                                              considerBufferDefrag);
  
  }
  CostScalar Exchange::getEstimatedRunTimeMemoryUsage(NABoolean perCPU)
  {
     //////////////////////////////////////
     // compute the buffer length (for both 
     // sendTop and sendBottom) first. 
     //////////////////////////////////////
     ULng32 upRowLength = 
           getGroupAttr()->getCharacteristicOutputs().getRowLength();
     ULng32 downRowLength = 
           getGroupAttr()->getCharacteristicInputs().getRowLength();
  
     // make sure the up buffer can fit at least one row
     ULng32 upSqlBufferLength =
        MAXOF((ULng32) ComTdbSendTop::minReceiveBufferSize(upRowLength),
              (ULng32) (upMessageBufferLength_.getValue() * 1024));
  
     // make sure the down buffer can fit at least one row
     ULng32 downSqlBufferLength =
        MAXOF((ULng32) ComTdbSendTop::minSendBufferSize(downRowLength),
              (ULng32) (downMessageBufferLength_.getValue() * 1024));
  
     ULng32 sqlBufferLengthUsed = 
        MAXOF(upSqlBufferLength, downSqlBufferLength);
  
     //////////////////////////////////////
     // compute the number of buffers
     //////////////////////////////////////
     ULng32 numUpBuffersSendT = getDefault(GEN_SNDT_NUM_BUFFERS);
     ULng32 numUpBuffersSendB = getDefault(GEN_SNDB_NUM_BUFFERS);
  
     ULng32 numDownBuffersSendT = 1; // only one down buffer allocated
     ULng32 numDownBuffersSendB = 1; // only one down buffer allocated
  
     double memoryRequired = 0;
  
     const PartitioningFunction *topPartFunc = getTopPartitioningFunction();
     Lng32  numTopEsps = topPartFunc->getCountOfPartitions();
  
     if ( isDP2Exchange() == FALSE ) {
      // regular ESP exchange
  
      // Compute for send top first.
  
      // Average it out because the memory for upper queue 
      // is allocated dynamically
      double topMemory = 
            (sqlBufferLengthUsed+1000) * numUpBuffersSendT / 2 +
            (sqlBufferLengthUsed+1000) * numDownBuffersSendT;
  
      if ( topPartFunc -> isAReplicationPartitioningFunction() == TRUE ) {
        topMemory *= numTopEsps;
      }
  
      memoryRequired = topMemory;
  
      // Compute for send bottom 
      double bottomMemory = (sqlBufferLengthUsed+1000) * numUpBuffersSendB / 2 +
                      (sqlBufferLengthUsed+1000) * numDownBuffersSendB;
  
      memoryRequired += bottomMemory;
  
    } else {
      // split top. 
    }
  
    if ( perCPU == FALSE ) {
  
      const PhysicalProperty* const phyProp = getPhysicalProperty();
      if (phyProp != NULL) {
        memoryRequired = numTopEsps * memoryRequired;
      }
    }
  
    return memoryRequired;
  }
  
  double Exchange::getEstimatedRunTimeMemoryUsage(ComTdb * tdb)
  {
    return (getEstimatedRunTimeMemoryUsage(FALSE)).value();
  }
  
  bool Exchange::thisExchangeCanUseSM(BindWA *bindWA) const
  {
    // SeaMonster can be enabled if all the following are true
    // * The SEAMONSTER default is ON or the env var SQ_SEAMONSTER is 1
    // * This is an ESP exchange
    // * This is not a parallel extract producer or consumer
    // * This is not an ESP access operator
    
    if (isEspExchange() &&
        !isExtractProducer_ && !isExtractConsumer_ &&
        !isAnESPAccess() &&
        bindWA->queryCanUseSeaMonster())
    {
      return true;
    }
  
    return false;
  }
  