LhxPlan Class Reference

#include <LhxPartition.h>

List of all members.

Public Member Functions
void	init (WeakLhxPlan parentPlanInit, uint partitionLevelInit, vector< SharedLhxPartition > &partitionsInit, bool enableSubPartStat)
	Initialize a plan, with its input partitions and parent plan.
void	init (WeakLhxPlan parentPlanInit, uint partitionLevelInit, vector< SharedLhxPartition > &partitionsInit, vector< shared_array< uint > > &subPartStats, shared_ptr< dynamic_bitset<> > filterInit, VectorOfUint &filteredRowsInit, bool enableSubPartStat, bool enableSwing)
	Initialize a plan.
LhxPartitionState	generatePartitions (LhxHashInfo const &hashInfo, LhxPartitionInfo &partInfo)
	Generate partitions for the child plans.
void	createChildren (LhxHashInfo const &hashInfo, bool enableSubPartStat)
	Partition this plan and create child plan.
void	createChildren (LhxPartitionInfo &partInfo, bool enableSubPartStat, bool enableSwing)
	Create child plan from partitions provided via partInfo.
uint	getPartitionLevel ()
	Get the partition level of this plan.
SharedLhxPartition	getBuildPartition ()
	Get the partition corresponding to inputIndex.
SharedLhxPartition	getProbePartition ()
SharedLhxPartition	getPartition (uint inputIndex)
uint	getBuildInput ()
uint	getProbeInput ()
uint	getJoinSide (uint inputIndex)
SharedLhxPlan	getFirstChild ()
	Get the first child plan.
LhxPlan *	getFirstLeaf ()
	Get first leaf plan in dfs order.
LhxPlan *	getNextLeaf ()
	Get next leaf plan in dfs order.
void	close ()
	Close the plan tree.
string	toString ()
	Print the content of the plan tree rooted at this plan.
Static Public Attributes
static const uint	LhxSubPartCount = 16
static const uint	LhxChildPartCount = 3
Private Member Functions
void	addSibling (SharedLhxPlan siblingPlan)
	Add sibling plan.
void	mapSubPartToChild (vector< shared_array< uint > > &subPartStats)
	Using sub partition stats gathered at the previous partition level, map sub partitions to child partitions.
uint	calculateChildIndex (uint hashKey, uint curInputIndex)
	Calculate the target child partition index based on the hashkey of a tuple.
bool	isBuildChildPart (uint childPartIndex)
bool	isProbeChildPart (uint childPartIndex)
uint	getBuildChildPart (uint childPartIndex)
uint	getProbeChildPart (uint childPartIndex)
Private Attributes
uint	partitionLevel
vector< SharedLhxPartition >	partitions
shared_array< uint >	joinSideToInputMap
shared_ptr< dynamic_bitset<> >	joinFilter
shared_array< uint >	filteredRowCount
shared_array< uint >	subPartToChildMap
vector< shared_array< uint > >	childPartSize
shared_array< uint >	inputSize
WeakLhxPlan	parentPlan
SharedLhxPlan	firstChildPlan
SharedLhxPlan	siblingPlan

---

Detailed Description

Definition at line 291 of file LhxPartition.h.

---

Member Function Documentation

void LhxPlan::addSibling

(

SharedLhxPlan

siblingPlan

)

[inline, private]

Add sibling plan.

Definition at line 495 of file LhxPartition.h.

References siblingPlan.

{
    siblingPlan = siblingPlanInit;
}

void LhxPlan::mapSubPartToChild

(

vector< shared_array< uint > > &

subPartStats

)

[private]

Using sub partition stats gathered at the previous partition level, map sub partitions to child partitions.

The objective is to come up with child partitions of similar size.

Definition at line 544 of file LhxPartition.cpp.

References childPartSize, getBuildInput(), getProbeInput(), LhxChildPartCount, LhxSubPartCount, partitions, and subPartToChildMap.

Referenced by init().

