// **********************************************************************
// @@@ 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 "ComOptIncludes.h"
#include "Generator.h"
#include "GenExpGenerator.h"
#include "GroupAttr.h"
#include "ExpCriDesc.h"
#include "DefaultConstants.h"
#include "ComTdbProbeCache.h"
#include "RelProbeCache.h"
#include "DefaultConstants.h"
#include "ExpSqlTupp.h"         // for sizeof(tupp_descriptor)
#include "ComDefs.h"            // to get common defines (ROUND8)

/////////////////////////////////////////////////////////////////////
//
// Contents:
//   
//   ProbeCache::codeGen()
//
//////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////////////////////
//
// Layout at this node:
// (not including the mandatory consts and temps entries)
//
// Returned atp layout (to parent of ProbeCache):
//
// |-------------|   |------------------------------|
// | 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: optional tupp for cached inner table result.  Note 
//             that there is no output row (and no tupp) if
//             ProbeCache is on the right of a semi-join or
//             anti-semi-join.
//             
//
// ATP layout of queue entries between ProbeCache node and its child:
//
// |-------------|   |-----------------------------|
// | input data  |   | input data |  output row    |
// | ( I tupps ) |   | ( I tupps )|  ( J tupps )   |
// |-------------|   |-----------------------------|
// <ATP to child >   <-- returned ATP to parent --->
//
// input data: same as atp input from this node's parent.
// output row: optional inner table result returned from child. 
//             See note above about semi-join and anti-semi-join.
// 
//
// Work atp layout:
//
// |----------------------------------------------------------|
// | Constants|   Temps  | Probe Hash| Probe Row | Inner Row  |
// | (1 tupp) | (1 tupp) | (1 tupp)  | (1 tupp)  | (1 tupp)   |
// |----------------------------------------------------------|
//
// Constants:         the constants tupp from the given Atp.
// Temps:             the temps tupp from the given Atp.
// Probe Hash:        the target for the hash of the probe input hash
// Probe Row:         the target for the probe input encode expr.
// Inner Row:         the target for the inner row's move expr,
//                    if ProbeCache is not on the right of a 
//                    semi-join or anti-semi-join.
//
////////////////////////////////////////////////////////////////////////////


