/**********************************************************************
// @@@ 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:         GenPartFunc.C
 * Description:  Generate code for a Partitioning Function
 * Created:      2/23/96
 * Modified:     $ $Date: 2006/11/01 01:27:56 $ (GMT)
 * Language:     C++
 *
 *
 *
 *
 *****************************************************************************
 */

// -----------------------------------------------------------------------

/////////////////////////////////////////////////////////////////////
//
// Contents:
//    
//   PartitioningFunction::codeGen()
//   HashPartitioningFunction::codeGen()
//   HashDistPartitioningFunction::codeGen()
//   Hash2PartitioningFunction::codeGen()
//   RangePartitionBoundaries::completePartitionBoundaries()
//   RangePartitioningFunction::codeGen()
//
//////////////////////////////////////////////////////////////////////


#include "ComOptIncludes.h"
#include "GroupAttr.h"
#include "PartFunc.h"
#include "Generator.h"
#include "GenExpGenerator.h"
//#include "ex_stdh.h"
#include "ExpCriDesc.h"
#include "PartInputDataDesc.h"


/////////////////////////////////////////////////////////
//
// PartitioningFunction::codeGen()
//
/////////////////////////////////////////////////////////

short PartitioningFunction::codeGen(Generator *, Lng32)
{
  ABORT("Need to override PartitioningFunction::codeGen()");
  return 0;
}

/////////////////////////////////////////////////////////
//
// PartitioningFunction::generatePivLayout()
//
/////////////////////////////////////////////////////////

void PartitioningFunction::generatePivLayout(
     Generator *generator,
     Lng32 &partitionInputDataLength,
     Lng32 atp,
     Lng32 atpIndex,
     Attributes ***pivAttrs)
{
  // assign offsets to the PIVs in a standard way

  ExpGenerator *expGen = generator->getExpGenerator();
  
  expGen->processValIdList(
       getPartitionInputValuesLayout(),
       ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
       (ULng32 &) partitionInputDataLength,
       atp,
       atpIndex,
       NULL,
       ExpTupleDesc::SHORT_FORMAT,
       0,
       pivAttrs);
}

/////////////////////////////////////////////////////////
//
// SinglePartitionPartitioningFunction::codeGen()
//
/////////////////////////////////////////////////////////

short SinglePartitionPartitioningFunction::codeGen(Generator *generator,
						   Lng32 partInputDataLength)
{
  GenAssert(partInputDataLength == 0,"Part input values for single part.");

  // there is no object generated for a single part. function
  generator->setGenObj(NULL, NULL);
  return 0;
}

/////////////////////////////////////////////////////////
//
// ReplicateViaBroadcastPartitioningFunction::codeGen()
//
/////////////////////////////////////////////////////////

short ReplicateViaBroadcastPartitioningFunction::codeGen
        (Generator *generator, Lng32 partInputDataLength)
{
  GenAssert(partInputDataLength == 0,"Part input values for replication");

  // there is no object generated for a replication part. function
  generator->setGenObj(NULL, NULL);
  return 0;
}

/////////////////////////////////////////////////////////
//
// ReplicateNoBroadcastPartitioningFunction::codeGen()
//
/////////////////////////////////////////////////////////

short ReplicateNoBroadcastPartitioningFunction::codeGen
        (Generator *generator, Lng32 partInputDataLength)
{
  GenAssert(partInputDataLength == 0,"Part input values for replication");

  // there is no object generated for a replication part. function
  generator->setGenObj(NULL, NULL);
  return 0;
}

/////////////////////////////////////////////////////////
//
// HashPartitioningFunction::codeGen()
//
/////////////////////////////////////////////////////////

short HashPartitioningFunction::codeGen(Generator *generator,
				        Lng32 /*partInputDataLength*/)
{
  // a hash partitioning scheme always produces a partition number as output
  ex_cri_desc *partInputCriDesc =
    new(generator->getSpace()) ex_cri_desc(1,generator->getSpace());
  // GenAssert(partInputDataLength == 4,"NOT partInputDataLength == 4");
  ExHashPartInputData *generatedObject =
    new(generator->getSpace()) ExHashPartInputData(partInputCriDesc,
						   numberOfHashPartitions_);
  generator->setGenObj(NULL,  (ComTdb*)generatedObject);
  return 0;
}

/////////////////////////////////////////////////////////
//
// HashDistPartitioningFunction::codeGen()
//
/////////////////////////////////////////////////////////