{
    uint numInputs = partitions.size();
    uint buildIndex =  getBuildInput();
    shared_array<uint> buildSubPartStat = subPartStats[buildIndex];

    if (!buildSubPartStat) {
        return;
    }

    uint i, j, k;

    subPartToChildMap.reset(new uint[LhxSubPartCount]);

    for (i = 0; i < numInputs; i ++) {
        childPartSize.push_back(
            shared_array<uint>(new uint[LhxChildPartCount]));
    }

    shared_array<uint> buildChildPartSize = childPartSize[buildIndex];

    for (i = 0; i < LhxChildPartCount; i ++) {
        buildChildPartSize[i] = 0;
    }

    j = 0;
    for (i = 0; i < LhxSubPartCount; i ++) {
        buildChildPartSize[j] += buildSubPartStat[i];
        subPartToChildMap[i] = j;

        k = 1;
        while (
            (buildChildPartSize[j]
                > buildChildPartSize[(j + k) % LhxChildPartCount])
            && k < LhxChildPartCount)
        {
            k ++;
        }

        if (k == LhxChildPartCount) {
            // If current child partition is bigger than all other child
            // partitions, move to the next child
            j = (j + 1) % LhxChildPartCount;
        }
    }

    /*
     * This is simply stats keeping for the probe side child partitions.
     */
    if (numInputs == 2) {
        uint probeIndex = getProbeInput();
        shared_array<uint> probeChildPartSize = childPartSize[probeIndex];
        shared_array<uint> probeSubPartStat = subPartStats[probeIndex];

        for (i = 0; i < LhxChildPartCount; i ++) {
            probeChildPartSize[i] = 0;
        }

        for (i = 0; i < LhxSubPartCount; i ++) {
            probeChildPartSize[subPartToChildMap[i]] += probeSubPartStat[i];
        }
    }
}

uint LhxPlan::calculateChildIndex	(	uint	hashKey,
		uint	curInputIndex
	)			[private]

Calculate the target child partition index based on the hashkey of a tuple.

Definition at line 609 of file LhxPartition.cpp.

References LhxChildPartCount, LhxSubPartCount, and subPartToChildMap.

Referenced by generatePartitions().

{
    if (subPartToChildMap) {
        return (subPartToChildMap[hashKey % LhxSubPartCount] +
            curInputIndex * LhxChildPartCount);
    } else {
        return (hashKey % LhxChildPartCount +
            curInputIndex * LhxChildPartCount);
    }
}

bool LhxPlan::isBuildChildPart

(

uint

childPartIndex

)

[inline, private]

Definition at line 547 of file LhxPartition.h.

References getBuildInput(), and LhxChildPartCount.

Referenced by generatePartitions().

{
    return ((childPartIndex / LhxChildPartCount) == getBuildInput());
}

bool LhxPlan::isProbeChildPart

(

uint

childPartIndex

)

[inline, private]

Definition at line 552 of file LhxPartition.h.

References getProbeInput(), and LhxChildPartCount.

{
    return ((childPartIndex / LhxChildPartCount) == getProbeInput());
}

uint LhxPlan::getBuildChildPart

(

uint

childPartIndex

)

[inline, private]

Definition at line 557 of file LhxPartition.h.

References getBuildInput(), and LhxChildPartCount.

Referenced by generatePartitions().

{
    return ((childPartIndex % LhxChildPartCount) +
            getBuildInput() * LhxChildPartCount);
}

uint LhxPlan::getProbeChildPart

(

uint

childPartIndex

)

[inline, private]

Definition at line 563 of file LhxPartition.h.

References getProbeInput(), and LhxChildPartCount.

Referenced by createChildren().

{
    return ((childPartIndex % LhxChildPartCount) +
            getProbeInput() * LhxChildPartCount);
}

void LhxPlan::init	(	WeakLhxPlan	parentPlanInit,
		uint	partitionLevelInit,
		vector< SharedLhxPartition > &	partitionsInit,
		bool	enableSubPartStat
	)

Initialize a plan, with its input partitions and parent plan.

Definition at line 445 of file LhxPartition.cpp.

{
    /*
     * No filter for this plan.
     */
    shared_ptr<dynamic_bitset<> > joinFilterInit =
        shared_ptr<dynamic_bitset<> >();
    vector<shared_array<uint> > subPartStatsInit;
    VectorOfUint filteredRows;

    for (uint i = 0; i < partitionsInit.size(); i ++) {
        subPartStatsInit.push_back(shared_array<uint>());
        filteredRows.push_back(0);
    }

    init(
        parentPlanInit, partitionLevelInit, partitionsInit,
        subPartStatsInit, joinFilterInit, filteredRows,
        enableSubPartStat, false);
}

void LhxPlan::init	(	WeakLhxPlan	parentPlanInit,
		uint	partitionLevelInit,
		vector< SharedLhxPartition > &	partitionsInit,
		vector< shared_array< uint > > &	subPartStats,
		shared_ptr< dynamic_bitset<> >	filterInit,
		VectorOfUint &	filteredRowsInit,
		bool	enableSubPartStat,
		bool	enableSwing
	)

Initialize a plan.

Definition at line 470 of file LhxPartition.cpp.

References filteredRowCount, inputSize, joinFilter, joinSideToInputMap, LhxSubPartCount, mapSubPartToChild(), parentPlan, partitionLevel, partitions, and subPartToChildMap.

