<commit_before>#include <ir/index_manager/index/PostingMerger.h>
#include <ir/index_manager/index/RTPostingWriter.h>
#include <ir/index_manager/index/EPostingWriter.h>
#include <ir/index_manager/store/FSIndexOutput.h>
#include <ir/index_manager/store/FSIndexInput.h>

#include <cassert>

NS_IZENELIB_IR_BEGIN

namespace indexmanager
{

PostingMerger::PostingMerger(
    int skipInterval,
    int maxSkipLevel,
    CompressionType compressType,
    bool optimize,
    bool requireIntermediateFileForMerging,
    IndexLevel indexLevel,
    size_t memPoolSizeForPostingMerger
)
    :skipInterval_(skipInterval)
    ,maxSkipLevel_(maxSkipLevel)
    ,compressType_(compressType)
    ,pOutputDescriptor_(NULL)
    ,pTmpPostingOutput_(NULL)
    ,pTmpPostingInput_(NULL)
    ,pVIntDataOutput_(NULL)
    ,requireIntermediateFileForMerging_(requireIntermediateFileForMerging)
    ,nPPostingLength_(0)
    ,nSkipIntervalBetweenBarrels_(0)
    ,bFirstPosting_(true)
    ,pSkipListMerger_(NULL)
    ,pFixedSkipListWriter_(NULL)
    ,pSkipListWriter_(NULL)
    ,compressedPos_(NULL)
    ,compressed_position_buffer_size_(0)
    ,doc_ids_offset_(0)
    ,positions_(NULL)
    ,curr_position_buffer_size_(0)
    ,position_buffer_pointer_(0)
    ,pPosDataPool_(NULL)
    ,pDocFreqDataPool_(NULL)
    ,block_buffer_(NULL)
    ,current_block_id_(0)
    ,optimize_(optimize)
    ,indexLevel_(indexLevel)
{
    // to avoid concurrent memory request,
    // such as by ChunkPostingWriter::pDocFreqDataPool_ in another thread,
    // posting merger use a seperated memory pool exclusively
    pMemCache_.reset(new MemCache(memPoolSizeForPostingMerger));

    init();
    reset();
}

PostingMerger::~PostingMerger()
{
    if(pSkipListMerger_)
        delete pSkipListMerger_;
    if(pTmpPostingOutput_)
    {
        delete pTmpPostingInput_;
        delete pTmpPostingOutput_;
        pOutputDescriptor_->getDirectory()->deleteFile(tmpPostingName_);
    }
    if(pVIntDataOutput_)
    {
        delete pVIntDataOutput_;
    }
    delete [] compressedPos_;
    delete [] positions_;
    if(block_buffer_) delete [] block_buffer_;
    if(pPosDataPool_)
        delete pPosDataPool_;
    delete pDocFreqDataPool_;
    delete pFixedSkipListWriter_;
    delete pSkipListWriter_;

}

void PostingMerger::reset()
{
    ///reset posting descriptor
    postingDesc_.length = 0;
    postingDesc_.plength = 0;
    postingDesc_.ctf = 0;//term frequence
    postingDesc_.df = 0;//doc frequence
    postingDesc_.maxTF = 0;
    postingDesc_.poffset = -1;
    chunkDesc_.lastdocid = 0;
    chunkDesc_.length = 0;
    termInfo_.reset();
    pMemCache_->flushMem();
    nSkipIntervalBetweenBarrels_ = 0;
    if(pSkipListMerger_)
    {
        pSkipListMerger_->reset();
    }
    if(pTmpPostingOutput_)
    {
        reinterpret_cast<FSIndexOutput*>(pTmpPostingOutput_)->trunc();
    }

    if(pVIntDataOutput_)
        pVIntDataOutput_->reset();
    if(pSkipListWriter_)
    {
        pSkipListWriter_->reset();
    }

    pFixedSkipListWriter_->reset();
    pDocFreqDataPool_->reset();
    if (pPosDataPool_)
        pPosDataPool_->reset();
    chunk_.reset();

    doc_ids_offset_ = 0;
    position_buffer_pointer_ = 0;
    blockEncoder_.reset();
    //current_block_id_ = 0;
}

void PostingMerger::init()
{
    if(isSkipParamValid())
    {
        pSkipListMerger_ = new SkipListMerger(skipInterval_,maxSkipLevel_,pMemCache_);
        pSkipListWriter_ = new SkipListWriter(skipInterval_,maxSkipLevel_,pMemCache_);
    }

    pVIntDataOutput_ = new VariantDataPool(pMemCache_);

    compressed_position_buffer_size_ = INIT_POS_CHUNK_SIZE;
    compressedPos_ = new uint32_t[compressed_position_buffer_size_];
    curr_position_buffer_size_ = INIT_POS_CHUNK_SIZE << 1;
    positions_ = new uint32_t[curr_position_buffer_size_];
    pFixedSkipListWriter_ = new FixedBlockSkipListWriter(pMemCache_);

    if (indexLevel_ == WORDLEVEL)
        pPosDataPool_ = new ChunkDataPool(pMemCache_) ;
    pDocFreqDataPool_ = new ChunkDataPool(pMemCache_);
}

void PostingMerger::setOutputDescriptor(OutputDescriptor* pOutputDescriptor)
{
    pOutputDescriptor_ = pOutputDescriptor;
    tmpPostingName_ = pOutputDescriptor_->getBarrelName() + ".merge";
    pTmpPostingOutput_ = pOutputDescriptor_->getDirectory()->createOutput(tmpPostingName_);
}

void PostingMerger::mergeWith(RTDiskPostingReader* pOnDiskPosting,BitVector* pFilter)
{
    if(optimize_)
    {
        optimize(pOnDiskPosting, pFilter);
    }
    else
    {
        if(pFilter)
            mergeWith_GC(pOnDiskPosting,pFilter);
        else
            mergeWith(pOnDiskPosting);
    }
}

void PostingMerger::mergeWith(MemPostingReader* pInMemoryPosting)
{
    ///flush last doc
    pInMemoryPosting->pPostingWriter_->flushLastDoc(true);

    IndexOutput* pDOutput = pOutputDescriptor_->getDPostingOutput();
    IndexOutput* pPOutput = pOutputDescriptor_->getPPostingOutput();

    if (bFirstPosting_)///first posting
    {
        reset();
        if(isSkipParamValid())
        {
            termInfo_.skipPointer_ = pDOutput->getFilePointer();
        }
        else
        {
            termInfo_.docPointer_ = pDOutput->getFilePointer();
            termInfo_.skipPointer_ = -1;
        }
        if (pPOutput)
            termInfo_.positionPointer_ = pPOutput->getFilePointer();
        nPPostingLength_ = 0;
        bFirstPosting_ = false;
        ///save position posting offset
        if(pPOutput)
            postingDesc_.poffset = pPOutput->getFilePointer();
    }

    OutputStream* pDocIndexOutput = 0;

    if(isSkipParamValid())
    {
        if(requireIntermediateFileForMerging_)
            /// we need an extra intermediate file to store merged posting data without skiplist
            pDocIndexOutput = pTmpPostingOutput_;
        else
            pDocIndexOutput = pVIntDataOutput_;
    }
    else
        pDocIndexOutput = pDOutput;

    fileoffset_t oldDOff = pDocIndexOutput->getLength();

    ///write chunk data, update the first doc id
    VariantDataChunk* pDChunk = pInMemoryPosting->pPostingWriter_->pDocFreqList_->pHeadChunk_;

    ///DPosting start for skip
    int32_t DOffsetStartForSkipping = 0;

    if (pDChunk)
    {
        uint8_t* bp = &(pDChunk->data[0]);
        docid_t firstDocID = VariantDataPool::decodeVData32(bp) - chunkDesc_.lastdocid;

        if(chunkDesc_.lastdocid == 0)
            DOffsetStartForSkipping = 0;
        else
            DOffsetStartForSkipping =
                postingDesc_.length - (IndexOutput::getVIntLength(firstDocID + chunkDesc_.lastdocid) - IndexOutput::getVIntLength(firstDocID));

        pDocIndexOutput->writeVInt(firstDocID);///write first doc id
        int32_t writeSize = pDChunk->size - (bp - &(pDChunk->data[0]));	//write the rest data of first chunk
        if (writeSize > 0)
            pDocIndexOutput->write((const char*)bp,writeSize);
        pDChunk = pDChunk->next;
    }
    ///write rest data
    while (pDChunk)
    {
        pDocIndexOutput->write((const char*)pDChunk->data,pDChunk->size);
        pDChunk = pDChunk->next;
    }

    chunkDesc_.length += (pDocIndexOutput->getLength() - oldDOff);

    ///write position posting
    if (indexLevel_ == WORDLEVEL && pInMemoryPosting->pPostingWriter_->pLocList_)
    {
        VariantDataChunk* pPChunk = pInMemoryPosting->pPostingWriter_->pLocList_->pHeadChunk_;
        while (pPChunk)
        {
            pPOutput->write((const char*)pPChunk->data,pPChunk->size);
            pPChunk = pPChunk->next;
        }
    }

    ///merge skiplist
    SkipListReader* pSkipReader = pInMemoryPosting->getSkipListReader();
    if(pSkipReader)
    {
        assert(isSkipParamValid() && pSkipListMerger_);

        pSkipListMerger_->setBasePoint(0, DOffsetStartForSkipping, postingDesc_.plength);
        pSkipListMerger_->addToMerge(pSkipReader,pInMemoryPosting->lastDocID(), nSkipIntervalBetweenBarrels_);
        nSkipIntervalBetweenBarrels_ = pInMemoryPosting->docFreq() - pSkipReader->getNumSkipped();
    }
    else
        nSkipIntervalBetweenBarrels_ += pInMemoryPosting->docFreq();

    ///update descriptors
    postingDesc_.ctf += pInMemoryPosting->getCTF();
    postingDesc_.df += pInMemoryPosting->docFreq();
    if(postingDesc_.maxTF < pInMemoryPosting->getMaxDocFreq())
        postingDesc_.maxTF = pInMemoryPosting->getMaxDocFreq();
    postingDesc_.length = chunkDesc_.length;
    postingDesc_.plength += pInMemoryPosting->getPPostingLen();
    chunkDesc_.lastdocid = pInMemoryPosting->lastDocID();
}

void PostingMerger::mergeWith(RTDiskPostingReader* pOnDiskPosting)
{
    IndexOutput* pDOutput = pOutputDescriptor_->getDPostingOutput();
    IndexOutput* pPOutput = pOutputDescriptor_->getPPostingOutput();
    IndexInput*	pDInput = pOnDiskPosting->getInputDescriptor()->getDPostingInput();
    IndexInput* pPInput = pOnDiskPosting->getInputDescriptor()->getPPostingInput();

    if (bFirstPosting_)///first posting
    {
        reset();
        if(isSkipParamValid())
        {
            termInfo_.skipPointer_ = pDOutput->getFilePointer();
        }
        else
        {
            termInfo_.docPointer_ = pDOutput->getFilePointer();
            termInfo_.skipPointer_ = -1;
        }
        if (pPOutput)
            termInfo_.positionPointer_ = pPOutput->getFilePointer();

        nPPostingLength_ = 0;
        bFirstPosting_ = false;

        ///save position offset
        if (pPOutput)
            postingDesc_.poffset = pPOutput->getFilePointer();
    }

    OutputStream* pDocIndexOutput = 0;

    if(isSkipParamValid())
    {
        if(requireIntermediateFileForMerging_)
            /// we need an extra intermediate file to store merged posting data without skiplist
            pDocIndexOutput = pTmpPostingOutput_;
        else
            pDocIndexOutput = pVIntDataOutput_;
    }
    else
        pDocIndexOutput = pDOutput;

    fileoffset_t oldDOff = pDocIndexOutput->getLength();
    docid_t firstDocID = pDInput->readVInt() - chunkDesc_.lastdocid;

    ///DPosting start for skip
    int32_t DOffsetStartForSkipping = 0;
    if(chunkDesc_.lastdocid == 0)
        DOffsetStartForSkipping = 0;
    else
        DOffsetStartForSkipping =
            postingDesc_.length - (IndexOutput::getVIntLength(firstDocID + chunkDesc_.lastdocid) - IndexOutput::getVIntLength(firstDocID));

    pDocIndexOutput->writeVInt(firstDocID);///write first doc id
    int64_t writeSize = pOnDiskPosting->postingDesc_.length - IndexOutput::getVIntLength(firstDocID + chunkDesc_.lastdocid);
    if (writeSize > 0)
        pDocIndexOutput->write(pDInput,writeSize);

    chunkDesc_.length += (pDocIndexOutput->getLength() - oldDOff);

    ///write position posting
    if (pPOutput && pPInput)
        pPOutput->write(pPInput,pOnDiskPosting->nPPostingLength_);

    ///merge skiplist
    SkipListReader* pSkipReader = pOnDiskPosting->getSkipListReader();

    if(pSkipReader)
    {
        assert(isSkipParamValid() && pSkipListMerger_);

        pSkipListMerger_->setBasePoint(0, DOffsetStartForSkipping, postingDesc_.plength);
        pSkipListMerger_->addToMerge(pSkipReader,pOnDiskPosting->chunkDesc_.lastdocid, nSkipIntervalBetweenBarrels_);
        nSkipIntervalBetweenBarrels_ = pOnDiskPosting->docFreq() - pSkipReader->getNumSkipped();
    }
    else
        nSkipIntervalBetweenBarrels_ += pOnDiskPosting->docFreq();

    ///update descriptors
    postingDesc_.ctf += pOnDiskPosting->postingDesc_.ctf;
    postingDesc_.df += pOnDiskPosting->postingDesc_.df;
    if(postingDesc_.maxTF < pOnDiskPosting->postingDesc_.maxTF)
        postingDesc_.maxTF = pOnDiskPosting->postingDesc_.maxTF;
    postingDesc_.length = chunkDesc_.length; ///currently,it's only one chunk
    postingDesc_.plength += pOnDiskPosting->nPPostingLength_;
    chunkDesc_.lastdocid = pOnDiskPosting->chunkDesc_.lastdocid;
}

void PostingMerger::mergeWith_GC(RTDiskPostingReader* pOnDiskPosting,BitVector* pFilter)
{
    IndexOutput* pDOutput = pOutputDescriptor_->getDPostingOutput();
    IndexOutput* pPOutput = pOutputDescriptor_->getPPostingOutput();
    IndexInput*	pDInput = pOnDiskPosting->getInputDescriptor()->getDPostingInput();
    IndexInput* pPInput = pOnDiskPosting->getInputDescriptor()->getPPostingInput();

    count_t nODDF = pOnDiskPosting->postingDesc_.df;
    if(nODDF <= 0)
        return;

    if (bFirstPosting_)///first posting
    {
        reset();
        if(isSkipParamValid())
        {
            termInfo_.skipPointer_ = pDOutput->getFilePointer();
        }
        else
        {
            termInfo_.docPointer_ = pDOutput->getFilePointer();
            termInfo_.skipPointer_ = -1;
        }
        if (pPOutput)
            termInfo_.positionPointer_ = pPOutput->getFilePointer();

        nPPostingLength_ = 0;
        bFirstPosting_ = false;

        ///save position offset
        if(pPOutput)
            postingDesc_.poffset = pPOutput->getFilePointer();
    }

    docid_t nDocID = 0;
    docid_t nLastDocID = chunkDesc_.lastdocid;
    freq_t	nTF = 0;
    count_t nCTF = 0;
    count_t nMtf = 0;
    count_t nDF = postingDesc_.df;
    count_t nPCount = 0;

    OutputStream* pDocIndexOutput = 0;

    if(isSkipParamValid())
    {
        if(requireIntermediateFileForMerging_)
            /// we need an extra intermediate file to store merged posting data without skiplist
            pDocIndexOutput = pTmpPostingOutput_;
        else
            pDocIndexOutput = pVIntDataOutput_;
    }
    else
        pDocIndexOutput = pDOutput;

    fileoffset_t oldDOff = pDocIndexOutput->getLength();
    fileoffset_t oldPOff = -1;
    if (pPOutput)
        oldPOff = pPOutput->getFilePointer();

    while (nODDF > 0)
    {
        nDocID += pDInput->readVInt();
        nTF = pDInput->readVInt();
        if(!pFilter->test((size_t)nDocID))///the document has not been deleted
        {
            pDocIndexOutput->writeVInt(nDocID - nLastDocID);
            pDocIndexOutput->writeVInt(nTF);

            nPCount += nTF;
            nDF++;
            if(nMtf < nTF)
                nMtf = nTF;
            nLastDocID = nDocID;
            if(pSkipListMerger_ && nDF > 0 && nDF % skipInterval_ == 0)
            {
                if(pPOutput)
                    pSkipListMerger_->addSkipPoint(nLastDocID,pDocIndexOutput->getLength(),pPOutput->getFilePointer());
                else
                    pSkipListMerger_->addSkipPoint(nLastDocID,pDocIndexOutput->getLength(),0);
            }
        }
        else ///this document has been deleted
        {
            nCTF += nPCount;
            ///write positions of documents
            while (nPCount > 0)
            {
                if (pPOutput && pPInput)
                    pPOutput->writeVInt(pPInput->readVInt());
                nPCount--;
            }
            ///skip positions of deleted documents
            while (nTF > 0)
            {
                if (pPInput)
                    pPInput->readVInt();
                nTF--;
            }
        }
        nODDF--;
    }
    if(nPCount > 0)
    {
        nCTF += nPCount;
        while (nPCount > 0)
        {
            if (pPOutput && pPInput)
                pPOutput->writeVInt(pPInput->readVInt());
            nPCount--;
        }
    }

    chunkDesc_.length += (pDocIndexOutput->getLength() - oldDOff);

    ///update descriptors
    postingDesc_.ctf += nCTF;
    postingDesc_.df += nDF;
    if((uint32_t)postingDesc_.maxTF < nMtf)
        postingDesc_.maxTF = nMtf;
    postingDesc_.length = chunkDesc_.length; ///currently,it's only one chunk
    if (pPOutput)
        postingDesc_.plength += pPOutput->getFilePointer() - oldPOff;
    chunkDesc_.lastdocid = nLastDocID;
}

void PostingMerger::mergeWith(BlockPostingReader* pPosting,BitVector* pFilter)
{
    IndexInput* pPInput = pPosting->inputDescriptorPtr_->getPPostingInput();

    if (bFirstPosting_)///first posting
    {
        reset();
        nPPostingLength_ = 0;
        bFirstPosting_ = false;
        if(! block_buffer_)
            block_buffer_ = new uint8_t[BLOCK_SIZE];
    }

    int num_docs_left = pPosting->df_;
    int size_of_positions = 0;
<commit_msg>Fix bug for PostingMerger: Skip point is not added to SkipMerger correctly within MergeWith_GC, the optimized indices might lead to corruption in future's merging<commit_after>#include <ir/index_manager/index/PostingMerger.h>
#include <ir/index_manager/index/RTPostingWriter.h>
#include <ir/index_manager/index/EPostingWriter.h>
#include <ir/index_manager/store/FSIndexOutput.h>
#include <ir/index_manager/store/FSIndexInput.h>

#include <cassert>

NS_IZENELIB_IR_BEGIN

namespace indexmanager
{

PostingMerger::PostingMerger(
    int skipInterval,
    int maxSkipLevel,
    CompressionType compressType,
    bool optimize,
    bool requireIntermediateFileForMerging,
    IndexLevel indexLevel,
    size_t memPoolSizeForPostingMerger
)
    :skipInterval_(skipInterval)
    ,maxSkipLevel_(maxSkipLevel)
    ,compressType_(compressType)
    ,pOutputDescriptor_(NULL)
    ,pTmpPostingOutput_(NULL)
    ,pTmpPostingInput_(NULL)
    ,pVIntDataOutput_(NULL)
    ,requireIntermediateFileForMerging_(requireIntermediateFileForMerging)
    ,nPPostingLength_(0)
    ,nSkipIntervalBetweenBarrels_(0)
    ,bFirstPosting_(true)
    ,pSkipListMerger_(NULL)
    ,pFixedSkipListWriter_(NULL)
    ,pSkipListWriter_(NULL)
    ,compressedPos_(NULL)
    ,compressed_position_buffer_size_(0)
    ,doc_ids_offset_(0)
    ,positions_(NULL)
    ,curr_position_buffer_size_(0)
    ,position_buffer_pointer_(0)
    ,pPosDataPool_(NULL)
    ,pDocFreqDataPool_(NULL)
    ,block_buffer_(NULL)
    ,current_block_id_(0)
    ,optimize_(optimize)
    ,indexLevel_(indexLevel)
{
    // to avoid concurrent memory request,
    // such as by ChunkPostingWriter::pDocFreqDataPool_ in another thread,
    // posting merger use a seperated memory pool exclusively
    pMemCache_.reset(new MemCache(memPoolSizeForPostingMerger));

    init();
    reset();
}

PostingMerger::~PostingMerger()
{
    if(pSkipListMerger_)
        delete pSkipListMerger_;
    if(pTmpPostingOutput_)
    {
        delete pTmpPostingInput_;
        delete pTmpPostingOutput_;
        pOutputDescriptor_->getDirectory()->deleteFile(tmpPostingName_);
    }
    if(pVIntDataOutput_)
    {
        delete pVIntDataOutput_;
    }
    delete [] compressedPos_;
    delete [] positions_;
    if(block_buffer_) delete [] block_buffer_;
    if(pPosDataPool_)
        delete pPosDataPool_;
    delete pDocFreqDataPool_;
    delete pFixedSkipListWriter_;
    delete pSkipListWriter_;

}

void PostingMerger::reset()
{
    ///reset posting descriptor
    postingDesc_.length = 0;
    postingDesc_.plength = 0;
    postingDesc_.ctf = 0;//term frequence
    postingDesc_.df = 0;//doc frequence
    postingDesc_.maxTF = 0;
    postingDesc_.poffset = -1;
    chunkDesc_.lastdocid = 0;
    chunkDesc_.length = 0;
    termInfo_.reset();
    pMemCache_->flushMem();
    nSkipIntervalBetweenBarrels_ = 0;
    if(pSkipListMerger_)
    {
        pSkipListMerger_->reset();
    }
    if(pTmpPostingOutput_)
    {
        reinterpret_cast<FSIndexOutput*>(pTmpPostingOutput_)->trunc();
    }

    if(pVIntDataOutput_)
        pVIntDataOutput_->reset();
    if(pSkipListWriter_)
    {
        pSkipListWriter_->reset();
    }

    pFixedSkipListWriter_->reset();
    pDocFreqDataPool_->reset();
    if (pPosDataPool_)
        pPosDataPool_->reset();
    chunk_.reset();

    doc_ids_offset_ = 0;
    position_buffer_pointer_ = 0;
    blockEncoder_.reset();
    //current_block_id_ = 0;
}

void PostingMerger::init()
{
    if(isSkipParamValid())
    {
        pSkipListMerger_ = new SkipListMerger(skipInterval_,maxSkipLevel_,pMemCache_);
        pSkipListWriter_ = new SkipListWriter(skipInterval_,maxSkipLevel_,pMemCache_);
    }

    pVIntDataOutput_ = new VariantDataPool(pMemCache_);

    compressed_position_buffer_size_ = INIT_POS_CHUNK_SIZE;
    compressedPos_ = new uint32_t[compressed_position_buffer_size_];
    curr_position_buffer_size_ = INIT_POS_CHUNK_SIZE << 1;
    positions_ = new uint32_t[curr_position_buffer_size_];
    pFixedSkipListWriter_ = new FixedBlockSkipListWriter(pMemCache_);

    if (indexLevel_ == WORDLEVEL)
        pPosDataPool_ = new ChunkDataPool(pMemCache_) ;
    pDocFreqDataPool_ = new ChunkDataPool(pMemCache_);
}

void PostingMerger::setOutputDescriptor(OutputDescriptor* pOutputDescriptor)
{
    pOutputDescriptor_ = pOutputDescriptor;
    tmpPostingName_ = pOutputDescriptor_->getBarrelName() + ".merge";
    pTmpPostingOutput_ = pOutputDescriptor_->getDirectory()->createOutput(tmpPostingName_);
}

void PostingMerger::mergeWith(RTDiskPostingReader* pOnDiskPosting,BitVector* pFilter)
{
    if(optimize_)
    {
        optimize(pOnDiskPosting, pFilter);
    }
    else
    {
        if(pFilter)
            mergeWith_GC(pOnDiskPosting,pFilter);
        else
            mergeWith(pOnDiskPosting);
    }
}

void PostingMerger::mergeWith(MemPostingReader* pInMemoryPosting)
{
    ///flush last doc
    pInMemoryPosting->pPostingWriter_->flushLastDoc(true);

    IndexOutput* pDOutput = pOutputDescriptor_->getDPostingOutput();
    IndexOutput* pPOutput = pOutputDescriptor_->getPPostingOutput();

    if (bFirstPosting_)///first posting
    {
        reset();
        if(isSkipParamValid())
        {
            termInfo_.skipPointer_ = pDOutput->getFilePointer();
        }
        else
        {
            termInfo_.docPointer_ = pDOutput->getFilePointer();
            termInfo_.skipPointer_ = -1;
        }
        if (pPOutput)
            termInfo_.positionPointer_ = pPOutput->getFilePointer();
        nPPostingLength_ = 0;
        bFirstPosting_ = false;
        ///save position posting offset
        if(pPOutput)
            postingDesc_.poffset = pPOutput->getFilePointer();
    }

    OutputStream* pDocIndexOutput = 0;

    if(isSkipParamValid())
    {
        if(requireIntermediateFileForMerging_)
            /// we need an extra intermediate file to store merged posting data without skiplist
            pDocIndexOutput = pTmpPostingOutput_;
        else
            pDocIndexOutput = pVIntDataOutput_;
    }
    else
        pDocIndexOutput = pDOutput;

    fileoffset_t oldDOff = pDocIndexOutput->getLength();

    ///write chunk data, update the first doc id
    VariantDataChunk* pDChunk = pInMemoryPosting->pPostingWriter_->pDocFreqList_->pHeadChunk_;

    ///DPosting start for skip
    int32_t DOffsetStartForSkipping = 0;

    if (pDChunk)
    {
        uint8_t* bp = &(pDChunk->data[0]);
        docid_t firstDocID = VariantDataPool::decodeVData32(bp) - chunkDesc_.lastdocid;

        if(chunkDesc_.lastdocid == 0)
            DOffsetStartForSkipping = 0;
        else
            DOffsetStartForSkipping =
                postingDesc_.length - (IndexOutput::getVIntLength(firstDocID + chunkDesc_.lastdocid) - IndexOutput::getVIntLength(firstDocID));

        pDocIndexOutput->writeVInt(firstDocID);///write first doc id
        int32_t writeSize = pDChunk->size - (bp - &(pDChunk->data[0]));	//write the rest data of first chunk
        if (writeSize > 0)
            pDocIndexOutput->write((const char*)bp,writeSize);
        pDChunk = pDChunk->next;
    }
    ///write rest data
    while (pDChunk)
    {
        pDocIndexOutput->write((const char*)pDChunk->data,pDChunk->size);
        pDChunk = pDChunk->next;
    }

    chunkDesc_.length += (pDocIndexOutput->getLength() - oldDOff);

    ///write position posting
    if (indexLevel_ == WORDLEVEL && pInMemoryPosting->pPostingWriter_->pLocList_)
    {
        VariantDataChunk* pPChunk = pInMemoryPosting->pPostingWriter_->pLocList_->pHeadChunk_;
        while (pPChunk)
        {
            pPOutput->write((const char*)pPChunk->data,pPChunk->size);
            pPChunk = pPChunk->next;
        }
    }

    ///merge skiplist
    SkipListReader* pSkipReader = pInMemoryPosting->getSkipListReader();
    if(pSkipReader)
    {
        assert(isSkipParamValid() && pSkipListMerger_);

        pSkipListMerger_->setBasePoint(0, DOffsetStartForSkipping, postingDesc_.plength);
        pSkipListMerger_->addToMerge(pSkipReader,pInMemoryPosting->lastDocID(), nSkipIntervalBetweenBarrels_);
        nSkipIntervalBetweenBarrels_ = pInMemoryPosting->docFreq() - pSkipReader->getNumSkipped();
    }
    else
        nSkipIntervalBetweenBarrels_ += pInMemoryPosting->docFreq();

    ///update descriptors
    postingDesc_.ctf += pInMemoryPosting->getCTF();
    postingDesc_.df += pInMemoryPosting->docFreq();
    if(postingDesc_.maxTF < pInMemoryPosting->getMaxDocFreq())
        postingDesc_.maxTF = pInMemoryPosting->getMaxDocFreq();
    postingDesc_.length = chunkDesc_.length;
    postingDesc_.plength += pInMemoryPosting->getPPostingLen();
    chunkDesc_.lastdocid = pInMemoryPosting->lastDocID();
}

void PostingMerger::mergeWith(RTDiskPostingReader* pOnDiskPosting)
{
    IndexOutput* pDOutput = pOutputDescriptor_->getDPostingOutput();
    IndexOutput* pPOutput = pOutputDescriptor_->getPPostingOutput();
    IndexInput* pDInput = pOnDiskPosting->getInputDescriptor()->getDPostingInput();
    IndexInput* pPInput = pOnDiskPosting->getInputDescriptor()->getPPostingInput();

    if (bFirstPosting_)///first posting
    {
        reset();
        if(isSkipParamValid())
        {
            termInfo_.skipPointer_ = pDOutput->getFilePointer();
        }
        else
        {
            termInfo_.docPointer_ = pDOutput->getFilePointer();
            termInfo_.skipPointer_ = -1;
        }
        if (pPOutput)
            termInfo_.positionPointer_ = pPOutput->getFilePointer();

        nPPostingLength_ = 0;
        bFirstPosting_ = false;

        ///save position offset
        if (pPOutput)
            postingDesc_.poffset = pPOutput->getFilePointer();
    }

    OutputStream* pDocIndexOutput = 0;

    if(isSkipParamValid())
    {
        if(requireIntermediateFileForMerging_)
            /// we need an extra intermediate file to store merged posting data without skiplist
            pDocIndexOutput = pTmpPostingOutput_;
        else
            pDocIndexOutput = pVIntDataOutput_;
    }
    else
        pDocIndexOutput = pDOutput;

    fileoffset_t oldDOff = pDocIndexOutput->getLength();
    docid_t firstDocID = pDInput->readVInt() - chunkDesc_.lastdocid;

    ///DPosting start for skip
    int32_t DOffsetStartForSkipping = 0;
    if(chunkDesc_.lastdocid == 0)
        DOffsetStartForSkipping = 0;
    else
        DOffsetStartForSkipping =
            postingDesc_.length - (IndexOutput::getVIntLength(firstDocID + chunkDesc_.lastdocid) - IndexOutput::getVIntLength(firstDocID));

    pDocIndexOutput->writeVInt(firstDocID);///write first doc id
    int64_t writeSize = pOnDiskPosting->postingDesc_.length - IndexOutput::getVIntLength(firstDocID + chunkDesc_.lastdocid);
    if (writeSize > 0)
        pDocIndexOutput->write(pDInput,writeSize);

    chunkDesc_.length += (pDocIndexOutput->getLength() - oldDOff);

    ///write position posting
    if (pPOutput && pPInput)
        pPOutput->write(pPInput,pOnDiskPosting->nPPostingLength_);

    ///merge skiplist
    SkipListReader* pSkipReader = pOnDiskPosting->getSkipListReader();

    if(pSkipReader)
    {
        assert(isSkipParamValid() && pSkipListMerger_);

        pSkipListMerger_->setBasePoint(0, DOffsetStartForSkipping, postingDesc_.plength);
        pSkipListMerger_->addToMerge(pSkipReader,pOnDiskPosting->chunkDesc_.lastdocid, nSkipIntervalBetweenBarrels_);
        nSkipIntervalBetweenBarrels_ = pOnDiskPosting->docFreq() - pSkipReader->getNumSkipped();
    }
    else
        nSkipIntervalBetweenBarrels_ += pOnDiskPosting->docFreq();

    ///update descriptors
    postingDesc_.ctf += pOnDiskPosting->postingDesc_.ctf;
    postingDesc_.df += pOnDiskPosting->postingDesc_.df;
    if(postingDesc_.maxTF < pOnDiskPosting->postingDesc_.maxTF)
        postingDesc_.maxTF = pOnDiskPosting->postingDesc_.maxTF;
    postingDesc_.length = chunkDesc_.length; ///currently,it's only one chunk
    postingDesc_.plength += pOnDiskPosting->nPPostingLength_;
    chunkDesc_.lastdocid = pOnDiskPosting->chunkDesc_.lastdocid;
}

void PostingMerger::mergeWith_GC(RTDiskPostingReader* pOnDiskPosting,BitVector* pFilter)
{
    IndexOutput* pDOutput = pOutputDescriptor_->getDPostingOutput();
    IndexOutput* pPOutput = pOutputDescriptor_->getPPostingOutput();
    IndexInput* pDInput = pOnDiskPosting->getInputDescriptor()->getDPostingInput();
    IndexInput* pPInput = pOnDiskPosting->getInputDescriptor()->getPPostingInput();

    count_t nODDF = pOnDiskPosting->postingDesc_.df;
    if(nODDF <= 0)
        return;

    if (bFirstPosting_)///first posting
    {
        reset();
        if(isSkipParamValid())
        {
            termInfo_.skipPointer_ = pDOutput->getFilePointer();
        }
        else
        {
            termInfo_.docPointer_ = pDOutput->getFilePointer();
            termInfo_.skipPointer_ = -1;
        }
        if (pPOutput)
            termInfo_.positionPointer_ = pPOutput->getFilePointer();

        nPPostingLength_ = 0;
        bFirstPosting_ = false;

        ///save position offset
        if(pPOutput)
            postingDesc_.poffset = pPOutput->getFilePointer();
    }

    docid_t nDocID = 0;
    docid_t nLastDocID = chunkDesc_.lastdocid;
    freq_t nTF = 0;
    count_t nCTF = 0;
    count_t nMtf = 0;
    count_t nPCount = 0;

    OutputStream* pDocIndexOutput = 0;

    if(isSkipParamValid())
    {
        if(requireIntermediateFileForMerging_)
            /// we need an extra intermediate file to store merged posting data without skiplist
            pDocIndexOutput = pTmpPostingOutput_;
        else
            pDocIndexOutput = pVIntDataOutput_;
    }
    else
        pDocIndexOutput = pDOutput;

    fileoffset_t oldDOff = pDocIndexOutput->getLength();
    fileoffset_t oldPOff = -1;
    if (pPOutput)
        oldPOff = pPOutput->getFilePointer();
    while (nODDF > 0)
    {
        nDocID += pDInput->readVInt();
        nTF = pDInput->readVInt();
		
        if(!pFilter->test((size_t)nDocID))///the document has not been deleted
        {
            pDocIndexOutput->writeVInt(nDocID - nLastDocID);
            pDocIndexOutput->writeVInt(nTF);

            nPCount += nTF;
            ++(postingDesc_.df);
            if(nMtf < nTF)
                nMtf = nTF;
            nLastDocID = nDocID;
            if(pSkipListMerger_ && postingDesc_.df > 0 && postingDesc_.df % skipInterval_ == 0)
            {
                if(pPOutput)
                    pSkipListMerger_->addSkipPoint(nLastDocID,pDocIndexOutput->getLength(),pPOutput->getFilePointer());
                else
                    pSkipListMerger_->addSkipPoint(nLastDocID,pDocIndexOutput->getLength(),0);
            }
        }
        else ///this document has been deleted
        {
            nCTF += nPCount;
            ///write positions of documents
            while (nPCount > 0)
            {
                if (pPOutput && pPInput)
                    pPOutput->writeVInt(pPInput->readVInt());
                nPCount--;
            }
            ///skip positions of deleted documents
            while (nTF > 0)
            {
                if (pPInput)
                    pPInput->readVInt();
                nTF--;
            }
        }
        nODDF--;
    }
    if(nPCount > 0)
    {
        nCTF += nPCount;
        while (nPCount > 0)
        {
            if (pPOutput && pPInput)
                pPOutput->writeVInt(pPInput->readVInt());
            nPCount--;
        }
    }

    chunkDesc_.length += (pDocIndexOutput->getLength() - oldDOff);

    ///update descriptors
    postingDesc_.ctf += nCTF;
    //postingDesc_.df += nDF;
    if((uint32_t)postingDesc_.maxTF < nMtf)
        postingDesc_.maxTF = nMtf;
    postingDesc_.length = chunkDesc_.length; ///currently,it's only one chunk
    if (pPOutput)
        postingDesc_.plength += pPOutput->getFilePointer() - oldPOff;
    chunkDesc_.lastdocid = nLastDocID;
}

void PostingMerger::mergeWith(BlockPostingReader* pPosting,BitVector* pFilter)
{
    IndexInput* pPInput = pPosting->inputDescriptorPtr_->getPPostingInput();

    if (bFirstPosting_)///first posting
    {
        reset();
        nPPostingLength_ = 0;
        bFirstPosting_ = false;
        if(! block_buffer_)
            block_buffer_ = new uint8_t[BLOCK_SIZE];
    }

    int num_docs_left = pPosting->df_;
    int size_of_positions = 0;

{'fraction_non_alphanumeric': 0.10088533927686974, 'fraction_numerical': 0.0047052996532937095, 'mean_word_length': 2.8642344497607657, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 277, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3946975', 'n_tokens_mistral': 10372, 'n_tokens_neox': 10154, 'n_words': 1614}