short HashDistPartitioningFunction::codeGen(Generator *generator,
                                            Lng32 /*partInputDataLength*/)
{
  // a HashDist partitioning scheme always produces a partition number
  // as output
  //
  ex_cri_desc *partInputCriDesc =
    new(generator->getSpace()) ex_cri_desc(1,generator->getSpace());

  // GenAssert(partInputDataLength == 4,"NOT partInputDataLength == 4");

  ExHashDistPartInputData *generatedObject =
    new(generator->getSpace()) 
    ExHashDistPartInputData(partInputCriDesc,
                            getCountOfPartitions(),
                            getCountOfOrigHashPartitions());

  generator->setGenObj(NULL,  (ComTdb*)generatedObject);
  return 0;
}

/////////////////////////////////////////////////////////
//
// Hash2PartitioningFunction::codeGen()
//
/////////////////////////////////////////////////////////

short Hash2PartitioningFunction::codeGen(Generator *generator,
                                            Lng32 /*partInputDataLength*/)
{
  ex_cri_desc *partInputCriDesc =
    new(generator->getSpace()) ex_cri_desc(1,generator->getSpace());

  // GenAssert(partInputDataLength == 4,"NOT partInputDataLength == 4");

  ExHash2PartInputData *generatedObject = new(generator->getSpace()) 
      ExHash2PartInputData(partInputCriDesc,
                           getCountOfPartitions());

  generator->setGenObj(NULL, (ComTdb*)generatedObject);
  return 0;
}

short SkewedDataPartitioningFunction::codeGen(Generator *generator,
                                            Lng32 partInputDataLength)
{
  return partialPartFunc_ -> codeGen(generator, partInputDataLength);
}


/////////////////////////////////////////////////////////
//
// LogPhysPartitioningFunction::codeGen()
//
/////////////////////////////////////////////////////////

short LogPhysPartitioningFunction::codeGen(Generator * /*generator*/,
					   Lng32 /*partInputDataLength*/)
{
  GenAssert(0,"Should not generate code for logphys part func.");
  return 0;
}

/////////////////////////////////////////////////////////
//
// RoundRobinPartitioningFunction::codeGen()
//
/////////////////////////////////////////////////////////

short RoundRobinPartitioningFunction::codeGen(Generator *generator,
					      Lng32 /*partInputDataLength*/)
{
  // A round-robin partitioning function produces a partition
  // number as output, for now at least.  
  //
  Space *space = generator->getSpace();

  ex_cri_desc *partInputCriDesc = new (space) ex_cri_desc(1, space);

  ExRoundRobinPartInputData *generatedObject =
    new (space) ExRoundRobinPartInputData(partInputCriDesc,
                                          getCountOfPartitions(),
                                          getCountOfOrigRRPartitions());

  generator->setGenObj(NULL, (ComTdb*)generatedObject);

  return 0;
}

/////////////////////////////////////////////////////////
//
// RangePartitioningFunction::codeGen()
//
/////////////////////////////////////////////////////////