{
    uint numInputs = partitionsInit.size();

    partitionLevel = partitionLevelInit;
    parentPlan   = parentPlanInit;

    /*
     * REVIEW(rchen 2006-08-08): if input sides swing, then the new build could
     * be the same as the probe input of the previous partition. This filter
     * will not filter any tuple as it tries to filter the very input the
     * filter is built on.
     */
    joinFilter = joinFilterInit;

    filteredRowCount.reset(new uint[numInputs]);
    inputSize.reset(new uint[numInputs]);
    joinSideToInputMap.reset(new uint[numInputs]);
    subPartToChildMap.reset();

    // REVIEW jvs 26-Aug-2006:  here "will be used" means once someone
    // gets around to true hybrid, right?
    /*
     * After support is added for repartitioning using stats (gathered during
     * previous round of partitioning), a bin-packing algorithm will be used to
     * keep in memory some subpartitions and output the rest to child
     * partitions of similar size.
     */
    for (int i = 0; i < numInputs; i ++) {
        partitions.push_back(partitionsInit[i]);
        filteredRowCount[i] = filteredRowsInit[i];
        joinSideToInputMap[i] = i;

        inputSize[i] = 0;
        shared_array<uint> inputSubPartStat = subPartStats[i];

        if (inputSubPartStat) {
            for (int k = 0; k < LhxSubPartCount; k ++) {
                inputSize[i] += inputSubPartStat[k];
            }
        }
    }

    /*
     * Map join side to input, using partitions stats associated with each
     * input. The input with the smaller side witll be the build side for the
     * join: joinSide for the build will map to the index of this input.
     */
    if (enableSwing &&
        (numInputs == 2) && (inputSize[0] < inputSize[1])) {
        joinSideToInputMap[0] = 1;
        joinSideToInputMap[1] = 0;
    }

    subPartToChildMap.reset();

    if (enableSubPartStat) {
        /*
         * Use build side sub part stat to divide both inputs into child
         * partitions. Needs to be called after the join sides have been
         * assigned.
         */
        mapSubPartToChild(subPartStats);
    }
}

LhxPartitionState LhxPlan::generatePartitions	(	LhxHashInfo const &	hashInfo,
		LhxPartitionInfo &	partInfo
	)

Generate partitions for the child plans.

Definition at line 620 of file LhxPartition.cpp.

References LhxPartitionInfo::buildTuple, calculateChildIndex(), LhxPartitionReader::close(), TupleData::compute(), LhxPartitionReader::consumeTuple(), LhxPartitionInfo::curInputIndex, LhxPartitionReader::demandData(), EXECBUF_EOS, LhxPartitionInfo::filteredRowCountList, getBuildChildPart(), getProbeInput(), LhxPartitionReader::getState(), LhxPartitionReader::getTupleDesc(), LhxHashGenerator::hash(), LhxPartitionInfo::hashTableReader, LhxHashGenerator::init(), LhxHashInfo::inputDesc, isBuildChildPart(), LhxHashInfo::isKeyColVarChar, LhxPartitionReader::isTupleConsumptionPending(), joinFilter, LhxPartitionInfo::joinFilterList, LhxHashInfo::keyProj, LhxSubPartCount, LhxPartitionInfo::numInputs, PartitionEndOfData, partitionLevel, LhxPartitionInfo::partitionMemory, PartitionUnderflow, LhxPartitionInfo::probeReader, LhxPartitionInfo::reader, LhxPartitionInfo::subPartStatList, LhxPartitionReader::unmarshalTuple(), LhxHashInfo::useJoinFilter, and LhxPartitionInfo::writerList.

Referenced by LhxJoinExecStream::execute(), and LhxAggExecStream::execute().

