LcsHash.h

Go to the documentation of this file.

/*
// $Id: //open/dev/fennel/lucidera/colstore/LcsHash.h#16 $
// Fennel is a library of data storage and processing components.
// Copyright (C) 2005-2009 LucidEra, Inc.
// Copyright (C) 2005-2009 The Eigenbase Project
//
// This program is free software; you can redistribute it and/or modify it
// under the terms of the GNU General Public License as published by the Free
// Software Foundation; either version 2 of the License, or (at your option)
// any later version approved by The Eigenbase Project.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

#ifndef Fennel_LcsHash_Included
#define Fennel_LcsHash_Included

#include "fennel/tuple/TupleData.h"
#include "fennel/tuple/TupleDataWithBuffer.h"
#include "fennel/tuple/TupleDescriptor.h"
#include "fennel/tuple/UnalignedAttributeAccessor.h"
#include "fennel/lucidera/colstore/LcsClusterNodeWriter.h"


FENNEL_BEGIN_NAMESPACE

/************************************************
  classes and defines used only by LcsHash class
 ************************************************/

class FENNEL_LCS_EXPORT LcsHashValOrd
{
private:

    uint16_t    valOrd;

public:

    LcsHashValOrd() {}
    ~LcsHashValOrd() {}

    inline explicit LcsHashValOrd(LcsHashValOrd const &other);

    inline LcsHashValOrd& operator=(uint16_t valOrdInit);

    inline uint16_t getValOrd();

    inline void setValOrd(uint16_t valOrdInit);

    inline bool isValueInBatch();

    inline void setValueInBatch();

    inline void clearValueInBatch();
};


struct LcsHashValueNode
{
    uint16_t      next;

    LcsHashValOrd valueOrd;

    uint16_t      valueOffset;

    uint16_t      sortedOrd;
};


class FENNEL_LCS_EXPORT LcsHashTable
{
private:

    PBuffer           hashBlock;

    uint              hashBlockSize;

    uint              hashTableSize;

    uint              nextValueNode;

public:

    uint16_t          *entry;

    LcsHashValueNode  *valueNodes;

    inline explicit LcsHashTable();

    void init(PBuffer hashBlockInit, uint hashBlockSizeInit);

    inline void resetHash();

    inline void resetBatch();

    inline uint numHashEntries();

    inline LcsHashValueNode* getNewValueNode();

    inline void insertNewValueNode(uint key,  LcsHashValueNode *newNode);

    inline void undoNewValueNode(uint key);

    inline LcsHashValueNode* getFirstValueNode(uint key);

    inline LcsHashValueNode* getNextValueNode(LcsHashValueNode *pValueNode);

    inline bool isFull(uint leftOvers = 0);
};


enum LcsUndoState { NOTHING, NEWENTRY, NEWBATCHVALUE };


struct LcsUndoType
{
    LcsUndoState     what;

    uint             key;

    uint             origMaxValueSize;

    LcsHashValueNode *vPtr;

    inline LcsUndoType();

    inline void set(
        LcsUndoState whatInit,
        uint keyInit,
        uint origMaxValueSizeInit,
        LcsHashValueNode *vPtrInit);

    inline void reset();
};


class FENNEL_LCS_EXPORT LcsCompareColKeyUsingOffsetIndex
{
private:

    LcsHashTable            *hashTable;

    SharedLcsClusterNodeWriter    clusterBlockWriter;

    uint                     columnId;

    TupleDataWithBuffer      colTuple1;
    TupleDataWithBuffer      colTuple2;

    TupleDescriptor          colTupleDesc;

    UnalignedAttributeAccessor attrAccessor;

public:

    explicit LcsCompareColKeyUsingOffsetIndex(
        SharedLcsClusterNodeWriter clusterBlockWriterInit,
        LcsHashTable *hashTableInit,
        TupleDescriptor const &colTupleDescInit,
        uint columnIdInit,
        UnalignedAttributeAccessor const &attrAccessorInit);

    ~LcsCompareColKeyUsingOffsetIndex() {}

    bool lessThan(
        const uint16_t colKeyOffsetIndex1,
        const uint16_t colKeyOffsetIndex2);
};

class FENNEL_LCS_EXPORT LcsCompare
{
private:

    SharedLcsCompareColKeyUsingOffsetIndex compareInstance;

public:

    inline explicit LcsCompare(
        SharedLcsCompareColKeyUsingOffsetIndex compareInstanceInit);

    inline bool operator()(
        const uint16_t colKeyOffsetIndex1,
        const uint16_t colKeyOffsetIndex2);
};


class FENNEL_LCS_EXPORT LcsHash
{
private:

    uint                  columnId;

    LcsHashTable          hash;

    SharedLcsClusterNodeWriter clusterBlockWriter;

    TupleDescriptor       colTupleDesc;

    TupleDataWithBuffer   colTuple;

    UnalignedAttributeAccessor attrAccessor;

    TupleDataWithBuffer   searchTuple;

    boost::scoped_array<FixedBuffer> colTupleBuffer;

    uint16_t              valCnt;

    uint                  maxValueSize;

    LcsUndoType           undo;

    uint8_t              *magicTable;

    uint                  numChecks;

    uint                  numMatches;

    SharedLcsCompareColKeyUsingOffsetIndex compareInst;

    uint computeKey(PBuffer dataWithLen);

    bool search(
        uint key, PBuffer dataWithLen,
        LcsHashValOrd *valOrd, LcsHashValueNode **v);

public:
    explicit LcsHash();
    ~LcsHash()
    {
    }

    void init(
        PBuffer hashBlockInit,
        SharedLcsClusterNodeWriter clusterBlockWriterInit,
        TupleDescriptor const &colTupleDescInit,
        uint columnIdInit,
        uint blockSizeInit);

