Google Git
Sign in
apache / trafodion / refs/tags/2.3.0rc1 / . / core / sql / optimizer / PartFunc.cpp
blob: 425a0b65c27c4c0c5ecabb1e5a77758dcb4340ec [file] [log] [blame]
/**********************************************************************
// @@@ 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: PartFunc.cpp
* Description: Partitioning Function
* Created: 11/16/1994
* Language: C++
*
*
*
*
**************************************************************************
*/
// -----------------------------------------------------------------------
#include "PartReq.h"
#include "PhyProp.h"
#include "ItemColRef.h"
#include "ItemLog.h"
#include "ItemFunc.h"
#include "ItemOther.h"
#include "ItemArith.h"
#include "opt.h"
#include "str.h"
#include "NumericType.h"
#include "MiscType.h"
#include "NAFileSet.h"
#include "SearchKey.h"
#include "GroupAttr.h"
#include "Generator.h"
// To test hash2 part number generated by a direct call to
// ExHDHash::hash() and compare the result, just uncomment out
// the following two lines.
//#include "Generator.h"
#include "exp_function.h"
// ***********************************************************************
// A function having an external linkage to allow display() to
// be called on a tree object. This is a workaround for bugs/missing
// functionality in ObjectCenter that cause display() to become
// an undefined symbol.
// ***********************************************************************
void displayPartitioningFunction(const PartitioningFunction& pf)
{
pf.display();
}
void displayPartitioningFunction(const PartitioningFunction* pf)
{
if (pf)
pf->display();
}
void displayPartitionBoundaries(const RangePartitionBoundaries& pb)
{
pb.display();
}
void displayPartitionBoundaries(const RangePartitionBoundaries* pb)
{
if (pb)
pb->display();
}
// ***********************************************************************
// PartitioningFunction
// ***********************************************************************
//------------------------------------------------------------------------
// PartitioningFunction destructor.
//------------------------------------------------------------------------
PartitioningFunction::~PartitioningFunction()
{
if (nodeMap_ != NULL)
{
delete nodeMap_;
}
} //PartitioningFunction destructor
// -----------------------------------------------------------------------
// Methods for perform type-safe pointer casts.
// -----------------------------------------------------------------------
const LogPhysPartitioningFunction*
PartitioningFunction::castToLogPhysPartitioningFunction() const
{ return NULL; }
const SinglePartitionPartitioningFunction*
PartitioningFunction::castToSinglePartitionPartitioningFunction() const
{ return NULL; }
const ReplicateViaBroadcastPartitioningFunction*
PartitioningFunction::castToReplicateViaBroadcastPartitioningFunction() const
{ return NULL; }
const ReplicateNoBroadcastPartitioningFunction*
PartitioningFunction::castToReplicateNoBroadcastPartitioningFunction() const
{ return NULL; }
const HashPartitioningFunction*
PartitioningFunction::castToHashPartitioningFunction() const
{ return NULL; }
const TableHashPartitioningFunction*
PartitioningFunction::castToTableHashPartitioningFunction() const
{ return NULL; }
const HashDistPartitioningFunction*
PartitioningFunction::castToHashDistPartitioningFunction() const
{ return NULL; }
const Hash2PartitioningFunction*
PartitioningFunction::castToHash2PartitioningFunction() const
{ return NULL; }
const RangePartitioningFunction*
PartitioningFunction::castToRangePartitioningFunction() const
{ return NULL; }
const RoundRobinPartitioningFunction*
PartitioningFunction::castToRoundRobinPartitioningFunction() const
{ return NULL; }
const SkewedDataPartitioningFunction*
PartitioningFunction::castToSkewedDataPartitioningFunction() const
{ return NULL; };
const HivePartitioningFunction*
PartitioningFunction::castToHivePartitioningFunction() const
{ return NULL; };
// ---------------------------------------------------------------------
// Method to test if the partitioning key contains any approximate
// numeric type columns. Necessary because in some cases certain
// parallel operations do not function properly if the partitioning
// key of the table contains approximate numeric columns.
// ---------------------------------------------------------------------
NABoolean PartitioningFunction::partKeyContainsFloatColumn() const
{
if (NOT getPartitioningKey().isEmpty())
{
ValueIdSet partKeyColumns = getPartitioningKey();
for (ValueId vid = partKeyColumns.init();
partKeyColumns.next(vid);
partKeyColumns.advance(vid))
{
const NAType& columnType = vid.getType();
if ((columnType.getTypeQualifier() == NA_NUMERIC_TYPE) AND
NOT ((NumericType&)columnType).isExact())
{
return TRUE;
}
} // end for
} // end if part key is not empty
return FALSE;
} // partKeyContainsFloatColumn()
// -----------------------------------------------------------------------
// PartitioningFunction::getNodeMap()
// Return base class node map by default.
// -----------------------------------------------------------------------
const NodeMap*
PartitioningFunction::getNodeMap() const { return nodeMap_; }
// use any existing nodemap from my req or my child (or synthesize one) that
// matches my partition count requirement
void
PartitioningFunction::useNodeMapFromReqOrChild(PartitioningRequirement *req,
PartitioningFunction *childPF,
NABoolean forESP)
{
ULng32 partCnt = (ULng32)getCountOfPartitions();
NodeMap *myNodeMap = NULL;
if (req->isRequirementFullySpecified() &&
(CmpCommon::getDefault(COMP_BOOL_87) != DF_ON) && getNodeMap() &&
(getNodeMap()->getNumEntries() == partCnt)) {
// we are a copy of a partitioning function that realizes a fully specified
// partitioning requirement. So, do nothing. our nodemap is good enough.
return;
} else if ((CmpCommon::getDefault(COMP_BOOL_87) != DF_ON) &&
(partCnt == (ULng32)childPF->getCountOfPartitions()) &&
childPF->getNodeMap() &&
(childPF->getNodeMap()->getNumEntries() == partCnt)) {
// we are a copy of a partitioning function that realizes a fuzzy
// partitioning requirement. childPF is the partitioning function of
// the synthesized physical property of the child of an Exchange. So,
// our child's partitioning function's nodemap is good enough.
myNodeMap = childPF->getNodeMap()->copy();
} else {
// only as a last resort do we synthesize a new node map
// Synthesize a nodemap based on the nodemap of the child and the
// desired number of ESPs. Using synthesizeLogicalMap() assumes
// that the lower and upper ESPs are associated via grouping. This
// assumption is not valid when considering the communication
// patterns between the upper and lower ESPs, but this assumption
// will lead to a reasonable nodemap for the upper ESPs.
myNodeMap = ((NodeMap*)childPF->getNodeMap())
->synthesizeLogicalMap(partCnt, forESP);
}
CMPASSERT(myNodeMap);
for(CollIndex i = 0; i < partCnt; i++) {
myNodeMap->setPartitionState(i, NodeMapEntry::ACTIVE);
}
CMPASSERT(myNodeMap->getNumActivePartitions() == partCnt);
replaceNodeMap(myNodeMap);
}
// -----------------------------------------------------------------------
// PartitioningFunction::replaceNodeMap()
// Replace node map with a newly specified node map.
// -----------------------------------------------------------------------
void
PartitioningFunction::replaceNodeMap(NodeMap* nodeMap)
{
if (nodeMap_ != NULL)
{
#ifndef NDEBUG
//assertion for NATable caching.
//assert if this object is not on the system heap i.e. (collHeap()!=0)
//and if the heap is not the statementHeap.
//The assertion would happen if this object is on a NATable heap,
//since we should not be changing the partitionins function associated
//with a cached NATable object
if((collHeap()) && (collHeap() != CmpCommon::statementHeap()))
CMPASSERT(FALSE);
#endif //NDEBUG
//Fix for solution 10-040120-2524
//Unconditionally deleting the nodeMap_
//was causing problems as the object pointed
//to by nodeMap_ was being used by other guys also.
//
//delete the object if is on the system heap.
//If it is on the the statement heap then it will
//be delete when the statement heap is deleted.
if(!nodeMap_->collHeap())delete nodeMap_;
}
nodeMap_ = nodeMap;
} // PartitioningFunction::replaceNodeMap()
// -----------------------------------------------------------------------
// PartitioningFunction::copy()
// Virtual copy constructor returns a copy of myself.
// -----------------------------------------------------------------------
PartitioningFunction* PartitioningFunction::copy() const
{
// illegal to call copy() of the base class
CMPABORT;
return NULL;
}
// -----------------------------------------------------------------------
// PartitioningFunction::normalizePartitioningKeys()
// Rewrite the partitioning keys of the partitioning function in
// terms of the VEGReference for the VEG to which the partitioning
// key column belongs.
// -----------------------------------------------------------------------
void PartitioningFunction::normalizePartitioningKeys(NormWA& normWARef)
{
partitioningKeyColumns_.normalizeNode(normWARef);
partitioningKeyPredicates_.normalizeNode(normWARef);
if(partitioningExpression_)
partitioningExpression_ =
partitioningExpression_->normalizeNode(normWARef);
if(partitionSelectionExpr_)
partitionSelectionExpr_ =
partitionSelectionExpr_->normalizeNode(normWARef);
} // PartitioningFunction::normalizePartitioningKeys
// --------------------------------------------------------------------
// A method that is used by optimizer for comparing partitioning
// function with the random number partitioning function i.e. it only
// compares the partitioning func type and the number of partitions.
// It does not compare partitioning key.
// --------------------------------------------------------------------
NABoolean
PartitioningFunction::isKnownReplicaPartFunc() const
{
return
(
(isASkewedDataPartitioningFunction() &&
((SkewedDataPartitioningFunction*)this)->getSkewProperty().isBroadcasted())
||
isAReplicateViaBroadcastPartitioningFunction()
||
isAReplicateNoBroadcastPartitioningFunction());
}
COMPARE_RESULT
PartitioningFunction::comparePartFuncsForUnion
(const PartitioningFunction &other) const
{
NABoolean myRepPartFunc = isKnownReplicaPartFunc();
NABoolean otherRepPartFunc = other.isKnownReplicaPartFunc();
if ( myRepPartFunc || otherRepPartFunc ) return INCOMPATIBLE;
if (getCountOfPartitions() == other.getCountOfPartitions())
return SAME;
return INCOMPATIBLE;
} // PartitioningFunction::comparePartFuncsForUnion
// -----------------------------------------------------------------------
// PartitioningFunction::comparePartFuncToFunc()
// Partitioning function comparison method for hash, round robin, replication,
// and single_partition partitioning functions i.e. everything except regular
// range partitioning functions.
// "Other" must be a partitioning function.
// -----------------------------------------------------------------------
COMPARE_RESULT
PartitioningFunction::comparePartFuncToFunc
(const PartitioningFunction &other) const
{
if ((getPartitioningFunctionType() == other.getPartitioningFunctionType()) AND
(getCountOfPartitions() == other.getCountOfPartitions()) AND
(getPartitioningKey() == other.getPartitioningKey())
)
return SAME;
else
return INCOMPATIBLE;
} // PartitioningFunction::comparePartFuncToFunc
static inline
PartitioningFunction* getPhysPartFunc(const PartitioningFunction* x)
{
PartitioningFunction* phys = 0;
if ( x -> isALogPhysPartitioningFunction() ) {
phys = x -> castToLogPhysPartitioningFunction() ->
getPhysPartitioningFunction();
}
return phys;
}
NABoolean PartitioningFunction::isAGroupingOf(
const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = 1;
// By default we are not smart and take no risk. However, two
// partitioning functions that are equal are always a grouping of
// each other.
return (comparePartFuncToFunc(other) == SAME);
}
PartitioningRequirement* PartitioningFunction::makePartitioningRequirement()
{
// Redefine PartitioningFunction::makePartitioningRequirement()
CMPABORT;
return NULL;
}
// -----------------------------------------------------------------------
// PartitioningFunction::scaleNumberOfPartitions
// -----------------------------------------------------------------------
PartitioningFunction * PartitioningFunction::scaleNumberOfPartitions(
Lng32 &suggestedNewNumberOfPartitions,
PartitionGroupingDistEnum partGroupDist)
{
// if we come here this means that the derived class did not
// override scaleNumberOfPartitions(). In other words, the derived
// class doesn't know how to change its number of partitions.
// Therefore, ignore the suggestion and return the current number
// of partitions, unless we scale down to 1.
if (suggestedNewNumberOfPartitions == 1)
return new(CmpCommon::statementHeap())
SinglePartitionPartitioningFunction();
// If an expression has been generated, then we want to discard it
// because it may no longer be correct.
storeExpression(NULL);
if ( suggestedNewNumberOfPartitions != getCountOfPartitions() ) {
NodeMap* newNodeMap =
new (heap_) NodeMap(heap_, suggestedNewNumberOfPartitions);
replaceNodeMap(newNodeMap);
}
suggestedNewNumberOfPartitions = getCountOfPartitions();
return this;
}
// -----------------------------------------------------------------------
// PartitioningFunction::copyAndRemap()
// -----------------------------------------------------------------------
PartitioningFunction*
PartitioningFunction::copyAndRemap
(ValueIdMap& map, NABoolean mapItUp) const
{
if (getPartitioningKey().entries() == 0)
return (PartitioningFunction*)this;
PartitioningFunction* newPartFunc = copy(); // invoke virtual copy constructor
newPartFunc->remapIt(this, map, mapItUp);
return newPartFunc;
} // PartitioningFunction::copyAndRemap()
// -----------------------------------------------------------------------
// PartitioningFunction::remapIt()
// -----------------------------------------------------------------------
void PartitioningFunction::remapIt
(const PartitioningFunction* opf,
ValueIdMap& map, NABoolean mapItUp)
{
// Clear because rewrite insists on it being so.
partitioningKeyColumns_.clear();
partitioningKeyPredicates_.clear();
partitionInputValues_.clear();
partitionInputValuesLayout_.clear();
if (mapItUp)
{
map.rewriteValueIdSetUp(partitioningKeyColumns_,
opf->partitioningKeyColumns_);
map.rewriteValueIdSetUp(partitioningKeyPredicates_,
opf->partitioningKeyPredicates_);
// Note the pivs don't need mapping, they just need to be copied.
map.rewriteValueIdSetUp(partitionInputValues_,
opf->partitionInputValues_);
map.rewriteValueIdListUp(partitionInputValuesLayout_,
opf->partitionInputValuesLayout_);
}
else
{
map.rewriteValueIdSetDown(opf->partitioningKeyColumns_,
partitioningKeyColumns_);
map.rewriteValueIdSetDown(opf->partitioningKeyPredicates_,
partitioningKeyPredicates_);
// Note the pivs don't need mapping, they just need to be copied.
map.rewriteValueIdSetDown(opf->partitionInputValues_,
partitionInputValues_);
map.rewriteValueIdListDown(opf->partitionInputValuesLayout_,
partitionInputValuesLayout_);
}
if (partitioningExpression_)
partitioningExpression_ = partitioningExpression_->
mapAndRewrite(map,NOT mapItUp).getItemExpr();
} // PartitioningFunction::remapIt()
// -----------------------------------------------------------------------
// PartitioningFunction::shouldUseSynchronousAccess()
// -----------------------------------------------------------------------
NABoolean PartitioningFunction::shouldUseSynchronousAccess(
const ReqdPhysicalProperty* rpp,
const EstLogPropSharedPtr& inputLogProp,
GroupAttributes* ga) const
{
// This default implementation handles all partitioning functions
// except range and single partition partitioning functions.
// Synchronous access only makes sense for range partitioned tables.
// So, unless the user is trying to force synchronous access
// (by disabling asynchronous access), then return FALSE.
NABoolean shouldUseSynchronousAccess = FALSE;
// NOTE: we cannot force synchronous access for non-range partitioned
// tables if there is a required logical order and/or arrangement!
// Doing so could lead to incorrect results! So, if the user is trying to
// force synchronous access, then they are out of luck!
if (NOT rpp->getLogicalOrderOrArrangementFlag())
{
const LogicalPartitioningRequirement *lpr =
rpp->getLogicalPartRequirement();
// Don't do synchronous access if the user is forcing a PAPA node
// and is not forcing the number of PAs
if ((lpr != NULL) AND lpr->getMustUsePapa() AND
(lpr->getNumClientsReq() == ANY_NUMBER_OF_PARTITIONS))
return FALSE;
// Get the value from the defaults table that specifies whether
// asynchronous access is ok. If it is not, then we need to force
// synchronous access. Also force synchronous access if the user
// forced it via C.Q. shape.
// See if the user is trying to force synchronous access.
if ((CmpCommon::getDefault(ATTEMPT_ASYNCHRONOUS_ACCESS) == DF_OFF) OR
((lpr != NULL) AND
(lpr->getNumClientsReq() != ANY_NUMBER_OF_PARTITIONS) AND
(lpr->getNumClientsReq() < getCountOfPartitions())))
{
shouldUseSynchronousAccess = TRUE;
}
}
return shouldUseSynchronousAccess;
}
// -----------------------------------------------------------------------
// Virtual functions that must be redefined for derived classes.
// -----------------------------------------------------------------------
Lng32 PartitioningFunction::getCountOfPartitions() const
{
// Redefine PartitioningFunction::getCountOfPartitions()
CMPABORT;
return 1;
}
NABoolean PartitioningFunction::canProducePartitioningKeyPredicates() const
{
return TRUE; // the most common case is TRUE
}
const ValueIdSet& PartitioningFunction::getPartitioningKeyPredicates() const
{
CMPASSERT(partKeyPredsCreated_);
return partitioningKeyPredicates_;
}
const ValueIdSet& PartitioningFunction::getPartitionInputValues() const
{
CMPASSERT(partKeyPredsCreated_);
return partitionInputValues_;
}
const ValueIdList& PartitioningFunction::getPartitionInputValuesLayout() const
{
CMPASSERT(partKeyPredsCreated_);
return partitionInputValuesLayout_;
}
void PartitioningFunction::createPartitioningKeyPredicates()
{
// Redefine PartitioningFunction::createPartitioningKeyPredicates()
CMPABORT;
}
void PartitioningFunction::replacePivs(
const ValueIdList& newPivs,
const ValueIdSet& newPartKeyPreds)
{
// Redefine PartitioningFunction::replacePivs()
CMPABORT;
}
PartitioningFunction*
PartitioningFunction::createPartitioningFunctionForIndexDesc
(IndexDesc *idesc) const
{
// Redefine PartitioningFunction::createPartitioningFunctionForIndexDesc()
CMPABORT;
return NULL;
}
ItemExpr* PartitioningFunction::createPartitioningExpression()
{
// Redefine PartitioningFunction::createPartitioningExpression()
CMPABORT;
return NULL;
}
void PartitioningFunction::createPartSelectionExprFromSearchKey(
const ValueId beginPartSelId,
const ValueId endPartSelId,
ValueIdList &partSelectionValIds) const
{
partSelectionValIds.insert(beginPartSelId);
partSelectionValIds.insert(endPartSelId);
}
// -----------------------------------------------------------------------
// PartitioningFunction::preCodeGen()
// Rewrite the partitioning keys of the partitioning function that
// are expressed using VEGReferences in terms of the available values.
// -----------------------------------------------------------------------
void PartitioningFunction::preCodeGen(const ValueIdSet& availableValues)
{
ValueIdSet noExternalInputs;
partitioningKeyColumns_.replaceVEGExpressions
(availableValues, noExternalInputs,
FALSE, NULL, TRUE);
if (partitioningExpression_)
{
partitioningExpression_ = partitioningExpression_->replaceVEGExpressions(
availableValues,
noExternalInputs,
FALSE,
NULL,
TRUE); // deep copy because copy constructor made a shallow copy
partitioningExpression_->synthTypeAndValueId(TRUE);
}
} // PartitioningFunction::preCodeGen()
const NAString PartitioningFunction::getText() const
{
CMPABORT;
return NAString("some type of partitioning function",
CmpCommon::statementHeap());
}
void PartitioningFunction::setupForStatement()
{
if(setupForStatement_)
return;
setupForStatement_ = TRUE;
resetAfterStatement_ = FALSE;
}
void PartitioningFunction::resetAfterStatement()
{
if(resetAfterStatement_)
return;
partitioningKeyColumns_.clear();
partitioningKeyPredicates_.clear();
partitionInputValues_.clear();
partitionInputValuesLayout_.clear();
partitionSelectionExprInputs_.clear();
partKeyPredsCreated_=FALSE;
assignPartition_ = FALSE;
partitioningExpression_=NULL;
dataConversionErrorFlag_=NULL;
resetAfterStatement_ = TRUE;
setupForStatement_ = FALSE;
}
// -----------------------------------------------------------------------
// Method for debugging
// -----------------------------------------------------------------------
void PartitioningFunction::print(FILE* ofd, const char* indent,
const char* title) const
{
BUMP_INDENT(indent);
fprintf(ofd,"%s--Partitioning-Function----------------\n",NEW_INDENT);
fprintf(ofd,"%s%s (%d)\n",
NEW_INDENT, title, getPartitioningFunctionType());
fprintf(ofd,"%snumber of partitions (%d)\n",
NEW_INDENT, getCountOfPartitions());
if (NOT partitioningKeyColumns_.isEmpty())
partitioningKeyColumns_.print(ofd,
NEW_INDENT,
"Partitioning Key Columns");
if(partKeyPredsCreated())
fprintf(ofd,"%sPartition Key Predicates Created\n",NEW_INDENT);
if (NOT partitioningKeyPredicates_.isEmpty())
partitioningKeyPredicates_.print(ofd,
NEW_INDENT,
"Partitioning Key Predicates");
if (NOT partitionInputValues_.isEmpty())
partitionInputValues_.print(ofd, NEW_INDENT,
"Partition Input Values");
if (NOT partitionInputValuesLayout_.isEmpty())
partitionInputValuesLayout_.print(ofd, NEW_INDENT,
"Partition Input Values Layout");
if (getPartitioningExpression()) {
NAString partExpr("Partitioning Expression\n", CmpCommon::statementHeap());
getPartitioningExpression()->
unparse(partExpr, DEFAULT_PHASE, EXPLAIN_FORMAT);
fprintf(ofd,partExpr);
}
if(assignPartition())
fprintf(ofd,"%sDoes Partition Assignment\n",NEW_INDENT);
fprintf(ofd,"%s--Partitioning-Function----------------\n",NEW_INDENT);
} // PartitioningFunction::print()
void PartitioningFunction::display() const { print(); }
// ***********************************************************************
// SinglePartitionPartitioningFunction
// ***********************************************************************
SinglePartitionPartitioningFunction::~SinglePartitionPartitioningFunction() {}
Lng32 SinglePartitionPartitioningFunction::getCountOfPartitions() const
{
return EXACTLY_ONE_PARTITION;
}
PartitioningRequirement*
SinglePartitionPartitioningFunction::makePartitioningRequirement()
{
return new (CmpCommon::statementHeap())
RequireExactlyOnePartition(this);
}
const SinglePartitionPartitioningFunction*
SinglePartitionPartitioningFunction::castToSinglePartitionPartitioningFunction()
const
{
return this;
}
PartitioningFunction*
SinglePartitionPartitioningFunction::copy() const
{
return new (CmpCommon::statementHeap())
SinglePartitionPartitioningFunction(*this, CmpCommon::statementHeap());
}
void SinglePartitionPartitioningFunction::createPartitioningKeyPredicates()
{
// do nothing, there aren't any partitioning key preds for a single
// partition
storePartitioningKeyPredicates(ValueIdSet());
}
void SinglePartitionPartitioningFunction::replacePivs(
const ValueIdList& newPivs,
const ValueIdSet& newPartKeyPreds)
{
// do nothing, there aren't any pivs for a single
// partition
}
// -----------------------------------------------------------------------
// SinglePartitionPartitioningFunction::createPartitioningExpression()
// -----------------------------------------------------------------------
ItemExpr*
SinglePartitionPartitioningFunction::createPartitioningExpression()
{
if (getExpression()) // already constructed?
return getExpression(); // reuse it!
// ---------------------------------------------------------------------
// Construct an expression of the form: ConstValue(0)
// Allocate ValueIds and type synthesize this partitioning function.
// ---------------------------------------------------------------------
ItemExpr * partFunc = new (CmpCommon::statementHeap())
Cast(new (CmpCommon::statementHeap()) SystemLiteral(0),
new (CmpCommon::statementHeap()) SQLInt(CmpCommon::statementHeap(), FALSE,FALSE));
partFunc->synthTypeAndValueId();
storeExpression(partFunc);
return partFunc;
}
NABoolean SinglePartitionPartitioningFunction::isAGroupingOf(
const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = other.getCountOfPartitions();
// A single partition is a grouping of about anything. There
// is only one exception: the replication partitioning "function".
// Combining the partitions of a replication partitioning
// function would get us multiple copies of the data which is
// different from the single partition.
return ((other.castToReplicateViaBroadcastPartitioningFunction() == NULL) AND
(other.castToReplicateNoBroadcastPartitioningFunction() == NULL));
}
// -----------------------------------------------------------------------
// SinglePartitionPartitioningFunction::shouldUseSynchronousAccess()
// -----------------------------------------------------------------------
NABoolean SinglePartitionPartitioningFunction::shouldUseSynchronousAccess(
const ReqdPhysicalProperty* rpp,
const EstLogPropSharedPtr& inputLogProp,
GroupAttributes* ga) const
{
// An unpartitioned table does not need to be accessed synchronously
// (or asynchronously, for that matter). So, always return FALSE.
return FALSE;
}
// -----------------------------------------------------------------------
// Push down checking: testing condition C1.2: all tables are not
// partitioned.
// -----------------------------------------------------------------------
NABoolean
SinglePartitionPartitioningFunction::partFuncAndFuncPushDownCompatible(
const PartitioningFunction& other) const
{
if ( other.castToSinglePartitionPartitioningFunction() == 0 )
return FALSE;
return TRUE;
}
const NAString SinglePartitionPartitioningFunction::getText() const
{
return "exactly 1 partition";
}
void SinglePartitionPartitioningFunction::print(FILE* ofd, const char* indent,
const char* title) const
{
PartitioningFunction::print(ofd, indent,
"SinglePartitionPartitioningFunction");
}
// ***********************************************************************
// ReplicateViaBroadcastPartitioningFunction
// ***********************************************************************
ReplicateViaBroadcastPartitioningFunction::
~ReplicateViaBroadcastPartitioningFunction()
{}
Lng32 ReplicateViaBroadcastPartitioningFunction::getCountOfPartitions() const
{ return numberOfPartitions_; }
const ReplicateViaBroadcastPartitioningFunction*
ReplicateViaBroadcastPartitioningFunction::
castToReplicateViaBroadcastPartitioningFunction() const
{ return this; }
PartitioningFunction *
ReplicateViaBroadcastPartitioningFunction::scaleNumberOfPartitions(
Lng32 &suggestedNewNumberOfPartitions,
PartitionGroupingDistEnum partGroupDist)
{
if (suggestedNewNumberOfPartitions == 1)
return new(CmpCommon::statementHeap())
SinglePartitionPartitioningFunction();
// If an expression has been generated, then we want to discard it
// because it may no longer be correct.
storeExpression(NULL);
numberOfPartitions_ = suggestedNewNumberOfPartitions;
return this;
}
PartitioningRequirement*
ReplicateViaBroadcastPartitioningFunction::makePartitioningRequirement()
{
return new (CmpCommon::statementHeap())
RequireReplicateViaBroadcast(this);
}
PartitioningFunction*
ReplicateViaBroadcastPartitioningFunction::copy() const
{
return new (CmpCommon::statementHeap())
ReplicateViaBroadcastPartitioningFunction(*this);
}
void ReplicateViaBroadcastPartitioningFunction::
createPartitioningKeyPredicates()
{
// Do nothing, the partitioning key preds of a replication "function"
// are the empty set. By doing nothing we will return all data for
// each partition.
storePartitioningKeyPredicates(ValueIdSet());
}
void ReplicateViaBroadcastPartitioningFunction::
replacePivs(const ValueIdList& newPivs,
const ValueIdSet& newPartKeyPreds)
{
// Do nothing, the pivs of a replication "function"
// are the empty set.
}
ItemExpr*
ReplicateViaBroadcastPartitioningFunction::createPartitioningExpression()
{
return NULL;
}
NABoolean ReplicateViaBroadcastPartitioningFunction::isAGroupingOf(
const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = 1;
// A replication partitioning function is never a grouping of any
// other partitioning function than itself. See the definition of a
// grouping: it is created by merging two partitions of "other"
// zero or more times.
return (comparePartFuncToFunc(other) == SAME);
}
const NAString ReplicateViaBroadcastPartitioningFunction::getText() const
{
char ntimes[20];
sprintf(ntimes,"%d",numberOfPartitions_);
return NAString("broadcast ") + ntimes + " times";
}
void ReplicateViaBroadcastPartitioningFunction::
print(FILE* ofd, const char* indent,
const char* title) const
{
PartitioningFunction::
print(ofd, indent, "ReplicateViaBroadcastPartitioningFunction");
}
// ***********************************************************************
// ReplicateNoBroadcastPartitioningFunction
// ***********************************************************************
ReplicateNoBroadcastPartitioningFunction::
~ReplicateNoBroadcastPartitioningFunction()
{}
Lng32 ReplicateNoBroadcastPartitioningFunction::getCountOfPartitions() const
{ return numberOfPartitions_; }
const ReplicateNoBroadcastPartitioningFunction*
ReplicateNoBroadcastPartitioningFunction::
castToReplicateNoBroadcastPartitioningFunction() const
{ return this; }
PartitioningFunction *
ReplicateNoBroadcastPartitioningFunction::scaleNumberOfPartitions(
Lng32 &suggestedNewNumberOfPartitions,
PartitionGroupingDistEnum partGroupDist)
{
if (suggestedNewNumberOfPartitions == 1)
return new(CmpCommon::statementHeap())
SinglePartitionPartitioningFunction();
// If an expression has been generated, then we want to discard it
// because it may no longer be correct.
storeExpression(NULL);
numberOfPartitions_ = suggestedNewNumberOfPartitions;
return this;
}
PartitioningRequirement*
ReplicateNoBroadcastPartitioningFunction::makePartitioningRequirement()
{
return new (CmpCommon::statementHeap())
RequireReplicateNoBroadcast(this);
}
PartitioningFunction*
ReplicateNoBroadcastPartitioningFunction::copy() const
{
return new (CmpCommon::statementHeap())
ReplicateNoBroadcastPartitioningFunction(*this);
}
void ReplicateNoBroadcastPartitioningFunction::
createPartitioningKeyPredicates()
{
// Do nothing, the partitioning key preds of a replication "function"
// are the empty set. By doing nothing we will return all data for
// each partition.
storePartitioningKeyPredicates(ValueIdSet());
}
void ReplicateNoBroadcastPartitioningFunction::
replacePivs(const ValueIdList& newPivs,
const ValueIdSet& newPartKeyPreds)
{
// Do nothing, the pivs of a replication "function"
// are the empty set.
}
ItemExpr*
ReplicateNoBroadcastPartitioningFunction::createPartitioningExpression()
{
return NULL;
}
NABoolean ReplicateNoBroadcastPartitioningFunction::isAGroupingOf(
const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = 1;
// A replication partitioning function is never a grouping of any
// other partitioning function than itself. See the definition of a
// grouping: it is created by merging two partitions of "other"
// zero or more times.
return (comparePartFuncToFunc(other) == SAME);
}
const NAString ReplicateNoBroadcastPartitioningFunction::getText() const
{
char ntimes[20];
sprintf(ntimes,"%d",numberOfPartitions_);
return NAString("replicate no broadcast ") + ntimes + " times";
}
void ReplicateNoBroadcastPartitioningFunction::
print(FILE* ofd, const char* indent,
const char* title) const
{
PartitioningFunction::
print(ofd, indent, "ReplicateNoBroadcastPartitioningFunction");
}
// ***********************************************************************
// HashPartitioningFunction
// ***********************************************************************
HashPartitioningFunction::~HashPartitioningFunction() {}
Lng32 HashPartitioningFunction::getCountOfPartitions() const
{ return numberOfHashPartitions_; }
const HashPartitioningFunction*
HashPartitioningFunction::castToHashPartitioningFunction() const
{ return this; }
PartitioningRequirement*
HashPartitioningFunction::makePartitioningRequirement()
{
return new (CmpCommon::statementHeap())
RequireHash(this);
}
PartitioningFunction*
HashPartitioningFunction::copy() const
{
return new (CmpCommon::statementHeap())
HashPartitioningFunction(*this, CmpCommon::statementHeap());
}
// -----------------------------------------------------------------------
// PartitioningFunction::createBetweenPartitioningKeyPredicates()
// -----------------------------------------------------------------------
void PartitioningFunction::createBetweenPartitioningKeyPredicates(
const char * pivLoName,
const char * pivHiName,
ItemExpr * partNumExpr,
NABoolean useHash2Split)
{
if (NOT partKeyPredsCreated() || partNumExpr != NULL)
{
ItemExpr * rootPtr;
ItemExpr * loPart;
ItemExpr * hiPart;
ValueIdSet setOfPartKeyPredicates;
// by default we use the partitioning function's expression
// to compute the partition number
if (partNumExpr == NULL)
partNumExpr = createPartitioningExpression();
// compute part input values if not already done so
if (partitionInputValues_.isEmpty())
{
ValueIdList partInputValues;
// must specify PIV names if they need to be created
CMPASSERT(pivLoName && pivHiName);
// the partition input values are two integer values: lo and hi part #
loPart = new (CmpCommon::statementHeap())
HostVar(pivLoName,
new (CmpCommon::statementHeap()) SQLInt(CmpCommon::statementHeap(), FALSE,FALSE),
TRUE);
hiPart = new (CmpCommon::statementHeap())
HostVar(pivHiName,
new (CmpCommon::statementHeap()) SQLInt(CmpCommon::statementHeap(), FALSE,FALSE),
TRUE);
loPart->synthTypeAndValueId();
hiPart->synthTypeAndValueId();
partInputValues.insert(loPart->getValueId());
partInputValues.insert(hiPart->getValueId());
storePartitionInputValues(partInputValues);
}
else
{
loPart = getPartitionInputValuesLayout()[0].getItemExpr();
hiPart = getPartitionInputValuesLayout()[1].getItemExpr();
}
// -----------------------------------------------------------------
// The partitioning key predicate for a hash partitioning selects
// a range of hash partitions with the following predicate:
// partNumExpr >= :loPart AND partNumExpr < :hiPart
// where the hash function is the partitioning function generated
// by the createPartitioningExpression() method and the host variables
// are generated here as partition input values.
// -----------------------------------------------------------------
if (useHash2Split)
{
// For a HASH2 function, the PIVs are expressed as min and max
// hash values to be retrieved. These need to be converted to
// partition numbers when we use them here. Note that we need to
// use the original number of HASH2 partitions here. For example,
// if we have a table salted with 64 salt buckets and we use
// 8 ESPs (8 partitions), the predicate needs to select 8 different
// salt values (original partitions) for each ESP.
CMPASSERT(isAHash2PartitioningFunction());
UInt32 numOfOrigPartns = castToHash2PartitioningFunction()->
getCountOfOrigHashPartitions();
char numPartsString[30];
snprintf(numPartsString,sizeof(numPartsString),"%d",numOfOrigPartns);
NAString numPartsLiteral(numPartsString);
ConstValue *numPartns = new (CmpCommon::statementHeap())
ConstValue(new (CmpCommon::statementHeap())
SQLInt(CmpCommon::statementHeap(), FALSE,
FALSE),
(void *) &numOfOrigPartns,
(Lng32) sizeof(numOfOrigPartns),
&numPartsLiteral,
CmpCommon::statementHeap());
loPart = new (CmpCommon::statementHeap())
Hash2Distrib(loPart, numPartns);
hiPart = new (CmpCommon::statementHeap())
Hash2Distrib(hiPart, numPartns);
}
// lower bound
rootPtr = new (CmpCommon::statementHeap())
BiRelat(ITM_GREATER_EQ,
partNumExpr,
loPart,
TRUE);
rootPtr->synthTypeAndValueId();
setOfPartKeyPredicates += rootPtr->getValueId();
// upper bound
rootPtr = new (CmpCommon::statementHeap())
BiRelat((useHash2Split ? ITM_LESS_EQ : ITM_LESS),
partNumExpr,
hiPart,
TRUE);
rootPtr->synthTypeAndValueId();
setOfPartKeyPredicates += rootPtr->getValueId();
// Store the set of key predicates in the partitioning attributes.
storePartitioningKeyPredicates(setOfPartKeyPredicates);
}
} // PartitioningFunction::createBetweenPartitioningKeyPredicates()
void HashPartitioningFunction::createPartitioningKeyPredicates()
{
createBetweenPartitioningKeyPredicates(
"_sys_HostVarLoHashPart",
"_sys_HostVarHiHashPart");
}
// -----------------------------------------------------------------------
// HashPartitioningFunction::replacePivs()
// -----------------------------------------------------------------------
void HashPartitioningFunction::replacePivs(
const ValueIdList& newPivs,
const ValueIdSet& newPartKeyPreds)
{
// Overwrite the old pivs, part key preds, and part expr. with the new ones.
storePartitionInputValues(newPivs);
storePartitioningKeyPredicates(newPartKeyPreds);
} // HashPartitioningFunction::replacePivs()
COMPARE_RESULT
HashPartitioningFunction::comparePartFuncToFunc
(const PartitioningFunction &other) const
{
COMPARE_RESULT c = PartitioningFunction::comparePartFuncToFunc(other);
if (c != SAME)
return INCOMPATIBLE;
return comparePartKeyToKey(other);
}
COMPARE_RESULT
HashPartitioningFunction::comparePartKeyToKey
(const PartitioningFunction &other) const
{
if (getPartitioningFunctionType() != other.getPartitioningFunctionType())
return INCOMPATIBLE;
const HashPartitioningFunction &oth =
(const HashPartitioningFunction &) other;
if (keyColumnList_.entries() != oth.keyColumnList_.entries())
return INCOMPATIBLE;
// compare the key columns and their order
for (CollIndex i = 0; i < keyColumnList_.entries(); i++)
{
if (keyColumnList_[i] != oth.keyColumnList_[i])
return INCOMPATIBLE;
if ( NOT (originalKeyColumnList_[i].getType() ==
oth.originalKeyColumnList_[i].getType() )
)
return INCOMPATIBLE;
}
return SAME;
}
// -----------------------------------------------------------------------
// HashPartitioningFunction::isAGroupingOf()
// -----------------------------------------------------------------------
// Right now we assume that the split function of a hash partitioning
// function is such that no two functions are a grouping of each other.
// The only exception is identity, of course. We use the base class'
// implementation.
//
// With more knowledge about the split function (e.g. by knowing it's
// a simple modulus), one could guarantee that a 4-way hash-partitioning
// scheme is actually a grouping of an 8-way scheme. This is not really
// necessary and therefore not done here.
// -----------------------------------------------------------------------
// HashPartitioningFunction::createPartitioningExpression()
// -----------------------------------------------------------------------
ItemExpr* HashPartitioningFunction::createPartitioningExpression()
{
if (getExpression()) // already constructed?
return getExpression(); // reuse it!
// ---------------------------------------------------------------------
// Construct an expression of the form:
//
// Modulus
// / \
// / \
// Hash <count of partitions>
// |
// (narrow(key1), narrow(key2), ..., narrow(keyN))
//
// Allocate ValueIds and type synthesize this partitioning function.
// ---------------------------------------------------------------------
ValueIdList typedKeyCols;
for (CollIndex i = 0; i < keyColumnList_.entries(); i++)
{
// cast the key column to the exact type of the original key column
const NAType &oType = originalKeyColumnList_[i].getType();
ItemExpr *c = getCastedItemExpre(keyColumnList_[i].getItemExpr(), oType,
CmpCommon::statementHeap());
c->synthTypeAndValueId();
typedKeyCols.insert(c->getValueId());
}
ItemExpr * partFunc =
new (CmpCommon::statementHeap())
Modulus(
buildHashingExpressionForExpr(
typedKeyCols.rebuildExprTree(ITM_ITEM_LIST)
),
// Hash(typedKeyCols.rebuildExprTree(ITM_ITEM_LIST)),
new (CmpCommon::statementHeap())
SystemLiteral(getCountOfPartitions()));
// once we support late binding, and/or changing the number of target
// partitions at run time, the number of partitions will have
// to become some sort of an input value $$$$
partFunc->synthTypeAndValueId();
storeExpression(partFunc);
return partFunc;
} // HashPartitioningFunction::createPartitioningExpression()
ItemExpr *
HashPartitioningFunction::buildHashingExpressionForExpr(ItemExpr* expr) const
{
return new (CmpCommon::statementHeap()) Hash(expr);
}
UInt32 HashPartitioningFunction::computeHashValue(char* data, UInt32 flags, Int32 len)
{
// Directly call the implementation function to compute the hash. NULL
// values and VARCHAR data types are not handled.
return FastHash(data, len);
}
ItemExpr * HashPartitioningFunction::getHashingExpression() const
{
ItemExpr* hashExpr = NULL;
ItemExpr* partExpr = getExpression();
if ( partExpr ) {
hashExpr = partExpr->child(0);
CMPASSERT(hashExpr AND
hashExpr->getOperatorType()== ITM_HASH);
}
return hashExpr;
}
// -----------------------------------------------------------------------
// HashPartitioningFunction::remapIt()
// -----------------------------------------------------------------------
void HashPartitioningFunction::remapIt
(const PartitioningFunction* opf,
ValueIdMap& map, NABoolean mapItUp)
{
PartitioningFunction::remapIt(opf, map,mapItUp);
// If we have arrived here, the original partitioning function (*opf)
// MUST be a HashPartitioningFunction().
CMPASSERT(opf->castToHashPartitioningFunction());
// Clear because rewrite insists on it being so.
keyColumnList_.clear();
if (mapItUp)
{
map.rewriteValueIdListUp(
keyColumnList_,
opf->castToHashPartitioningFunction()->keyColumnList_);
}
else
{
map.rewriteValueIdListDown(
opf->castToHashPartitioningFunction()->keyColumnList_,
keyColumnList_);
}
// do NOT map the originalKeyColumnList_, that's why it's called ORIGINAL
} // HashPartitioningFunction::remapIt()
PartitioningFunction * HashPartitioningFunction::scaleNumberOfPartitions(
Lng32 &suggestedNewNumberOfPartitions,
PartitionGroupingDistEnum partGroupDist)
{
if (suggestedNewNumberOfPartitions == 1)
return new(CmpCommon::statementHeap())
SinglePartitionPartitioningFunction();
// If an expression has been generated, then we want to discard it
// because it may no longer be correct.
storeExpression(NULL);
if ( suggestedNewNumberOfPartitions != getCountOfPartitions() ) {
NodeMap* newNodeMap =
new (heap_) NodeMap(heap_, suggestedNewNumberOfPartitions);
replaceNodeMap(newNodeMap);
}
numberOfHashPartitions_ = suggestedNewNumberOfPartitions;
return this;
}
// -----------------------------------------------------------------------
// HashPartitioningFunction::preCodeGen()
// Rewrite the partitioning keys of the partitioning function that
// are expressed using VEGReferences in terms of the available values.
// -----------------------------------------------------------------------
void HashPartitioningFunction::preCodeGen(const ValueIdSet& availableValues)
{
ValueIdSet noExternalInputs;
PartitioningFunction::preCodeGen(availableValues);
keyColumnList_.replaceVEGExpressions(availableValues, noExternalInputs,
FALSE, NULL, TRUE);
} // HashPartitioningFunction::preCodeGen()
// -----------------------------------------------------------------------
// Method for debugging.
// -----------------------------------------------------------------------
const NAString HashPartitioningFunction::getTextImp(const char* hashType) const
{
char nparts[20];
NAString result(hashType, CmpCommon::statementHeap());
result.append(" partitioned ");
sprintf(nparts,"%d",numberOfHashPartitions_);
result += nparts;
result += " ways on (";
//getPartitioningKey().unparse(result,DEFAULT_PHASE,EXPLAIN_FORMAT);
getKeyColumnList().unparse(result,DEFAULT_PHASE,EXPLAIN_FORMAT);
result += ")";
if (result.contains("randomNum"))
result = NAString("round robin partitioned");
return result;
}
const NAString HashPartitioningFunction::getText() const
{
return getTextImp("hash");
}
void HashPartitioningFunction::print(FILE* ofd, const char* indent,
const char* title) const
{
PartitioningFunction::print(ofd, indent, "HashPartitioningFunction");
} // HashPartitioningFunction::print()
// Return an expression casting an encoded skew value to oType.
static
ItemExpr*
getCastedSkewValueExpr(const EncodedValue& ev, const NAType& oType, CollHeap* heap)
{
double x = ev.getDblValue();
return new (heap) Cast(
new (heap) ConstValue(
new (heap) SQLDoublePrecision(heap, FALSE /* no SQL NULL*/),
(char*)&x, sizeof(x)
),
&oType
);
}
// ***********************************************************************
// TableHashPartitioningFunction
// - The externalized Hash Partitioning functions.
// ***********************************************************************
// -----------------------------------------------------------------------
// TableHashPartitioningFunction Destructor
// -----------------------------------------------------------------------
TableHashPartitioningFunction::~TableHashPartitioningFunction() {}
void TableHashPartitioningFunction::setupForStatement()
{
if(setupForStatement_)
return;
PartitioningFunction::setupForStatement();
setupForStatement_ = TRUE;
resetAfterStatement_ = FALSE;
}
void TableHashPartitioningFunction::resetAfterStatement()
{
if(resetAfterStatement_)
return;
PartitioningFunction::resetAfterStatement();
keyColumnList_.clear();
originalKeyColumnList_.clear();
numberOfPartitions_ = numberOfOrigHashPartitions_;
setupForStatement_ = FALSE;
resetAfterStatement_ = TRUE;
}
// -----------------------------------------------------------------------
// TableHashPartitioningFunction Safe down cast.
// -----------------------------------------------------------------------
const TableHashPartitioningFunction*
TableHashPartitioningFunction::castToTableHashPartitioningFunction() const
{
return this;
}
Lng32
TableHashPartitioningFunction::getCountOfPartitions() const
{
return numberOfPartitions_;
}
PartitioningRequirement*
TableHashPartitioningFunction::makePartitioningRequirement()
{
// Redefine PartitioningFunction::makePartitioningRequirement()
CMPABORT;
return NULL;
}
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
void
TableHashPartitioningFunction::createPartitionSelectionExprInputs()
{
if (partitionSelectionExprInputs().entries() != 0)
return;
CollHeap *heap = CmpCommon::statementHeap();
// Use a host var to provide access to numParts, this will be
// mapped to a specific ATP:ATPIndex:offset in PartitionAccess::codeGen()
//
char hvFabricatedName[50];
sprintf(hvFabricatedName, "_sys_hostVarNumParts_%p", this);
ItemExpr *numParts = new (heap)
HostVar(hvFabricatedName,
// int not null
new (heap) SQLInt(CmpCommon::statementHeap(), FALSE, FALSE),
// is system-supplied
TRUE);
numParts->synthTypeAndValueId();
// Use a host var to provide access to partNum, this will be
// mapped to a specific ATP:ATPIndex:offset in PartitionAccess::codeGen()
//
sprintf(hvFabricatedName, "_sys_hostVarPartNo_%p", this);
ItemExpr *partNum = new (heap)
HostVar(hvFabricatedName,
// int not null
new (heap) SQLInt(CmpCommon::statementHeap(), FALSE, FALSE),
// is system-supplied
TRUE);
partNum->synthTypeAndValueId();
// Record these hostvars as the inputs to the partitionSelectionExpr.
//
partitionSelectionExprInputs().insert(partNum->getValueId());
partitionSelectionExprInputs().insert(numParts->getValueId());
}
// -----------------------------------------------------------------------
// TableHashPartitioningFunction::normalizePartitioningKeys()
// Rewrite the partitioning keys of the partitioning function in
// terms of the VEGReference for the VEG to which the partitioning
// key column belongs.
// -----------------------------------------------------------------------
void
TableHashPartitioningFunction::normalizePartitioningKeys(NormWA& normWARef)
{
PartitioningFunction::normalizePartitioningKeys(normWARef);
keyColumnList_.normalizeNode(normWARef);
// don't normalize original key col list, avoid VEGies which could
// cause data type changes.
}
// -----------------------------------------------------------------------
// TableHashPartitioningFunction::createPartitioningKeyPredicates()
// Since hash dist partitioning can not create any Key Predicates,
// this method simply creates the partition input values (PIVs).
// -----------------------------------------------------------------------
void TableHashPartitioningFunction::createPartitioningKeyPredicates()
{
if (NOT partKeyPredsCreated())
createBetweenPartitioningKeyPredicates("_sys_HostVarLoHashPart",
"_sys_HostVarHiHashPart");
// Create the partition selection input values (needed by hash2).
createPartitionSelectionExprInputs();
} // TableHashPartitioningFunction::createPartitioningKeyPredicates()
// -----------------------------------------------------------------------
// For salted tables, since we store the hash partition value, we
// can generate a range predicate for the _SALT_ column
// -----------------------------------------------------------------------
void TableHashPartitioningFunction::createPartitioningKeyPredicatesForSaltedTable(
ValueId saltCol)
{
// For now we only allow this call after calling the regular
// createPartitioningKeyPredicates() method
CMPASSERT(partKeyPredsCreated());
// this allows us to specify NULL for the names here
createBetweenPartitioningKeyPredicates(NULL,
NULL,
saltCol.getItemExpr(),
isAHash2PartitioningFunction());
// and we don't need to call this
// createPartitionSelectionExprInputs();
} // TableHashPartitioningFunction::createPartitioningKeyPredicates()
// -----------------------------------------------------------------------
// TableHashPartitioningFunction::replacePivs()
// -----------------------------------------------------------------------
void TableHashPartitioningFunction::replacePivs(
const ValueIdList& newPivs,
const ValueIdSet& newPartKeyPreds)
{
// Overwrite the old pivs, part key preds, and part expr. with the new ones.
storePartitionInputValues(newPivs);
storePartitioningKeyPredicates(newPartKeyPreds);
} // TableHashPartitioningFunction::replacePivs()
PartitioningFunction*
TableHashPartitioningFunction::createPartitioningFunctionForIndexDesc
(IndexDesc *idesc) const
{
// Redefine PartitioningFunction::createPartitioningFunctionForIndexDesc()
CMPABORT;
return NULL;
}
// -----------------------------------------------------------------------
// TableHashPartitioningFunction::remapIt()
// -----------------------------------------------------------------------
void
TableHashPartitioningFunction::remapIt(const PartitioningFunction* opf,
ValueIdMap& map,
NABoolean mapItUp)
{
PartitioningFunction::remapIt(opf, map,mapItUp);
// If we have arrived here, the original partitioning function (*opf)
// MUST be a TableHashPartitioningFunction().
//
const TableHashPartitioningFunction *oth =
opf->castToTableHashPartitioningFunction();
CMPASSERT(oth);
// Clear because rewrite insists on it being so.
//
keyColumnList_.clear();
if (mapItUp)
map.rewriteValueIdListUp(keyColumnList_, oth->keyColumnList_);
else
map.rewriteValueIdListDown(oth->keyColumnList_, keyColumnList_);
// do NOT map the originalKeyColumnList_, that's why it's called ORIGINAL
} // TableHashPartitioningFunction::remapIt()
// -----------------------------------------------------------------------
// TableHashPartitioningFunction::preCodeGen()
// Rewrite the partitioning keys of the partitioning function that
// are expressed using VEGReferences in terms of the available values.
// -----------------------------------------------------------------------
void
TableHashPartitioningFunction::preCodeGen(const ValueIdSet& availableValues)
{
ValueIdSet noExternalInputs;
PartitioningFunction::preCodeGen(availableValues);
keyColumnList_.replaceVEGExpressions(availableValues, noExternalInputs,
FALSE, NULL, TRUE);
} // TableHashPartitioningFunction::preCodeGen()
// -----------------------------------------------------------------------
// TableHashPartitioningFunction::createPartitioningExpression()
// -----------------------------------------------------------------------
ItemExpr* TableHashPartitioningFunction::createPartitioningExpression()
{
return TableHashPartitioningFunction::createPartitioningExpressionImp(FALSE);
} // TableHashPartitioningFunction::createPartitioningExpression()
ItemExpr*
TableHashPartitioningFunction::createPartitioningExpressionImp(NABoolean doVarCharCast)
{
// already constructed with the same doVarCharCast argument?
if (getExpression() && doVarCharCast == doVarCharCast_ )
return getExpression(); // reuse it!
doVarCharCast_ = doVarCharCast;
ValueIdList typedKeyCols;
const ValueIdList &keyColumnList = getKeyColumnList();
const ValueIdList &originalKeyColumnList = getOriginalKeyColumnList();
CollHeap *heap = CmpCommon::statementHeap();
for (CollIndex i = 0; i < keyColumnList.entries(); i++)
{
const NAType &oType = originalKeyColumnList[i].getType();
ItemExpr *c;
ItemExpr *dataConversionErrorFlag = getConvErrorExpr();
if (dataConversionErrorFlag == 0)
{
dataConversionErrorFlag =
new (heap) HostVar("_sys_repartConvErrorFlg",
new (heap) SQLInt(heap, TRUE,FALSE),
TRUE);
storeConvErrorExpr(dataConversionErrorFlag);
}
// Note that we always generate a Narrow operator here, even
// if it's not necessary. Because keyColumnList[i] may be a
// VEGReference it may not be possible to determine its
// exact type at this time. We could eliminate some cases of
// Narrow in the preCodeGen phase but this is not done yet.
NAType* finalTypePtr = oType.newCopy(heap);
if ( oType.getTypeQualifier() == NA_CHARACTER_TYPE &&
doVarCharCast == TRUE ) {
ValueId vId = originalKeyColumnList[i];
ValueIdSet vidSet;
CharType *maxCharType = NULL;
switch (vId.getItemExpr()->getOperatorType()) {
case ITM_VEG_PREDICATE:
case ITM_VEG_REFERENCE:
{
vId.getItemExpr()->findAll(ITM_BASECOLUMN, vidSet, TRUE, TRUE);
for (ValueId x=vidSet.init(); vidSet.next(x); vidSet.advance(x))
{
const CharType &ctype = (CharType&)(x.getType());
if ( maxCharType == NULL OR
maxCharType->getNominalSize() < ctype.getNominalSize() OR
(maxCharType->getNominalSize() == ctype.getNominalSize() AND
maxCharType->getStrCharLimit() < ctype.getStrCharLimit())
)
maxCharType = &(CharType&)x.getType() ;
}
finalTypePtr = maxCharType->equivalentVarCharType(heap);
}
break;
case ITM_INDEXCOLUMN:
case ITM_BASECOLUMN:
// For skewbuster+OCR nested join, the partition key column
// is not a VEG or VEGref since the equality predicates have
// been pushed down to the inner table. Here we cast the
// column type to equivalent VarCHAR, regardless of the length
// of the other column in the join predicate. The optimization
// works fine because we just want to trim the trailing
// spaces and then hash the trimmed value.
finalTypePtr = ((CharType&)(vId.getType())).equivalentVarCharType(heap);
break;
default:
break;
}
}
// leave the statement heap to delete the SQLVarChar object, if created
c = new (heap) Narrow(keyColumnList[i].getItemExpr(),
dataConversionErrorFlag,
finalTypePtr,
ITM_NARROW,
FALSE /* reverseDataErrorConversionFlag */,
TRUE /* match child nullability */);
c->synthTypeAndValueId();
typedKeyCols.insert(c->getValueId());
}
ItemExpr * partFunc = buildPartitioningExpression(typedKeyCols);
#if 0
//===================================================================
// Test the classification expression
//===================================================================
ItemExpr* isSkewExpr = createClassificationExpressionForSkewedValues();
if ( isSkewExpr ) {
ValueIdList exprs;
exprs.insert(isSkewExpr->getValueId());
char resultBuf[2000];
ex_expr::exp_return_type evalReturnCode = exprs.evalAtCompileTime
(0, ExpTupleDesc::SQLARK_EXPLODED_FORMAT, resultBuf, 2000);
}
#endif
// once we support late binding, and/or changing the number of target
// partitions at run time, the number of partitions will have
// to become some sort of an input value $$$$
partFunc->synthTypeAndValueId();
storeExpression(partFunc);
#if 0 // test
buildHashListForSkewedValues();
#endif
return partFunc;
} // TableHashPartitioningFunction::createPartitioningExpressionImp()
// -----------------------------------------------------------------------
// TableHashPartitioningFunction::createPartitionSelectionExpr()
// Create the partition selection expression. 'Partition selection'
// means that an expression is used to determine the partition to
// access as opposed to using the File System to determine the range
// of partitions to access based on a set of partitioning key
// predicates. Partition selection is currently used for Hash Dist and
// Round Robin Partitioning. And the File System is used for Range
// Partitioning. If a partitioning selection expression is created,
// it is cached in the data member 'partitionSelectionExpr_' and the
// partition selection inputs are generated and stored in
// 'partitionSelectionExprInputs_'. This method is redefined for
// TableHashPartitioningFunction and RoundRobinPartitioningFunction.
// -----------------------------------------------------------------------
ItemExpr *
TableHashPartitioningFunction::
createPartitionSelectionExpr(const SearchKey *partSearchKey,
const ValueIdSet &availableValues)
{
// If it has already been created, return cached version.
//
if(partitionSelectionExpr())
return partitionSelectionExpr();
// For now, only support a partition selection expression when the
// partition search key is unique (identifies exactly one partition).
//
if(!partSearchKey->isUnique()) {
return NULL;
} else {
CollHeap *heap = CmpCommon::statementHeap();
const ValueIdList &keyColumns = partSearchKey->getKeyColumns();
const ValueIdList &keyValues = partSearchKey->getBeginKeyValues();
ValueIdList newValues;
createPartitionSelectionExprInputs();
const ValueIdList &partSelExprInputs = partitionSelectionExprInputs();
ItemExpr *numParts = partSelExprInputs[1].getItemExpr();
// Construct the list of key values. Make sure that they are cast
// to the exact type of the original key columns since the Hash
// function is sensitive to the type of its inputs. Must use the
// original key columns, since the keyColumnList may have been
// changed.
//
ValueIdList typedKeyCols;
const ValueIdList &originalKeyColumnList = getOriginalKeyColumnList();
for (CollIndex i = 0; i < keyColumns.entries(); i++)
{
// cast the key value to the exact type of the original key column
const NAType &oType = originalKeyColumnList[i].getType();
ItemExpr *c;
ItemExpr *dataConversionErrorFlag = getConvErrorExpr();
if (dataConversionErrorFlag == 0)
{
dataConversionErrorFlag =
new (heap) HostVar("_sys_repartConvErrorFlg",
new (heap) SQLInt(heap, TRUE,FALSE),
TRUE);
storeConvErrorExpr(dataConversionErrorFlag);
}
// Note that we always generate a Narrow operator here, even
// if it's not necessary. Because keyColumnList[i] may be a
// VEGReference it may not be possible to determine its
// exact type at this time. We could eliminate some cases of
// Narrow in the preCodeGen phase but this is not done yet.
//
NAType * finalTypePtr = oType.newCopy(heap);
c = new (heap) Narrow(keyValues[i].getItemExpr(),
dataConversionErrorFlag,
finalTypePtr);
c->synthTypeAndValueId();
typedKeyCols.insert(c->getValueId());
}
// Construct and cache the partitionSelectionExpr for Table Hash.
//
partitionSelectionExpr() =
buildPartitioningSelectionExpr(typedKeyCols, numParts);
// Bind the expression
//
partitionSelectionExpr()->synthTypeAndValueId();
// PreCodeGen the expression (This maybe should go in
// TableHashPartitioningFunction::preCodeGen(), but preCodeGen() is
// typically called before this expression is generated.
//
partitionSelectionExpr()->replaceVEGExpressions(availableValues,
availableValues);
return partitionSelectionExpr();
}
} // TableHashPartitioningFunction::createPartitionSelectionExpr()
// -----------------------------------------------------------------------
// Make a new partSearchKey that indicates that PA_PARTITION_GROUPING
// is being done. Note that a search key can not be generated which
// can group hashed partitions. For TableHashPartitioning, a flag in
// the search key is used to indicate that PA_PARTITION_GROUPING is
// being done and the begin/end key values of the search key are set
// to the partition input values of the partitioning function.
// -----------------------------------------------------------------------
SearchKey *
TableHashPartitioningFunction::createSearchKey(const IndexDesc *indexDesc,
ValueIdSet availInputs,
ValueIdSet additionalPreds) const
{
ValueIdSet preds(getPartitioningKeyPredicates());
ValueIdSet nonKeyColumnSet; // empty set
SearchKey *partSearchKey = NULL;
availInputs += getPartitionInputValues();
preds += additionalPreds;
if (indexDesc->getPrimaryTableDesc()->getNATable()->isHbaseTable())
{
// The HbaseAccess executor operator doesn't have a separate
// place to handle partitioning key predicates. Instead, we can
// only use them as regular key or executor predicates.
// For salted tables, we have a chance to read a range of salt
// values through a begin/end key.
partSearchKey = new (CmpCommon::statementHeap())
SearchKey(indexDesc->getIndexKey(),
indexDesc->getOrderOfKeyValues(),
availInputs,
TRUE,
preds,
nonKeyColumnSet,
indexDesc);
}
else
{
// Call this special constructor that constructs a search key for a
// TableHashPartitioningFunction.
//
partSearchKey = new (CmpCommon::statementHeap())
SearchKey(indexDesc->getPartitioningKey(),
indexDesc->getOrderOfPartitioningKeyValues(),
availInputs,
preds,
this,
nonKeyColumnSet,
indexDesc);
}
return partSearchKey;
} // TableHashPartitioningFunction::createSearchKey()
ItemExpr *
TableHashPartitioningFunction::buildHashingExpressionForExpr(ItemExpr* expr)
const
{
CollHeap *heap = CmpCommon::statementHeap();
return new (heap) HashDistPartHash(expr);
}
UInt32 TableHashPartitioningFunction::computeHashValue(char* data, UInt32 flags, Int32 len)
{
// Directly call the implementation function to compute the hash. NULL
// values and VARCHAR data types are not handled.
return ExHDPHash::hash(data, flags, len);
}
ItemExpr * TableHashPartitioningFunction::getHashingExpression() const
{
ItemExpr* hashExpr = NULL;
ItemExpr* partExpr = getExpression();
if ( partExpr ) {
hashExpr = partExpr->child(0);
CMPASSERT(hashExpr);
if ( hashExpr->getOperatorType() == ITM_PROGDISTRIB ) {
hashExpr = hashExpr->child(0);
CMPASSERT(hashExpr);
}
CMPASSERT ( hashExpr->getOperatorType() == ITM_HDPHASH);
}
return hashExpr;
}
// -----------------------------------------------------------------------
// HashDistPartitioningFunction Destructor
// -----------------------------------------------------------------------
HashDistPartitioningFunction::~HashDistPartitioningFunction() {}
// -----------------------------------------------------------------------
// HashDistPartitioningFunction Safe down cast.
// -----------------------------------------------------------------------
const HashDistPartitioningFunction*
HashDistPartitioningFunction::castToHashDistPartitioningFunction() const
{
return this;
}
PartitioningRequirement*
HashDistPartitioningFunction::makePartitioningRequirement()
{
return new (CmpCommon::statementHeap())
RequireHashDist(this);
}
PartitioningFunction*
HashDistPartitioningFunction::copy() const
{
return new (CmpCommon::statementHeap()) HashDistPartitioningFunction(*this);
}
// -----------------------------------------------------------------------
// Method for debugging.
// -----------------------------------------------------------------------
const NAString HashDistPartitioningFunction::getText() const
{
NAString result("hash1 partitioned ");
char nparts[40];
sprintf(nparts,"%d (%d) ways on (", numberOfPartitions_,
numberOfOrigHashPartitions_);
result += nparts;
getKeyColumnList().unparse(result,DEFAULT_PHASE,EXPLAIN_FORMAT);
result += ")";
return result;
}
void HashDistPartitioningFunction::print(FILE* ofd, const char* indent,
const char* title) const
{
PartitioningFunction::print(ofd, indent, "HashDistPartitioningFunction");
} // TableHashPartitioningFunction::print()
// -----------------------------------------------------------------------
// HashDistPartitioningFunction::createPartitioningFunctionForIndexDesc()
// -----------------------------------------------------------------------
PartitioningFunction*
HashDistPartitioningFunction::
createPartitioningFunctionForIndexDesc(IndexDesc *idesc) const
{
const NAFileSet * fileSet = idesc->getNAFileSet();
const NAColumnArray & allColumns = fileSet->getAllColumns();
const NAColumnArray & partKeyColumns = fileSet->getPartitioningKeyColumns();
CollIndex ixColNumber;
ValueId keyValueId;
ValueIdSet partitioningKey;
ValueIdList partitioningKeyList;
for (CollIndex i = 0; i < partKeyColumns.entries(); i++)
{
// which column of the index is this (usually this will be == i)
ixColNumber = allColumns.index(partKeyColumns[i]);
// insert the value id of the index column into the partitioning
// key column value id set
keyValueId = idesc->getIndexColumns()[ixColNumber];
partitioningKey += keyValueId;
partitioningKeyList.insertAt(i,keyValueId);
} // end loop over partitioning key columns
// -----------------------------------------------------------------
// Allocate a new HashPartitioningFunction.
// -----------------------------------------------------------------
HashDistPartitioningFunction *partFunc = new(idesc->wHeap())
HashDistPartitioningFunction (partitioningKey,
partitioningKeyList,
getCountOfPartitions(),
getNodeMap()->copy(idesc->wHeap()));
// -----------------------------------------------------------------
// Construct the partitioning key predicates.
// -----------------------------------------------------------------
partFunc->createPartitioningKeyPredicates();
return partFunc;
} // HashDistPartitioningFunction::createPartitioningFunctionForIndexDesc()
// -----------------------------------------------------------------------
// HashDistPartitioningFunction::comparePartFuncToFunc(): Compare this
// partitioning function to another hash dist function. To be 'SAME'
// must have the same number and order of partitioning key columns and
// have the same number of partitions (scaled and original).
// -----------------------------------------------------------------------
COMPARE_RESULT
HashDistPartitioningFunction::
comparePartFuncToFunc(const PartitioningFunction &other) const
{
COMPARE_RESULT c = PartitioningFunction::comparePartFuncToFunc(other);
if (c != SAME)
return INCOMPATIBLE;
const HashDistPartitioningFunction *oth =
other.castToHashDistPartitioningFunction();
// Since they compared 'SAME', oth should always exist, so this
// test is redundant.
//
if(!oth)
return INCOMPATIBLE;
// They must be based on the same physical partitioning.
//
if (getCountOfOrigHashPartitions() != oth->getCountOfOrigHashPartitions())
return INCOMPATIBLE;
// Make sure that the keys are in the same order.
//
if (keyColumnList_.entries() != oth->keyColumnList_.entries())
return INCOMPATIBLE;
for (CollIndex i = 0; i < keyColumnList_.entries(); i++)
{
if (keyColumnList_[i] != oth->keyColumnList_[i])
return INCOMPATIBLE;
if ( NOT (originalKeyColumnList_[i].getType() ==
oth->originalKeyColumnList_[i].getType() )
)
return INCOMPATIBLE;
}
return SAME;
} // HashDistPartitioningFunction::comparePartFuncToFunc()
PartitioningFunction *
HashDistPartitioningFunction::
scaleNumberOfPartitions(Lng32 &suggestedNewNumberOfPartitions,
PartitionGroupingDistEnum partGroupDist)
{
if (suggestedNewNumberOfPartitions == 1)
return new(CmpCommon::statementHeap())
SinglePartitionPartitioningFunction();
// If an expression has been generated, then we want to discard it
// because it may no longer be correct.
storeExpression(NULL);
// Allow arbitrary scaling down of HashDistPartitioningFunction.
// (The runtime will handle the mapping of physical partitions
// to logical partitions.)
// Scaling up (to a greater number of partitions is not allowed for
// HashDistPartitioningFunction.
//
suggestedNewNumberOfPartitions =
(suggestedNewNumberOfPartitions > numberOfPartitions_)
? numberOfPartitions_
: suggestedNewNumberOfPartitions;
numberOfPartitions_ = suggestedNewNumberOfPartitions;
return this;
} // HashDistPartitioningFunction::scaleNumberOfPartitions()
// -----------------------------------------------------------------------
// HashDistPartitioningFunction::isAGroupingOf()
// -----------------------------------------------------------------------
NABoolean
HashDistPartitioningFunction::isAGroupingOf(const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = 1;
const HashDistPartitioningFunction *oth =
other.castToHashDistPartitioningFunction();
// If other is not a HashDistPartitioningFunction, then it cannot
// be a grouping of...
if (oth == NULL)
return FALSE;
if (keyColumnList_.entries() != oth->keyColumnList_.entries())
return FALSE;
// compare the key columns and their order
for (CollIndex i = 0; i < keyColumnList_.entries(); i++)
{
if (keyColumnList_[i] != oth->keyColumnList_[i])
return FALSE;
if ( NOT (originalKeyColumnList_[i].getType() ==
oth->originalKeyColumnList_[i].getType() )
)
return FALSE;
}
// If this function has more partitions than other,
// then it cannot be a grouping of.
// Eg. this.numParts: 10 this.origNumParts: 20
// oth.numParts: 5 oth.origNumParts: 20
//
// If the two functions are not based on the same physical function,
// then it cannot be a grouping of.
// Eg. this.numParts: 10 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 30
//
if((getCountOfPartitions() > oth->getCountOfPartitions()) ||
(getCountOfOrigHashPartitions() != oth->getCountOfOrigHashPartitions())) {
return FALSE;
}
// Here the following is known to be TRUE:
//
// (getCountOfPartitions() <= oth->getCountOfPartitions()
//
// AND
//
// (getCountOfOrigHashPartitions() == oth->getCountOfOrigHashPartitions())
//
// Eg. this.numParts: 10 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// this.numParts: 5 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
// If other has not been scaled (allow arbitrary scaling of one function):
// Eg. this.numParts: 7 this.origNumParts: 20
// oth.numParts: 20 oth.origNumParts: 20
// OR
// If they have both been scaled to the same number of partitions:
// then it is a grouping of.
// Eg. this.numParts: 7 this.origNumParts: 20
// oth.numParts: 7 oth.origNumParts: 20
//
if((oth->getCountOfPartitions() == oth->getCountOfOrigHashPartitions()) ||
(getCountOfPartitions() == oth->getCountOfPartitions())) {
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup =
((oth->getCountOfPartitions() + getCountOfPartitions() - 1)
/ getCountOfPartitions());
return TRUE;
}
// WARNING..... I am not sure if the current code can ever produce
// a situation that would bring control to here. Also, I am not
// sure if the semantics of GROUPING implemented below are correct
// for these situations.
//
// Here the following is known to be TRUE:
//
// both functions have been scaled. (I DON'T THINK THIS CAN HAPPEN)
//
// AND
//
// They are scaled to different sizes.
//
// Eg. this.numParts: 5 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// this.numParts: 7 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// Under these conditions, three things must be true for it to be a
// grouping of:
//
// - the scaled number of partitions must evenly divide the scaled
// number of partitions of other
// Eg. this.numParts: 5 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// - the other scaling must be a multiple of or evenly divide the
// original number of partitions
// Eg. this.numParts: 5 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// this.numParts: 5 this.origNumParts: 20
// oth.numParts: 40 oth.origNumParts: 20
//
// - this scaling must also be a multiple of or evenly divide the
// original number of partitions
// Eg. this.numParts: 5 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// this.numParts: 40 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// If the scaled number of partitions evenly divides the scaled
// number of partitions of other...
//
if((oth->getCountOfPartitions() % getCountOfPartitions()) != 0) {
return FALSE;
}
// AND the other scaling is a multiple of or evenly divides the
// original number of partitions...
//
if(oth->getCountOfOrigHashPartitions() >= oth->getCountOfPartitions()) {
if(oth->getCountOfOrigHashPartitions() % oth->getCountOfPartitions()) {
return FALSE;
}
} else {
if(oth->getCountOfPartitions() % oth->getCountOfOrigHashPartitions()) {
return FALSE;
}
}
// AND this scaling is a multiple of or evenly divides the original
// number of partitions ...
//
if(getCountOfOrigHashPartitions() >= getCountOfPartitions()) {
if(getCountOfOrigHashPartitions() % getCountOfPartitions()) {
return FALSE;
}
} else {
if(getCountOfPartitions() % getCountOfOrigHashPartitions()) {
return FALSE;
}
}
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup =
((oth->getCountOfPartitions() + getCountOfPartitions() - 1)
/ getCountOfPartitions());
// THEN it is a grouping of...
//
return TRUE;
} // HashDistPartitioningFunction::isAGroupingOf()
// -----------------------------------------------------------------------
// - Create expressions for the constant values for the original and
// scale number of partitions. Create the hash1 (hash distrib)
// function, PAGroup, and return the result.
// -----------------------------------------------------------------------
ItemExpr *
HashDistPartitioningFunction::buildPartitioningExpression(
const ValueIdList &keyCols) const
{
CollHeap *heap = CmpCommon::statementHeap();
NAType *numPartsType = new (heap) SQLInt(heap, FALSE,FALSE);
char buffer[20];
// Create a ConstValue expression containing the original number of hash
// partitions.
Lng32 numParts = getCountOfOrigHashPartitions();
sprintf(buffer, "%d", numParts);
NAString numPartsStr("origNumParts");
ItemExpr *origNumParts =
new (heap) ConstValue(numPartsType, &numParts, sizeof(Lng32), &numPartsStr);
// Create a ConstValue expression containing the scaled number of hash
// partitions.
numParts = getCountOfPartitions();
sprintf(buffer, "%d", numParts);
numPartsStr = "scaledNumParts";
ItemExpr *scaledNumParts = new (heap)
ConstValue(numPartsType, &numParts, sizeof(Lng32), &numPartsStr);
// Create the hash distribution function.
// partitions.
ItemExpr *partFunc =
new (heap)
ProgDistrib(new (heap) HashDistPartHash(keyCols.rebuildExprTree(ITM_ITEM_LIST)), origNumParts);
// Add a PAGroup expression and return the partitioning function
partFunc = new (heap) PAGroup(partFunc, scaledNumParts, origNumParts);
return partFunc;
}
// -----------------------------------------------------------------------
// HashDistPartitioningFunction::buildPartitioningSelectionExpr
// - build the expression used during partition selection
// -----------------------------------------------------------------------
ItemExpr *
HashDistPartitioningFunction::buildPartitioningSelectionExpr(
const ValueIdList &keyCols,
ItemExpr *numParts) const
{
CollHeap *heap = CmpCommon::statementHeap();
ItemExpr *partFunc =
new (heap)
ProgDistrib(new (heap)
HashDistPartHash(keyCols.rebuildExprTree(ITM_ITEM_LIST)),
numParts);
return partFunc;
}
// -----------------------------------------------------------------------
// Hash2PartitioningFunction Destructor
// -----------------------------------------------------------------------
Hash2PartitioningFunction::~Hash2PartitioningFunction() {}
// -----------------------------------------------------------------------
// Hash2PartitioningFunction Safe down cast.
// -----------------------------------------------------------------------
const Hash2PartitioningFunction*
Hash2PartitioningFunction::castToHash2PartitioningFunction() const
{
return this;
}
PartitioningRequirement*
Hash2PartitioningFunction::makePartitioningRequirement()
{
return new (CmpCommon::statementHeap())
RequireHash2(this);
}
PartitioningFunction*
Hash2PartitioningFunction::copy() const
{
return new (CmpCommon::statementHeap()) Hash2PartitioningFunction(*this);
}
// -----------------------------------------------------------------------
// Method for debugging.
// -----------------------------------------------------------------------
const NAString Hash2PartitioningFunction::getText() const
{
NAString result("hash2 partitioned ");
char nparts[32];
sprintf(nparts,"%d ways on (", numberOfPartitions_);
result += nparts;
getKeyColumnList().unparse(result,DEFAULT_PHASE,EXPLAIN_FORMAT);
result += ")";
return result;
}
void Hash2PartitioningFunction::print(FILE* ofd, const char* indent,
const char* title) const
{
PartitioningFunction::print(ofd, indent, "Hash2PartitioningFunction");
} // TableHashPartitioningFunction::print()
// -----------------------------------------------------------------------
// Hash2PartitioningFunction::createPartitioningFunctionForIndexDesc()
// -----------------------------------------------------------------------
PartitioningFunction*
Hash2PartitioningFunction::
createPartitioningFunctionForIndexDesc(IndexDesc *idesc) const
{
const NAFileSet * fileSet = idesc->getNAFileSet();
const NAColumnArray & allColumns = fileSet->getAllColumns();
const NAColumnArray & partKeyColumns = fileSet->getPartitioningKeyColumns();
CollIndex ixColNumber;
ValueId keyValueId;
ValueIdSet partitioningKey;
ValueIdList partitioningKeyList;
for (CollIndex i = 0; i < partKeyColumns.entries(); i++)
{
// which column of the index is this (usually this will be == i)
ixColNumber = allColumns.index(partKeyColumns[i]);
// insert the value id of the index column into the partitioning
// key column value id set
keyValueId = idesc->getIndexColumns()[ixColNumber];
partitioningKey += keyValueId;
partitioningKeyList.insertAt(i,keyValueId);
} // end loop over partitioning key columns
// -----------------------------------------------------------------
// Allocate a new HashPartitioningFunction.
// -----------------------------------------------------------------
Hash2PartitioningFunction *partFunc = new(idesc->wHeap())
Hash2PartitioningFunction (partitioningKey,
partitioningKeyList,
getCountOfPartitions(),
getNodeMap()->copy(idesc->wHeap()));
partFunc->setRestrictedBeginPartNumber(getRestrictedBeginPartNumber());
partFunc->setRestrictedEndPartNumber(getRestrictedEndPartNumber());
// -----------------------------------------------------------------
// Construct the partitioning key predicates.
// -----------------------------------------------------------------
partFunc->createPartitioningKeyPredicates();
return partFunc;
} // Hash2PartitioningFunction::createPartitioningFunctionForIndexDesc()
// -----------------------------------------------------------------------
// Hash2PartitioningFunction::createPartSelectionExprFromSearchKey
// is called from PartitionAccess::preCodeGen() to create expressions
// for determining the beginning and ending partition numbers that
// partition access must deal with. Other partitioning schemes
// pass partition numbers, but Hash2 passes hash boundaries. Hash2
// passes hash boundaries instead to allow a single ESP to work with
// groupings of tables with different numbers of physical partitions.
// -----------------------------------------------------------------------
void Hash2PartitioningFunction::createPartSelectionExprFromSearchKey(
const ValueId beginPartSelId,
const ValueId endPartSelId,
ValueIdList &partSelectionValIds) const
{
ItemExpr *beginPartSelExpr = new (CmpCommon::statementHeap())
Hash2Distrib(beginPartSelId.getItemExpr(),
partitionSelectionExprInputs()[1].getItemExpr());
beginPartSelExpr->synthTypeAndValueId();
ItemExpr *endPartSelExpr = new (CmpCommon::statementHeap())
Hash2Distrib(endPartSelId.getItemExpr(),
partitionSelectionExprInputs()[1].getItemExpr());
endPartSelExpr->synthTypeAndValueId();
partSelectionValIds.insert(beginPartSelExpr->getValueId());
partSelectionValIds.insert(endPartSelExpr->getValueId());
}
// -----------------------------------------------------------------------
// Hash2PartitioningFunction::comparePartFuncToFunc(): Compare this
// partitioning function to another hash2 function. To be 'SAME'
// must have the same number and order of partitioning key columns and
// the same number of scaled partitions.
// -----------------------------------------------------------------------
COMPARE_RESULT
Hash2PartitioningFunction::
comparePartFuncToFunc(const PartitioningFunction &other) const
{
COMPARE_RESULT c = PartitioningFunction::comparePartFuncToFunc(other);
if (c != SAME)
return INCOMPATIBLE;
const Hash2PartitioningFunction *oth =
other.castToHash2PartitioningFunction();
// Since they compared 'SAME', oth should always exist, so this
// test is redundant.
//
if(!oth)
return INCOMPATIBLE;
// If OLD_HASH2_GROUPING is turned on, then the original number of
// partitions must be equal. The OLD_HASH2_GROUPING CQD is for test
// purposes and will most likely be removed after the new hash2
// grouping has been tested extensively.
if (CmpCommon::getDefault(OLD_HASH2_GROUPING) == DF_ON &&
getCountOfOrigHashPartitions() != oth->getCountOfOrigHashPartitions())
return INCOMPATIBLE;
// The normal behavior of hash2 grouping is that the scaled
// number of partitions must be equal.
if (getCountOfPartitions() != oth->getCountOfPartitions())
return INCOMPATIBLE;
// Make sure that the keys are in the same order.
//
if (keyColumnList_.entries() != oth->keyColumnList_.entries())
return INCOMPATIBLE;
for (CollIndex i = 0; i < keyColumnList_.entries(); i++)
{
if (keyColumnList_[i] != oth->keyColumnList_[i])
return INCOMPATIBLE;
if ( NOT (originalKeyColumnList_[i].getType() ==
oth->originalKeyColumnList_[i].getType() )
)
return INCOMPATIBLE;
}
return SAME;
} // Hash2PartitioningFunction::comparePartFuncToFunc()
// -----------------------------------------------------------------------
// Hash2PartitioningFunction::scaleNumberOfPartitions()
// -----------------------------------------------------------------------
PartitioningFunction *
Hash2PartitioningFunction::
scaleNumberOfPartitions(Lng32 &suggestedNewNumberOfPartitions,
PartitionGroupingDistEnum partGroupDist)
{
// Do not allow suggestedNewNumberOfPartitions greater than
// the number of partitions.
suggestedNewNumberOfPartitions =
(suggestedNewNumberOfPartitions > numberOfPartitions_)
? numberOfPartitions_
: suggestedNewNumberOfPartitions;
// Only allow a suggestedNewNumberOfPartitions that evenly
// divides into the number of partitions.
CMPASSERT(suggestedNewNumberOfPartitions > 0);
while (numberOfPartitions_ % suggestedNewNumberOfPartitions != 0)
suggestedNewNumberOfPartitions--;
if (suggestedNewNumberOfPartitions == 1)
return new(CmpCommon::statementHeap())
SinglePartitionPartitioningFunction();
// If an expression has been generated, then we want to discard it
// because it may no longer be correct.
storeExpression(NULL);
if ( suggestedNewNumberOfPartitions != getCountOfPartitions() ) {
NodeMap* newNodeMap =
new (heap_) NodeMap(heap_, suggestedNewNumberOfPartitions);
replaceNodeMap(newNodeMap);
}
numberOfPartitions_ = suggestedNewNumberOfPartitions;
return this;
} // Hash2PartitioningFunction::scaleNumberOfPartitions()
// -----------------------------------------------------------------------
// Hash2PartitioningFunction::isAGroupingOf()
// -----------------------------------------------------------------------
NABoolean
Hash2PartitioningFunction::isAGroupingOf(const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = 1;
const Hash2PartitioningFunction *oth =
other.castToHash2PartitioningFunction();
// If other is not a Hash2PartitioningFunction, then it cannot
// be a grouping of...
if (oth == NULL)
return FALSE;
if (getCountOfPartitions() > oth->getCountOfPartitions() ||
oth->getCountOfPartitions() % getCountOfPartitions() != 0)
return FALSE;
// For testing purposes, it was requested that the ability to
// provide grouping similar to hash1 be provided. If the
// OLD_HASH2_GROUPING CQD is true, then only allow grouping
// when the number of original partitions is the same.
if (CmpCommon::getDefault(OLD_HASH2_GROUPING) == DF_ON &&
getCountOfPartitions() != oth->getCountOfOrigHashPartitions())
return FALSE;
if (keyColumnList_.entries() != oth->keyColumnList_.entries())
return FALSE;
// compare the key columns and their order
for (CollIndex i = 0; i < keyColumnList_.entries(); i++)
{
if (keyColumnList_[i] != oth->keyColumnList_[i])
return FALSE;
if ( NOT (originalKeyColumnList_[i].getType() ==
oth->originalKeyColumnList_[i].getType() )
)
return FALSE;
}
// This is a grouping of. Set the maxPartsPerGroup and return TRUE.
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = oth->getCountOfPartitions() / getCountOfPartitions();
return TRUE;
} // Hash2PartitioningFunction::isAGroupingOf()
// -----------------------------------------------------------------------
// Hash2PartitioningFunction::buildPartitioningExpression()
// - Create an expression for the Hash2 partitioning function.
// -----------------------------------------------------------------------
ItemExpr *
Hash2PartitioningFunction::buildPartitioningExpression(const ValueIdList &keyCols) const
{
CollHeap *heap = CmpCommon::statementHeap();
NAType *numPartsType = new (heap) SQLInt(heap, FALSE,FALSE);
// Build a ConstValue expression of the scaled number of partitions.
Lng32 numParts = getCountOfPartitions();
char buffer[20];
sprintf(buffer, "%d", numParts);
NAString numPartsStr("scaledNumParts");
ItemExpr *scaledNumParts = new (heap)
ConstValue(numPartsType, &numParts, sizeof(Lng32), &numPartsStr);
// Create the hash2 partitioning function expression and return it.
ItemExpr *partFunc = new (heap) Hash2Distrib(
new (heap) HashDistPartHash(keyCols.rebuildExprTree(ITM_ITEM_LIST)), scaledNumParts
);
return partFunc;
}
// -----------------------------------------------------------------------
// HashDistPartitioningFunction::buildPartitioningSelectionExpr
// - build the expression used during partition selection
// -----------------------------------------------------------------------
ItemExpr *
Hash2PartitioningFunction::buildPartitioningSelectionExpr(
const ValueIdList &keyCols,
ItemExpr *numParts) const
{
CollHeap *heap = CmpCommon::statementHeap();
ItemExpr *partFunc = new (heap)
Hash2Distrib(new (heap)
HashDistPartHash(keyCols.rebuildExprTree(ITM_ITEM_LIST)),
numParts);
return partFunc;
}
// A refactored method whose implementaion is originally contained inside
// method HashPartitioningFunction::createPartitioningExpression().
ItemExpr*
PartitioningFunction::getCastedItemExpre(ItemExpr* iv, const NAType& oType, CollHeap* heap)
{
ItemExpr *dataConversionErrorFlag = getConvErrorExpr();
if (dataConversionErrorFlag == 0)
{
dataConversionErrorFlag =
new (CmpCommon::statementHeap()) HostVar(
"_sys_repartConvErrorFlg",
new (CmpCommon::statementHeap()) SQLInt(CmpCommon::statementHeap(), TRUE,FALSE),
TRUE);
storeConvErrorExpr(dataConversionErrorFlag);
}
// Note that we always generate a Narrow operator here, even
// if it's not necessary. Because keyColumnList_[i] may be a
// VEGReference it may not be possible to determine its
// exact type at this time. We could eliminate some cases of
// Narrow in the preCodeGen phase but this is not done yet.
// Begin_Fix 10-040114-2431
// 02/18/2004
// changed statement below for above mentioned solution
ItemExpr* c = new (CmpCommon::statementHeap()) Narrow(
iv,
dataConversionErrorFlag,
&oType,
ITM_NARROW,
FALSE,
TRUE); // TRUE => make nullability of Narrow same as child's
// End_Fix 10-040114-2431
((Narrow*)c)->setMatchChildType(TRUE);
return c;
}
// ***********************************************************************
// SkewedDataPartitioningFunction
// ***********************************************************************
SkewedDataPartitioningFunction::SkewedDataPartitioningFunction(
PartitioningFunction* partFuncForUnskewed,
const skewProperty& sk,
NAMemory* heap
)
: PartitioningFunction(SKEWEDDATA_PARTITIONING_FUNCTION, NULL, heap),
partialPartFunc_(partFuncForUnskewed), skewProperty_(sk),
skewHashList_(NULL)
{
CMPASSERT(NOT sk.isAnySkew());
CMPASSERT(partFuncForUnskewed->canHandleSkew());
ValueIdList pivl;
ValueIdSet pivs;
// Create the partition input variable and partitioning key for this
// function. The main purpose of this is to produce a partitioning key.
// The skew buster partitioning function doesn't have a real
// partitioning key. So, one cannot really decide from a row in
// which partition it is (at least not for a skew element row).
// Because we need to have some partitioning key, we basically say
// "you have to know the partition number in order to compute the
// partition number". This is another way of saying that the
// partition number cannot be computed from the row. We achieve this
// by making the PIV the partitioning key.
// If needed, we can also make the PIV the partitioning expression. We
// cannot make a partitioning key expression with this method, however.
createPIV(pivl);
pivs = pivl;
setPartKey(pivs);
}
void SkewedDataPartitioningFunction::createPIV(ValueIdList &partInputValues)
{
// Create a single partition input value
//
ItemExpr *dummyPIV = new (CmpCommon::statementHeap())
HostVar("_sys_dummySkewBusterPartNo",
new (CmpCommon::statementHeap()) SQLInt(CmpCommon::statementHeap(), FALSE, FALSE),
TRUE);
dummyPIV->synthTypeAndValueId();
// the partition input value is one integer, which is also used as
// the partitioning key
partInputValues.insert(dummyPIV->getValueId());
// Store the partition input values.
//
storePartitionInputValues(partInputValues);
}
void SkewedDataPartitioningFunction::replacePivs(
const ValueIdList& newPivs,
const ValueIdSet& newPartKeyPreds)
{
partialPartFunc_ -> replacePivs(newPivs, newPartKeyPreds);
}
SkewedDataPartitioningFunction::SkewedDataPartitioningFunction(
const SkewedDataPartitioningFunction& other, NAMemory* heap)
: PartitioningFunction(other, heap),
skewHashList_(NULL), // recompute this
partialPartFunc_(other.partialPartFunc_), // may need a deep copy
skewProperty_(other.skewProperty_) // share the skew property
{
}
Lng32 SkewedDataPartitioningFunction::getCountOfPartitions() const
{
return partialPartFunc_->getCountOfPartitions();
}
void SkewedDataPartitioningFunction::createPartitioningKeyPredicates()
{
// do nothing, there aren't any partitioning key preds for a skew
// partition
storePartitioningKeyPredicates(ValueIdSet());
partialPartFunc_ -> createPartitioningKeyPredicates();
}
PartitioningRequirement*
SkewedDataPartitioningFunction::makePartitioningRequirement()
{
return new (CmpCommon::statementHeap()) RequireSkewed(this);
}
PartitioningFunction*
SkewedDataPartitioningFunction::copy() const
{
SkewedDataPartitioningFunction *result;
result = new (CmpCommon::statementHeap())
SkewedDataPartitioningFunction(*this);
result->partialPartFunc_ = partialPartFunc_->copy();
return result;
}
// -----------------------------------------------------------------------
// Method for debugging.
// -----------------------------------------------------------------------
const NAString SkewedDataPartitioningFunction::getText() const
{
NAString result = partialPartFunc_->getText();
// SKEW_EXPLAIN = off --> do no display any details on skew-processing
// method or skew value list.
if ( CmpCommon::getDefault(SKEW_EXPLAIN) == DF_OFF )
return result;
NAString pat, suffix;
switch ( skewProperty_.getIndicator() ) {
case skewProperty::UNIFORM_DISTRIBUTE:
suffix = "-ud";
break;
case skewProperty::BROADCAST:
suffix = "-br";
break;
default:
break;
}
switch (partialPartFunc_ -> getPartitioningFunctionType()) {
case HASH_PARTITIONING_FUNCTION:
pat = "hash";
break;
case HASH_DIST_PARTITIONING_FUNCTION:
pat = "hash1";
break;
case HASH2_PARTITIONING_FUNCTION:
pat = "hash2";
break;
default:
break;
}
size_t loc = result.index(pat);
if ( loc != NA_NPOS ) {
result.insert(loc+pat.length(), suffix);
}
result += skewProperty_.getText();
return result;
}
void SkewedDataPartitioningFunction::print(FILE* ofd, const char* indent,
const char* title) const
{
PartitioningFunction::print(ofd, indent, "SkewedDataPartitioningFunction");
}
// -----------------------------------------------------------------------
// SkewedDataPartitioningFunction::comparePartFuncToFunc(): Compare this
// partitioning function to another hash2 function. To be 'SAME'
// must have the same number and order of partitioning key columns and
// the same number of scaled partitions.
// -----------------------------------------------------------------------
COMPARE_RESULT
SkewedDataPartitioningFunction::
comparePartFuncToFunc(const PartitioningFunction &other) const
{
// the other has to be a SkewedDataPartitioningFunction
const SkewedDataPartitioningFunction *oth =
other.castToSkewedDataPartitioningFunction();
if(!oth) return INCOMPATIBLE;
// compare the two partial partfuncs.
COMPARE_RESULT c = partialPartFunc_->comparePartFuncToFunc(
*(oth->getPartialPartitioningFunction())
);
if (c != SAME) return INCOMPATIBLE;
// compare the skew property last.
if ( NOT (skewProperty_ == oth->skewProperty_) )
return INCOMPATIBLE;
return SAME;
} // SkewedDataPartitioningFunction::comparePartFuncToFunc()
// -----------------------------------------------------------------------
// SkewedDataPartitioningFunction::scaleNumberOfPartitions()
// -----------------------------------------------------------------------
//::scaleNUmberOfPartitions() are called in following locations
//
// OptPhysRelExpr.cpp
// 3536 NJ::genLeftChild()
// 4892 NJ:createContextForChild() (rightPartFunc->)
// 14166 and 14262 synthDP2PhysicalProperty()
//
// GP.cpp
// 955 GroupAttributes::recommendedOrderForNJProbing()
//
//As a result, the following version will not be called.
PartitioningFunction *
SkewedDataPartitioningFunction::
scaleNumberOfPartitions(Lng32 &suggestedNewNumberOfPartitions,
PartitionGroupingDistEnum partGroupDist)
{
PartitioningFunction *mayBeNewPartfunc =
partialPartFunc_->scaleNumberOfPartitions(
suggestedNewNumberOfPartitions, partGroupDist);
if ( mayBeNewPartfunc != partialPartFunc_ )
partialPartFunc_ = mayBeNewPartfunc;
return this;
} // SkewedDataPartitioningFunction::scaleNumberOfPartitions()
// -----------------------------------------------------------------------
// SkewedDataPartitioningFunction::isAGroupingOf()
// -----------------------------------------------------------------------
NABoolean
SkewedDataPartitioningFunction::isAGroupingOf(const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
return FALSE;
} // SkewedDataPartitioningFunction::isAGroupingOf()
// -----------------------------------------------------------------------
// SkewedDataPartitioningFunction::copyAndRemap()
// -----------------------------------------------------------------------
PartitioningFunction*
SkewedDataPartitioningFunction::copyAndRemap
(ValueIdMap& map, NABoolean mapItUp) const
{
SkewedDataPartitioningFunction *newPartFunc =
(SkewedDataPartitioningFunction *)
copy()->castToSkewedDataPartitioningFunction();
CMPASSERT(skewProperty_.getIndicator() != skewProperty::BROADCAST);
newPartFunc->remapIt(this, map, mapItUp);
newPartFunc->partialPartFunc_->remapIt(partialPartFunc_, map, mapItUp);
return newPartFunc;
} // SkewedDataPartitioningFunction::copyAndRemap()
void
SkewedDataPartitioningFunction::preCodeGen(const ValueIdSet& availableValues)
{
PartitioningFunction::preCodeGen(availableValues);
partialPartFunc_->preCodeGen(availableValues);
}
ItemExpr* SkewedDataPartitioningFunction::createPartitioningExpression()
{
CMPASSERT(partialPartFunc_ -> isAHash2PartitioningFunction() == TRUE);
// cast to TableHashPartitioningFunction class (the parent class
// of Hash2PartitioningFunction) which has the method
// createPartitioningExpressionImp() defined.
ItemExpr* partExpr = ((TableHashPartitioningFunction*)partialPartFunc_)
-> createPartitioningExpressionImp(TRUE /*do varchar cast*/);
storeExpression(partExpr);
return partExpr;
}
// ***********************************************************************
// RangePartitionBoundaries
// ***********************************************************************
// -----------------------------------------------------------------------
// Constructor for RangePartitionBoundaries
// -----------------------------------------------------------------------
RangePartitionBoundaries::RangePartitionBoundaries
(Lng32 numberOfPartitions,
Lng32 numberOfPartitioningKeyColumns, NAMemory *h)
: partKeyColumnCount_(numberOfPartitioningKeyColumns),
origPartKeyColumnCount_(numberOfPartitioningKeyColumns),
partitionCount_(numberOfPartitions),
origPartitionCount_(numberOfPartitions),
boundaryStringsList_(h,numberOfPartitions+1),
boundaryValuesList_(h,numberOfPartitions+1),
boundaryValues_(h,numberOfPartitions+1),
binaryBoundaryValues_(h,numberOfPartitions+1),
encodedBoundaryKeyLength_(-1),
setBinaryBoundaryFirstLastKey_(FALSE),
setupForStatement_(FALSE),
resetAfterStatement_(FALSE),
heap_(h)
{
// MUST be given non zero counts
// make two dummy entries for the first and last key
boundaryValuesList_.insertAt(0,NULL);
boundaryValuesList_.insertAt(numberOfPartitions,NULL);
boundaryStringsList_.insertAt(0, NULL);
boundaryStringsList_.insertAt(numberOfPartitions,NULL);
} // RangePartitionBoundaries::RangePartitionBoundaries()
RangePartitionBoundaries::~RangePartitionBoundaries() {}
void RangePartitionBoundaries::defineUnboundBoundary
(Lng32 partitionNumber,
const ItemExpr * unboundBoundaryValue,
const char *encodedKeyValue)
{
// must insert a 'true' boundary between the first and last (pre-allocated)
// entry, this gives us n insertion points for a table with n partitions,
// new entry delimits partition <partitionNumber>
CMPASSERT(partitionNumber > 0 AND
partitionNumber <= partitionCount_ AND
unboundBoundaryValue != NULL AND
encodedKeyValue != NULL);
boundaryValuesList_.insertAt(partitionNumber, unboundBoundaryValue);
// encodedKeyValue should be NULL only for SQL/MP tables
if (encodedKeyValue)
binaryBoundaryValues_.insertAt(partitionNumber, encodedKeyValue);
}
void RangePartitionBoundaries::bindAddBoundaryValue(Lng32 partitionNumber)
{
if(!boundaryValuesList_[partitionNumber]){
boundaryValues_.insertAt(partitionNumber, NULL);
return;
}
//get unbound boundary value
ItemExpr * unboundBoundaryValue = ((ItemExpr *)boundaryValuesList_[partitionNumber])->
copyTree(CmpCommon::statementHeap());
//new (CmpCommon::statementHeap())
//ItemExpr(*boundaryValuesList_[partitionNumber]);
//bind the boundary value
unboundBoundaryValue->synthTypeAndValueId();
ItemExprList * boundBoundaryValue = new (CmpCommon::statementHeap())
ItemExprList(unboundBoundaryValue,
CmpCommon::statementHeap());
boundaryValues_.insertAt(partitionNumber, boundBoundaryValue);
if (!boundaryStringsList_.used(partitionNumber)) {
NAString result;
// use QUERY_FORMAT to obtain the full SQL text.
boundBoundaryValue->unparse(result, OPTIMIZER_PHASE, QUERY_FORMAT);
boundaryStringsList_.insertAt(partitionNumber, new (heap_) NAString(result));
}
}
// -----------------------------------------------------------------------
// RangePartitionBoundaries::defineBoundary()
// -----------------------------------------------------------------------
void RangePartitionBoundaries::defineBoundary
(Lng32 partitionNumber,
const ItemExprList* boundaryValue,
const char *encodedKeyValue)
{
// must insert a 'true' boundary between the first and last (pre-allocated)
// entry, this gives us n insertion points for a table with n partitions,
// new entry delimits partition <partitionNumber>
CMPASSERT(partitionNumber > 0 AND
partitionNumber <= partitionCount_ AND
boundaryValue != NULL AND
encodedKeyValue != NULL);
boundaryValues_.insertAt(partitionNumber, boundaryValue);
// encodedKeyValue should be NULL only for SQL/MP tables
if (encodedKeyValue)
binaryBoundaryValues_.insertAt(partitionNumber, encodedKeyValue);
} // RangePartitionBoundaries::defineBoundary()
// -----------------------------------------------------------------------
// RangePartitionBoundaries::checkConsistency()
// -----------------------------------------------------------------------
void RangePartitionBoundaries::checkConsistency(
const Lng32 numberOfPartitions) const
{
// If n = numberOfPartitions then entries 1 through n-1 have to be
// present. Each entry delimits the boundary between two partitions.
// Entries 0 and n contain the minimum and maximum permissible
// values, respectively. Entries 0 and n are initialized to NULL in
// the constructor and later added by method
// RangePartitioningFunction::completePartitionBoundaries()
CMPASSERT(numberOfPartitions == partitionCount_ AND
(Lng32)boundaryValues_.entries() == partitionCount_ + 1);
// check the actual boundaries after the key length of the encoded
// key and the key column value ids have been entered, but not before
if (encodedBoundaryKeyLength_ > 0)
{
CMPASSERT(boundaryValues_.entries() == binaryBoundaryValues_.entries());
for (CollIndex index = 1;
index < (CollIndex)numberOfPartitions;
index++)
{
// boundary values in SQL format and encoded key format must be
// filled in
CMPASSERT(boundaryValues_[index]);
CMPASSERT(binaryBoundaryValues_[index]);
// encoded key values must be in ascending order
// CMPASSERT(str_cmp(binaryBoundaryValues_[index-1],
// binaryBoundaryValues_[index],
// encodedBoundaryKeyLength_) < 0);
}
}
} // RangePartitionBoundaries::checkConsistency()
// -----------------------------------------------------------------------
// RangePartitionBoundaries::compareRangePartitionBoundaries()
// -----------------------------------------------------------------------
NABoolean RangePartitionBoundaries::compareRangePartitionBoundaries(
const RangePartitionBoundaries& other,
NABoolean groupingAllowed,
Lng32* maxPartsPerGroup) const
{
NABoolean match = FALSE;
CollIndex i;
CollIndex thisNumPartKeyCols,otherNumPartKeyCols;
const ItemExprList *thisBoundaryValue;
const ItemExprList *otherBoundaryValue;
const ConstValue *thisConstValue;
const ConstValue *otherConstValue;
NABoolean thisIsNegated = FALSE;
NABoolean otherIsNegated = FALSE;
Lng32 currentPartsPerGroup = 1;
Lng32 numOfPartsInLastGroup = 1;
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = 1; // start with the minimum
// Give up now if "other" has fewer partitions than "this"
if ( other.partitionCount_ < partitionCount_)
{
return FALSE;
}
// NOTE: We CANNOT compare the encoded values! Two values that are
// the SAME will compare as NOT THE SAME if the types of the
// underlying columns are different.
//
// So, we must compare the boundaryValues_ instead. The
// boundaryValues_ are an array of pointers to ItemExpr Lists
// which contain pointers to ConstValue nodes.
//
// We go through the array of ItemExpr Lists, and
// first compare the ValueId's of the ItemExpr Lists which
// contain pointers to the ItemExpr ConstValue nodes that represent
// the first key values. If these match, great -
// the first keys must be the same. But, they will only have the same
// ValueId if they are from the same table, or if the datatypes of
// the partitioning key columns are exactly the same.
// If they don't have the same ValueId, then we need to compare
// the original raw ASCII text as was present in the DDL.
// We go through the ConstValue nodes in ItemExpr lists one
// by one, and compare them. We use the "getConstStr" method
// of the ConstValue class to do this. This method gets the
// original raw text as typed in by the user, and also appends
// the character set type. This may be important if a user
// is joining two tables where the character set types of the
// partitioning key columns are different.
//
// Go through the boundaries of "this" and try to find them
// in "other" Don't bother with the first and last entry,
// they are always the same. When all is said and done, we
// must have found a match in other for every partition boundary
// from "this". The opposite assertion - that every partition
// boundary in "other" found a match in "this" - is not necessary
// if grouping is allowed.
CollIndex myix = 1;
CollIndex otherix = 1;
while (myix < (CollIndex) partitionCount_ AND
otherix < (CollIndex) other.partitionCount_)
{
match = FALSE;
thisBoundaryValue = boundaryValues_[myix];
otherBoundaryValue = other.boundaryValues_[otherix];
// Check for the simple case: are the ValueId's the same?
if (*thisBoundaryValue == *otherBoundaryValue)
{
match = TRUE;
}
else
{
// Traverse the ItemExprList and compare each ConstValue node
// in "this" to the one in "other". The first keys are the
// same only if they both have the same number of first keys,
// they are either both negated or neither are,
// and the string text of each ConstValue node is
// exactly the same.
thisNumPartKeyCols = thisBoundaryValue->entries();
otherNumPartKeyCols = otherBoundaryValue->entries();
if (thisNumPartKeyCols != otherNumPartKeyCols)
{
// If the two partitioning functions specified a different
// number of first key values, then they cannot possibly
// have equivalent partitioning schemes.
return FALSE;
}
else
{
// Hope for the best...
match = TRUE;
for (i = 0;i < thisNumPartKeyCols AND match; i++)
{
thisIsNegated = FALSE;
otherIsNegated = FALSE;
thisConstValue =
((*thisBoundaryValue)[i])->castToConstValue(thisIsNegated);
otherConstValue =
((*otherBoundaryValue)[i])->castToConstValue(otherIsNegated);
if ((thisIsNegated != otherIsNegated) OR
thisConstValue == NULL OR
otherConstValue == NULL OR
(thisConstValue->getConstStr(FALSE) !=
otherConstValue->getConstStr(FALSE)))
match = FALSE;
}
}
}
if (match)
{
// Found this boundary in "other". Advance to the next
// start key in both "this" and "other".
myix++;
otherix++;
if ((maxPartsPerGroup != NULL) AND
(currentPartsPerGroup > *maxPartsPerGroup))
{
*maxPartsPerGroup = currentPartsPerGroup;
}
currentPartsPerGroup = 1; // clear
}
else if (groupingAllowed)
{
// No match. Advance to the next start key in "other"
// to see if that will match.
otherix++;
currentPartsPerGroup++;
}
else
{
// These boundaries don't match, and since grouping is not allowed,
// we are out of luck. Give up now.
return FALSE;
}
} // end while more partitions
// Since we didn't look at the last group, we need to compute the
// number of partitions in it and see if it is the group with
// the largest number of partitions.
numOfPartsInLastGroup = (other.partitionCount_ - otherix) + 1;
if ((maxPartsPerGroup != NULL) AND
(numOfPartsInLastGroup > *maxPartsPerGroup))
{
*maxPartsPerGroup = numOfPartsInLastGroup;
}
if (myix == (CollIndex) partitionCount_)
{
// We have found all of our partition boundaries
// in the other range partition boundaries object.
return TRUE;
}
else
{
// Even with grouping, the partition boundaries don't match. Too bad.
return FALSE;
}
} // RangePartitionBoundaries::compareRangePartitionBoundaries()
//<pb>
//==============================================================================
// Merge two range partition boundaries to produce new range partition
// boundaries. Also produce a node map corresponding to the new range partition
// boundaries.
//
// Input:
// other -- other partition boundary map with which to merge.
// thisNodeMap -- node map associated with this RangePartitionBoundaries
// object.
//
// Output:
// resultNodeMap -- node map associated with resulting RangepartitionBoundaries
// object.
//
// Return:
// Pointer to merged partition boundaries.
//
//==============================================================================
RangePartitionBoundaries*
RangePartitionBoundaries::merge(const RangePartitionBoundaries& other,
const NodeMap& thisNodeMap,
NodeMap& resultNodeMap) const
{
// need binary boundary values to do a merge
if (encodedBoundaryKeyLength_ == 0 OR
other.encodedBoundaryKeyLength_ == 0)
return NULL;
// left is "this", right is "other",
// merge boundaries 0...entries of "this" with boundaries 1...entries-1
// of "other" (don't use -infinity and +infinity of right)
CollIndex thisix = 0;
CollIndex otherix = 1;
CollIndex resultix = 0;
CollIndex maxthis = boundaryValues_.entries();
CollIndex maxother = other.boundaryValues_.entries()-1;
CollIndex maxboth = maxthis + maxother;
RangePartitionBoundaries *result = new(CmpCommon::statementHeap())
RangePartitionBoundaries(1,partKeyColumnCount_,CmpCommon::statementHeap());
Lng32 compResult;
CMPASSERT(partKeyColumnCount_ == other.partKeyColumnCount_ AND
encodedBoundaryKeyLength_ == other.encodedBoundaryKeyLength_);
// continue merging boundaries while at least one index has not reached its
// ending value.
while (thisix < maxthis OR otherix < maxother)
{
// The final boundary has no corresponding node map entry.
// In other words, the number of boundaries is always one more than
// the number of node map entries.
if (maxboth - thisix - otherix > 1)
{
// Since node maps have one less entry than boundary maps, ensure
// that the node map's index does not fall off the end of the node
// map.
CollIndex nodeMapIx = MINOF(thisix, thisNodeMap.getNumEntries() - 1);
// Store node map entry associated with current boundary.
const NodeMapEntry* entry = thisNodeMap.getNodeMapEntry(nodeMapIx);
resultNodeMap.setNodeMapEntry(resultix,
*entry,
CmpCommon::statementHeap());
}
// other index reached max, so comparison is automatically "less"
if (otherix >= maxother)
{
compResult = -1;
}
else
{
// this index reached max, so comparison is automatically "more"
if (thisix >= maxthis)
{
compResult = 1;
}
else
{
// neither index reached max, so do an actual comparison of all
// 'real' boundaries in the encoded representation (the first
// and last boundaries are always the same)
compResult = str_cmp(binaryBoundaryValues_[thisix],
other.binaryBoundaryValues_[otherix],
encodedBoundaryKeyLength_);
}
}
// take the smallest boundary value and insert it into result
if (compResult <= 0)
{
// "this" has the smaller (or equal) boundary value
result->boundaryValues_.insertAt(
resultix,
boundaryValues_[thisix]);
result->binaryBoundaryValues_.insertAt(
resultix,
binaryBoundaryValues_[thisix]);
thisix++;
// if the boundaries are identical, skip both of them
if (compResult == 0)
otherix++;
}
else
{
// "other" has the smaller boundary value
result->boundaryValues_.insertAt(
resultix,
other.boundaryValues_[otherix]);
result->binaryBoundaryValues_.insertAt(
resultix,
other.binaryBoundaryValues_[otherix]);
otherix++;
}
// Increment result index for next boundary.
resultix++;
}
result->encodedBoundaryKeyLength_ = encodedBoundaryKeyLength_;
result->partitionCount_ = resultix - 1;
return result;
} // RangePartitionBoundaries::merge
Lng32 RangePartitionBoundaries::getOptimizedNumberOfPartKeys()
{
// Do an optimization of the partitioning key: what is passed in is
// currently the clustering key. Count the actual number of first
// keys that are specified in the DDL and only use the max. number
// of items instead of all clustering key columns. Example:
//
// create table t (a int, b int, c int, primary key (a,b desc,c)
// partition (add first key (1), add first key (2,2));
//
// In this case, the partitioning key order is not (a,b desc,c), but
// (a, b desc). This optimization should be done in CATMAN. Remove
// this method once that is done.
//
CollIndex numPartKeyCols = 0;
for (CollIndex i = 1; i < (CollIndex) partitionCount_; i++)
{
// Are there more values listed than we assume as partitioning
// key columns? Adjust our assumption if needed.
if ( boundaryValues_[i] ) {
CollIndex numKeySpecs = boundaryValues_[i]->entries();
if (numKeySpecs > numPartKeyCols)
numPartKeyCols = numKeySpecs;
} else {
// no boundary values specified in SQL syntax
// (e.g. when we got binary region boundaries only)
numPartKeyCols = partKeyColumnCount_;
break;
}
}
return numPartKeyCols;
}
Lng32 RangePartitionBoundaries::scaleNumberOfPartitions(
Lng32 suggestedNewNumberOfPartitions,
const NodeMap* nodeMap,
PartitionGroupingDistEnum partGroupDist)
{
// * * * * * * * * * * * * * I M P O R T A N T * * * * * * * * * * * * * *
// * *
// * THIS MEMBER FUNCTION USES THE SAME GROUPING ALGORITHM AS MEMBER *
// * FUNCTION NodeMap::deriveGrouping(). ANY CHANGES TO THIS MEMBER *
// * FUNCTION WOULD NECESSITATE CHANGES TO MEMBER FUNCTION *
// * NodeMap::deriveGrouping() AS WELL. *
// * *
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
// For now, the only way of scaling the number of partitions
// is by eliminating part of the boundaries (deleting all but
// every i-th boundary where i is an integer number).
// There are two problems with this limitation: the suggestion may
// ask for an increase in the number of partitions, or the suggestion
// may cause an imbalance because the suggested number times i is not
// equal to the existing number of partitions.
// Here are the heuristics that we are using until we get smarter:
//
// H1: Treat the suggested number as an upper limit of new partitions.
// Often, the suggestion is the number of CPUs available and plans
// with 13 partitions don't run very well on 12 CPUs.
//
// H2: Distribute the old partitions such that the number of old partitions
// for each new partition varies at most by one (a new partition has
// either o or o+1 old partitions in it).
//
// We have two ways of distributing the partitions: "uniformly" (off
// by at most one) distribute the number of physical partitions
// amongst the groups, or uniformly distribute the number of active
// partitions amongst the groups. The latter is preferable if there
// are many inactive partitions and the active partitions are not
// uniformly distributed amongst them. Actually, if the active
// partition estimate is correct, we cannot go wrong distributing
// the partitions based on the number of active partitions.
// In cases where this would combine partitions from two or more clusters
// it is better not to form such groups, as this would limit autonomy
// and also lead to more messages between clusters. In such cases, this
// method will return with the original partition boundary arrays.
// Make copies in case we need to undo.
ARRAY(const ItemExprList *) undoBoundaryValues(boundaryValues_);
ARRAY(const char *) undoBinaryBoundaryValues(binaryBoundaryValues_);
// As already said, we can't increase the number of partitions
if (suggestedNewNumberOfPartitions >= partitionCount_)
return partitionCount_;
CMPASSERT(suggestedNewNumberOfPartitions > 1); // sanity check
NABoolean useAPDistribution = FALSE;
// Determine partition grouping distribution algorithm to use
if (partGroupDist == DEFAULT_PARTITION_GROUPING)
{
if (CmpCommon::getDefault(BASE_NUM_PAS_ON_ACTIVE_PARTS) == DF_ON)
useAPDistribution = TRUE;
else
useAPDistribution = FALSE;
}
else if (partGroupDist == UNIFORM_PHYSICAL_PARTITION_GROUPING)
{
useAPDistribution = FALSE;
}
else if (partGroupDist == UNIFORM_ACTIVE_PARTITION_GROUPING)
{
useAPDistribution = TRUE;
}
Lng32 newix = 1;
Lng32 prevPartStart = 0;
if (useAPDistribution)
{
// Uniform number of active partitions grouping
CostScalar activePartitions =
((NodeMap *)(nodeMap))->getNumActivePartitions();
Lng32 numActiveParts = (Lng32)activePartitions.getValue();
if (suggestedNewNumberOfPartitions > numActiveParts)
return partitionCount_;
Lng32 currentNumActiveParts = 0;
Lng32 numActivePartsPerGroup = numActiveParts / suggestedNewNumberOfPartitions;
Lng32 remainder = numActiveParts % suggestedNewNumberOfPartitions;
Lng32 oldix = 1;
Lng32 partNum = 0;
NodeMapEntry::PartitionState partState;
NABoolean isPartActive = FALSE;
while ((partNum < partitionCount_) AND
(newix != suggestedNewNumberOfPartitions))
{
// Determine if the current partition is active
partState = nodeMap->getNodeMapEntry((CollIndex)partNum)->
getPartitionState();
isPartActive = ((partState == NodeMapEntry::ACTIVE) OR
(partState == NodeMapEntry::ACTIVE_NO_DATA));
if ((remainder != 0) AND
((newix - 1) + remainder == suggestedNewNumberOfPartitions))
{
// all the remaining groups get one extra active partition
numActivePartsPerGroup++;
remainder = 0; // so we won't do it again
}
if (isPartActive)
{
// one more active partition for this group
currentNumActiveParts++;
if (currentNumActiveParts == numActivePartsPerGroup)
{
// full group - time to end it, so check that partns on same cluster
if (nodeMap->isMultiCluster(prevPartStart, oldix, TRUE))
{
// More than one cluster in a group - undo.
boundaryValues_ = undoBoundaryValues;
binaryBoundaryValues_ = undoBinaryBoundaryValues;
return partitionCount_;
}
// Transfer the values over
boundaryValues_[newix] = boundaryValues_[oldix];
binaryBoundaryValues_[newix] = binaryBoundaryValues_[oldix];
// advance to prepare for next new partition boundary
newix++;
// reset number of active partitions in the current group
currentNumActiveParts = 0;
// remember where that previous group started
prevPartStart = oldix;
}
} // end if current partition is active
// advance to next old partition boundary
oldix++;
// advance to next partition
partNum++;
} // end while more partitions and more new boundaries to produce
}
else // Uniform number of physical partitions grouping
{
// Determine an integer value i such that partitionCount_ / i is
// approximately equal to suggestedNewNumberOfPartitions.
Lng32 numPartsPerGroup = partitionCount_ / suggestedNewNumberOfPartitions;
Lng32 remainder = partitionCount_ % suggestedNewNumberOfPartitions;
// If the remainder is not zero we will end up with an imbalanced new
// partitioning scheme. To reduce the imbalance, distribute the remainder
// partitions such that at most one of them goes into each new partition.
// The easiest way to do this is to give one extra old partition to the
// last <remainder> of the new partitions.
// Remember that entries 0 and partitionCount_ stand for the min and
// max key and should be preserved. For the entries in between, take
// only every <numPartsPerGroup> one and make a new, contiguous array.
Lng32 oldix = numPartsPerGroup;
while (oldix < partitionCount_)
{
if (nodeMap->isMultiCluster(prevPartStart, oldix, FALSE))
{
// More than one cluster in a group - undo.
boundaryValues_ = undoBoundaryValues;
binaryBoundaryValues_ = undoBinaryBoundaryValues;
return partitionCount_;
}
boundaryValues_[newix] = boundaryValues_[oldix];
binaryBoundaryValues_[newix] = binaryBoundaryValues_[oldix];
if (newix + remainder == suggestedNewNumberOfPartitions)
numPartsPerGroup++; // rest of the new partitions get one extra
prevPartStart = oldix;
oldix += numPartsPerGroup;
newix += 1;
}
} // end if uniform number of physical partitions grouping
// move the last entry for the max key
if (nodeMap->isMultiCluster(prevPartStart, partitionCount_,
useAPDistribution))
{
// More than one cluster in a group - undo.
boundaryValues_ = undoBoundaryValues;
binaryBoundaryValues_ = undoBinaryBoundaryValues;
return partitionCount_;
}
boundaryValues_[newix] = boundaryValues_[partitionCount_];
binaryBoundaryValues_[newix] = binaryBoundaryValues_[partitionCount_];
// Sanity check, we should now have exactly suggestedNewNumberOfPartitions+1
// new entries in the boundaryValues_ array.
CMPASSERT(newix == suggestedNewNumberOfPartitions);
// to be nice, make the rest of the array empty
for (newix = newix+1; newix <= partitionCount_; newix++)
{
boundaryValues_.remove(newix);
binaryBoundaryValues_.remove(newix);
}
partitionCount_ = suggestedNewNumberOfPartitions;
return partitionCount_;
}
NABoolean RangePartitionBoundaries::isAGroupingOf(
const RangePartitionBoundaries &other,
Lng32* maxPartsPerGroup) const
{
// Call the method that compares range partition boundaries, with
// the optional parameter that indicates if grouping is OK set to TRUE.
return compareRangePartitionBoundaries(other,TRUE,maxPartsPerGroup);
} // RangePartitionBoundaries::isAGroupingOf()
// -----------------------------------------------------------------------
// RangePartitionBoundaries::getBoundaryValues()
// -----------------------------------------------------------------------
const ItemExprList* RangePartitionBoundaries::getBoundaryValues(Lng32 index) const
{
// It is legal to index entries in the range [0, partitionCount_].
CMPASSERT( (index >= 0) AND (index <= partitionCount_) );
return boundaryValues_[index];
} // RangePartitionBoundaries::getBoundaryValues()
// -----------------------------------------------------------------------
// RangePartitionBoundaries::getBinaryBoundaryValue()
// -----------------------------------------------------------------------
const char * RangePartitionBoundaries::getBinaryBoundaryValue(
Lng32 index) const
{
// It is legal to index entries in the range [0, partitionCount_].
CMPASSERT( (index >= 0) AND (index <= partitionCount_) );
return binaryBoundaryValues_[index];
} // RangePartitionBoundaries::getBinaryBoundaryValue()
// -----------------------------------------------------------------------
// RangePartitionBoundaries::completePartitionBoundaries()
// -----------------------------------------------------------------------
void RangePartitionBoundaries::completePartitionBoundaries(
const ValueIdList& partitioningKeyOrder,
Lng32 encodedBoundaryKeyLength)
{
// recognize from the fact that the encoded key length is
// not set yet that this method hasn't been called before
CMPASSERT(encodedBoundaryKeyLength_ == -1);
NABoolean ascending_key_column = TRUE;
ItemExprList *firstKey = new(CmpCommon::statementHeap())
ItemExprList(CmpCommon::statementHeap());
ItemExprList *lastKey = new(CmpCommon::statementHeap())
ItemExprList(CmpCommon::statementHeap());
// set some data members that could not be set before
partKeyColumnCount_ = partitioningKeyOrder.entries();
encodedBoundaryKeyLength_ = encodedBoundaryKeyLength;
Lng32 i;
// for each partitioning key column
for (i = 0; i < partKeyColumnCount_; i++)
{
// get the type of the partitioning key column
const NAType &prototypeType = partitioningKeyOrder[i].getType();
ItemExpr *minVal = new(CmpCommon::statementHeap())
SystemLiteral(&prototypeType,
TRUE /*want min*/,
TRUE /*allow NULL*/);
ItemExpr *maxVal = new(CmpCommon::statementHeap())
SystemLiteral(&prototypeType,
FALSE,
TRUE);
ItemExpr * orderExpr = partitioningKeyOrder[i].getItemExpr();
// add the min/max values to the first and last key
if (orderExpr->getOperatorType() != ITM_INVERSE)
{
// Ascending key column, create min value for first key and
// max value for last key
ascending_key_column = TRUE;
firstKey->insert(minVal);
lastKey->insert(maxVal);
}
else
{
// Descending key column, create max value for first key and
// min value for last key
ascending_key_column = FALSE;
firstKey->insert(maxVal);
lastKey->insert(minVal);
}
// Check if any first key values don't specify this partitioning
// key column. If so, add the MIN or the MAX value for that
// column as the missing first key value.
for (CollIndex j = 1; j < (CollIndex)partitionCount_; j++)
{
if (boundaryValues_[j]->entries() == (CollIndex)i)
{
// This first key does not have enough entries to have
// specified a value for this column. Add it. Add the
// MIN value if ascending, add the MAX value if descending.
// Need to cast away the const-ness to do this - sorry.
if (ascending_key_column)
((ItemExprList*)boundaryValues_[j])->insert(minVal);
else
((ItemExprList*)boundaryValues_[j])->insert(maxVal);
}
}
} // for each partitioning key column
// finally, add the created begin and end key as the first key and
// after the last start key
boundaryValues_[0] = firstKey;
boundaryValues_[partitionCount_] = lastKey;
// do we have any binary boundary values at all?
if (binaryBoundaryValues_.entries())
{
if(!setBinaryBoundaryFirstLastKey_)
{
// yes, make entries for binary encoded keys with all zeroes and
// all ones for the first and last key, respectively
char * binaryMin =
new(heap_) char[encodedBoundaryKeyLength_];
char * binaryMax =
new(heap_) char[encodedBoundaryKeyLength_];
str_pad(binaryMin,encodedBoundaryKeyLength_,'\0');
str_pad(binaryMax,encodedBoundaryKeyLength_,'\377');
binaryBoundaryValues_.insertAt(0,binaryMin);
binaryBoundaryValues_.insertAt(partitionCount_,binaryMax);
setBinaryBoundaryFirstLastKey_ = TRUE;
}
}
else
{
// indicate that there are no binary key values (happens for SQL/MP)
encodedBoundaryKeyLength_ = 0;
}
checkConsistency(partitionCount_);
} // RangePartitionBoundaries::completePartitionBoundaries()
ItemExpr * getRangePartitionBoundaryValues
(const char * keyValueBuffer,
const Lng32 keyValueBufferSize,
NAMemory* heap,
CharInfo::CharSet strCharSet = CharInfo::UTF8
);
void RangePartitionBoundaries::setupForStatement(NABoolean useStringVersion)
{
if(setupForStatement_)
return;
if ( useStringVersion ) {
for(UInt32 i=0; i < boundaryStringsList_.entries(); i++)
{
if (!boundaryStringsList_.used(i) || !boundaryStringsList_[i])
{
boundaryValuesList_.insertAt(i,NULL);
} else {
boundaryValuesList_.insertAt(i,
getRangePartitionBoundaryValues
( boundaryStringsList_[i]->data(),
boundaryStringsList_[i]->length(),
// Warning: use stmt heap so that the returned itemExpr is NOT
// allocated on heap_, which can be the context heap.
// If the allocation is on the context heap, the memory
// will not be deleted (no way to delete a complicated itemExpr)
// during RangePartitionBoundaries::resetAfterStatement() call and
// we can fail the assertion test
// CMPASSERT(sizeAfterLastStatement_ <= initialSize_ )
// in RangePartitionBoundaries::resetAfterStatement()
CmpCommon::statementHeap()
));
}
}
}
for(UInt32 i=0; i < boundaryValuesList_.entries(); i++)
{
bindAddBoundaryValue(i);
}
setupForStatement_ = TRUE;
resetAfterStatement_ = FALSE;
}
void RangePartitionBoundaries::resetAfterStatement()
{
if(resetAfterStatement_)
return;
boundaryValuesList_.clear();
boundaryValues_.clear();
partKeyColumnCount_ = origPartKeyColumnCount_;
partitionCount_ = origPartitionCount_;
setupForStatement_ = FALSE;
resetAfterStatement_ = TRUE;
}
// -----------------------------------------------------------------------
// Method for debugging.
// -----------------------------------------------------------------------
void RangePartitionBoundaries::print(FILE* ofd, const char* indent,
const char* title) const
{
BUMP_INDENT(indent);
char S[500];
int index;
fprintf(ofd,"%s %s\n",NEW_INDENT,title);
for (index = 0; index < partitionCount_; index++)
{
const ItemExprList* iel = boundaryValues_[index];
snprintf(S, sizeof(S), "boundary[%d]: ", index);
if (iel)
iel->print(ofd, indent, S);
else
fprintf(ofd,"%s %s is empty\n",NEW_INDENT,S);
}
for (int index2 = 0; index2 < partitionCount_; index2++)
if (binaryBoundaryValues_.used(index2) &&
binaryBoundaryValues_[index2])
{
const char *binaryVal = binaryBoundaryValues_[index2];
fprintf(ofd, "binary boundary[%d]: 0x", // %#0*",
index2);
for (int b=0; b<encodedBoundaryKeyLength_; b++)
fprintf(ofd, "%02hhx", binaryVal[b]);
fprintf(ofd, "\n");
}
fprintf(ofd,"%s %s (in binary form)\n",NEW_INDENT,title);
Lng32 keyLen = getEncodedBoundaryKeyLength();
for (index = 0; index < partitionCount_; index++)
{
const char* bValues = getBinaryBoundaryValue(index);
for (Int32 j = 0; j < keyLen; j++) {
fprintf(ofd,"%04x ", (int)bValues[j]);
}
}
} // RangePartitionBoundaries::print()
// ***********************************************************************
// RangePartitioningFunction
// ***********************************************************************
RangePartitioningFunction::RangePartitioningFunction(
const RangePartitioningFunction& other,
NAMemory* heap)
: PartitioningFunction(other, heap),
keyColumnList_(other.keyColumnList_),
orderOfKeyValues_(other.orderOfKeyValues_),
originalKeyColumnList_(other.originalKeyColumnList_),
setupForStatement_(FALSE),
resetAfterStatement_(FALSE)
{
partitionBoundaries_ = new(heap)
RangePartitionBoundaries(*(other.partitionBoundaries_),heap);
}
RangePartitioningFunction::~RangePartitioningFunction() {}
Lng32 RangePartitioningFunction::getCountOfPartitions() const
{ return partitionBoundaries_->getCountOfPartitions(); }
const RangePartitioningFunction*
RangePartitioningFunction::castToRangePartitioningFunction() const
{ return this; }
PartitioningRequirement*
RangePartitioningFunction::makePartitioningRequirement()
{
return new (CmpCommon::statementHeap())
RequireRange(this);
}
PartitioningFunction*
RangePartitioningFunction::copy() const
{ return new(CmpCommon::statementHeap()) RangePartitioningFunction(*this); }
// -----------------------------------------------------------------------
// RangePartitioningFunction::normalizePartitioningKeys()
// Rewrite the partitioning keys of the partitioning function in
// terms of the VEGReference for the VEG to which the partitioning
// key column belongs.
// -----------------------------------------------------------------------
void RangePartitioningFunction::normalizePartitioningKeys(NormWA& normWARef)
{
PartitioningFunction::normalizePartitioningKeys(normWARef);
keyColumnList_.normalizeNode(normWARef);
orderOfKeyValues_.normalizeNode(normWARef);
// don't normalize original key col list, avoid VEGies which could
// cause data type changes.
} // RangePartitioningFunction::normalizePartitioningKeys
// -----------------------------------------------------------------------
// RangePartitioningFunction::comparePartFuncToFunc()
// -----------------------------------------------------------------------
COMPARE_RESULT
RangePartitioningFunction::comparePartFuncToFunc
(const PartitioningFunction &other) const
{
// Make use of the implementation of isAGroupingOf. According to the
// definition of a grouping, two partitioning functions must be the
// same if one is a grouping of the other and if they have the same
// number of partitions. This is not a very straightforward way to
// compare two range partitioning functions but it avoids
// duplication of code.
if (isAGroupingOf(other) AND
getCountOfPartitions() == other.getCountOfPartitions())
return SAME;
else
return INCOMPATIBLE;
} // RangePartitioningFunction::comparePartFuncToFunc()
NABoolean RangePartitioningFunction::isAGroupingOf(
const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
const RangePartitioningFunction *oth =
other.castToRangePartitioningFunction();
// a range partitioning function can only be a grouping of
// another range partitioning function
if (oth == NULL)
return FALSE;
// to be a grouping of the other function, this function has to be
// partitioned on the same columns in the same sequence and order
// (a prefix might work in some cases which we don't recognize
// at this point, sorry)
if (orderOfKeyValues_.entries() != oth->orderOfKeyValues_.entries())
return FALSE;
// compare the key columns and their order
for (CollIndex i = 0; i < orderOfKeyValues_.entries(); i++)
{
if (orderOfKeyValues_[i] != oth->orderOfKeyValues_[i])
{
// value ids are different, but this may just be that we
// have different inversion expressions
if (orderOfKeyValues_[i].getItemExpr()->getOperatorType() !=
ITM_INVERSE OR
oth->orderOfKeyValues_[i].getItemExpr()->getOperatorType() !=
ITM_INVERSE OR
orderOfKeyValues_[i].getItemExpr()->child(0) !=
oth->orderOfKeyValues_[i].getItemExpr()->child(0))
return FALSE; // really different columns
}
}
// -----------------------------------------------------------------
// Compare the partition boundaries
// -----------------------------------------------------------------
return partitionBoundaries_->isAGroupingOf(*oth->partitionBoundaries_,
maxPartsPerGroup);
}
// -----------------------------------------------------------------------
// RangePartitioningFunction::createPartitioningKeyPredicates()
// -----------------------------------------------------------------------
void RangePartitioningFunction::createPartitioningKeyPredicates()
{
if (NOT partKeyPredsCreated())
{
char fabricatedName[20]; // a name for the host variable
char* S = fabricatedName;
CollIndex nCols = keyColumnList_.entries();
ValueIdSet setOfKeyPredicates;
ValueIdList partInputValues(2*nCols+1);
ValueIdList loValues;
ValueIdList hiValues;
// -----------------------------------------------------------------
// create partition input variables: one low value for each range
// key column, one hi value for each range key column, and one
// inclusion indicator column
// -----------------------------------------------------------------
for (CollIndex beginEnd = 0; beginEnd < 2; beginEnd++)
for (CollIndex i = 0; i < nCols; i++)
{
ValueId vid = originalKeyColumnList_[i];
// -- Fabricate a name for the first host variable
if (beginEnd == 0)
sprintf(S,"_sys_HostVarLo%d",i);
else
sprintf(S,"_sys_HostVarHi%d",i);
// -- Build a host var node with the key column's type
HostVar *hv = new(CmpCommon::statementHeap())
HostVar(S, &(vid.getType()), TRUE);
hv->synthTypeAndValueId();
// insert the partition input variable
partInputValues.insert(hv->getValueId());
if (beginEnd == 0)
loValues.insert(hv->getValueId());
else
hiValues.insert(hv->getValueId());
}
// The inclusion indicator is a 4-byte integer, no NULLs allowed.
// At execution time it is set to a non-zero value if the end
// key is excluded and to a zero value if it is included.
ItemExpr *intervalExclusionIndicator =
new(CmpCommon::statementHeap()) HostVar(
"_sys_hostVarExclRange",
new(CmpCommon::statementHeap()) SQLInt(CmpCommon::statementHeap(), TRUE,FALSE),
TRUE);
intervalExclusionIndicator->synthTypeAndValueId();
partInputValues.insert(intervalExclusionIndicator->getValueId());
// -----------------------------------------------------------------
// Create the semantic equivalent of multi-valued predicates, taking
// the order of keys into account (which can't be expressed in
// "normal" multi-valued predicates). Set the "special nulls"
// property for all comparison operators that involve nullable
// columns, since we do not want to filter out any NULL values.
// -----------------------------------------------------------------
// -----------------------------------------------------------------
// a GREATER_EQ multi-valued predicate for the begin key
// -----------------------------------------------------------------
CollIndex i = keyColumnList_.entries()-1;
CollHeap *h = CmpCommon::statementHeap();
// the comparison operator for the current column, taking the
// order of the column into account
OperatorTypeEnum compOp,extraOp;
ItemExpr *kc = keyColumnList_[i].getItemExpr();
NABoolean nullable =
keyColumnList_[i].getType().supportsSQLnullLogical();
// start with a greater or equal comparison for the last key
// column (may be the only key column), use less equal if the
// column is descending
if (orderOfKeyValues_[i].getItemExpr()->getOperatorType() ==
ITM_INVERSE)
{
compOp = ITM_LESS_EQ;
extraOp = ITM_GREATER_EQ;
}
else
{
compOp = ITM_GREATER_EQ;
extraOp = ITM_LESS_EQ;
}
ItemExpr * beginPred = new(h) BiRelat(compOp,
kc,
loValues[i].getItemExpr(),
nullable,
TRUE);
// There are now two end predicates generated. The tri-relational predicate
// is the one that enforces the inclusive comparison for the last partition,
// and only exclusive comparison for the other partitions. Thus this
// accurately models way partitions are setup.
// The extraEndPred has been added (HL 6/20/2001) to form a convenient key predicate
// for access by a B-Tree index. Although this is always inclusive, and thus
// will select extra data (one extra uec) for all partitions except the last,
// the extra data will be filtered out by the tri-relational operator.
ItemExpr * extraEndPred = new(h) BiRelat(extraOp,
kc,
hiValues[i].getItemExpr(),
nullable,
TRUE);
// for all other key columns, add (ai > bi) or (ai = bi) and (tfm)
while (i--)
{
if (orderOfKeyValues_[i].getItemExpr()->getOperatorType() ==
ITM_INVERSE)
{
compOp = ITM_LESS;
extraOp = ITM_GREATER;
}
else
{
compOp = ITM_GREATER;
extraOp = ITM_LESS;
}
kc = keyColumnList_[i].getItemExpr();
nullable =
keyColumnList_[i].getType().supportsSQLnullLogical();
beginPred = new(h) BiLogic(
ITM_OR,
new(h) BiRelat(compOp,
kc,
loValues[i].getItemExpr(),
nullable,
TRUE),
new(h) BiLogic(ITM_AND,
new(h) BiRelat(
ITM_EQUAL,
kc,
loValues[i].getItemExpr(),
nullable,
TRUE),
beginPred));
extraEndPred = new(h) BiLogic(
ITM_OR,
new(h) BiRelat(extraOp,
kc,
hiValues[i].getItemExpr(),
nullable,
TRUE),
new(h) BiLogic(ITM_AND,
new(h) BiRelat(
ITM_EQUAL,
kc,
hiValues[i].getItemExpr(),
nullable,
TRUE),
extraEndPred));
}
// -----------------------------------------------------------------
// a less or less-equal predicate for the end key
// -----------------------------------------------------------------
// reset to the end of the key column list
i = keyColumnList_.entries()-1;
// Start with a tri-relational comparison for the last key
// column (may be the only key column), it's a LESS comparison
// if the exclusion indicator is not zero, or LESS_EQUAL otherwise.
// Note: replace "less" with "greater" for descending columns.
if (orderOfKeyValues_[i].getItemExpr()->getOperatorType() ==
ITM_INVERSE)
compOp = ITM_GREATER_OR_GE;
else
compOp = ITM_LESS_OR_LE;
kc = keyColumnList_[i].getItemExpr();
nullable =
keyColumnList_[i].getType().supportsSQLnullLogical();
TriRelational * tr = new(h) TriRelational(
compOp,
kc,
hiValues[i].getItemExpr(),
new(h) BiRelat(ITM_NOT_EQUAL,
new(h) SystemLiteral(0),
intervalExclusionIndicator));
tr->setSpecialNulls(nullable);
ItemExpr *endPred = tr;
// for all other key columns, add (ai < bi) or (ai = bi) and (tfm)
while (i--)
{
if (orderOfKeyValues_[i].getItemExpr()->getOperatorType() ==
ITM_INVERSE)
compOp = ITM_GREATER;
else
compOp = ITM_LESS;
kc = keyColumnList_[i].getItemExpr();
nullable =
keyColumnList_[i].getType().supportsSQLnullLogical();
endPred = new(h) BiLogic(
ITM_OR,
new(h) BiRelat(compOp,
kc,
hiValues[i].getItemExpr(),
nullable),
new(h) BiLogic(ITM_AND,
new(h) BiRelat(
ITM_EQUAL,
kc,
hiValues[i].getItemExpr(),
nullable),
endPred));
}
endPred->synthTypeAndValueId();
beginPred->synthTypeAndValueId();
extraEndPred->synthTypeAndValueId();
setOfKeyPredicates += beginPred->getValueId();
setOfKeyPredicates += endPred->getValueId();
setOfKeyPredicates += extraEndPred->getValueId();
// -- Store the set of key predicates and the partition input values
// in the partitioning attributes.
storePartitioningKeyPredicates(setOfKeyPredicates);
storePartitionInputValues(partInputValues);
}
} // RangePartitioningFunction::createPartitioningKeyPredicates()
// -----------------------------------------------------------------------
// RangePartitioningFunction::replacePivs()
// -----------------------------------------------------------------------
void RangePartitioningFunction::replacePivs(
const ValueIdList& newPivs,
const ValueIdSet& newPartKeyPreds)
{
// Overwrite the old pivs, part key preds, and part expr. with the new ones.
storePartitionInputValues(newPivs);
storePartitioningKeyPredicates(newPartKeyPreds);
} // RangePartitioningFunction::replacePivs()
PartitioningFunction * RangePartitioningFunction::scaleNumberOfPartitions(
Lng32 &suggestedNewNumberOfPartitions,
PartitionGroupingDistEnum partGroupDist)
{
if (suggestedNewNumberOfPartitions == 1)
return new(CmpCommon::statementHeap())
SinglePartitionPartitioningFunction();
// If an expression has been generated, then we want to discard it
// because it may no longer be correct.
storeExpression(NULL);
// the power of delegation
suggestedNewNumberOfPartitions =
partitionBoundaries_->scaleNumberOfPartitions(
suggestedNewNumberOfPartitions,
getNodeMap(),
partGroupDist);
return this;
}
// -----------------------------------------------------------------------
// RangePartitioningFunction::remapIt()
// -----------------------------------------------------------------------
void RangePartitioningFunction::remapIt
(const PartitioningFunction* opf,
ValueIdMap& map, NABoolean mapItUp)
{
PartitioningFunction::remapIt(opf, map,mapItUp);
// If we have arrived here, the original partitioning function (*opf)
// MUST be a RangePartitioningFunction().
CMPASSERT(opf->castToRangePartitioningFunction());
// Clear because rewrite insists on it being so.
keyColumnList_.clear();
if (mapItUp)
{
map.rewriteValueIdListUp(keyColumnList_,
opf->castToRangePartitioningFunction()->
keyColumnList_);
}
else
{
map.rewriteValueIdListDown(opf->castToRangePartitioningFunction()->
keyColumnList_,
keyColumnList_);
}
// Clear because rewrite insists on it being so.
orderOfKeyValues_.clear();
if (mapItUp)
{
map.rewriteValueIdListUp(orderOfKeyValues_,
opf->castToRangePartitioningFunction()->
orderOfKeyValues_);
}
else
{
map.rewriteValueIdListDown(opf->castToRangePartitioningFunction()->
orderOfKeyValues_,
orderOfKeyValues_);
}
// do NOT map the originalKeyColumnList_, that's why it's called ORIGINAL
} // RangePartitioningFunction::remapIt()
// -----------------------------------------------------------------------
// RangePartitioningFunction::createPartitioningFunctionForIndexDesc()
// -----------------------------------------------------------------------
PartitioningFunction*
RangePartitioningFunction::createPartitioningFunctionForIndexDesc
(IndexDesc *idesc) const
{
const NAFileSet * fileSet = idesc->getNAFileSet();
const NAColumnArray & allColumns = fileSet->getAllColumns();
const NAColumnArray & partKeyColumns = fileSet->getPartitioningKeyColumns();
Lng32 ixColNumber;
ValueId keyValueId;
ValueIdSet partitioningKey;
ValueIdList partitioningKeyList;
ValueIdList orderOfPartKeyValues;
RangePartitionBoundaries * partBounds;
CollIndex i = 0;
for (i = 0; i < partKeyColumns.entries(); i++)
{
// which column of the index is this (usually this will be == i)
ixColNumber = allColumns.index(partKeyColumns[i]);
// insert the value id of the index column into the partitioning
// key column value id list
keyValueId = idesc->getIndexColumns()[ixColNumber];
partitioningKey += keyValueId;
partitioningKeyList.insertAt(i,keyValueId);
// insert the same value id into the order list, if the column
// is in ascending order, otherwise insert the inverse of the
// column
if (partKeyColumns.isAscending(i))
{
orderOfPartKeyValues.insert(keyValueId);
}
else
{
InverseOrder *invExpr = new(idesc->wHeap())
InverseOrder(keyValueId.getItemExpr());
invExpr->synthTypeAndValueId();
orderOfPartKeyValues.insert(invExpr->getValueId());
}
} // end loop over partitioning key columns
// -----------------------------------------------------------------
// Allocate new range partition boundaries.
// -----------------------------------------------------------------
partBounds = new(idesc->wHeap()) RangePartitionBoundaries
(*getRangePartitionBoundaries(),
idesc->wHeap());
// ---------------------------------------------------------------------
// Determine the minimum number of partitioning keys based on the
// start key values that are specified. Columns for which no explicit
// start key values are specified need not be part of the part key.
// Adjust our bookkeeping if necessary. Remove this logic if it can
// be moved to CATMAN.
// ---------------------------------------------------------------------
CollIndex numPartKeyCols = partBounds->getOptimizedNumberOfPartKeys();
if (partitioningKeyList.entries() > numPartKeyCols)
{
while (partitioningKeyList.entries() > numPartKeyCols)
{
partitioningKeyList.removeAt(numPartKeyCols);
orderOfPartKeyValues.removeAt(numPartKeyCols);
}
partitioningKey.clear();
partitioningKey.insertList(partitioningKeyList);
}
// -----------------------------------------------------------------
// now compute the encoded key length of all the boundary key columns
// (there are no fillers between the encoded key values, so it's ok
// to just add the encoded lengths of the individual columns)
// -----------------------------------------------------------------
Lng32 encodedBoundaryKeyLength = 0;
for (i = 0; i < orderOfPartKeyValues.entries(); i++)
{
encodedBoundaryKeyLength +=
partitioningKeyList[i].getType().getEncodedKeyLength();
}
// -----------------------------------------------------------------
// Add the first and the last boundary (0 and numberOfPartitions)
// at the ends that do not separate two partitions
// -----------------------------------------------------------------
partBounds->completePartitionBoundaries(
orderOfPartKeyValues,
encodedBoundaryKeyLength);
// -----------------------------------------------------------------
// Allocate a new RangePartitioningFunction.
// -----------------------------------------------------------------
RangePartitioningFunction *partFunc
= new(idesc->wHeap()) RangePartitioningFunction
(partitioningKey,
partitioningKeyList,
orderOfPartKeyValues,
partBounds,
getNodeMap()->copy(idesc->wHeap()));
partFunc->createPartitioningKeyPredicates();
return partFunc;
} // RangePartitioningFunction::createPartitioningFunctionForIndexDesc()
// -----------------------------------------------------------------------
// RangePartitioningFunction::createPartitioningExpression()
// -----------------------------------------------------------------------
ItemExpr* RangePartitioningFunction::createPartitioningExpression()
{
if (getExpression()) // already constructed?
return getExpression(); // reuse it!
// create a lookup function that takes the partitioning key and looks
// up its partition number in the array of split ranges provided by
// this partitioning function.
// first, encode all key values (for asc or desc order) and concatenate
// the encoded strings
ItemExpr *encKey = NULL;
for (CollIndex i=0; i < keyColumnList_.entries(); i++)
{
NABoolean descOrder =
(orderOfKeyValues_[i].getItemExpr()->getOperatorType() == ITM_INVERSE);
ItemExpr *dataConversionErrorFlag = getConvErrorExpr();
if (dataConversionErrorFlag == 0)
{
dataConversionErrorFlag =
new (CmpCommon::statementHeap()) HostVar(
"_sys_repartConvErrorFlg",
new (CmpCommon::statementHeap()) SQLInt(CmpCommon::statementHeap(), TRUE,FALSE),
TRUE);
storeConvErrorExpr(dataConversionErrorFlag);
}
// cast the key column to the exact type of the original key column
const NAType &oType = originalKeyColumnList_[i].getType();
ItemExpr *c = new (CmpCommon::statementHeap()) Narrow(
keyColumnList_[i].getItemExpr(),
dataConversionErrorFlag,
&oType);
// form the key encoding of the key column (a character string)
ItemExpr *e = new (CmpCommon::statementHeap()) CompEncode (c,descOrder);
// concatenate the individual key encodings of the key columns
if (encKey == NULL)
encKey = e;
else
encKey = new (CmpCommon::statementHeap()) Concat(encKey,e);
}
// give the concatenated key encodings as an input to the range lookup function
ItemExpr * partFunc = new (CmpCommon::statementHeap()) RangeLookup(
encKey,this);
partFunc->synthTypeAndValueId();
storeExpression(partFunc);
return partFunc;
}
// -----------------------------------------------------------------------
// RangePartitioningFunction::shouldUseSynchronousAccess()
// -----------------------------------------------------------------------
NABoolean RangePartitioningFunction::shouldUseSynchronousAccess(
const ReqdPhysicalProperty* rpp,
const EstLogPropSharedPtr& inputLogProp,
GroupAttributes* ga) const
{
NABoolean shouldUseSynchronousAccess = FALSE;
NABoolean synchronousAccessForced = FALSE;
NABoolean tooManyPAs = FALSE;
ValueIdList partKey = getOrderOfKeyValues();
// Remove from the partitioning key any columns that are covered by
// constants or input values.
partKey.removeCoveredExprs(ga->getCharacteristicInputs());
const LogicalPartitioningRequirement *lpr =
rpp->getLogicalPartRequirement();
// Don't do synchronous access if the user is forcing a PAPA node
// and is not forcing the number of PAs
if ((lpr != NULL) AND lpr->getMustUsePapa() AND
(lpr->getNumClientsReq() == ANY_NUMBER_OF_PARTITIONS))
return FALSE;
// See if the user is trying to force synchronous access.
if ((CmpCommon::getDefault(ATTEMPT_ASYNCHRONOUS_ACCESS) == DF_OFF) OR
((lpr != NULL) AND
(lpr->getNumClientsReq() != ANY_NUMBER_OF_PARTITIONS) AND
(lpr->getNumClientsReq() < getCountOfPartitions())))
{
shouldUseSynchronousAccess = TRUE;
synchronousAccessForced = TRUE;
}
// Synchronous access is a good idea and is ok if there is a required order
// and/or arrangement that came from an operator that is not in DP2,
// and the required order and/or arrangement columns are a leading
// prefix of the partitioning key columns, and enough data is being
// returned that the PA buffers will fill up and so the PA operators
// will block until all data from all preceding partitions (in the
// required order) has been returned. Note that whether the required
// order or arrangement is a leading prefix of the clustering key
// columns will be checked later (by the satisfied method).
// NOTE: we cannot force synchronous access if there is a required
// logical order and/or arrangement! If the code below decides that
// synchronous access is not possible, and the user is trying to
// force synchronous access, then they are out of luck!
if (rpp->getLogicalOrderOrArrangementFlag())
{
if ((rpp->getSortKey() != NULL) AND
(rpp->getArrangedCols() != NULL))
{
// We have both a required order and an arrangement.
if ((partKey.satisfiesReqdOrder(*rpp->getSortKey()) == SAME_ORDER) AND
partKey.satisfiesReqdArrangement(*rpp->getArrangedCols()))
shouldUseSynchronousAccess = TRUE;
else
shouldUseSynchronousAccess = FALSE;
}
else if (rpp->getSortKey() != NULL)
{
// Only a required order.
OrderComparison oc = partKey.satisfiesReqdOrder(*rpp->getSortKey());
NABoolean okToUseSerialOrder = FALSE;
if (CmpCommon::getDefault(ATTEMPT_REVERSE_SYNCHRONOUS_ORDER) != DF_ON)
{
okToUseSerialOrder = (oc == SAME_ORDER);
}
else
{
okToUseSerialOrder = (oc == SAME_ORDER || oc == INVERSE_ORDER);
}
if (okToUseSerialOrder)
shouldUseSynchronousAccess = TRUE;
else
shouldUseSynchronousAccess = FALSE;
}
else if (rpp->getArrangedCols() != NULL)
{
// Only a required arrangement.
if (partKey.satisfiesReqdArrangement(*rpp->getArrangedCols()))
shouldUseSynchronousAccess = TRUE;
else
shouldUseSynchronousAccess = FALSE;
}
}
if (shouldUseSynchronousAccess AND NOT synchronousAccessForced)
{
// Don't do synchronous access if there is a dp2 sort order
// partitioning requirement, as this means we want a DP2 sort
// order, and the main reason for requiring a DP2 sort order
// is to avoid having to access the data synchronously.
if (rpp->getDp2SortOrderPartReq() != NULL)
shouldUseSynchronousAccess = FALSE;
}
if (shouldUseSynchronousAccess AND NOT synchronousAccessForced)
{
// There was a required order or arrangement that could potentially
// need us to access the data synchronously. If synch. access is not
// forced, see if it is a good idea or not.
// get the maximum number of PAs per process that can be allowed.
Int32 maxPAsPerProcess =
(Int32) CmpCommon::getDefaultNumeric(MAX_ACCESS_NODES_PER_ESP);
if (getCountOfPartitions() > maxPAsPerProcess)
{
// Get the logical part requirement, if one exists.
PartitioningRequirement *logPartReq = NULL;
PartitioningFunction *requiredPartFunc = NULL;
if (lpr != NULL)
{
logPartReq = lpr->getLogReq();
if (logPartReq AND logPartReq->
castToFullySpecifiedPartitioningRequirement())
requiredPartFunc = logPartReq->
castToFullySpecifiedPartitioningRequirement()->
getPartitioningFunction();
}
// Determine if we might exceed the maximum number of PAs allowed
// to a process. If we will, then we definitely want to use
// synchronous access if there is an order to preserve.
// The first check is the most restrictive - this is for logical
// subpartitioning. If we do logical subpartitioning, it is possible
// one process will have to access all the partitions, and there
// is nothing we can do about it later.
if (logPartReq != NULL AND
requiredPartFunc AND
requiredPartFunc->castToLogPhysPartitioningFunction() AND
requiredPartFunc->castToLogPhysPartitioningFunction()->
getLogPartType() ==
LogPhysPartitioningFunction::LOGICAL_SUBPARTITIONING)
tooManyPAs = TRUE;
else if ((logPartReq != NULL) AND
(getCountOfPartitions() >
(logPartReq->getCountOfPartitions() * maxPAsPerProcess)))
tooManyPAs = TRUE;
else if ((getCountOfPartitions() >
(rpp->getCountOfPipelines() * maxPAsPerProcess)))
tooManyPAs = TRUE;
}
if (NOT tooManyPAs)
{
// Now check to see how much data will be returned. If it is less
// than twice the amount of data that it takes to fill up all
// the PA buffers, than we will be able to access the majority
// of the data asynchronously before the buffers fill up. So,
// in this case assume that it is not a good idea to access the
// data synchronously. Otherwise, synchronous access is a good
// idea (if it passed all the other checks) because we will be
// accessing the majority of the data after the PA buffers fill up
// and block, and so we will end up accessing the data synchronously
// anyway, so we would rather not pay for all those PA buffers.
// Compute size of data returned in KB. Allocate 16 bytes per
// record for overhead.
CostScalar totalsize =
(ga->outputLogProp(inputLogProp)->getResultCardinality() *
(ga->getRecordLength() + 16.0));
// Compute the size of all the PA buffers.
double buffersize = CmpCommon::getDefaultNumeric(GEN_PA_BUFFER_SIZE);
// A PA buffer can not be larger than 56K. If the table has remote
// partitions, then the actual limit is 31K, so we really should
// check for remote partitions. That would require the index
// descriptor, which we don't have.
buffersize = MINOF (buffersize, 56000);
// number of PA output buffers is always at least one less than the
// total number of pa buffers, as at least one is reserved for input.
double numOutputBuffers =
MAXOF(CmpCommon::getDefaultNumeric(GEN_PA_NUM_BUFFERS) - 1,1);
// Subtract 2000 bytes per buffer for headers.
buffersize = (buffersize * numOutputBuffers) -
(2000 * numOutputBuffers);
if (totalsize < (getCountOfPartitions() * 2.0 * buffersize))
shouldUseSynchronousAccess = FALSE;
}
} // end if order or arrangement that could be preserved with synch. access
return shouldUseSynchronousAccess;
}
SearchKey *
RangePartitioningFunction::createSearchKey(const IndexDesc *indexDesc,
ValueIdSet availInputs,
ValueIdSet additionalPreds) const
{
ValueIdSet preds(getPartitioningKeyPredicates());
ValueIdSet nonKeyColumnSet; // empty set
availInputs += getPartitionInputValues();
preds += additionalPreds;
// make a search key from all that
SearchKey *partSearchKey = new (CmpCommon::statementHeap())
SearchKey(
indexDesc->getPartitioningKey(),
indexDesc->getOrderOfPartitioningKeyValues(),
availInputs,
TRUE, // there isn't really a scan direction for partKey
preds,
this,
nonKeyColumnSet,
indexDesc);
return partSearchKey;
}
/////////////////////////////////////////////////////////////
// Test condition C2.2 (co-partitioned), which is defined as:
// 1. All tables share the same number of partitioned key columns;
// 2. Corresponding partition key columns across all tables are
// of same type;
// 3. All tables should be partitioned into the same number of ranges;
// 4. Corresponding partitions are bounded by identical upper and
// lower boundaries and reside on the same DP2.
/////////////////////////////////////////////////////////////
NABoolean
RangePartitioningFunction::partFuncAndFuncPushDownCompatible(
const PartitioningFunction& x) const
{
const RangePartitioningFunction* other =
x.castToRangePartitioningFunction();
if ( other == NULL ) return FALSE;
Lng32 thisKeyCount = getPartitioningKey().entries();
Lng32 otherKeyCount = other->getPartitioningKey().entries();
// same key count
if (thisKeyCount != otherKeyCount)
return FALSE;
Lng32 thisPartCount = getCountOfPartitions();
Lng32 otherPartCount = other->getCountOfPartitions();
// same part count
if (thisPartCount != otherPartCount)
return FALSE;
// same part column type
for (Lng32 index = 0; index < thisKeyCount; index++)
{
if ( NOT getKeyColumnList()[index].getType().isCompatible
(
other->getKeyColumnList()[index].getType()
)
)
return FALSE;
}
// same boundaries
if ( getRangePartitionBoundaries()->
compareRangePartitionBoundaries
(*(other->getRangePartitionBoundaries())) == FALSE
)
return FALSE;
return TRUE;
}
// -----------------------------------------------------------------------
// RangePartitioningFunction::preCodeGen()
// Rewrite the partitioning keys of the partitioning function that
// are expressed using VEGReferences in terms of the available values.
// -----------------------------------------------------------------------
void RangePartitioningFunction::preCodeGen(const ValueIdSet& availableValues)
{
ValueIdSet noExternalInputs;
PartitioningFunction::preCodeGen(availableValues);
keyColumnList_.replaceVEGExpressions(availableValues, noExternalInputs,
FALSE, NULL, TRUE);
} // RangePartitioningFunction::preCodeGen()
void RangePartitioningFunction::setupForStatement()
{
if(setupForStatement_)
return;
PartitioningFunction::setupForStatement();
if(partitionBoundaries_)
partitionBoundaries_->setupForStatement(TRUE /* use string */);
setupForStatement_ = TRUE;
resetAfterStatement_ = FALSE;
}
void RangePartitioningFunction::resetAfterStatement()
{
if(resetAfterStatement_)
return;
PartitioningFunction::resetAfterStatement();
keyColumnList_.clear();
orderOfKeyValues_.clear();
originalKeyColumnList_.clear();
if(partitionBoundaries_)
partitionBoundaries_->resetAfterStatement();
setupForStatement_ = FALSE;
resetAfterStatement_ = TRUE;
}
// -----------------------------------------------------------------------
// Methods for debugging.
// -----------------------------------------------------------------------
const NAString RangePartitioningFunction::getText() const
{
NAString result("range partitioned ", CmpCommon::statementHeap());
char nparts[20];
sprintf(nparts,"%d",getCountOfPartitions());
result += nparts;
result += " ways on (";
orderOfKeyValues_.unparse(result,DEFAULT_PHASE,EXPLAIN_FORMAT);
result += ")";
if ( partitionBoundaries_ ) {
result += " with boundaries(";
for (Int32 index = 0; index < partitionBoundaries_->getCountOfPartitions(); index++)
{
const ItemExprList* iel = partitionBoundaries_->getBoundaryValues(index);
for (Int32 j=0; j<iel->entries(); j++) {
ItemExpr* ie = (*iel)[j];
if ( ie->getOperatorType() == ITM_CONSTANT ) {
ConstValue* cv = (ConstValue*)ie;
result += "c(";
result += cv->getTextForQuery(QUERY_FORMAT);
result += ")";
} else {
result += "nc(";
result += ie->getText();
result += ")";
}
result += " ";
}
if ( index < partitionBoundaries_->getCountOfPartitions()-1 )
result += ";";
}
result += ")";
/* enable this for debugging of binary key problems
result += " binary (";
Lng32 encodedBoundaryKeyLength = partitionBoundaries_->getEncodedBoundaryKeyLength();
char hexDigits[4];
for (Int32 index2 = 0; index2 < partitionBoundaries_->getCountOfPartitions(); index2++)
{
const char * binaryVal = partitionBoundaries_->getBinaryBoundaryValue(index2);
if (binaryVal)
{
if (index2 > 0)
result += ", ";
result += "b(0x";
for (int b=0; b<encodedBoundaryKeyLength; b++)
{
snprintf(hexDigits, sizeof(hexDigits), "%02hhx", binaryVal[b]);
result += hexDigits;
}
result += ")";
}
}
result += ")";
end of code for binary keys */
}
return result;
}
void RangePartitioningFunction::print(FILE* ofd, const char* indent,
const char* title) const
{
PartitioningFunction::print(ofd, indent, "RangePartitioningFunction");
if (orderOfKeyValues_.entries() > 0)
{
fprintf(ofd,"Ascending/descending order :\n");
for (CollIndex index = 0; index < orderOfKeyValues_.entries(); index++)
{
if (orderOfKeyValues_[index].getItemExpr()->getOperatorType()
== ITM_INVERSE)
fprintf(ofd,"D ");
else
fprintf(ofd,"A ");
if (((index/9)* 9 == index) OR (index == orderOfKeyValues_.entries()-1))
fprintf(ofd,"\n");
}
}
partitionBoundaries_->print(ofd, indent, title);
} // RangePartitioningFunction::print()
NABoolean
compareEncodedKey(const char* low, const char* key, const char* high, Int32 keyLen, NABoolean checkLast)
{
Int32 cmpLow = memcmp(low, key, keyLen);
Int32 cmpHigh = memcmp(key, high, keyLen);
if ( cmpLow <= 0 && cmpHigh < 0 )
return TRUE;
return (checkLast && cmpLow <= 0 && cmpHigh <= 0);
}
NABoolean
compareAsciiKey(const char* low, const char* key, const char* high, Int32, NABoolean checkLast)
{
Int32 cmpLow = strverscmp(low, key);
Int32 cmpHigh = strverscmp(key, high);
if ( cmpLow <= 0 && cmpHigh < 0 )
return TRUE;
return (checkLast && cmpLow <= 0 && cmpHigh <= 0);
}
// find a boundary pair [low, high) with smallest low value in which keys fall, and return the
// index of the boundary low. Return -1 otherwise, or the key lengths are different.
Int32 RangePartitionBoundaries::findBeginBoundary(char* encodedKey, Int32 keyLen,
compFuncPtrT compFunc) const
{
// boundaries are stored in entries in the range [0, partitionCount_]
for (Lng32 i=0; i<=partitionCount_-1; i++ ) {
const char* low = getBinaryBoundaryValue(i);
const char* high = getBinaryBoundaryValue(i+1);
// test if encodedKey is in [low, high)
if ( (*compFunc)(low, encodedKey, high, keyLen, i==partitionCount_-1) )
return i;
}
return -1;
}
// find a boundary pair [low, high) with the largest low value in which keys fall, and return the
// index of the boundary low. Return -1 otherwise, or the key lengths are different.
Int32 RangePartitionBoundaries::findEndBoundary(char* encodedKey, Int32 keyLen,
compFuncPtrT compFunc) const
{
// boundaries are stored in entries in the range [0, partitionCount_]
for (Lng32 i=partitionCount_-1; i>= 0; i--) {
const char* low = getBinaryBoundaryValue(i);
const char* high = getBinaryBoundaryValue(i+1);
// test if encodedKey is in [low, high)
if ( (*compFunc)(low, encodedKey, high, keyLen, i==partitionCount_-1) )
return i;
}
return -1;
}
Int32
RangePartitioningFunction::computeNumOfActivePartitions(SearchKey* skey, const TableDesc* tDesc) const
{
const RangePartitionBoundaries* boundaries = getRangePartitionBoundaries();
Int32 origPartitions = getCountOfPartitions();
Int32 partitions = origPartitions;
Int32 bIndex = 0;
const NATable* naTable = tDesc->getNATable();
if ( naTable->isHiveTable() )
return origPartitions;
NABoolean isNativeHbase = (naTable->isHbaseCellTable() || naTable->isHbaseRowTable());
compFuncPtrT compFuncPtr = ( isNativeHbase ) ? compareAsciiKey: compareEncodedKey;
char* buf = NULL;
Int32 len = 0;
const ValueIdList& beginKey = skey->getBeginKeyValues();
if ( beginKey.computeEncodedKey(tDesc, FALSE, buf, len) ) {
bIndex = boundaries->findBeginBoundary(buf, len, compFuncPtr);
if ( bIndex < 0 )
return origPartitions; // error in deciding the partiton
else
partitions -= bIndex; // bIndex is 0 based.
}
const ValueIdList& endKey = skey->getEndKeyValues();
if ( endKey.computeEncodedKey(tDesc, TRUE, buf, len) ) {
Int32 eIndex = boundaries->findEndBoundary(buf, len, compFuncPtr);
if ( eIndex < 0 ) // error in deciding the partition.
return origPartitions;
if ( eIndex >= bIndex ) //eIndex is also 0 based
partitions -= (getCountOfPartitions() - eIndex - 1);
else
return origPartitions; // end partition is preceeding the start partition!
}
return partitions;
}
// ***********************************************************************
// LogPhysPartitioningFunction
// ***********************************************************************
LogPhysPartitioningFunction::LogPhysPartitioningFunction(
PartitioningFunction * logPartFunc,
PartitioningFunction * physPartFunc,
logPartType logPartType,
Lng32 numOfClients,
NABoolean usePapa,
NABoolean synchronousAccess,
NAMemory* heap)
: PartitioningFunction(LOGPHYS_PARTITIONING_FUNCTION,heap),
logPartFunc_(logPartFunc),physPartFunc_(physPartFunc),
realPartFunc_(NULL),logPartType_(logPartType),
numOfClients_(numOfClients),usePapa_(usePapa),
synchronousAccess_(synchronousAccess)
{
ValueIdSet partKey(logPartFunc->getPartitioningKey());
partKey += physPartFunc->getPartitioningKey();
setPartKey(partKey);
}
LogPhysPartitioningFunction::~LogPhysPartitioningFunction()
{}
const LogPhysPartitioningFunction *
LogPhysPartitioningFunction::castToLogPhysPartitioningFunction() const
{
return this;
}
Lng32 LogPhysPartitioningFunction::getCountOfPartitions() const
{
return physPartFunc_->getCountOfPartitions();
}
PartitioningRequirement*
LogPhysPartitioningFunction::makePartitioningRequirement()
{
CMPASSERT(physPartFunc_);
return physPartFunc_ -> makePartitioningRequirement();
}
// -----------------------------------------------------------------------
// LogPhysPartitioningFunction::getNodeMap()
// Return appropriate node map from underlying partitioning function.
// -----------------------------------------------------------------------
const NodeMap*
LogPhysPartitioningFunction::getNodeMap() const
{
// --------------------------------------------------------------------------
// If no "real" partitioning function exists, extract node map from
// underlying physical partitioning function; otherwise extract node map from
// "real" partitioning function.
// --------------------------------------------------------------------------
if (realPartFunc_ == NULL)
{
return physPartFunc_->getNodeMap();
}
else
{
return realPartFunc_->getNodeMap();
}
} // PartitioningFunction::getNodeMap()
// get any existing (logical or physical) nodemap (or synthesize one) that
// matches logPartFunc_'s partition count requirement
NodeMap* LogPhysPartitioningFunction::getOrMakeSuitableNodeMap
(NABoolean forESP) const
{
// we are the LogPhysPartitioningFunction of the synthesized physical
// property of an Exchange's child (that executes in DP2).
// if logical partn func's nodemap entry count == log partn func partn count
// then return logical partn func's nodemap
NodeMap *nodemap = (NodeMap*)logPartFunc_->getNodeMap();
ULng32 partCnt=(ULng32)logPartFunc_->getCountOfPartitions();
if (nodemap && nodemap->getNumEntries() == partCnt) {
// the logical partitioning of the result already has a suitable nodemap.
// fall thru and use it.
}
else {
// logical partitioning function has no nodemap. try to find one in the
// physical partitioning function of the child.
// if phys partn func's nodemap entry count == log partn func partn count
// then return phys partn func's nodemap.
nodemap = (NodeMap*)physPartFunc_->getNodeMap();
if (nodemap && nodemap->getNumEntries() == partCnt) {
// this can happen in "select * from t035t6;" and other queries against
// vertically partitioned tables like t035t6 (see regress/core/test035).
// fall thru and use it.
}
else {
// only as a last resort do we synthesize a new nodemap
nodemap = nodemap->synthesizeLogicalMap(partCnt, forESP);
}
}
if (logPartFunc_->castToReplicateNoBroadcastPartitioningFunction()) {
for(CollIndex i = 0; i < partCnt; i++) {
nodemap->setPartitionState(i, NodeMapEntry::ACTIVE);
}
}
return nodemap;
}
PartitioningFunction* LogPhysPartitioningFunction::copy() const
{
return new(CmpCommon::statementHeap()) LogPhysPartitioningFunction(*this);
}
NABoolean
LogPhysPartitioningFunction::canProducePartitioningKeyPredicates() const
{
// this is an exception to the rule: for a LogPhysPartitioningFunction
// we assume that there is no need to enforce the physical partitioning
// key predicates, since each DP2 partition has only its own data
// anyway. Therefore, for the purpose of evaluating partitioning key
// predicates and partition input values, we deal with the logical
// partitioning function only, and only in the case where the PA
// node isn't doing the grouping.
// we can do this if we can produce the "real" partitioning function
// and if the real one can produce part key preds
return (logPartFunc_->canProducePartitioningKeyPredicates() OR
logPartType_ == PA_PARTITION_GROUPING OR
logPartType_ == PA_GROUPED_REPARTITIONING);
}
void LogPhysPartitioningFunction::createPartitioningKeyPredicates()
{
CMPASSERT(logPartFunc_->canProducePartitioningKeyPredicates());
// see comment in canProducePartitioningKeyPredicates() above
storePartitioningKeyPredicates(logPartFunc_->getPartitioningKeyPredicates());
storePartitionInputValues(logPartFunc_->getPartitionInputValuesLayout());
}
// -----------------------------------------------------------------------
// LogPhysPartitioningFunction::replacePivs()
// -----------------------------------------------------------------------
void LogPhysPartitioningFunction::replacePivs(
const ValueIdList& newPivs,
const ValueIdSet& newPartKeyPreds)
{
// Overwrite the old pivs, part key preds, and part expr. with the new ones.
storePartitionInputValues(newPivs);
storePartitioningKeyPredicates(newPartKeyPreds);
// Also overwrite them in the logical partitioning function.
logPartFunc_->replacePivs(newPivs,newPartKeyPreds);
} // LogPhysPartitioningFunction::replacePivs()
ItemExpr* LogPhysPartitioningFunction::createPartitioningExpression()
{
CMPASSERT(0); // should never reach here, it's not legal to do a
// repartitioning for a logphys part func
return NULL;
}
PartitioningFunction*
LogPhysPartitioningFunction::createRealPartitioningFunction()
{
if (realPartFunc_)
return realPartFunc_;
switch (logPartType_)
{
case PA_PARTITION_GROUPING:
case PA_GROUPED_REPARTITIONING:
// for partition grouping or repartitioning we never split any
// physical partitions and we are processing only physical
// partitions at this point (logical partitions come into the
// picture once we reach the DP2 exchange)
realPartFunc_ = physPartFunc_;
break;
case LOGICAL_SUBPARTITIONING:
{
// Merge the split ranges of logical and physical partitioning
// functions, so we can count the exact number of partitions.
const RangePartitioningFunction *lpf =
logPartFunc_->castToRangePartitioningFunction();
const RangePartitioningFunction *ppf =
physPartFunc_->castToRangePartitioningFunction();
// Log and phys partitioning functions must have the same key
// to be able to produce a common function
if (lpf AND ppf AND
lpf->getOrderOfKeyValues() == ppf->getOrderOfKeyValues())
{
// Make a new RangePartitioningFunction object with the combined
// split boundaries of the two. Also produce a corresponding
// node map for the new boundaries.
NodeMap *mergedMap = new (CmpCommon::statementHeap())
NodeMap(CmpCommon::statementHeap());
RangePartitionBoundaries *rb =
ppf->getRangePartitionBoundaries()->merge(
*lpf->getRangePartitionBoundaries(),
*ppf->getNodeMap(),
*mergedMap);
// return the merged partitioning function if we could merge
if (rb)
realPartFunc_ = new(CmpCommon::statementHeap())
RangePartitioningFunction(
ppf->getPartitioningKey(),
ppf->getKeyColumnList(),
ppf->getOrderOfKeyValues(),
rb,
mergedMap );
}
else
{
// Special case of single partition physical partitioning function.
if (physPartFunc_->castToSinglePartitionPartitioningFunction())
{
realPartFunc_ = logPartFunc_;
// Create a new node map.
NodeMap *nodeMap = new(CmpCommon::statementHeap())
NodeMap(CmpCommon::statementHeap());
// Add a copy of the single physical node map entry for
// each partition in the logical node map.
const NodeMapEntry *entry = physPartFunc_->getNodeMapEntry(0);
for (CollIndex nodeIdx = 0;
nodeIdx < (CollIndex) logPartFunc_->getCountOfPartitions();
nodeIdx++)
{
nodeMap->setNodeMapEntry(nodeIdx,
*entry,
CmpCommon::statementHeap());
}
realPartFunc_->replaceNodeMap(nodeMap);
}
}
}
break;
default:
break;
}
return realPartFunc_; // is NULL if we aren't smart enough to do this
}
COMPARE_RESULT
LogPhysPartitioningFunction::comparePartFuncToFunc
(const PartitioningFunction &other) const
{
COMPARE_RESULT c = PartitioningFunction::comparePartFuncToFunc(other);
if (c != SAME)
return INCOMPATIBLE;
const LogPhysPartitioningFunction &lppfOther =
(const LogPhysPartitioningFunction &) other;
if (getLogPartType() != lppfOther.getLogPartType())
return INCOMPATIBLE;
if (getNumOfClients() != lppfOther.getNumOfClients())
return INCOMPATIBLE;
if (getUsePapa() != lppfOther.getUsePapa())
return INCOMPATIBLE;
if (getSynchronousAccess() != lppfOther.getSynchronousAccess())
return INCOMPATIBLE;
if (getLogPartitioningFunction()->comparePartFuncToFunc(
*lppfOther.getLogPartitioningFunction()) != SAME)
return INCOMPATIBLE;
if (getPhysPartitioningFunction()->comparePartFuncToFunc(
*lppfOther.getPhysPartitioningFunction()) != SAME)
return INCOMPATIBLE;
return SAME;
}
NABoolean LogPhysPartitioningFunction::isAGroupingOf(
const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = 1;
// Handle the trivial case of two functions that are equal. The only
// other way to check for a grouping is to actually form the "real"
// partitioning function and use it for the check.
return (comparePartFuncToFunc(other) == SAME OR
(realPartFunc_ AND
realPartFunc_->isAGroupingOf(other,maxPartsPerGroup)));
}
// -----------------------------------------------------------------------
// LogPhysPartitioningFunction::remapIt()
// -----------------------------------------------------------------------
void LogPhysPartitioningFunction::remapIt
(const PartitioningFunction* opf,
ValueIdMap& map, NABoolean mapItUp)
{
// Currently, the only consumer of this method is
// MapValueIds::synthPhysicalProperty, when it executes in DP2
// Map myself
PartitioningFunction::remapIt(opf,map,mapItUp);
// Map the logical function
logPartFunc_ = logPartFunc_->copyAndRemap(map,mapItUp);
// Map the physical function
physPartFunc_ = physPartFunc_->copyAndRemap(map,mapItUp);
} // LogPhysPartitioningFunction::remapIt()
// LogPhysPartitioningFunction::canMaintainSortOrder()
// Can this logPhys partitioning function maintain the order of an
// individual partition of the physical partitioning function. In
// order to maintain the order, a merge expression may be required.
//
NABoolean
LogPhysPartitioningFunction::canMaintainSortOrder(const ValueIdList& sortOrder)
const
{
// The sortOrder is the sort order of an individual partition of the
// physical partitioning function.
//
CollIndex numSortKeys = sortOrder.entries();
// We can always maintain, no sort order.
//
if(numSortKeys == 0) {
return TRUE;
}
// For now, only concerned with table/index partitioning functions.
// (Single, Range or TableHash)
//
const SinglePartitionPartitioningFunction *sppf =
physPartFunc_->castToSinglePartitionPartitioningFunction();
// A single partition can always maintain its order.
//
if(sppf) {
return TRUE;
}
const RangePartitioningFunction *rpf =
physPartFunc_->castToRangePartitioningFunction();
const TableHashPartitioningFunction *hdpf =
physPartFunc_->castToTableHashPartitioningFunction();
CMPASSERT(rpf || hdpf);
if(!getSynchronousAccess()) {
// Can always merge the streams, even if (in the case of
// non-decoupled range) this results in serial access to the
// partitions.
//
return TRUE;
}
if(rpf) {
// For range partitioning, if the partitioning keys are a prefix
// of the clustering key, then in order to maintain the sort order
// the partitions must be accessed synchronously.
//
// If the partitioning keys are not a prefix of the clustering
// key, then the partitions can be merged if the partitions are
// accessed asynchronously. (This case is handled above)
//
// Determine if the sortkey is a prefix of the partitioning key or
// if the partitioning key is a prefix of the sortkey. If either
// of these is true, the sort order can be maintained if the
// partitions are accessed synchronously.
//
const ValueIdList &partKeyCols = rpf->getOrderOfKeyValues();
// This logic may have to change when we become more sophisticated
// in our sort keys and sort requirements. For example, if the
// sort req is (a/2) and the sort key is a
//
if (partKeyCols.entries() <= numSortKeys) {
OrderComparison oc= sortOrder.satisfiesReqdOrder(partKeyCols);
if(oc == SAME_ORDER ||
(CmpCommon::getDefault(ATTEMPT_REVERSE_SYNCHRONOUS_ORDER) == DF_ON &&
oc == INVERSE_ORDER)
)
{
// Range, non-decoupled, sort order == partition order,
// synchronous access: therefore can maintain order.
//
return TRUE;
}
else {
// Not a prefix, so in order to maintain order, must be
// accessing the partitions asynchronously so that a merge
// expression can be used.
//
return FALSE;
}
} else {
if(partKeyCols.satisfiesReqdOrder(sortOrder) == SAME_ORDER) {
// Range, non-decoupled, sort order == partition order,
// synchronous access: therefore can maintain order.
//
return TRUE;
} else {
// Not a prefix, so in order to maintain order, must be
// accessing the partitions asynchronously so that a merge
// expression can be used.
//
return FALSE;
}
}
} else {
// For Hash, synchronous (and all other partitioning functions)
// cannot maintain order.
//
return FALSE;
}
} // LogPhysPartitioningFunction::canMaintainSortOrder()
const NAString LogPhysPartitioningFunction::getText() const
{
NAString result("logphys partitioned(", CmpCommon::statementHeap());
switch (logPartType_)
{
case PA_PARTITION_GROUPING:
result += "grouping";
break;
case LOGICAL_SUBPARTITIONING:
result += "subpartitioning";
break;
case HORIZONTAL_PARTITION_SLICING:
result += "horizontal slicing";
break;
case PA_GROUPED_REPARTITIONING:
result += "repartitioning";
break;
default:
result += "???";
break;
}
if (usePapa_)
{
char numPAs[20];
result += ", PAPA with ";
sprintf(numPAs,"%d",numOfClients_);
result += numPAs;
result += " PA(s)";
}
result += ", log=" + logPartFunc_->getText() +
", phys=" + physPartFunc_->getText() + ")";
return result;
}
const NAString LogPhysPartitioningFunction::getLogForSplitTop() const
{
NAString result("", CmpCommon::statementHeap());
switch (logPartType_)
{
case PA_PARTITION_GROUPING:
//whenever the logical partitioning function is replicate no broadcast
//it is not truly the case of grouping
if (logPartFunc_->isAReplicateNoBroadcastPartitioningFunction())
result += "no grouping, ";
else {
char numParts[20];
result += "grouped ";
sprintf(numParts, "%d to %d,", logPartFunc_->getCountOfPartitions(), physPartFunc_->getCountOfPartitions());
result += numParts;
}
break;
case LOGICAL_SUBPARTITIONING:
result += "subpartitioned, ";
break;
case HORIZONTAL_PARTITION_SLICING:
result += "horizontally sliced, ";
break;
case PA_GROUPED_REPARTITIONING:
result += "repartitioned, ";
break;
default:
result += "???, ";
break;
}
if (usePapa_)
{
char numPAs[20];
result += "PAPA with ";
sprintf(numPAs,"%d",numOfClients_);
result += numPAs;
result += " PA(s)";
}
result += ", " + logPartFunc_->getText();
return result;
}
const NAString LogPhysPartitioningFunction::getPhysForSplitTop() const
{
return physPartFunc_->getText();
}
void LogPhysPartitioningFunction::print(
FILE* ofd,
const char* indent,
const char* title) const
{
PartitioningFunction::print(ofd,indent,"LogPhysPartitioningFunction");
logPartFunc_->print(ofd,"logical: ");
physPartFunc_->print(ofd,"physical: ");
}
// ***********************************************************************
// RoundRobinPartitioningFunction
// ***********************************************************************
PartitioningRequirement*
RoundRobinPartitioningFunction::makePartitioningRequirement()
{
return new (CmpCommon::statementHeap())
RequireRoundRobin(this);
}
PartitioningFunction*
RoundRobinPartitioningFunction::copy() const
{
return new(CmpCommon::statementHeap()) RoundRobinPartitioningFunction(*this);
}
// -----------------------------------------------------------------------
// RoundRobinPartitioningFunction::comparePartFuncToFunc(): Compare
// this partitioning function to another round robin function. To be
// 'SAME' must have the same number and order of partitioning key
// columns and have the same number of partitions (scaled and
// original).
// -----------------------------------------------------------------------
COMPARE_RESULT
RoundRobinPartitioningFunction::
comparePartFuncToFunc(const PartitioningFunction &other) const
{
COMPARE_RESULT c = PartitioningFunction::comparePartFuncToFunc(other);
if (c != SAME)
return INCOMPATIBLE;
const RoundRobinPartitioningFunction *oth =
other.castToRoundRobinPartitioningFunction();
// Since they compared 'SAME', oth should always exist, so this
// test is redundant.
//
if(!oth)
return INCOMPATIBLE;
// They must be based on the same physical partitioning.
//
if (getCountOfOrigRRPartitions() != oth->getCountOfOrigRRPartitions())
return INCOMPATIBLE;
return SAME;
} // RoundRobinPartitioningFunction::comparePartFuncToFunc()
// -----------------------------------------------------------------------
// RoundRobinPartitioningFunction::createPartitioningKeyPredicates()
// -----------------------------------------------------------------------
void RoundRobinPartitioningFunction::createPartitioningKeyPredicates()
{
if (NOT partKeyPredsCreated())
{
// Create the partition input values.
//
ItemExpr *loPart = new (CmpCommon::statementHeap())
HostVar("_sys_HostVarLoRoundRobinPart",
new (CmpCommon::statementHeap()) SQLInt(CmpCommon::statementHeap(), FALSE, FALSE),
TRUE);
ItemExpr *hiPart = new (CmpCommon::statementHeap())
HostVar("_sys_HostVarHiRoundRobinPart",
new (CmpCommon::statementHeap()) SQLInt(CmpCommon::statementHeap(), FALSE, FALSE),
TRUE);
loPart->synthTypeAndValueId();
hiPart->synthTypeAndValueId();
ValueIdSet setOfKeyPredicates;
ValueIdList partInputValues;
// -----------------------------------------------------------------
// The partitioning key predicate is never used for a round
// robin partitioning so none is generated.
// -----------------------------------------------------------------
// the partition input values are two integer values: lo and hi part #
partInputValues.insert(loPart->getValueId());
partInputValues.insert(hiPart->getValueId());
// Store the empty set of key predicate. This will set the
// boolean 'partKeyPredsCreated' to TRUE.
//
storePartitioningKeyPredicates(setOfKeyPredicates);
// Store the partition input values.
//
storePartitionInputValues(partInputValues);
}
} // RoundRobinPartitioningFunction::createPartitioningKeyPredicates()
// -----------------------------------------------------------------------
// RoundRobinPartitioningFunction::replacePivs()
// -----------------------------------------------------------------------
void RoundRobinPartitioningFunction::replacePivs(
const ValueIdList& newPivs,
const ValueIdSet& newPartKeyPreds)
{
// Overwrite the old pivs, part key preds, and part expr. with the new ones.
storePartitionInputValues(newPivs);
storePartitioningKeyPredicates(newPartKeyPreds);
} // RoundRobinPartitioningFunction::replacePivs()
// -----------------------------------------------------------------------
// RoundRobinPartitioningFunction::createPartitioningFunctionForIndexDesc()
// -----------------------------------------------------------------------
PartitioningFunction*
RoundRobinPartitioningFunction::
createPartitioningFunctionForIndexDesc(IndexDesc *idesc) const
{
const NAFileSet * fileSet = idesc->getNAFileSet();
const NAColumnArray & allColumns = fileSet->getAllColumns();
const NAColumnArray & partKeyColumns = fileSet->getPartitioningKeyColumns();
Lng32 ixColNumber;
ValueId keyValueId;
ValueIdSet partitioningKey;
for (CollIndex i = 0; i < partKeyColumns.entries(); i++)
{
// which column of the index is this (usually this will be == i)
ixColNumber = allColumns.index(partKeyColumns[i]);
// insert the value id of the index column into the partitioning
// key column value id set
keyValueId = idesc->getIndexColumns()[ixColNumber];
partitioningKey += keyValueId;
} // end loop over partitioning key columns
// -----------------------------------------------------------------
// Allocate a new RoundRobinPartitioningFunction.
// -----------------------------------------------------------------
RoundRobinPartitioningFunction *partFunc
= new(idesc->wHeap()) RoundRobinPartitioningFunction
(getCountOfPartitions(),
partitioningKey,
getNodeMap()->copy(idesc->wHeap()));
partFunc->createPartitioningKeyPredicates();
return partFunc;
} // RoundRobinPartitioningFunction::createPartitioningFunctionForIndexDesc()
// -----------------------------------------------------------------------
// RoundRobinPartitioningFunction::createPartitioningExpression()
// -----------------------------------------------------------------------
ItemExpr* RoundRobinPartitioningFunction::createPartitioningExpression()
{
if (getExpression()) // already constructed?
return getExpression(); // reuse it!
CollHeap *heap = CmpCommon::statementHeap();
// The Partitioning Key (SYSKEY)
//
ValueIdList keyValue = getPartitioningKey();
// The type of the partitioning key for round robin is always
// SQLLargeInt (the type of SYSKEY)
//
NAType *desiredType = new (heap) SQLLargeInt(heap, TRUE, FALSE);
// The layout of the SYSKEY is
//
ItemExpr *partKey =
new (heap)
Cast(new (heap)
Shift(ITM_SHIFT_RIGHT,
new (heap)
Cast(keyValue[0].getItemExpr(),
desiredType),
new (heap)
ConstValue(32)),
new (heap) SQLInt(heap, FALSE,FALSE));
NAType *numPartsType = new (heap) SQLInt(heap, FALSE,FALSE);
Lng32 numParts = getCountOfOrigRRPartitions();
char buffer[20];
sprintf(buffer, "%d", numParts);
NAString numPartsStr = buffer;
ItemExpr *origNumParts =
new (heap) ConstValue(numPartsType, &numParts, sizeof(Lng32), &numPartsStr);
numParts = getCountOfPartitions();
sprintf(buffer, "%d", numParts);
numPartsStr = buffer;
ItemExpr *scaledNumParts =
new (heap) ConstValue(numPartsType, &numParts, sizeof(Lng32), &numPartsStr);
ItemExpr *partFunc =
new (heap)
PAGroup(new (heap)
ProgDistrib(partKey, origNumParts),
scaledNumParts,
origNumParts);
partFunc->synthTypeAndValueId();
storeExpression(partFunc);
return partFunc;
} // RoundRobinPartitioningFunction::createPartitioningExpression()
// -----------------------------------------------------------------------
// RoundRobinPartitioningFunction::createPartitionSelectionExpr()
// Create the partition selection expression. 'Partition selection'
// means that an expression is used to determine the partition to
// access as opposed to using the File System to determine the range
// of partitions to access based on a set of partitioning key
// predicates. Partition selection is currently used for Hash Dist and
// Round Robin Partitioning. And the File System is used for Range
// Partitioning. If a partitioning selection expression is created,
// it is cached in the data member 'partitionSelectionExpr_' and the
// partition selection inputs are generated and stored in
// 'partitionSelectionExprInputs_'. This method is redefined for
// HashDistPartitioningFunction and RoundRobinPartitioningFunction.
// -----------------------------------------------------------------------
ItemExpr *
RoundRobinPartitioningFunction::
createPartitionSelectionExpr(const SearchKey *partSearchKey,
const ValueIdSet &availableValues)
{
// If it has already been created, return cached version.
//
if(partitionSelectionExpr())
return partitionSelectionExpr();
CollHeap *heap = CmpCommon::statementHeap();
// Use a host var to provide access to numParts, this will be
// mapped to a specific ATP:ATPIndex:offset in PartitionAccess::codeGen()
//
ItemExpr *numParts = new (heap) HostVar("_sys_hostVarNumParts",
// int not null
new (heap) SQLInt(heap, FALSE, FALSE),
// is system-supplied
TRUE);
numParts->synthTypeAndValueId();
// Use a host var to provide access to partNum, this will be
// mapped to a specific ATP:ATPIndex:offset in PartitionAccess::codeGen()
//
ItemExpr *partNum = new (heap) HostVar("_sys_hostVarPartNo",
// int not null
new (heap) SQLInt(heap, FALSE, FALSE),
// is system-supplied
TRUE);
partNum->synthTypeAndValueId();
// Record these hostvars as the inputs to the partitionSelectionExpr.
//
CMPASSERT(partitionSelectionExprInputs().entries() == 0);
partitionSelectionExprInputs().insert(partNum->getValueId());
partitionSelectionExprInputs().insert(numParts->getValueId());
if(assignPartition()) {
// If partitionAssignment is to be done, the partition selection
// expression is Modulus(Cast((partNum+1),SQLInt),numParts). The
// cast (to SQLInt) is there because the type synthesized for
// partNum+1 is a large int, which Modulus can't handle at the
// moment. So, we just cast the sum to a SQLInt for now.
// Probably the best thing to do here is to somehow force the type
// synthesis for BiArith(ITM_PLUS) so that when we add two SQLInts
// we get a result of SQLInt, rather than something bigger. This
// doesn't follow SQL type synthesis rules, but that is OK, since
// the expression is generated internally. $$ TBD $$
// Construct and cache the partitionSelectionExpr for Round Robin
// (partition assignment).
//
partitionSelectionExpr() =
new (heap) Modulus(new (heap) Cast(new (heap)
BiArith(ITM_PLUS, partNum,
new (heap) SystemLiteral(1)),
new (heap) SQLInt(heap, TRUE,FALSE)),
numParts);
// Bind the expression.
//
partitionSelectionExpr()->synthTypeAndValueId();
// PreCodeGen the expression (This maybe should go in
// RoundRobinPartitioningFunction::preCodeGen(), but preCodeGen()
// is typically called before this expression is generated.
//
partitionSelectionExpr()->replaceVEGExpressions(availableValues,
availableValues);
return partitionSelectionExpr();
} else if(partSearchKey->isUnique()) {
// For now, only support a partition selection expression when the
// partition search key is unique (identifies exactly one partition).
//
const ValueIdList &keyValues = partSearchKey->getBeginKeyValues();
// The type of the partitioning key for round robin is always
// SQLLargeInt (the type of SYSKEY)
//
NAType *desiredType = new (heap) SQLLargeInt(heap,TRUE, FALSE);
// The partition selection expression is:
//
// ProgDistrib(Cast(ShiftRight(Cast(<keyvalue>, SQLLargeInt), 32), SQLInt),
// numParts)
//
// First, the <keyvalue> is cast to a SQLLargeInt to make it the
// same type as the partitioning key (SYSKEY). This value is then
// shifted right 32 bits to remove the 16 bits of padding
// introduced for packing and 16 bits of random data used to make
// sure the key is unique, but not used to compute the partition
// number. This is then cast to a SQLInt to make it compatible
// with ProgDistrib. Then ProgDistrib calculates the partNum.
// Construct and cache the partitionSelectionExpr for Round Robin
// (partition assignment).
//
partitionSelectionExpr() =
new (heap) ProgDistrib(new (heap)
Cast(new (heap)
Shift(ITM_SHIFT_RIGHT,
new (heap)
Cast(keyValues[0].getItemExpr(),
desiredType),
new (heap)
SystemLiteral(0x1000)),
new (heap) SQLInt(heap, FALSE,FALSE)),
numParts);
// Bind the expression.
//
partitionSelectionExpr()->synthTypeAndValueId();
// PreCodeGen the expression (This maybe should go in
// RoundRobinPartitioningFunction::preCodeGen(), but preCodeGen()
// is typically called before this expression is generated.
//
partitionSelectionExpr()->replaceVEGExpressions(availableValues,
availableValues);
return partitionSelectionExpr();
} else {
return NULL;
}
} // RoundRobinPartitioningFunction::createPartitionSelectionExpr()
//::scaleNUmberOfPartitions() are called in following locations
//
// OptPhysRelExpr.cpp
// 3536 NJ::genLeftChild()
// 4892 NJ:createContextForChild() (rightPartFunc->)
// 14166 and 14262 synthDP2PhysicalProperty()
//
// GP.cpp
// 955 GroupAttributes::recommendedOrderForNJProbing()
//
//As a result, the following version will not be called.
PartitioningFunction *
RoundRobinPartitioningFunction::
scaleNumberOfPartitions(Lng32 &suggestedNewNumberOfPartitions,
PartitionGroupingDistEnum partGroupDist)
{
if (suggestedNewNumberOfPartitions == 1)
return new(CmpCommon::statementHeap())
SinglePartitionPartitioningFunction();
// If an expression has been generated, then we want to discard it
// because it may no longer be correct.
storeExpression(NULL);
// Allow arbitrary scaling of RoundRobinPartitioningFunctions.
// (The runtime will handle the mapping of physical partitions
// to logical partitions.)
//
if ( suggestedNewNumberOfPartitions != getCountOfPartitions() ) {
NodeMap* newNodeMap =
new (heap_) NodeMap(heap_, suggestedNewNumberOfPartitions);
replaceNodeMap(newNodeMap);
}
partitionCount_ = suggestedNewNumberOfPartitions;
return this;
} // RoundRobinPartitioningFunction::scaleNumberOfPartitions()
// -----------------------------------------------------------------------
// RoundRobinPartitioningFunction::isAGroupingOf()
// -----------------------------------------------------------------------
NABoolean
RoundRobinPartitioningFunction::
isAGroupingOf(const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = 1;
const RoundRobinPartitioningFunction *oth =
other.castToRoundRobinPartitioningFunction();
// If other is not a RoundRobinPartitioningFunction, then it cannot
// be a grouping of...
//
if(!oth)
return FALSE;
// To be a grouping of, the key column (SYSKEY) must be the same.
//
if (! (getPartitioningKey() == oth->getPartitioningKey()))
return FALSE;
// If this function has more partitions than other,
// then it cannot be a grouping of.
// Eg. this.numParts: 10 this.origNumParts: 20
// oth.numParts: 5 oth.origNumParts: 20
//
// If the two functions are not based on the same physical function,
// then it cannot be a grouping of.
// Eg. this.numParts: 10 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 30
//
if((getCountOfPartitions() > oth->getCountOfPartitions()) ||
(getCountOfOrigRRPartitions() != oth->getCountOfOrigRRPartitions())) {
return FALSE;
}
// Here the following is known to be TRUE:
//
// (getCountOfPartitions() <= oth->getCountOfPartitions()
//
// AND
//
// (getCountOfOrigRRPartitions() == oth->getCountOfOrigRRPartitions())
//
// Eg. this.numParts: 10 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// this.numParts: 5 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
// If other has not been scaled (allow arbitrary scaling of one function):
// Eg. this.numParts: 7 this.origNumParts: 20
// oth.numParts: 20 oth.origNumParts: 20
// OR
// If they have both been scaled to the same number of partitions:
// then it is a grouping of.
// Eg. this.numParts: 7 this.origNumParts: 20
// oth.numParts: 7 oth.origNumParts: 20
//
if((oth->getCountOfPartitions() == oth->getCountOfOrigRRPartitions()) ||
(getCountOfPartitions() == oth->getCountOfPartitions())) {
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup =
((oth->getCountOfPartitions() + getCountOfPartitions() - 1)
/ getCountOfPartitions());
return TRUE;
}
// WARNING..... I am not sure if the current code can ever produce
// a situation that would bring control to here. Also, I am not
// sure if the semantics of GROUPING implemented below are correct
// for these situations.
//
// Here the following is known to be TRUE:
//
// both functions have been scaled. (I DON'T THINK THIS CAN HAPPEN)
//
// AND
//
// They are scaled to different sizes.
//
// Eg. this.numParts: 5 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// this.numParts: 7 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// Under these conditions, three things must be true for it to be a
// grouping of:
//
// - the scaled number of partitions must evenly divide the scaled
// number of partitions of other
// Eg. this.numParts: 5 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// - the other scaling must be a multiple of or evenly divide the
// original number of partitions
// Eg. this.numParts: 5 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// this.numParts: 5 this.origNumParts: 20
// oth.numParts: 40 oth.origNumParts: 20
//
// - this scaling must also be a multiple of or evenly divide the
// original number of partitions
// Eg. this.numParts: 5 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// this.numParts: 40 this.origNumParts: 20
// oth.numParts: 10 oth.origNumParts: 20
//
// If the scaled number of partitions evenly divides the scaled
// number of partitions of other...
//
if((oth->getCountOfPartitions() % getCountOfPartitions()) != 0) {
return FALSE;
}
// AND the other scaling is a multiple of or evenly divides the
// original number of partitions...
//
if(oth->getCountOfOrigRRPartitions() >= oth->getCountOfPartitions()) {
if(oth->getCountOfOrigRRPartitions() % oth->getCountOfPartitions()) {
return FALSE;
}
} else {
if(oth->getCountOfPartitions() % oth->getCountOfOrigRRPartitions()) {
return FALSE;
}
}
// AND this scaling is a multiple of or evenly divides the original
// number of partitions ...
//
if(getCountOfOrigRRPartitions() >= getCountOfPartitions()) {
if(getCountOfOrigRRPartitions() % getCountOfPartitions()) {
return FALSE;
}
} else {
if(getCountOfPartitions() % getCountOfOrigRRPartitions()) {
return FALSE;
}
}
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup =
((oth->getCountOfPartitions() + getCountOfPartitions() - 1)
/ getCountOfPartitions());
// THEN it is a grouping of...
//
return TRUE;
} // RoundRobinPartitioningFunction::isAGroupingOf()
// -----------------------------------------------------------------------
// RoundRobinPartitioningFunction::codeGen() is defined in
// generator/GenPartFunc.cpp
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
// Make a new partSearchKey that indicates that PA_PARTITION_GROUPING
// is being done. Note that a search key can not be generated which
// can group RR partitions. For RoundRobinPartitioning, a flag in the
// search key is used to indicate that PA_PARTITION_GROUPING is being
// done and the begin/end key values of the search key are set to the
// partition input values of the partitioning function.
// -----------------------------------------------------------------------
SearchKey *
RoundRobinPartitioningFunction::createSearchKey(const IndexDesc *indexDesc,
ValueIdSet availInputs,
ValueIdSet additionalPreds) const
{
ValueIdSet preds(getPartitioningKeyPredicates());
ValueIdSet nonKeyColumnSet; // empty set
availInputs += getPartitionInputValues();
preds += additionalPreds;
// Call this special constructor that constructs a search key for a
// RoundRobinPartitioningFunction.
//
SearchKey *partSearchKey = new (CmpCommon::statementHeap())
SearchKey(indexDesc->getPartitioningKey(),
indexDesc->getOrderOfPartitioningKeyValues(),
availInputs,
preds,
this,
nonKeyColumnSet,
indexDesc);
return partSearchKey;
} // RoundRobinPartitioningFunction::createSearchKey()
void RoundRobinPartitioningFunction::setupForStatement()
{
if(setupForStatement_)
return;
PartitioningFunction::setupForStatement();
setupForStatement_ = TRUE;
resetAfterStatement_ = FALSE;
}
void RoundRobinPartitioningFunction::resetAfterStatement()
{
if(resetAfterStatement_)
return;
PartitioningFunction::resetAfterStatement();
partitionCount_ = numberOfOrigRRPartitions_;
setupForStatement_ = FALSE;
resetAfterStatement_ = TRUE;
}
const NAString RoundRobinPartitioningFunction::getText() const
{
NAString result("round robin partitioned ", CmpCommon::statementHeap());
char nparts[20];
sprintf(nparts,"%d (%d)", partitionCount_, numberOfOrigRRPartitions_);
result += nparts;
result += " ways on (";
getPartitioningKey().unparse(result,DEFAULT_PHASE,EXPLAIN_FORMAT);
result += ")";
return result;
}
void RoundRobinPartitioningFunction::print(
FILE* ofd,
const char* indent,
const char* title) const
{
PartitioningFunction::print(ofd,indent,"RoundRobinPartitioningFunction");
}
const skewProperty ANY_SKEW_PROPERTY(skewProperty::ANY, NULL);
NABoolean skewProperty::operator==(const skewProperty& other) const
{
if ( indicator_ != other.indicator_ )
return FALSE;
if (broadcastOneRow_ != other.broadcastOneRow_)
return FALSE;
if ( skewValues_ == other.skewValues_ )
return TRUE;
if ( skewValues_ != NULL && other.skewValues_ != NULL &&
NOT (*skewValues_ == *(other.skewValues_)) )
return FALSE;
if ( numESPs_ == other.numESPs_ )
return TRUE;
return FALSE;
}
const NAString skewProperty::getText(NABoolean abbre) const
{
NAString result;
char esps[20];
NAString suffix;
if ( abbre == FALSE && getAntiSkewESPs() > 0 ) {
suffix += "(via ";
str_itoa(getAntiSkewESPs(), esps);
suffix += esps;
suffix += " ESP(s)) ";
}
switch ( getIndicator() )
{
case skewProperty::UNIFORM_DISTRIBUTE:
if ( abbre == FALSE ) {
result += " - uniformly distribute ";
result += suffix;
result += "skewed ";
if ( skewValues_ )
result += skewValues_->getText();
} else {
result += "-ud";
}
break;
case skewProperty::BROADCAST:
if ( abbre == FALSE ) {
result += " - broadcast ";
result += suffix;
result += "skewed ";
if ( skewValues_ )
result += skewValues_->getText();
if (getBroadcastOneRow())
result += " and one row ";
} else {
result += "-br";
}
break;
default:
break;
}
return result;
}
Int64List*
SkewedDataPartitioningFunction::buildHashListForSkewedValues()
{
if ( skewHashList_ )
return skewHashList_;
const SkewedValueList* svlist = getSkewProperty().getSkewValues();
if ( svlist == NULL )
return NULL;
skewHashList_ = new (STMTHEAP) Int64List(STMTHEAP, svlist->entries());
CollHeap *heap = CmpCommon::statementHeap();
char data[10]; Int32 len = 0;
ConstValue* cvExp = NULL;
ItemExpr* expForSkewedValue = NULL;
ValueIdList exprs;
UInt32 hashval;
if ( !svlist->needToComputeFinalHash() ) {
// for TRUE MC-SB. hash value for the composite skew value is pre-computed.
for (CollIndex i = 0; i < svlist->entries(); i++) {
hashval = (UInt32)((*svlist)[i].getDblValue());
skewHashList_ -> insertAt(i, hashval);
}
} else {
const NAType* naType = svlist->getNAType();
NABoolean useHash = naType->useHashRepresentation();
UInt32 flags = ExHDPHash::NO_FLAGS;
for (CollIndex i = 0; i < svlist->entries(); i++)
{
if ( (*svlist)[i].getValue().isNull() ) {
// an untyped NULL constant. All null values hash to this constant.
// Ref. module exp_function.cpp, method ExHDPHash::eval() and
// ex_function_hash::eval().
hashval = ExHDPHash::nullHashValue; //666654765;
} else {
if ( useHash ) {
// The skew value for any character or exact numeric data type is
// the hash itself. So here we just copy and cast it back to UInt32.
hashval = (UInt32)((*svlist)[i].getDblValue());
} else {
// If we hit this branch, it means the boundary values are stored in
// the skew list. We can assume these values can be safely casted
// back (without loosing precision). Otherwise, storing-hash-
// value method should be used.
(*svlist)[i].outputToBufferToComputeRTHash(naType,data,len,flags);
hashval = computeHashValue(data, flags, len);
}
}
skewHashList_ -> insertAt(i, hashval);
}
}
return skewHashList_;
}
//============================
HivePartitioningFunction::~HivePartitioningFunction() {}
PartitioningRequirement*
HivePartitioningFunction::makePartitioningRequirement()
{
return new (CmpCommon::statementHeap())
RequireHive(this);
}
PartitioningFunction*
HivePartitioningFunction::copy() const
{
return new (CmpCommon::statementHeap())
HivePartitioningFunction(*this, CmpCommon::statementHeap());
}
// -----------------------------------------------------------------------
// HivePartitioningFunction::createPartitioningKeyPredicates()
// -----------------------------------------------------------------------
void HivePartitioningFunction::createPartitioningKeyPredicates()
{
createBetweenPartitioningKeyPredicates("_sys_HostVarLoHivePart",
"_sys_HostVarHiHivePart");
} // HivePartitioningFunction::createPartitioningKeyPredicates()
// -----------------------------------------------------------------------
// HivePartitioningFunction::isAGroupingOf()
// -----------------------------------------------------------------------
// Right now we assume that the split function of a hash partitioning
// function is such that no two functions are a grouping of each other.
// The only exception is identity, of course. We use the base class'
// implementation.
//
// With more knowledge about the split function (e.g. by knowing it's
// a simple modulus), one could guarantee that a 4-way hash-partitioning
// scheme is actually a grouping of an 8-way scheme. This is not really
// necessary and therefore not done here.
ItemExpr *
HivePartitioningFunction::buildHashingExpressionForExpr(ItemExpr* expr) const
{
return new (CmpCommon::statementHeap()) HiveHash(expr);
}
UInt32 HivePartitioningFunction::computeHashValue(char* data, UInt32 flags, Int32 len)
{
// need a Hive()
// Directly call the implementation function to compute the hash. NULL
// values and VARCHAR data types are not handled.
return FastHash(data, len);
}
ItemExpr * HivePartitioningFunction::getHashingExpression() const
{
// need a Hive()
ItemExpr* hashExpr = NULL;
ItemExpr* partExpr = getExpression();
if ( partExpr ) {
hashExpr = partExpr->child(0);
CMPASSERT(hashExpr AND
hashExpr->getOperatorType()== ITM_HASH);
}
return hashExpr;
}
// -----------------------------------------------------------------------
// Method for debugging.
// -----------------------------------------------------------------------
const NAString HivePartitioningFunction::getText() const
{
return getTextImp("hive");
}
void HivePartitioningFunction::print(FILE* ofd, const char* indent,
const char* title) const
{
PartitioningFunction::print(ofd, indent, "HivePartitioningFunction");
} // HivePartitioningFunction::print()
PartitioningFunction*
HivePartitioningFunction::
createPartitioningFunctionForIndexDesc(IndexDesc *idesc) const
{
const NAFileSet * fileSet = idesc->getNAFileSet();
const NAColumnArray & allColumns = fileSet->getAllColumns();
const NAColumnArray & partKeyColumns = fileSet->getPartitioningKeyColumns();
CollIndex ixColNumber;
ValueId keyValueId;
ValueIdSet partitioningKey;
ValueIdList partitioningKeyList;
for (CollIndex i = 0; i < partKeyColumns.entries(); i++)
{
// which column of the index is this (usually this will be == i)
ixColNumber = allColumns.index(partKeyColumns[i]);
// insert the value id of the index column into the partitioning
// key column value id set
keyValueId = idesc->getIndexColumns()[ixColNumber];
partitioningKey += keyValueId;
partitioningKeyList.insertAt(i,keyValueId);
} // end loop over partitioning key columns
// -----------------------------------------------------------------
// Allocate a new HashPartitioningFunction.
// -----------------------------------------------------------------
HivePartitioningFunction *partFunc = new(idesc->wHeap())
HivePartitioningFunction (partitioningKey,
partitioningKeyList,
getCountOfPartitions(),
getNodeMap()->copy(idesc->wHeap()));
// -----------------------------------------------------------------
// Construct the partitioning key predicates.
// -----------------------------------------------------------------
partFunc->createPartitioningKeyPredicates();
return partFunc;
} // HivePartitioningFunction::createPartitioningFunctionForIndexDesc()
NABoolean HivePartitioningFunction::isAGroupingOf(
const PartitioningFunction &other,
Lng32* maxPartsPerGroup) const
{
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = 1;
if ( comparePartKeyToKey(other) == INCOMPATIBLE )
return FALSE;
// assume we can repartition a hive table with <m> partitions to
// <n> partitions. No question asked.
if ( getCountOfPartitions() < other.getCountOfPartitions() &&
other.getCountOfPartitions() % getCountOfPartitions() != 0 )
return FALSE;
if ( other.getCountOfPartitions() < getCountOfPartitions() &&
getCountOfPartitions() % other.getCountOfPartitions() != 0 )
return FALSE;
// This is a grouping of. Set the maxPartsPerGroup and return TRUE.
if (maxPartsPerGroup != NULL)
*maxPartsPerGroup = other.getCountOfPartitions() / getCountOfPartitions();
return TRUE;
}
void
HivePartitioningFunction::normalizePartitioningKeys(NormWA& normWARef)
{
HashPartitioningFunction::normalizePartitioningKeys(normWARef);
keyColumnList_.normalizeNode(normWARef);
// don't normalize original key col list, avoid VEGies which could
// cause data type changes.
}