{
    // REVIEW jvs 26-Aug-2006:  modulo on this is computed below; make
    // sure compiler is optimizing to bitmask, or do it by hand
    uint filterSize = 4096;
    bool isAggregate = (partInfo.numInputs == 1);

    LhxHashGenerator hashGenPrev;
    LhxHashGenerator hashGen;
    LhxHashGenerator hashGenNext;

    hashGenPrev.init(partitionLevel);
    hashGen.init(partitionLevel + 1);
    hashGenNext.init(partitionLevel + 2);

    LhxPartitionReader *&reader = partInfo.reader;
    vector<SharedLhxPartitionWriter> &writerList = partInfo.writerList;
    vector<shared_ptr<dynamic_bitset<> > > &joinFilterList =
        partInfo.joinFilterList;
    vector<shared_array<uint> > &subPartStatList = partInfo.subPartStatList;
    shared_array<uint> &filteredRowCountList = partInfo.filteredRowCountList;

    uint &curInputIndex = partInfo.curInputIndex;
    uint otherInputIndex = partInfo.numInputs - curInputIndex - 1;

    TupleData inputTuple;
    TupleDescriptor inputTupleDesc = reader->getTupleDesc();
    inputTuple.compute(inputTupleDesc);

    uint prevHashKey;
    uint hashKey;
    uint nextHashKey;

    uint childPartIndex;
    bool writeToPartition;

    uint statIndex;
    shared_array<uint> curSubPartStat;

    /*
     * If partition source is from memory,
     * i.e. there's hash table to read from.
     */
    if (partInfo.partitionMemory) {
        TupleData hashTableTuple;
        TupleDescriptor hashTableTupleDesc = hashInfo.inputDesc[curInputIndex];

        hashTableTuple.compute(hashTableTupleDesc);

        while ((partInfo.hashTableReader)->getNext(hashTableTuple)) {
            writeToPartition = false;

            hashKey = hashGen.hash(
                hashTableTuple,
                hashInfo.keyProj[curInputIndex],
                hashInfo.isKeyColVarChar[curInputIndex]);

            childPartIndex = calculateChildIndex(hashKey, curInputIndex);

            if (hashInfo.useJoinFilter[curInputIndex]) {
                /*
                 * Use input filter if there is one. Note top level build input
                 * does not have input filter.
                 */
                if (partitionLevel == 0) {
                    writeToPartition = true;
                } else {
                    prevHashKey =
                        hashGenPrev.hash(
                            hashTableTuple,
                            hashInfo.keyProj[curInputIndex],
                            hashInfo.isKeyColVarChar[curInputIndex]);
                    if (joinFilter &&
                        joinFilter->test(prevHashKey % filterSize)) {
                        writeToPartition = true;
                    } else {
                        filteredRowCountList[childPartIndex]++;
                    }
                }
            } else {
                /*
                 * Not using filter.
                 */
                writeToPartition = true;
            }

            if (writeToPartition) {
                writerList[childPartIndex]->marshalTuple(hashTableTuple);

                nextHashKey = hashGenNext.hash(
                    hashTableTuple,
                    hashInfo.keyProj[curInputIndex],
                    hashInfo.isKeyColVarChar[curInputIndex]);

                statIndex = nextHashKey % LhxSubPartCount;
                curSubPartStat = subPartStatList[childPartIndex];
                curSubPartStat[statIndex]++;

                /*
                 * Set output filter for the other input.
                 * Note: if the filter is used on the next level, "the other
                 * input" could be the same if join sides are switched.
                 */
                if (!joinFilterList[childPartIndex]) {
                    /*
                     * Filter not allocated yet.
                     */
                    joinFilterList[childPartIndex].reset(
                        new dynamic_bitset<>(filterSize));
                }
                joinFilterList[childPartIndex]->set(hashKey % filterSize);
            }
        }

        if (isAggregate) {
            /*
             * release the agg exec stream hash table scratch pages
             * and initialize writer scratch pages.
             */
            ((partInfo.hashTableReader)->getHashTable())->releaseResources();
            for (int i = 0; i < writerList.size();i ++) {
                writerList[i]->allocateResources();
            }
        }

        /*
         * Done with tuples in the hash table. Next tuples will come from a
         * partition (stream or disk).
         */
        partInfo.partitionMemory = false;

        /*
         * The tuple for which hash table full is detected is not in the hash
         * table yet.
         */
        inputTuple = partInfo.buildTuple;
    }

    for (;;) {
        /*
         * Note that partInfo.buildTuple is an unconsumed tuple from the
         * reader. So when first time in this loop, isTupleConsumptionPending
         * returns true.
         */
        if (!reader->isTupleConsumptionPending()) {
            if (reader->getState() == EXECBUF_EOS) {
                if (curInputIndex == getProbeInput()) {
                    /*
                     * Done with partitioning the 0th input.
                     * The current plan is completely partitioned.
                     */
                    return PartitionEndOfData;
                } else {
                    curInputIndex = getProbeInput();
                    otherInputIndex = partInfo.numInputs - curInputIndex - 1;
                    reader->close();
                    reader = &partInfo.probeReader;
                    inputTupleDesc = reader->getTupleDesc();
                    inputTuple.compute(inputTupleDesc);
                    continue;
                }
            }

            if (!reader->demandData()) {
                if (partitionLevel == 0) {
                    /*
                     * Reading from a stream: indicate underflow to producer.
                     */
                    return PartitionUnderflow;
                } else {
                    if (curInputIndex == getProbeInput()) {
                        /*
                         * Done with partitioning the 0th input.
                         * The current plan is completely partitioned.
                         */
                        return PartitionEndOfData;
                    } else {
                        curInputIndex = getProbeInput();
                        reader->close();
                        reader = &partInfo.probeReader;
                        inputTupleDesc = reader->getTupleDesc();
                        inputTuple.compute(inputTupleDesc);
                        continue;
                    }
                }
            }

            reader->unmarshalTuple(inputTuple);
        }

        writeToPartition = false;

        hashKey = hashGen.hash(
            inputTuple,
            hashInfo.keyProj[curInputIndex],
            hashInfo.isKeyColVarChar[curInputIndex]);

        childPartIndex = calculateChildIndex(hashKey, curInputIndex);

        nextHashKey = hashGenNext.hash(
            inputTuple,
            hashInfo.keyProj[curInputIndex],
            hashInfo.isKeyColVarChar[curInputIndex]);

        statIndex = nextHashKey % LhxSubPartCount;

        if (!isAggregate) {
            if (hashInfo.useJoinFilter[curInputIndex]) {
                /*
                 * Use input filter if there exists one
                 */
                if (isBuildChildPart(childPartIndex)) {
                    /*
                     * Build input.  Note top level build input does not have
                     * input filter.  Use joinfilter from the probe input of
                     * the previous level.
                     */
                    if (partitionLevel == 0) {
                        writeToPartition = true;
                    } else {
                        prevHashKey =
                            hashGenPrev.hash(
                                inputTuple,
                                hashInfo.keyProj[curInputIndex],
                                hashInfo.isKeyColVarChar[curInputIndex]);
                        if (joinFilter &&
                            joinFilter->test(prevHashKey % filterSize)) {
                            writeToPartition = true;
                        } else {
                            filteredRowCountList[childPartIndex]++;
                        }
                    }
                } else {
                    /*
                     * Probe input.
                     * Use join filter from build input of the same
                     * partitioning level.
                     */
                    if (joinFilterList[getBuildChildPart(childPartIndex)] &&
                        joinFilterList[getBuildChildPart(childPartIndex)]->
                        test(hashKey % filterSize)) {
                        writeToPartition = true;
                    } else {
                        filteredRowCountList[childPartIndex]++;
                    }
                }
            } else {
                /*
                 * Not using filter.
                 */
                writeToPartition = true;
            }

            if (writeToPartition) {
                writerList[childPartIndex]->marshalTuple(inputTuple);
                curSubPartStat = subPartStatList[childPartIndex];
                curSubPartStat[statIndex]++;

                /*
                 * Set output filter to be used by the other input.
                 * Note: if the filter is used on the next level, "the other
                 * input" could be the same if join sides are switched.
                 */
                if (!joinFilterList[childPartIndex]) {
                    /*
                     * Filter not allocated yet.
                     */
                    joinFilterList[childPartIndex].reset(
                        new dynamic_bitset<>(filterSize));
                }
                joinFilterList[childPartIndex]->set(hashKey % filterSize);
            }
        } else {
            writerList[childPartIndex]->aggAndMarshalTuple(inputTuple);
            (subPartStatList[childPartIndex])[statIndex]++;
        }

        reader->consumeTuple();
    }
}