    void insert(
        TupleDatum &colTupleDatum,
        LcsHashValOrd *valOrd,
        bool *undoInsert);

    void insert(
        PBuffer dataWithLen,
        LcsHashValOrd *valOrd,
        bool *undoInsert);

    void undoInsert(TupleDatum &colTupleDatum);

    void undoInsert(PBuffer dataWithLen);


    void prepareFixedOrVariableBatch(uint8_t *rowArray, uint numRows);

    void prepareCompressedBatch(
        uint8_t *rowArray,
        uint     numRows,
        uint16_t *numVals,
        uint16_t *offsetIndexVector);

    void clearFixedEntries();

    void startNewBatch(uint leftOvers);

    void restore(uint numVals, uint16_t lastValOff);

    inline uint getMaxValueSize();

    inline bool isHashFull(uint leftOvers = 0);
};


/*******************************************************
  Definitions of inline methods for class LcsHashValOrd
 *******************************************************/

inline LcsHashValOrd::LcsHashValOrd(LcsHashValOrd const &other)
{
    valOrd = other.valOrd;
}

inline LcsHashValOrd& LcsHashValOrd::operator=(uint16_t valOrdInit)
{
    setValOrd(valOrdInit);
    return *this;
}

inline uint16_t LcsHashValOrd::getValOrd()
{
    return (uint16_t)(valOrd & ~(1<<15));
}

inline void LcsHashValOrd::setValOrd(uint16_t valOrdInit)
{
    valOrd = (valOrdInit & ~(1<<15));
}

inline bool LcsHashValOrd::isValueInBatch()
{
    return (valOrd & (1<<15));
}

inline void LcsHashValOrd::setValueInBatch()
{
    valOrd |= 1<<15;
}

inline void LcsHashValOrd::clearValueInBatch()
{
    valOrd &= ~(1<<15);
}


/*****************************************************
  Definitions of inline methods for class LcsUndoType
 *****************************************************/

inline LcsUndoType::LcsUndoType()
{
    reset();
}

inline void LcsUndoType::set(
    LcsUndoState whatInit,
    uint keyInit,
    uint origMaxValueSizeInit,
    LcsHashValueNode *vPtrInit)
{
    what = whatInit;
    key = keyInit;
    origMaxValueSize = origMaxValueSizeInit;;
    vPtr = vPtrInit;
}

inline void LcsUndoType::reset()
{
    what = NOTHING;
    key = 0;
    origMaxValueSize = 0;
    vPtr = 0;
}


/******************************************************
  Definitions of inline methods for class LcsHashTable
 ******************************************************/

inline LcsHashTable::LcsHashTable()
{
    hashBlock = NULL;
    hashTableSize = 0;
    nextValueNode = 0;
}

inline void LcsHashTable::resetHash()
{
    memset(hashBlock, 0, hashBlockSize);
    nextValueNode = 0;
}

inline void LcsHashTable::resetBatch()
{
    for (int i = 0; i < nextValueNode; i++) {
        (&(valueNodes[i].valueOrd))->clearValueInBatch();
    }
}

inline uint LcsHashTable::numHashEntries()
{
    return hashTableSize;
}

inline LcsHashValueNode* LcsHashTable::getNewValueNode()
{
    return &(valueNodes[nextValueNode]);
}

inline void LcsHashTable::insertNewValueNode(
    uint key,
    LcsHashValueNode *newNode)
{
    newNode->next = (uint16_t)entry[key];

    /*
      Insert at the beginning.
    */
    entry[key] = (uint16_t)((uint8_t*)newNode - hashBlock);

    /*
      Bump up the value for the next node to give out.
    */
    nextValueNode++;
}

inline void LcsHashTable::undoNewValueNode(uint key)
{
    entry[key] = (uint16_t)
        ((LcsHashValueNode*)(hashBlock + entry[key]))->next;
    nextValueNode--;
}

inline LcsHashValueNode* LcsHashTable::getFirstValueNode(uint key)
{
    uint16_t offset = entry[key];

    if (offset) {
        return (LcsHashValueNode*) (hashBlock + offset);
    } else {
        return NULL;
    }
}

inline LcsHashValueNode* LcsHashTable::getNextValueNode(
    LcsHashValueNode *pValueNode)
{
    uint16_t offset = pValueNode->next;

    if (offset) {
        return (LcsHashValueNode*) (hashBlock + offset);
    } else {
        return NULL;
    }
}

inline bool LcsHashTable::isFull(uint leftOvers)
{
    /*
      need to leave one spot for sorting and leftsOvers that
      will be added
    */
    return ((PBuffer)(&(valueNodes[nextValueNode])
            + (leftOvers + 1) * sizeof(LcsHashValueNode))
        >= (PBuffer) (hashBlock + hashBlockSize));
}


/****************************************************
  Definitions of inline methods for class LcsCompare
 ****************************************************/

inline LcsCompare::LcsCompare(
    SharedLcsCompareColKeyUsingOffsetIndex compareInstanceInit)
{
    compareInstance = compareInstanceInit;
}

inline bool LcsCompare::operator()(
    const uint16_t colKeyOffsetIndex1,
    const uint16_t colKeyOffsetIndex2)
{
    return compareInstance->lessThan(colKeyOffsetIndex1, colKeyOffsetIndex2);
}


/*************************************************
  Definitions of inline methods for class LcsHash
 *************************************************/

inline uint LcsHash::getMaxValueSize()
{
    return maxValueSize;
}

inline bool LcsHash::isHashFull(uint leftOvers)
{
    return hash.isFull(leftOvers);
}

FENNEL_END_NAMESPACE

#endif

// End LcsHash.h

---