short RangePartitioningFunction::codeGen(Generator *generator,
					 Lng32 partInputDataLength)
{
  ExpGenerator         * exp_gen = generator->getExpGenerator();
  Lng32                 myOwnPartInputDataLength;

  const Int32            pivMoveAtp = 0; // only one atp is used for this expr
  const Int32            pivMoveAtpIndex = 2; // 0: consts, 1: temps, 2: result
  const ExpTupleDesc::TupleDataFormat pivFormat = // format of PIVs
                          ExpTupleDesc::SQLARK_EXPLODED_FORMAT;
  ex_cri_desc          *partInputCriDesc = new(generator->getSpace())
                           ex_cri_desc(pivMoveAtpIndex+1,
				       generator->getSpace());
  ExpTupleDesc         *partInputTupleDesc;
  ExRangePartInputData *generatedObject = NULL;

  // get the list of partition input variables
  ValueIdList          piv(getPartitionInputValuesLayout());

  CollIndex numPartInputs = piv.entries();
  CollIndex numPartKeyCols = (numPartInputs - 1) / 2;
  // the number of partition input variables must be odd
  GenAssert(2*numPartKeyCols+1 == numPartInputs,
	    "NOT 2*numPartKeyCols+1 == numPartInputs");

  Attributes **begEndAttrs;
  Int32 alignedPartKeyLen;

  // make a layout of the partition input data record
  generatePivLayout(
       generator,
       myOwnPartInputDataLength,
       pivMoveAtp,
       pivMoveAtpIndex,
       &begEndAttrs);
  
  // the aligned part key length is where the end key values start
  alignedPartKeyLen = (Int32) begEndAttrs[numPartKeyCols]->getOffset();

  if (begEndAttrs[numPartKeyCols]->getNullIndicatorLength() > 0)
    alignedPartKeyLen = MINOF(
	 alignedPartKeyLen,
	 (Int32)begEndAttrs[numPartKeyCols]->getNullIndOffset());
    
  if (begEndAttrs[numPartKeyCols]->getVCIndicatorLength() > 0)
    alignedPartKeyLen = MINOF(
	 alignedPartKeyLen,
	 begEndAttrs[numPartKeyCols]->getVCLenIndOffset());
    
  // generate a tuple desc for the whole PIV record and a cri desc
  partInputTupleDesc = new(generator->getSpace()) ExpTupleDesc(
       numPartInputs,
       begEndAttrs,
       myOwnPartInputDataLength,
       pivFormat,
       ExpTupleDesc::LONG_FORMAT,
       generator->getSpace());
  partInputCriDesc->setTupleDescriptor(pivMoveAtpIndex,partInputTupleDesc);

  // make sure we fulfill the assertions we made

  // optimizer and generator should agree on the part input data length
  GenAssert(partInputDataLength == (Lng32) myOwnPartInputDataLength,
	    "NOT partInputDataLength == myOwnPartInputDataLength");
  // the length of the begin key and the end key must be the same
  // (compare offsets of their last fields)
// Commented out because this check does not work. The check needs
// to compute the LENGTH of each key field, by subtracting the current
// offset from the next offset, taking into account varchar length
// and null indicator fields (which are not part of the length but
// increase the offset).
//GenAssert(begEndAttrs[numPartKeyCols-1]->getOffset() + alignedPartKeyLen ==
//  begEndAttrs[2*numPartKeyCols-1]->getOffset(),
//    "begin/end piv keys have different layouts");

  generatedObject = new(generator->getSpace()) ExRangePartInputData(
       partInputCriDesc,
       partInputDataLength,
       alignedPartKeyLen, //len of one part key + filler
       begEndAttrs[numPartInputs-1]->getOffset(),//offset of last field
       getCountOfPartitions(),
       generator->getSpace(),
       TRUE); // uses expressions to calculate ranges in the executor
  generatedObject->setPartitionExprAtp(pivMoveAtp);
  generatedObject->setPartitionExprAtpIndex(pivMoveAtpIndex);

  // now fill in the individual partition boundaries
  // (NOTE: there is one more than there are partitions)
  ULng32 boundaryDataLength = 0;
  for (Lng32 i = 0; i <= getCountOfPartitions(); i++)
    {
      const ItemExprList *iel = partitionBoundaries_->getBoundaryValues(i);
      ex_expr * generatedExpr = NULL;

      ValueIdList boundaryColValues;
      ULng32 checkedBoundaryLength;

      // convert the ItemExpressionList iel into a ValueIdList
      for (CollIndex kc = 0; kc < iel->entries(); kc++)
	{
	  ItemExpr *boundaryVal = (*iel)[kc];

	  // create a cast node to convert the boundary value to the
	  // data type of the column
	  ItemExpr *castBoundaryVal =
	    new(generator->wHeap()) Cast(boundaryVal,&piv[kc].getType());

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

	  boundaryColValues.insert(castBoundaryVal->getValueId());
	}

      // Now generate a contiguous move expression. Only for the first time
      // generate a tuple desc, since all tuples should be the same.
      exp_gen->generateContiguousMoveExpr(
	   boundaryColValues,
	   0, // cast nodes created above will do the move, no conv nodes
	   pivMoveAtp,
	   pivMoveAtpIndex,
	   pivFormat,
	   checkedBoundaryLength,
	   &generatedExpr);

      if (i == 0)
	{
	  // first time set the actual part key data length
	  boundaryDataLength = checkedBoundaryLength;
	}
      else
	{
	  // all boundary values (piv tuples) must have the same layout
	  // and therefore the same length
	  GenAssert(boundaryDataLength == checkedBoundaryLength,
		    "Partition boundary tuple layout mismatch");
	}

      generatedObject->setPartitionStartExpr(i,generatedExpr);
    }

  NADELETEBASIC(begEndAttrs, generator->wHeap());
  generator->setGenObj(NULL, (ComTdb*)generatedObject);
  return 0;
}