void LhxPlan::createChildren	(	LhxHashInfo const &	hashInfo,
		bool	enableSubPartStat
	)

Partition this plan and create child plan.

This is used in testing only.

Definition at line 903 of file LhxPartition.cpp.

References LhxPartitionReader::close(), close(), TupleData::compute(), LhxPartitionReader::consumeTuple(), LhxPartitionReader::demandData(), EXECBUF_EOS, firstChildPlan, LhxPartitionReader::getState(), LhxHashGenerator::hash(), LhxHashGenerator::init(), LhxHashInfo::inputDesc, LhxHashInfo::isKeyColVarChar, LhxPartitionReader::isTupleConsumptionPending(), LhxHashInfo::keyProj, LhxChildPartCount, LhxPartitionWriter::marshalTuple(), LhxPartitionWriter::open(), LhxPartitionReader::open(), partitionLevel, partitions, and LhxPartitionReader::unmarshalTuple().

Referenced by LhxJoinExecStream::execute(), LhxAggExecStream::execute(), and LhxHashTableTest::testInsert().

{
    LhxHashGenerator hashGen;
    hashGen.init(partitionLevel + 1);

    uint numInputs = hashInfo.inputDesc.size();

    vector<SharedLhxPartition> destPartitionList(LhxChildPartCount * numInputs);

    LhxPartitionReader reader;
    LhxPartitionWriter writerList[LhxChildPartCount];
    uint childNum, i, j;
    TupleData outputTuple;

    /*
     * Generate partitions for each input.
     */
    for (j = 0; j < numInputs; j ++) {
        reader.open(partitions[j], hashInfo);
        outputTuple.compute(hashInfo.inputDesc[j]);

        for (i = 0; i < LhxChildPartCount; i ++) {
            uint index = j * LhxChildPartCount + i;
            destPartitionList[index].reset(
                new LhxPartition(partitions[j]->pExecStream));
            destPartitionList[index]->inputIndex = j;
            writerList[i].open(destPartitionList[index], hashInfo);
        }

        for (;;) {
            if (!reader.isTupleConsumptionPending()) {
                if (reader.getState() == EXECBUF_EOS) {
                    /*
                     * The current plan is completely partitioned.
                     */
                    break;
                }
                if (!reader.demandData()) {
                    break;
                }
                reader.unmarshalTuple(outputTuple);
            }

            childNum =
                hashGen.hash(
                    outputTuple,
                    hashInfo.keyProj[j],
                    hashInfo.isKeyColVarChar[j]) % LhxChildPartCount;

            writerList[childNum].marshalTuple(outputTuple);
            reader.consumeTuple();
        }

        for (i = 0; i < LhxChildPartCount; i ++) {
            writerList[i].close();
        }

        /*
         * Partial aggregate hash tables used inside the writers(one HT
         * for each writer) share the same scratch buffer space.
         * Release the buffer pages used by these hash tables at the end,
         * after all writers have been closed(so that there will be no more
         * scratch page alloc calls).
         */
        for (i = 0; i < LhxChildPartCount; i ++) {
            writerList[i].releaseResources();
        }

        reader.close();
    }

    /*
     * Create child plans consisting of one partition from each input.
     */
    for (i = 0; i < LhxChildPartCount; i ++) {
        SharedLhxPlan newChildPlan = SharedLhxPlan(new LhxPlan());
        vector<SharedLhxPartition> partitionList;
        partitionList.push_back(destPartitionList[i]);
        partitionList.push_back(destPartitionList[i + LhxChildPartCount]);

        newChildPlan->init(
            WeakLhxPlan(shared_from_this()),
            partitionLevel + 1,
            partitionList,
            enableSubPartStat);

        newChildPlan->addSibling(firstChildPlan);
        firstChildPlan = newChildPlan;
    }
}

