text: /*
      $Id: divideskipmerger.h 4887 2009-12-16 22:01:55Z abehm $
  
      Copyright (C) 2010 by The Regents of the University of California
  
      Redistribution of this file is permitted under the terms of
      the BSD license.
  
      This imeplementation merges multiple lists by
      the DividedSkip algorithm.
  
      The lists are assumed to
      be sorted in an ascending order.
  
      Author: Jiaheng Lu, Alexander Behm
      Date: 05/23/2007
  
  */
  
  #ifndef _divideskipmerger_h_
  #define _divideskipmerger_h_
  
  #include <cmath>
  #include <iostream>
  
  #include "listmerger.h"
  #include "../util/misc.h"
  
  // DivideSkip algorithm,
  // which combine MergeOpt and MergeSkip algorithm.
  
  template <class InvList = vector<unsigned> >
  class DivideSkipMerger : public ListsMerger<DivideSkipMerger<InvList>, InvList> {
  private:
  
    // for estimation purposes
    float shortListsSlope;
    float shortListsIntercept;
    float longListsSlope;
    float longListsIntercept;
  
    // for calculating L
    float mu;
  
    // no list weights
    void mergeSkipWithoutDupls(const vector<InvList*> &invLists,
  			     unsigned threshold,
  			     unsigned longLists,
  			     vector<Candi>& candis);
  
    // no list weights
    void mergeWithoutDupls(vector<InvList*> &invLists,
  			 unsigned threshold,
  			 vector<unsigned> &results);
  
    // using list weights
    void mergeSkipWithDupls(const vector<InvList*> &invLists,
  			  const vector<unsigned> &weights,
  			  unsigned threshold,
  			  unsigned longLists,
  			  vector<Candi>& candis);
  
    // using list weights
    void mergeWithDupls(vector<InvList*> &invLists,
  		      unsigned threshold,
  		      vector<unsigned> &results);
  
    // helper for query time estimatiom
    void getQueryStats(vector<InvList*> &invLists,
  		     unsigned threshold,
  		     float& shortListsVar,
  		     float& shortListsTime,
  		     float& longListsVar,
  		     float& longListsTime);
  
  public:
    DivideSkipMerger(bool hasDuplicateLists = false)
      : ListsMerger<DivideSkipMerger<InvList>, InvList>(hasDuplicateLists),
        shortListsSlope(0.0f), shortListsIntercept(0.0f),
        longListsSlope(0.0f), longListsIntercept(0.0f), mu(0.01f) { }
  
    void merge_Impl(vector<InvList*> &invLists,
  		  unsigned threshold,
  		  vector<unsigned> &results);
  
    void fillEstimationParams(vector<vector<InvList* >* >& invLists,
  			    vector<unsigned>& thresholds);
  
    float getEstimatedQueryTime(vector<InvList*>& invLists,
  			      unsigned threshold) const;
  
    string getName() {
      return "DivideSkipMerger";
    }
  };
  
  template <class InvList>
  void
  DivideSkipMerger<InvList>::
  merge_Impl(vector<InvList*> &invLists,
  	   unsigned threshold,
  	   vector<unsigned> &results) {
  
    if(threshold > invLists.size() || threshold == 0) return;
  
    if(this->hasDuplicateLists)
      mergeWithDupls(invLists, threshold, results);
    else
      mergeWithoutDupls(invLists, threshold, results);
  }
  
  template <class InvList>
  void
  DivideSkipMerger<InvList>::
  mergeWithoutDupls(vector<InvList*> &invLists,
  		  unsigned threshold,
  		  vector<unsigned> &results) {
  
    sort(invLists.begin(), invLists.end(), DivideSkipMerger<InvList>::cmpInvList);
  
    unsigned longestSize = invLists.back()->size();
    unsigned longListsSize = (unsigned)floor(threshold / (mu*log2(longestSize) + 1)) ;
    if(longListsSize >= threshold) longListsSize = threshold - 1;
  
    // use mergeskip to search the short lists
    vector<Candi> candis;
    mergeSkipWithoutDupls(invLists,
  			threshold-longListsSize,
  			longListsSize,
  			candis);
  
    unsigned stop = invLists.size();
    unsigned start = stop - longListsSize;
    typename InvList::iterator longListsPointers [longListsSize];
    for(unsigned j = start; j < stop; j++) longListsPointers[j - start] = invLists.at(j)->begin();
  
    // use exponential probe + binary search for long lists
    for(unsigned i = 0; i < candis.size(); i++) {
  
      if(candis[i].count >= threshold) {
        results.push_back(candis[i].id);
        continue;
      }
  
      unsigned targetCount = threshold - candis[i].count;
      unsigned count = 0;
      for(unsigned j = start; j < stop; j++) {
        unsigned llindex = j - start;
        typename InvList::iterator start, end;
        this->expProbe(longListsPointers[llindex], invLists[j]->end(), start, end, candis[i].id);
        typename InvList::iterator iter = lower_bound(start, end, candis[i].id);
        longListsPointers[llindex] = start;
        if(iter != invLists[j]->end() && *iter == candis[i].id) {
  	count++;
  	if(count >= targetCount) {
  	  results.push_back(candis[i].id);
  	  break;
  	}
        }
        else {
  	if(count + stop - j - 1 < targetCount) break; // early termination
        }
      }
    }
  }
  
  
  template <class InvList>
  void
  DivideSkipMerger<InvList>::
  mergeSkipWithoutDupls(const vector<InvList*> &invLists,
  		      unsigned threshold,
  		      unsigned longLists,
  		      vector<Candi>& candis) {
  
    const typename InvList::value_type maxUnsigned = (typename InvList::value_type)0xFFFFFFFF;
  
    unsigned numShortLists = invLists.size() - longLists;
  
    if(threshold > numShortLists) return;
  
    // add maxUnsigned
    for(unsigned i = 0; i < numShortLists; i++)
      invLists[i]->push_back(maxUnsigned);
  
    vector<HeapElement<InvList> > heap;
    heap.reserve(numShortLists);
  
    // make initial heap
    make_heap(heap.begin(), heap.end());
    for(unsigned i = 0; i < numShortLists; i++) {
      heap.push_back(HeapElement<InvList>(invLists[i]->begin(), i));
      push_heap(heap.begin(), heap.end());
    }
  
    unsigned pivot = threshold - 1;
    HeapElement<InvList> popped[numShortLists];
  
    while(*(heap.front().iter) != maxUnsigned) {
  
      unsigned count = 0;
      unsigned currentId = *(heap.front().iter);
  
      while(currentId == *(heap.front().iter) && !heap.empty()) {
        popped[count++] = heap.front();
        pop_heap(heap.begin(), heap.end());
        heap.pop_back();
      }
  
      if(count >= threshold) {
        candis.push_back(Candi(currentId, count));
  
        // move to the next element
        for(unsigned i = 0; i < count; i++) {
  	popped[i].iter++;
  	heap.push_back(HeapElement<InvList>(popped[i].iter, popped[i].invListIndex));
  	push_heap(heap.begin(), heap.end());
        }
        continue;
      }
  
      // pop more elements from heap
      while(count < pivot) {
        popped[count++] = heap.front();
        pop_heap(heap.begin(), heap.end());
        heap.pop_back();
      }
  
      // skip
      for(unsigned i = 0; i < count; i++) {
        popped[i].iter = lower_bound(invLists[popped[i].invListIndex]->begin(),
  				   invLists[popped[i].invListIndex]->end(),
  				   *(heap.front().iter));
      }
  
      // insert to heap
      for(unsigned i = 0; i < count; i++) {
        heap.push_back(HeapElement<InvList>(popped[i].iter, popped[i].invListIndex));
        push_heap(heap.begin(), heap.end());
      }
    }
  
    // remove maxUnsigned
    for(unsigned i = 0; i < numShortLists; i++)
      invLists[i]->pop_back();
  }
  
  template <class InvList>
  void
  DivideSkipMerger<InvList>::
  mergeWithDupls(vector<InvList*> &invLists,
  	       unsigned threshold,
  	       vector<unsigned> &results) {
  
    vector<InvList*> newInvLists;
    vector<unsigned> newWeights;
    this->detectDuplicateLists(invLists, newInvLists, newWeights);
  
    // assume that newArray should be sorted according to the length increasing
    unsigned longestSize = newInvLists.at(newInvLists.size()-1)->size();
    unsigned targetLongListWeight = (unsigned)floor(threshold / (mu*log2(longestSize) + 1));
    if(targetLongListWeight >= threshold) targetLongListWeight = threshold - 1;
  
    unsigned longListsSize = 0;
    unsigned totalLongListsWeight = 0;
    for(unsigned i = 0; i < newInvLists.size(); i++) {
      unsigned currentWeight = newWeights[newInvLists.size() - 1 - i];
      totalLongListsWeight += currentWeight;
      if(totalLongListsWeight >= targetLongListWeight) {
        if(totalLongListsWeight >= threshold) {
  	longListsSize = i;
  	totalLongListsWeight -= currentWeight;
        }
        else {
  	longListsSize = i + 1;
        }
        break;
      }
    }
  
    vector<Candi> candis;
    mergeSkipWithDupls(newInvLists,
  		     newWeights,
  		     threshold-totalLongListsWeight,
  		     longListsSize,
  		     candis);
  
    unsigned stop = newInvLists.size();
    unsigned start = stop - longListsSize;
    typename InvList::iterator longListsPointers [longListsSize];
  
    unsigned weightRemaining[longListsSize]; // for early termination
    unsigned cumulWeight = 0;
    for(unsigned j = start; j < stop; j++) {
      longListsPointers[j - start] = newInvLists.at(j)->begin();
      cumulWeight += newWeights[j];
      weightRemaining[j - start] = totalLongListsWeight - cumulWeight;
    }
  
    // use exponential probe + binary search for long lists
    for(unsigned i = 0; i < candis.size(); i++) {
  
      if(candis[i].count >= threshold) {
        results.push_back(candis[i].id);
        continue;
      }
  
      unsigned targetCount = threshold - candis[i].count;
      unsigned count = 0;
      for(unsigned j = start; j < stop; j++) {
        unsigned llindex = j - start;
        typename InvList::iterator start, end;
        this->expProbe(longListsPointers[llindex], newInvLists[j]->end(), start, end, candis[i].id);
  
        typename InvList::iterator iter = lower_bound(start, end, candis[i].id);
        longListsPointers[llindex] = iter;
        if(iter != invLists[j]->end() && *iter == candis[i].id) {
  	count += newWeights[j];
  	if(count >= targetCount) {
  	  results.push_back(candis[i].id);
  	  break;
  	}
        }
        else {
  	if(count + weightRemaining[llindex] < targetCount) break; // early termination
        }
      }
    }
  }
  
  template <class InvList>
  void
  DivideSkipMerger<InvList>::
  mergeSkipWithDupls(const vector<InvList*> &invLists,
  		   const vector<unsigned> &weights,
  		   unsigned threshold,
  		   unsigned longLists,
  		   vector<Candi>& candis) {
  
    const typename InvList::value_type maxUnsigned = (typename InvList::value_type)0xFFFFFFFF;
  
    unsigned numShortLists = invLists.size() - longLists;
    if(threshold > numShortLists) return;
  
    // add maxUnsigned
    for(unsigned i = 0; i < numShortLists; i++)
      invLists[i]->push_back(maxUnsigned);
  
    vector<HeapElement<InvList> > heap;
    heap.reserve(numShortLists);
  
    // make initial heap
    make_heap(heap.begin(), heap.end());
    for(unsigned i = 0; i < numShortLists; i++) {
      heap.push_back(HeapElement<InvList>(invLists[i]->begin(), i));
      push_heap(heap.begin(), heap.end());
    }
  
    unsigned pivot = threshold - 1;
    HeapElement<InvList> popped[numShortLists];
  
    while(*(heap.front().iter) != maxUnsigned) {
  
      unsigned weight = 0, count = 0;
      unsigned currentId = *(heap.front().iter);
  
      while(currentId == *(heap.front().iter) && !heap.empty()) {
        popped[count++] = heap.front();
        weight += weights[heap.front().invListIndex];
        pop_heap(heap.begin(), heap.end());
        heap.pop_back();
      }
  
      if(weight >= threshold) {
        candis.push_back(Candi(currentId, weight));
  
        // move to the next element
        for(unsigned i = 0; i < count; i++) {
  	popped[i].iter++;
  	heap.push_back(HeapElement<InvList>(popped[i].iter, popped[i].invListIndex));
  	push_heap(heap.begin(), heap.end());
        }
        continue;
      }
  
      // pop more elements from heap
      while(weight < pivot) {
        // it is possible for a list to have a weight equals to T
        if(weight+ weights[heap.front().invListIndex] > pivot) break;
        popped[count++] = heap.front();
        weight += weights[heap.front().invListIndex];
        pop_heap(heap.begin(), heap.end());
        heap.pop_back();
      }
  
      // skip
      for(unsigned i = 0; i < count; i++) {
        popped[i].iter = lower_bound(invLists[popped[i].invListIndex]->begin(),
  				   invLists[popped[i].invListIndex]->end(),
  				   *(heap.front().iter));
      }
  
      // insert to heap
      for(unsigned i = 0; i < count; i++) {
        heap.push_back(HeapElement<InvList>(popped[i].iter, popped[i].invListIndex));
        push_heap(heap.begin(), heap.end());
      }
    }
  
    // remove maxUnsigned
    for(unsigned i = 0; i < numShortLists; i++)
      invLists[i]->pop_back();
  }
  
  template <class InvList>
  void
  DivideSkipMerger<InvList>::
  getQueryStats(vector<InvList*> &invLists,
  	      unsigned threshold,
  	      float& shortListsVar,
  	      float& shortListsTime,
  	      float& longListsVar,
  	      float& longListsTime) {
  
    struct timeval t1, t2;
    struct timezone tz;
  
    if(threshold >= invLists.size() || threshold == 0) return;
  
    sort(invLists.begin(), invLists.end(), DivideSkipMerger<InvList>::cmpInvList);
  
    unsigned longestSize = invLists.back()->size();
    unsigned longListsSize = (unsigned)floor(threshold / (mu*log2(longestSize) + 1));
    if(longListsSize >= threshold) longListsSize = threshold - 1;
  
    vector<Candi> candis;
    gettimeofday(&t1, &tz);
    mergeSkipWithoutDupls(invLists,
  			threshold-longListsSize,
  			longListsSize,
  			candis);
  
    gettimeofday(&t2, &tz);
    shortListsTime = ((t2.tv_sec - t1.tv_sec) * 1000000 + (t2.tv_usec - t1.tv_usec)) / 1000.0f;
  
    // use binary search or hash search for long lists
    vector<unsigned> results;
    gettimeofday(&t1, &tz);
  
    unsigned stop = invLists.size();
    unsigned start = stop - longListsSize;
    typename InvList::iterator longListsPointers [longListsSize];
    for(unsigned j = start; j < stop; j++) longListsPointers[j - start] = invLists.at(j)->begin();
  
    // use exponential probe + binary search for long lists
    for(unsigned i = 0; i < candis.size(); i++) {
  
      if(candis[i].count >= threshold) {
        results.push_back(candis[i].id);
        continue;
      }
  
      unsigned targetCount = threshold - candis[i].count;
      unsigned count = 0;
      for(unsigned j = start; j < stop; j++) {
        unsigned llindex = j - start;
        typename InvList::iterator start, end;
        this->expProbe(longListsPointers[llindex], invLists[j]->end(), start, end, candis[i].id);
        typename InvList::iterator iter = lower_bound(start, end, candis[i].id);
        longListsPointers[llindex] = start;
        if(iter != invLists[j]->end() && *iter == candis[i].id) {
  	count++;
  	if(count >= targetCount) {
  	  results.push_back(candis[i].id);
  	  break;
  	}
        }
        else {
  	if(count + stop - j - 1 < targetCount) break; // early termination
        }
      }
    }
    gettimeofday(&t2, &tz);
    longListsTime = ((t2.tv_sec - t1.tv_sec) * 1000000 + (t2.tv_usec - t1.tv_usec)) / 1000.0f;
  
    // calculate the total sizes of short lists
    float log2 = log(2);
    unsigned shortListsSize = 0;
    for(unsigned i = 0; i < start; i++)
      shortListsSize += invLists[i]->size();
  
    // calculate the log2 sum of the total size of long lists
    float logLongLists = 0.0f;
    for(unsigned i = start; i < stop; i++)
      logLongLists += log(invLists[i]->size()) / log2;
  
    shortListsVar = ((float)shortListsSize / (float)((start))) * logLongLists;
    longListsVar = shortListsSize * ( log(invLists.size()) / log2 );
  }
  
  
  template <class InvList>
  void
  DivideSkipMerger<InvList>::
  fillEstimationParams(vector<vector<InvList* >* >& invLists,
  		     vector<unsigned>& thresholds) {
  
    vector<float> shortListsXVals;
    vector<float> shortListsYVals;
    vector<float> longListsXVals;
    vector<float> longListsYVals;
  
    float shortListsXVal = 0.0f;
    float shortListsYVal = 0.0f;
    float longListsXVal = 0.0f;
    float longListsYVal = 0.0f;
  
    for(unsigned i = 0; i < invLists.size(); i++) {
      vector<InvList*>* currentArrs = invLists.at(i);
      if(currentArrs->size() <= 0) continue;
      getQueryStats(*currentArrs,
  		  thresholds.at(i),
  		  shortListsXVal,
  		  shortListsYVal,
  		  longListsXVal,
  		  longListsYVal);
  
      shortListsXVals.push_back(shortListsXVal);
      shortListsYVals.push_back(shortListsYVal);
      longListsXVals.push_back(longListsXVal);
      longListsYVals.push_back(longListsYVal);
    }
  
    // perform linear regressions
    linearRegression(shortListsXVals, shortListsYVals, shortListsSlope, shortListsIntercept);
    linearRegression(longListsXVals, longListsYVals, longListsSlope, longListsIntercept);
  }
  
  
  template <class InvList>
  float
  DivideSkipMerger<InvList>::
  getEstimatedQueryTime(vector<InvList*>& invLists,
  		      unsigned threshold) const {
  
    if(threshold >= invLists.size()) return 0.0f;
  
    sort(invLists.begin(), invLists.end(), DivideSkipMerger<InvList>::cmpInvList);
  
    unsigned longestSize = invLists.back()->size();
    unsigned longListsSize = (unsigned)floor(threshold / (mu*log2(longestSize) + 1));
    if(longListsSize >= threshold) longListsSize = threshold - 1;
  
    // calculate the total sizes of short lists
    float log2 = log(2);
    unsigned shortListsSize = 0;
    unsigned numShortLists = invLists.size() - longListsSize;
    for(unsigned i = 0; i < numShortLists; i++)
      shortListsSize += invLists[i]->size();
  
    // calculate the log2 sum of the total size of long lists
    float logLongLists = 0.0f;
    for(unsigned i = numShortLists; i < invLists.size();i++)
      logLongLists += log(invLists[i]->size()) / log2;
  
    float shortListsVar = ((float)shortListsSize / (float)((threshold - longListsSize))) * logLongLists;
    float longListsVar = shortListsSize * ( log(invLists.size()) / log2 );
  
    return shortListsVar*shortListsSlope + longListsVar*longListsSlope + shortListsIntercept + longListsIntercept;
  }
  
  #endif
  