void RangePartitioningFunction::generatePivLayout(
     Generator *generator,
     Lng32 &partitionInputDataLength,
     Lng32 atp,
     Lng32 atpIndex,
     Attributes ***pivAttrs)
{
  // Make a layout of the partition input data record such that
  // begin and end key are aligned in the same way.
  // (layout = ((beg. key) (filler1) (end key) (filler2) (exclusion flag)))

  ExpGenerator *expGen = generator->getExpGenerator();
  CollIndex numPartInputs = getPartitionInputValuesLayout().entries();
  CollIndex numPartKeyCols = (numPartInputs - 1) / 2;
  // the number of partition input variables must be odd
  GenAssert(2*numPartKeyCols+1 == numPartInputs,
	    "NOT 2*numPartKeyCols+1 == numPartInputs");


  // ---------------------------------------------------------------------
  // Start by processing the begin key PIVs
  // ---------------------------------------------------------------------
  ValueIdList partialPivs;
  Attributes **returnedAttrs = NULL;
  Attributes **localPartialAttrs;
  Lng32 maxAlignment = 1;
  Lng32 alignedPartKeyLen;

  if (pivAttrs)
    {
      returnedAttrs = new(generator->wHeap()) Attributes *[numPartInputs];
    }

  CollIndex i = 0;
  for (i = 0; i < numPartKeyCols; i++)
    partialPivs.insert(getPartitionInputValuesLayout()[i]);
  
  expGen->processValIdList(
       partialPivs,
       ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
       (ULng32 &) partitionInputDataLength,
       atp,
       atpIndex,
       NULL,
       ExpTupleDesc::SHORT_FORMAT,
       0,
       &localPartialAttrs);

  if (returnedAttrs)
    for (i = 0; i < numPartKeyCols; i++)
      returnedAttrs[i] = localPartialAttrs[i];

  // ---------------------------------------------------------------------
  // Now find out the max. alignment that is needed in the begin key,
  // make sure that the end key starts on an offset that is a
  // multiple of the max. alignment in the partition input values
  // ---------------------------------------------------------------------
  for (i = 0; i < numPartKeyCols; i++)
    {
      if (localPartialAttrs[i]->getDataAlignmentSize() > maxAlignment)
	maxAlignment = localPartialAttrs[i]->getDataAlignmentSize();
      if (localPartialAttrs[i]->getVCIndicatorLength() > maxAlignment)
	maxAlignment = localPartialAttrs[i]->getVCIndicatorLength();
      if (localPartialAttrs[i]->getNullIndicatorLength() > maxAlignment)
	maxAlignment = localPartialAttrs[i]->getNullIndicatorLength();
    }

  alignedPartKeyLen = partitionInputDataLength;
  while (alignedPartKeyLen % maxAlignment != 0)
    alignedPartKeyLen++;

  // ---------------------------------------------------------------------
  // Now that we are starting on a good offset, process the end key
  // ---------------------------------------------------------------------
  partialPivs.clear();
  for (i = numPartKeyCols; i < numPartInputs-1; i++)
    partialPivs.insert(getPartitionInputValuesLayout()[i]);
  
  expGen->processValIdList(
       partialPivs,
       ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
       (ULng32 &) partitionInputDataLength,
       atp,
       atpIndex,
       NULL,
       ExpTupleDesc::SHORT_FORMAT,
       alignedPartKeyLen,
       &localPartialAttrs);

  if (returnedAttrs)
    for (i = numPartKeyCols; i < numPartInputs-1; i++)
      returnedAttrs[i] = localPartialAttrs[i-numPartKeyCols];

  // ---------------------------------------------------------------------
  // Process the exclusion flag at offset 2*alignedPartKeyLen
  // ---------------------------------------------------------------------
  partialPivs.clear();
  partialPivs.insert(getPartitionInputValuesLayout()[numPartInputs-1]);
  
  expGen->processValIdList(
       partialPivs,
       ExpTupleDesc::SQLARK_EXPLODED_FORMAT,
       (ULng32 &) partitionInputDataLength,
       atp,
       atpIndex,
       NULL,
       ExpTupleDesc::SHORT_FORMAT,
       2*alignedPartKeyLen,
       &localPartialAttrs);

  // set up return values

  if (returnedAttrs)
    returnedAttrs[numPartInputs-1] = localPartialAttrs[0];

  partitionInputDataLength += 2*alignedPartKeyLen;

  if (pivAttrs)
    {
      *pivAttrs = returnedAttrs;
    }
  else
    {
      NADELETEBASIC(returnedAttrs, generator->wHeap());
    }
}