void LhxPlan::createChildren	(	LhxPartitionInfo &	partInfo,
		bool	enableSubPartStat,
		bool	enableSwing
	)

Create child plan from partitions provided via partInfo.

Definition at line 996 of file LhxPartition.cpp.

References LhxPartitionInfo::destPartitionList, LhxPartitionInfo::filteredRowCountList, firstChildPlan, getProbeChildPart(), LhxPartitionInfo::joinFilterList, LhxChildPartCount, LhxPartitionInfo::numInputs, partitionLevel, and LhxPartitionInfo::subPartStatList.

{
    uint i, j;

    for (i = 0; i < LhxChildPartCount; i ++) {
        SharedLhxPlan newChildPlan = SharedLhxPlan(new LhxPlan());
        vector<SharedLhxPartition> partitionList;
        vector<shared_array<uint> > subPartStats;
        VectorOfUint filteredRows;
        for (j = 0; j < partInfo.numInputs; j ++) {
            partitionList.push_back(
                partInfo.destPartitionList[i + LhxChildPartCount * j]);
            subPartStats.push_back(
                partInfo.subPartStatList[i + LhxChildPartCount * j]);
            filteredRows.push_back(
                partInfo.filteredRowCountList[i + LhxChildPartCount * j]);
        }
        newChildPlan->init(
            WeakLhxPlan(shared_from_this()),
            partitionLevel + 1,
            partitionList,
            subPartStats,
            partInfo.joinFilterList[getProbeChildPart(i)],
            filteredRows,
            enableSubPartStat,
            enableSwing);

        newChildPlan->addSibling(firstChildPlan);
        firstChildPlan = newChildPlan;
    }
    partInfo.destPartitionList.clear();
    partInfo.joinFilterList.clear();
}

uint LhxPlan::getPartitionLevel

(

)

[inline]

Get the partition level of this plan.

Definition at line 505 of file LhxPartition.h.

References partitionLevel.

Referenced by LhxJoinExecStream::execute(), LhxAggExecStream::execute(), and LhxJoinExecStream::open().

{
    return partitionLevel;
}

SharedLhxPartition LhxPlan::getBuildPartition

(

)

[inline]

Get the partition corresponding to inputIndex.

Definition at line 525 of file LhxPartition.h.

References getBuildInput(), and partitions.

Referenced by LhxJoinExecStream::execute(), and LhxJoinExecStream::open().

{
    return partitions[getBuildInput()];
}

SharedLhxPartition LhxPlan::getProbePartition

(

)