short ProbeCache::codeGen(Generator *generator)
{
  ExpGenerator * exp_gen = generator->getExpGenerator();
  Space * space = generator->getSpace();

  MapTable * last_map_table = generator->getLastMapTable();

  ex_cri_desc * given_desc
    = generator->getCriDesc(Generator::DOWN);

  ex_cri_desc * returned_desc
    = new(space) ex_cri_desc(given_desc->noTuples() + 1, space);

  // cri descriptor for work atp has 5 entries:
  // entry #0 for const
  // entry #1 for temp
  // entry #2 for hash value of probe input data in Probe Cache Manager
  // entry #3 for encoded probe input data in Probe Cache Manager
  // enrry #4 for inner table row data in this operator's cache buffer
  Int32 work_atp = 1;
  ex_cri_desc * work_cri_desc = new(space) ex_cri_desc(5, space);
  unsigned short hashValIdx          = 2; 
  unsigned short encodedProbeDataIdx = 3;
  unsigned short innerRowDataIdx     = 4;

    // generate code for child tree, and get its tdb and explain tuple.
  child(0)->codeGen(generator);
  ComTdb * child_tdb = (ComTdb *)(generator->getGenObj());
  ExplainTuple *childExplainTuple = generator->getExplainTuple();


  //////////////////////////////////////////////////////
  // Generate up to 4 runtime expressions.
  //////////////////////////////////////////////////////

  // Will use child's char. inputs (+ execution count) for the next
  // two runtime expressions.
  ValueIdList inputsToUse = child(0).getGroupAttr()->getCharacteristicInputs();
  
  inputsToUse.insert(generator->getOrAddStatementExecutionCount());

  // Expression #1 gets the hash value of the probe input data
  ValueIdList hvAsList;

  // Executor has hard-coded assumption that the result is long, 
  // so add a Cast node to convert result to a long.

  ItemExpr *probeHashAsIe = new (generator->wHeap())
    HashDistPartHash(inputsToUse.rebuildExprTree(ITM_ITEM_LIST));

  probeHashAsIe->bindNode(generator->getBindWA());

  NumericType &nTyp = (NumericType &)probeHashAsIe->getValueId().getType();
  GenAssert(nTyp.isSigned() == FALSE,
            "Unexpected signed HashDistPartHash.");

  GenAssert(probeHashAsIe->getValueId().getType().supportsSQLnullLogical()
            == FALSE, "Unexpected nullable HashDistPartHash.");

  ItemExpr *hvAsIe = new (generator->wHeap()) Cast(
       probeHashAsIe, 
       new (generator->wHeap()) 
            SQLInt(generator->wHeap(), FALSE,   // false == unsigned.
                   FALSE    // false == not nullable.
                  ));

  hvAsIe->bindNode(generator->getBindWA());

  hvAsList.insert(hvAsIe->getValueId());

  ex_expr *hvExpr   = NULL;
  ULng32 hvLength;
  exp_gen->generateContiguousMoveExpr(
              hvAsList,
              0, // don't add convert node
              work_atp,
              hashValIdx,
              ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
              hvLength,
              &hvExpr);

  GenAssert(hvLength == sizeof(Lng32),
            "Unexpected length of result of hash function.");

  // Expression #2 encodes the probe input data for storage in 
  // the ProbeCacheManager.

  ValueIdList encodeInputAsList;

  CollIndex inputListIndex;
  for (inputListIndex = 0; 
       inputListIndex < inputsToUse.entries(); 
       inputListIndex++) {   

    ItemExpr *inputIe = 
         (inputsToUse[inputListIndex].getValueDesc())->getItemExpr();

    if (inputIe->getValueId().getType().getVarLenHdrSize() > 0)
      {
        // This logic copied from Sort::codeGen().
        // Explode varchars by moving them to a fixed field
        // whose length is equal to the max length of varchar.
        // 5/8/98: add support for VARNCHAR

        const CharType& char_type =
          (CharType&)(inputIe->getValueId().getType());

	if (!CollationInfo::isSystemCollation(char_type.getCollation()))
	{
	  inputIe = new(generator->wHeap())
              Cast (inputIe,
                    (new(generator->wHeap())
                      SQLChar(
		          generator->wHeap(), CharLenInfo(char_type.getStrCharLimit(), char_type.getDataStorageSize()),
                          char_type.supportsSQLnull(),
                          FALSE, FALSE, FALSE,
                          char_type.getCharSet(),
                          char_type.getCollation(),
                          char_type.getCoercibility()
                              )
                    )
                   );
	}
      }

    CompEncode * enode = new(generator->wHeap()) 
      CompEncode(inputIe, FALSE /* ascend/descend doesn't matter*/);

    enode->bindNode(generator->getBindWA());
    encodeInputAsList.insert(enode->getValueId());
  }

  ex_expr *encodeInputExpr = NULL;
  ULng32 encodedInputLength;
  exp_gen->generateContiguousMoveExpr(encodeInputAsList, 
                              0, //don't add conv nodes
                              work_atp, encodedProbeDataIdx,
                              ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
                              encodedInputLength, &encodeInputExpr);


  // Expression #3 moves the inner table data into a buffer pool.  
  // This is also the tuple returned to ProbeCache's parent. 

  ex_expr * innerRecExpr = NULL;
  ValueIdList innerTableAsList = getGroupAttr()->getCharacteristicOutputs();

  //////////////////////////////////////////////////////
  // Describe the returned row and add the returned 
  // values to the map table.
  //////////////////////////////////////////////////////

  // determine internal format
  NABoolean useCif = FALSE;
  ExpTupleDesc::TupleDataFormat tupleFormat = generator->getInternalFormat();
  //tupleFormat = determineInternalFormat( innerTableAsList, this, useCif,generator);

  ULng32 innerRecLength = 0;
  ExpTupleDesc * innerRecTupleDesc = 0;
  MapTable * returnedMapTable = NULL;

  exp_gen->generateContiguousMoveExpr(innerTableAsList, 
                              -1, // do add conv nodes 
			      work_atp, innerRowDataIdx,
			      tupleFormat,
			      innerRecLength, &innerRecExpr,
			      &innerRecTupleDesc, ExpTupleDesc::SHORT_FORMAT,
                              &returnedMapTable);

  returned_desc->setTupleDescriptor(returned_desc->noTuples() - 1, 
        innerRecTupleDesc);

  // remove all appended map tables and return the returnedMapTable
  generator->removeAll(last_map_table);
  generator->appendAtEnd(returnedMapTable);
  // This returnedMapTable will contain the value ids that are being returned 
  // (the inner table probed).

  // Massage the atp and atp_index of the innerTableAsList.
  for (CollIndex i = 0; i < innerTableAsList.entries(); i++)
    {
      ValueId innerValId = innerTableAsList[i];

      Attributes *attrib =
	generator->getMapInfo(innerValId)->getAttr();

      // All reference to the returned values from this point on
      // will be at atp = 0, atp_index = last entry in returned desc.
      attrib->setAtp(0);
      attrib->setAtpIndex(returned_desc->noTuples() - 1);
    }

  // Expression #4 is a selection predicate, to be applied
  // before returning rows to the parent

  ex_expr * selectPred = NULL;

  if (!selectionPred().isEmpty())
    {
      ItemExpr * selPredTree =
        selectionPred().rebuildExprTree(ITM_AND,TRUE,TRUE);
      exp_gen->generateExpr(selPredTree->getValueId(),
                            ex_expr::exp_SCAN_PRED,
                            &selectPred);
    }

  //////////////////////////////////////////////////////
  // Prepare params for ComTdbProbeCache.
  //////////////////////////////////////////////////////

  queue_index pDownSize = (queue_index)getDefault(GEN_PROBE_CACHE_SIZE_DOWN);
  queue_index pUpSize   = (queue_index)getDefault(GEN_PROBE_CACHE_SIZE_UP);

  // Make sure that the ProbeCache queues can support the childs queues.
  if(pDownSize < child_tdb->getInitialQueueSizeDown()) {
    pDownSize = child_tdb->getInitialQueueSizeDown();
    pDownSize = MINOF(pDownSize, 32768);
  }
  if(pUpSize < child_tdb->getInitialQueueSizeUp()) {
    pUpSize = child_tdb->getInitialQueueSizeUp();
    pUpSize = MINOF(pUpSize, 32768);
  }

  ULng32 pcNumEntries = numCachedProbes_;
  
  // Number of entries in the probe cache cannot be less than 
  // max parent down queue size.  Before testing and adjusting the 
  // max queue size, it is necessary to make sure it is a power of
  // two, rounding up if necessary.  This is to match the logic in
  // ex_queue::resize.

  queue_index pdq2 = 1;
  queue_index bits = pDownSize;
  while (bits && pdq2 < pDownSize) {
    bits = bits  >> 1;
    pdq2 = pdq2 << 1;
  }
  if (pcNumEntries < pdq2)
    pcNumEntries = pdq2;

  numInnerTuples_ = getDefault(GEN_PROBE_CACHE_NUM_INNER);

  if (innerRecExpr == NULL)
    {
      // For semi-join and anti-semi-join, executor need not allocate
      // a buffer.  Set the tdb's buffer size to 0 to be consistent.
      numInnerTuples_ = 0;
    }
  else if (numInnerTuples_ == 0)
    {
      // Handle special value, 0, which tells code gen to 
      // decided on buffer size: i.e., large enough to accomodate
      // all parent up queue entries and all probe cache entries 
      // having a different inner table row.

      // As we did for the down queue, make sure the up queue size 
      // specified is a power of two.

      queue_index puq2 = 1;
      queue_index bits = pUpSize;
      while (bits && puq2 < pUpSize) {
        bits = bits  >> 1;
        puq2 = puq2 << 1;
      }
      numInnerTuples_ = puq2 + pcNumEntries;
    }
  else 
    {
      // Just use the default.  ExSimpleSqlBuffer takes care
      // that there is room enough for one row.
    }

   double  memoryLimitPerInstance =
      ActiveSchemaDB()->getDefaults().getAsLong(EXE_MEMORY_FOR_PROBE_CACHE_IN_MB) * 1024 * 1024;
   double estimatedMemory;
   
   if (numInnerTuples_ > 0) {
      estimatedMemory = numInnerTuples_ * innerRecLength;
      if (estimatedMemory > memoryLimitPerInstance) {
          numInnerTuples_ = memoryLimitPerInstance / innerRecLength;
          queue_index pUpSize_calc;
 
          pUpSize_calc = numInnerTuples_ ;
          queue_index pq2 = 1;
          queue_index bits = pUpSize_calc;
          while (bits && pq2 < pUpSize_calc) {
              bits = bits  >> 1;
              pq2 = pq2 << 1;
          }
          pUpSize = MINOF(pq2/2, 4); 
          pDownSize = MINOF(pq2/2, 4);
          pcNumEntries = MINOF(numInnerTuples_, 4);
          numInnerTuples_ = pcNumEntries;
          numCachedProbes_ = pcNumEntries;
      }
   } 


  ///////////////////////////////////////////////////////
  // Create and initialize the ComTdbProbeCache, generate
  // the explain tuple, set the new up queue cri desc 
  // and give the generator this newly generated object.
  ///////////////////////////////////////////////////////
  
  ComTdbProbeCache *probeCacheTdb = new (space) ComTdbProbeCache (
                 hvExpr,
		 encodeInputExpr,
		 innerRecExpr,
                 selectPred,
		 encodedInputLength,
		 innerRecLength,
                 pcNumEntries,
		 returned_desc->noTuples() - 1,
                 hashValIdx,
                 encodedProbeDataIdx,
                 innerRowDataIdx,
		 child_tdb,
		 given_desc,
		 returned_desc,
		 pDownSize,
		 pUpSize,
		 (Cardinality) 
                   (getInputCardinality() * getEstRowsUsed()).getValue(),
		 numInnerTuples_) ;

  generator->initTdbFields(probeCacheTdb);

  // Make sure that the ProbeCache queues were not set too large
  if ((probeCacheTdb->getInitialQueueSizeDown() > pDownSize) ||
      (probeCacheTdb->getInitialQueueSizeUp() > pUpSize)) {
    probeCacheTdb->setQueueResizeParams(pDownSize,
                                        pUpSize,
                                        probeCacheTdb->getQueueResizeLimit(),
                                        probeCacheTdb->getQueueResizeFactor());
  }

  double probeCacheMemEst = getEstimatedRunTimeMemoryUsage(generator, probeCacheTdb);
  generator->addToTotalEstimatedMemory(probeCacheMemEst);
  Lng32 pcMemEstInKBPerNode = getEstimatedRunTimeMemoryUsage(generator, TRUE).value() / 1024;
  if(!generator->explainDisabled()) {
    generator->setExplainTuple(
       addExplainInfo(probeCacheTdb, childExplainTuple, 0, generator));
  }

  generator->setCriDesc(returned_desc, Generator::UP);

  generator->setGenObj(this, probeCacheTdb);

  return 0;
}

