/**********************************************************************
// @@@ 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 "RelPackedRows.h"
#include "Generator.h"
#include "GenExpGenerator.h"
#include "ExpCriDesc.h"
#include "ComTdbUnPackRows.h"
#include "ex_queue.h"

// PhysUnPackRows::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.
//
// PhysUnPackRows::preCodeGen() - is basically the same as the RelExpr::
// preCodeGen() except that here we replace the VEG references in the
// unPackExpr() and packingFactor(), 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 * PhysUnPackRows::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 unPackExpr() and packingFactor() expressions have access 
  // to only the Input Values. These can come from the parent or be
  // the outputs of the child.
  //
  unPackExpr().
    replaceVEGExpressions(availableValues,
                          getGroupAttr()->getCharacteristicInputs());

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

  if (rowwiseRowset())
    {
      if (rwrsInputSizeExpr())
	((ItemExpr*)rwrsInputSizeExpr())->
	  replaceVEGExpressions(availableValues,
				getGroupAttr()->getCharacteristicInputs());

      if (rwrsMaxInputRowlenExpr())
	((ItemExpr*)rwrsMaxInputRowlenExpr())->
	  replaceVEGExpressions(availableValues,
				getGroupAttr()->getCharacteristicInputs());

      if (rwrsBufferAddrExpr())
	((ItemExpr*)rwrsBufferAddrExpr())->
	  replaceVEGExpressions(availableValues,
				getGroupAttr()->getCharacteristicInputs());
    }

  // The selectionPred has access to only the output values generated by
  // UnPackRows 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());

  Lng32 memLimit = (Lng32)CmpCommon::getDefaultNumeric(MEMORY_LIMIT_ROWSET_IN_MB);
  if (memLimit > 0)
  {
    Lng32 rowLength = getGroupAttr()->getCharacteristicOutputs().getRowLength();
    Lng32 rowsetSize = getGroupAttr()->getOutputLogPropList()[0]->getResultCardinality().value() ;
    Lng32 memNeededMB = (rowLength * rowsetSize)/(1024 * 1024);
    if (memLimit < memNeededMB)
    {
      *CmpCommon::diags() << DgSqlCode(-30050) << DgInt0(memNeededMB) 
                          << DgInt1(memLimit) << DgInt2(rowLength) 
                          << DgInt3(rowsetSize);
      GenExit();
      return NULL;
    }
  }

  generator->oltOptInfo()->setMultipleRowsReturned(TRUE);

  markAsPreCodeGenned();

  return this;
} // PhysUnPackRows::preCodeGen


short
PhysUnPackRows::codeGen(Generator *generator) 
{
  // Get handles on expression generator, map table, and heap allocator
  //
  ExpGenerator *expGen = generator->getExpGenerator();
  Space *space = generator->getSpace();

  // Allocate a new map table for this operation
  //
  MapTable *localMapTable = generator->appendAtEnd();

  // Generate the child and capture the task definition block and a description
  // of the reply composite row layout and the explain information.
  //
  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. Since they are
  // not needed above, they will not be in the work ATP (0).
  // (Later, they will be removed from the map table)
  //
  localMapTable->setAllAtp(1);

  // Generate the given and returned composite row descriptors.
  // unPackRows adds a tupp (for the generated outputs) to the
  // row given by the parent. The workAtp will have the 2 more 
  // tupps (1 for the generated outputs and another for the
  // indexValue) than the given.
  //
  ex_cri_desc *givenCriDesc = generator->getCriDesc(Generator::DOWN);

  ex_cri_desc *returnedCriDesc = 
    new(space) ex_cri_desc(givenCriDesc->noTuples() + 1, space);

  ex_cri_desc *workCriDesc = 
    new(space) ex_cri_desc(givenCriDesc->noTuples() + 2, space);


  // unPackCols is the next to the last Tp in Atp 0, the work ATP.
  // and the last Tp in the returned ATP.
  //
  const Int32 unPackColsAtpIndex = workCriDesc->noTuples() - 2;
  const Int32 unPackColsAtp = 0;

  // The length of the new tuple which will contain the columns
  // generated by unPackRows
  //
  ULng32 unPackColsTupleLen;

  // The Tuple Desc describing the tuple containing the new unPacked columns
  // It is generated when the expression is generated.
  //
  ExpTupleDesc *unPackColsTupleDesc = 0;

  // indexValue is the last Tp in Atp 0, the work ATP.
  //
  const Int32 indexValueAtpIndex = workCriDesc->noTuples() - 1;
  const Int32 indexValueAtp = 0;

  // The length of the work tuple which will contain the value
  // of the index.  This should always be sizeof(int).
  //
  ULng32 indexValueTupleLen = 0;

  // The Tuple Desc describing the tuple containing the new unPacked columns
  // It is generated when the expression is generated.
  //
  ExpTupleDesc *indexValueTupleDesc = 0;

  ValueIdList indexValueList;

  if (indexValue() != NULL_VALUE_ID)
    {
      indexValueList.insert(indexValue());
      
      expGen->processValIdList(indexValueList,
			       ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
			       indexValueTupleLen,
			       indexValueAtp,
			       indexValueAtpIndex,
			       &indexValueTupleDesc,
			       ExpTupleDesc::SHORT_FORMAT);
      
      GenAssert(indexValueTupleLen == sizeof(Int32),
		"UnPackRows::codeGen: Internal Error");
    }

  // If a packingFactor exists, generate a move expression for this.
  // It is assumed that the packingFactor expression evaluates to a
  // 4 byte integer.
  //
  ex_expr *packingFactorExpr = 0;
  ULng32 packingFactorTupleLen;
  
  if(packingFactor().entries() > 0) {
    expGen->generateContiguousMoveExpr(packingFactor(),
                                       -1, 
                                       unPackColsAtp,
                                       unPackColsAtpIndex,
                                       ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
                                       packingFactorTupleLen,
                                       &packingFactorExpr);

    GenAssert(packingFactorTupleLen == sizeof(Int32),
              "UnPackRows::codeGen: Internal Error");
  }

  // Generate the UnPack expressions.
  //
  // characteristicOutputs() - refers to the list of expressions
  // to be move to another tuple.
  //
  // 0 - Do not add conv. nodes.
  //
  // unPackColsAtp - this expression will move data to the
  // unPackColsAtp (0) ATP.
  //
  // unPackColsAtpIndex - within the unPackColsAtp (0) ATP, the destination
  // for this move expression will be the unPackColsAtpIndex TP. This should
  // be the next to the last TP of the work ATP. (The indexValue will be in
  // the last position)
  //
  // SQLARK_EXPLODED_FORMAT - generate the move expression to construct
  // the destination tuple in EXPLODED FORMAT.
  //
  // unPackColsTupleLen - This is an output which will contain the length
  // of the destination Tuple.
  //
  // &unPackColsExpr - The address of the pointer to the expression
  // which will be generated.
  //
  // &unPackColsTupleDesc - The address of the tuple descriptor which is
  // generated.  This describes the destination tuple of the move expression.
  //
  // SHORT_FORMAT - generate the unPackColsTupleDesc in the SHORT FORMAT.
  //
  ex_expr *unPackColsExpr = 0;

  expGen->
    genGuardedContigMoveExpr(selectionPred(),
			     getGroupAttr()->getCharacteristicOutputs(),
                             0, // No Convert Nodes added
                             unPackColsAtp,
                             unPackColsAtpIndex,
                             ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
                             unPackColsTupleLen,
                             &unPackColsExpr,
                             &unPackColsTupleDesc,
                             ExpTupleDesc::SHORT_FORMAT);

  workCriDesc->setTupleDescriptor(unPackColsAtpIndex,
                                  unPackColsTupleDesc);

  returnedCriDesc->setTupleDescriptor(unPackColsAtpIndex,
                                      unPackColsTupleDesc);


  // expressions for rowwise rowset implementation.
  ex_expr * rwrsInputSizeExpr = 0;
  ex_expr * rwrsMaxInputRowlenExpr = 0;
  ex_expr * rwrsBufferAddrExpr = 0;
  ULng32 rwrsInputSizeExprLen = 0;
  ULng32 rwrsMaxInputRowlenExprLen = 0;
  ULng32 rwrsBufferAddrExprLen = 0;

  const Int32 rwrsAtp = 1;
  const Int32 rwrsAtpIndex = workCriDesc->noTuples() - 2;
  ExpTupleDesc * rwrsTupleDesc = 0;
  ValueIdList rwrsVidList;
  if (rowwiseRowset())
    {
      rwrsVidList.insert(this->rwrsInputSizeExpr()->getValueId());
      expGen->generateContiguousMoveExpr(rwrsVidList, 
					 0 /*don't add conv nodes*/,
					 rwrsAtp, rwrsAtpIndex,
					 ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
					 rwrsInputSizeExprLen, 
					 &rwrsInputSizeExpr,
					 &rwrsTupleDesc,ExpTupleDesc::SHORT_FORMAT);
      
      rwrsVidList.clear();
      rwrsVidList.insert(this->rwrsMaxInputRowlenExpr()->getValueId());
      expGen->generateContiguousMoveExpr(rwrsVidList, 
					 0 /*don't add conv nodes*/,
					 rwrsAtp, rwrsAtpIndex,
					 ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
					 rwrsMaxInputRowlenExprLen, 
					 &rwrsMaxInputRowlenExpr,
					 &rwrsTupleDesc,ExpTupleDesc::SHORT_FORMAT);

      rwrsVidList.clear();
      rwrsVidList.insert(this->rwrsBufferAddrExpr()->getValueId());
      expGen->generateContiguousMoveExpr(rwrsVidList, 
					 0 /*don't add conv nodes*/,
					 rwrsAtp, rwrsAtpIndex,
					 ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
					 rwrsBufferAddrExprLen, 
					 &rwrsBufferAddrExpr,
					 &rwrsTupleDesc,ExpTupleDesc::SHORT_FORMAT);

      expGen->assignAtpAndAtpIndex(rwrsOutputVids(), 
				   unPackColsAtp, unPackColsAtpIndex);
    }

  // Move the generated maptable entries, to the localMapTable,
  // so that all other entries can later be removed.
  //
  for(ValueId outputValId = getGroupAttr()->getCharacteristicOutputs().init();
      getGroupAttr()->getCharacteristicOutputs().next(outputValId);
      getGroupAttr()->getCharacteristicOutputs().advance(outputValId)) {
    
    generator->addMapInfoToThis(localMapTable, outputValId,
				generator->getMapInfo(outputValId)->
				getAttr());

    // Indicate that code was generated for this map table entry.
    //
    generator->getMapInfoFromThis(localMapTable, outputValId)->codeGenerated();
  }

  NABoolean tolerateNonFatalError = FALSE;

  if (isRowsetIterator() && (generator->getTolerateNonFatalError())) {
    tolerateNonFatalError = TRUE;
    setTolerateNonFatalError(RelExpr::NOT_ATOMIC_);
  }


  // Allocate the UnPack TDB
  //
  ComTdbUnPackRows *unPackTdb = NULL;

  if (rowwiseRowset())
    {
      unPackTdb =
	new (space) ComTdbUnPackRows(NULL, //childTdb,
				     rwrsInputSizeExpr,
				     rwrsMaxInputRowlenExpr,
				     rwrsBufferAddrExpr,
				     rwrsAtpIndex,
				     givenCriDesc,
				     returnedCriDesc,
				     workCriDesc,
				     16,
				     1024,
				     (Cardinality) getGroupAttr()->
				     getOutputLogPropList()[0]->
				     getResultCardinality().value(),
				     2, 20000);
    }
  else
    {

      // Base the initial queue size on the est. cardinality.
      // UnPackRows does not do dyn queue resize, so passed in
      // queue size values represent initial (and final) queue
      // sizes (not max queue sizes).
      //
      ULng32 rowsetSize = 
          getGroupAttr()->getOutputLogPropList()[0]->getResultCardinality().value();
      double  memoryLimitPerInstance =
        (ActiveSchemaDB()->getDefaults().getAsLong(EXE_MEMORY_FOR_UNPACK_ROWS_IN_MB) * 1024 * 1024)/2; // At runtime (ExUnpackRowsTcb::ExUnpackRowsTcb) we allocate twice the size of the queue. So ensure it fits in to half the specified limit.

      rowsetSize = (rowsetSize < 1024 ? 1024 : rowsetSize);
     
      double estimatedMemory = (Int64)rowsetSize * (Int64)unPackColsTupleLen;
 
      if (estimatedMemory > memoryLimitPerInstance)
      {
         estimatedMemory = memoryLimitPerInstance;
         rowsetSize = estimatedMemory / unPackColsTupleLen;
         rowsetSize = MAXOF(rowsetSize, 1);
      }

      queue_index upQueueSize = ex_queue::roundUp2Power((queue_index)rowsetSize);

      unPackTdb =
	new (space) ComTdbUnPackRows(childTdb,
				     packingFactorExpr,
				     unPackColsExpr,
				     unPackColsTupleLen,
				     unPackColsAtpIndex,
				     indexValueAtpIndex,
				     givenCriDesc,
				     returnedCriDesc,
				     workCriDesc,
				     16,
				     upQueueSize,
				     (Cardinality) getGroupAttr()->
				     getOutputLogPropList()[0]->
				     getResultCardinality().value(),
				     isRowsetIterator(),
				     tolerateNonFatalError);
    }

  generator->initTdbFields(unPackTdb);

  // Remove child's outputs from mapTable, They are not needed
  // above.
  //
  generator->removeAll(localMapTable);

  //Turn off override of queue sizes for unpack. Once set they should remain.
  generator->setMakeOnljLeftQueuesBig(FALSE);
  // Add the explain Information for this node to the EXPLAIN
  // Fragment.  Set the explainTuple pointer in the generator so
  // the parent of this node can get a handle on this explainTuple.
  //
  if(!generator->explainDisabled()) {
    generator->setExplainTuple(addExplainInfo(unPackTdb,
                                              childExplainTuple,
                                              0,
                                              generator));
  }     
        
  // Restore the Cri Desc's and set the return object.
  //
  generator->setCriDesc(givenCriDesc, Generator::DOWN);
  generator->setCriDesc(returnedCriDesc, Generator::UP);
  generator->setGenObj(this, unPackTdb);

  
  return 0;
}