[inline]

Definition at line 520 of file LhxPartition.h.

References getProbeInput(), and partitions.

Referenced by LhxJoinExecStream::execute().

{
    return partitions[getProbeInput()];
}

SharedLhxPartition LhxPlan::getPartition

(

uint

inputIndex

)

[inline]

Definition at line 530 of file LhxPartition.h.

References partitions.

Referenced by LhxAggExecStream::execute(), LhxAggExecStream::open(), and LhxHashTableTest::testInsert().

{
    return partitions[inputIndex];
}

uint LhxPlan::getBuildInput

(

)

[inline]

Definition at line 515 of file LhxPartition.h.

References joinSideToInputMap, and partitions.

Referenced by LhxJoinExecStream::execute(), getBuildChildPart(), getBuildPartition(), isBuildChildPart(), mapSubPartToChild(), LhxJoinExecStream::open(), LhxJoinExecStream::returnBuildInner(), and LhxJoinExecStream::returnBuildOuter().

{
    return joinSideToInputMap[partitions.size() - 1];
}

uint LhxPlan::getProbeInput

(

)

[inline]

Definition at line 510 of file LhxPartition.h.

References joinSideToInputMap.

Referenced by LhxJoinExecStream::execute(), generatePartitions(), getProbeChildPart(), getProbePartition(), isProbeChildPart(), mapSubPartToChild(), LhxJoinExecStream::returnProbeInner(), and LhxJoinExecStream::returnProbeOuter().

{
    return joinSideToInputMap[0];
}

uint LhxPlan::getJoinSide

(

uint

inputIndex

)

[inline]

Definition at line 535 of file LhxPartition.h.

References joinSideToInputMap, and partitions.

Referenced by toString().

{
    uint i = 0;
    while ((joinSideToInputMap[i] != inputIndex)
        && (i < partitions.size()))
    {
        i ++;
    }

    return i;
}

SharedLhxPlan LhxPlan::getFirstChild

(

)

[inline]

Get the first child plan.

Definition at line 500 of file LhxPartition.h.

References firstChildPlan.

Referenced by LhxJoinExecStream::execute(), and LhxAggExecStream::execute().

{
    return firstChildPlan;
}

LhxPlan * LhxPlan::getFirstLeaf

(

)

Get first leaf plan in dfs order.

Definition at line 1033 of file LhxPartition.cpp.

References firstChildPlan.

Referenced by LhxHashTableTest::testInsert().

{
    if (!firstChildPlan) {
        return this;
    } else {
        return firstChildPlan->getFirstLeaf();
    }
}

LhxPlan * LhxPlan::getNextLeaf

(

)

Get next leaf plan in dfs order.

Definition at line 1042 of file LhxPartition.cpp.

References parentPlan, and siblingPlan.

Referenced by LhxJoinExecStream::execute(), LhxAggExecStream::execute(), and LhxHashTableTest::testInsert().

{
    if (siblingPlan) {
        return siblingPlan->getFirstLeaf();
    } else {
        WeakLhxPlan parent = this->parentPlan;
        SharedLhxPlan shared_parent = parent.lock();

        if (shared_parent) {
            return shared_parent->getNextLeaf();
        } else {
            return NULL;
        }
    }
}

void LhxPlan::close

(

)

Close the plan tree.

Release any resource pointed to from this plan tree.

Definition at line 1058 of file LhxPartition.cpp.

References firstChildPlan, partitions, and siblingPlan.

Referenced by createChildren().

{
    if (firstChildPlan) {
        firstChildPlan->close();
    }

    if (siblingPlan) {
        siblingPlan->close();
    }

    for (uint i = 0; i < partitions.size(); i ++) {
        if (partitions[i] && partitions[i]->segStream) {
            partitions[i]->segStream->close();
        }
    }
}

string LhxPlan::toString

(

)

Print the content of the plan tree rooted at this plan.

Returns:

the string representation of this plan tree.

Definition at line 1075 of file LhxPartition.cpp.

References childPartSize, filteredRowCount, firstChildPlan, getJoinSide(), inputSize, joinFilter, LhxChildPartCount, LhxSubPartCount, parentPlan, partitionLevel, partitions, siblingPlan, and subPartToChildMap.

Referenced by LhxJoinExecStream::execute(), and LhxAggExecStream::execute().