CostScalar ProbeCache::getEstimatedRunTimeMemoryUsage(Generator *generator, NABoolean perNode, Lng32 *numStreams)
{
  const Lng32 probeSize = 
      getGroupAttr()->getCharacteristicInputs().getRowLength();
  const Lng32 numCacheEntries = 
    ActiveSchemaDB()->getDefaults().getAsLong(GEN_PROBE_CACHE_NUM_ENTRIES);
  const Int32 perProbeOverhead = 32 ; // bytes
  const double cacheSize = (probeSize + perProbeOverhead) * numCacheEntries;  // in bytes

  const Lng32 resultSize = 
      getGroupAttr()->getCharacteristicOutputs().getRowLength();
  Lng32 numBufferPoolEntries = 
    ActiveSchemaDB()->getDefaults().getAsLong(GEN_PROBE_CACHE_NUM_INNER);
  if (numBufferPoolEntries == 0)
  {
      numBufferPoolEntries = numCacheEntries + 
            ActiveSchemaDB()->getDefaults().getAsLong(GEN_PROBE_CACHE_SIZE_UP);
  }
  const double outputBufferSize = resultSize * numBufferPoolEntries;  // in bytes

  // totalMemory is perNode at this point of time.
  double totalMemory = cacheSize + outputBufferSize;
  Lng32 numOfStreams = 1;
  const PhysicalProperty* const phyProp = getPhysicalProperty();
  if (phyProp)
  {
     PartitioningFunction * partFunc = phyProp -> getPartitioningFunction() ;
     numOfStreams = partFunc->getCountOfPartitions();
     if (numOfStreams <= 0)
        numOfStreams = 1;
     double  memoryLimitPerInstance =
          ActiveSchemaDB()->getDefaults().getAsLong(EXE_MEMORY_FOR_PROBE_CACHE_IN_MB) * 1024 * 1024;
     if (totalMemory > memoryLimitPerInstance)
        totalMemory = memoryLimitPerInstance;          
     totalMemory *= numOfStreams;
  }
  if (numStreams != NULL)
     *numStreams = numOfStreams;
  if (perNode)
     totalMemory /= MINOF(MAXOF(((NAClusterInfoLinux*)gpClusterInfo)->getTotalNumberOfCPUs(), 1), numOfStreams);
  else
     totalMemory /= numOfStreams;
  return totalMemory;
}

double ProbeCache::getEstimatedRunTimeMemoryUsage(Generator *generator, ComTdb * tdb)
{
  // tdb is ignored for ProbeCache because this operator
  // does not participate in the BMO quota system.
  Lng32 numOfStreams = 1;
  CostScalar totalMemory = getEstimatedRunTimeMemoryUsage(generator, FALSE, &numOfStreams);
  totalMemory = totalMemory * numOfStreams ;
  return totalMemory.value();
}