{
    ostringstream planTrace;

    planTrace << "\n"
              << "[Plan : addr       = " << this                 << "]\n"
              << "[       level      = " << partitionLevel       << "]\n"
              << "[       parent     = " << parentPlan.lock().get()  << "]\n"
              << "[       firstChild = " << firstChildPlan.get() << "]\n"
              << "[       sibling    = " << siblingPlan.get()    << "]\n";

    /*
     * Print out the partitions.
     */
    planTrace << "[joinFilter = ";
    if (joinFilter) {
        planTrace << joinFilter.get();
    }
    planTrace << "]\n";

    for (uint i = 0; i < partitions.size(); i ++) {
        planTrace << "[Partition(" << i << ")]\n"
                  << "[       inputIndex     = " << partitions[i]->inputIndex << "]\n"
                  << "[       join side      = " << getJoinSide(partitions[i]->inputIndex) << "]\n"
                  << "[       filteredRows   = " << filteredRowCount[i] << "]\n"
                  << "[       inputSize      = " << inputSize[i] << "]\n";
        planTrace << "[       childPartSize  = ";
        if (childPartSize.size() > i) {
            shared_array<uint> oneChildPartSize = childPartSize[i];
            if (oneChildPartSize) {
                for (uint j = 0; j < LhxChildPartCount; j ++) {
                    planTrace << oneChildPartSize[j] << " ";
                }
            }
        }
        planTrace << "]\n";
    }

    planTrace << "[subPartToChildMap = ";
    if (subPartToChildMap) {
        for (uint i = 0; i < LhxSubPartCount; i ++) {
            planTrace << subPartToChildMap[i] << " ";
        }
    }
    planTrace << "]\n";

    SharedLhxPlan childPlan = firstChildPlan;

    while (childPlan) {
        planTrace << childPlan->toString();
        childPlan = childPlan->siblingPlan;
    }

    return planTrace.str();
}

---

Member Data Documentation

uint LhxPlan::partitionLevel [private]

Definition at line 294 of file LhxPartition.h.

Referenced by createChildren(), generatePartitions(), getPartitionLevel(), init(), and toString().

vector<SharedLhxPartition> LhxPlan::partitions [private]

Definition at line 295 of file LhxPartition.h.

Referenced by close(), createChildren(), getBuildInput(), getBuildPartition(), getJoinSide(), getPartition(), getProbePartition(), init(), mapSubPartToChild(), and toString().

shared_array<uint> LhxPlan::joinSideToInputMap [private]

Definition at line 296 of file LhxPartition.h.

Referenced by getBuildInput(), getJoinSide(), getProbeInput(), and init().

shared_ptr<dynamic_bitset<> > LhxPlan::joinFilter [private]

Definition at line 298 of file LhxPartition.h.

Referenced by generatePartitions(), init(), and toString().

shared_array<uint> LhxPlan::filteredRowCount [private]

Definition at line 299 of file LhxPartition.h.

Referenced by init(), and toString().

shared_array<uint> LhxPlan::subPartToChildMap [private]

Definition at line 306 of file LhxPartition.h.

Referenced by calculateChildIndex(), init(), mapSubPartToChild(), and toString().

vector<shared_array<uint> > LhxPlan::childPartSize [private]

Definition at line 307 of file LhxPartition.h.

Referenced by mapSubPartToChild(), and toString().

shared_array<uint> LhxPlan::inputSize [private]

Definition at line 309 of file LhxPartition.h.

Referenced by init(), and toString().

WeakLhxPlan LhxPlan::parentPlan [private]

Definition at line 322 of file LhxPartition.h.

Referenced by getNextLeaf(), init(), and toString().

SharedLhxPlan LhxPlan::firstChildPlan [private]

Definition at line 323 of file LhxPartition.h.

Referenced by close(), createChildren(), getFirstChild(), getFirstLeaf(), and toString().

SharedLhxPlan LhxPlan::siblingPlan [private]

Definition at line 324 of file LhxPartition.h.

Referenced by addSibling(), close(), getNextLeaf(), and toString().

const uint LhxPlan::LhxSubPartCount = 16 [static]

Definition at line 358 of file LhxPartition.h.

Referenced by calculateChildIndex(), generatePartitions(), init(), mapSubPartToChild(), LhxPartitionInfo::open(), and toString().

const uint LhxPlan::LhxChildPartCount = 3 [static]

Definition at line 359 of file LhxPartition.h.

Referenced by calculateChildIndex(), createChildren(), getBuildChildPart(), getProbeChildPart(), LhxAggExecStream::getResourceRequirements(), LhxPartitionInfo::init(), isBuildChildPart(), isProbeChildPart(), mapSubPartToChild(), LhxPartitionInfo::open(), LhxJoinExecStream::prepare(), LhxAggExecStream::prepare(), LhxHashTableTest::testInsert(), and toString().

---

The documentation for this class was generated from the following files:

• /home/pub/open/dev/fennel/hashexe/LhxPartition.h
• /home/pub/open/dev/fennel/hashexe/LhxPartition.cpp

---