snapshot

arangod/Aql/ExecutionBlock.cpp

@@ -855,10 +855,18 @@ IndexRangeBlock::IndexRangeBlock (ExecutionEngine* engine,
     _posInDocs(0),
     _anyBoundVariable(false),
     _skiplistIterator(nullptr),
-    _condition(&en->_ranges),
+    _condition(new IndexOrCondition()),
     _posInRanges(0),
-    _freeCondition(false) {
-   
+    _freeCondition(true) {
+  
+  for (size_t i = 0; i < en->_ranges.size(); i++) {
+    _condition->emplace_back(IndexAndCondition());
+    for (auto ri: en->_ranges[i]) {
+      _condition->at(i).emplace_back(ri.clone());
+    }
+  }
+  removeOverlapsIndexOr(*_condition);
+
   std::vector<std::vector<RangeInfo>> const& orRanges = en->_ranges;
   TRI_ASSERT(en->_index != nullptr);
 
@@ -980,16 +988,12 @@ bool IndexRangeBlock::initRanges () {
   ENTER_BLOCK
   _flag = true; 
   auto en = static_cast<IndexRangeNode const*>(getPlanNode());
-  freeCondition();
-  _condition = &en->_ranges;
-  _freeCondition = false;
   
   TRI_ASSERT(en->_index != nullptr);
 
   // Find out about the actual values for the bounds in the variable bound case:
 
   if (_anyBoundVariable) {
-    
     size_t posInExpressions = 0;
       
     // The following are needed to evaluate expressions with local data from
@@ -999,7 +1003,6 @@ bool IndexRangeBlock::initRanges () {
     vector<TRI_document_collection_t const*>& docColls(cur->getDocumentCollections());
     RegisterId nrRegs = cur->getNrRegs();
     
-
     // must have a V8 context here to protect Expression::execute()
     auto engine = _engine;
     triagens::basics::ScopeGuard guard{
@@ -1178,7 +1181,10 @@ bool IndexRangeBlock::initRanges () {
         throw;
       }
     }
+    // remove duplicates . . .
+    removeOverlapsIndexOr(*_condition);
   }
+  
    
   if (en->_index->type == TRI_IDX_TYPE_PRIMARY_INDEX) {
     return true; //no initialization here!
@@ -1217,11 +1223,11 @@ void IndexRangeBlock::orCombineIndexOrs (IndexOrCondition* lhs,
   for (IndexAndCondition indexAnd: *rhs) {
     std::vector<RangeInfo> newIndexAnd;
     for (RangeInfo& ri: indexAnd) {
-      differenceIndexOrRangeInfo(lhs, ri);
-      if (ri.isValid()) { 
+      //differenceIndexOrRangeInfo(lhs, ri);
+      //if (ri.isValid()) { 
         // if ri is invalid, then don't insert it
         newIndexAnd.emplace_back(ri);
-      }
+      //}
     }
     if (! newIndexAnd.empty()) {
       lhs->emplace_back(newIndexAnd);

arangod/Aql/ExecutionBlock.h

@@ -694,7 +694,7 @@ namespace triagens {
 /// incoming block. 
 ////////////////////////////////////////////////////////////////////////////////
         
-        IndexOrCondition const* _condition;
+        IndexOrCondition* _condition;
         
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief _flag: since readIndex for primary, hash, edges indexes reads the

arangod/Aql/RangeInfo.cpp

@@ -557,7 +557,8 @@ RangeInfoMapVec* triagens::aql::orCombineRangeInfoMapVecs (RangeInfoMapVec* lhs,
         for (auto y: x.second) {
           // take the difference of 
           RangeInfo ri = y.second.clone();
-          lhs->differenceRangeInfo(ri);
+          //lhs->differenceRangeInfo(ri); //TODO delete this line, we remove
+          //duplicates later in the IndexRangeBlock
           if (ri.isValid()) { 
             // if ri is not valid, then y.second is contained in an existing ri 
             rim->insert(ri);
@@ -777,3 +778,94 @@ void triagens::aql::differenceIndexOrRangeInfo(IndexOrCondition const* ioc,
     }
   }
 }
+
+// 3 way comparison
+
+/*int triagens::aql::compareIndexAndConditions (IndexAndCondition& and1, IndexAndCondition& and2) {
+}*/
+
+bool triagens::aql::areDisjointIndexAndConditions (IndexAndCondition& and1, IndexAndCondition& and2) {
+  
+  for (auto ri1: and1) {
+    for (auto ri2: and2) {
+      if (ri2._attr == ri1._attr) {
+        if (areDisjointRangeInfos(ri1, ri2)) {
+          return true;
+        }
+      }
+    }
+  }
+  return false;
+}
+
+// is and1 contained in and2
+bool triagens::aql::isContainedIndexAndConditions (IndexAndCondition& and1, IndexAndCondition& and2) {
+  
+  for (auto ri1: and1) {
+    bool contained = false;
+    for (auto ri2: and2) {
+      if (ri2._attr == ri1._attr) {
+        if (containmentRangeInfos(ri2, ri1) == 1){
+          contained = true;
+          break;
+        }
+      }
+    }
+    
+    if (! contained) {
+      return false;
+    }
+  }    
+  return true;
+}
+
+void triagens::aql::differenceIndexAnd (IndexAndCondition& and1, IndexAndCondition& and2) {
+
+  if (! areDisjointIndexAndConditions(and1, and2)) {
+    if (isContainedIndexAndConditions(and1, and2)) { 
+      // and1 is a subset of and2, disregard and1
+      and1.clear();
+    } 
+    else if (isContainedIndexAndConditions(and2, and1)) {
+      // and2 is a subset of and1, disregard and2
+      and2.clear();
+    } 
+    else {
+      // and1 and and2 have non-empty intersection
+      for (auto& ri1: and1) {
+        for (auto& ri2: and2) {
+          if (ri2._attr == ri1._attr && containmentRangeInfos(ri1, ri2) == 0){
+            int LoLo = CompareRangeInfoBound(ri1._lowConst, ri2._lowConst, -1);
+            if (LoLo == 1) { // replace low bound of new with high bound of old
+              ri2._lowConst.assign(ri1._highConst);
+              ri2._lowConst.setInclude(! ri1._highConst.inclusive());
+            } 
+            else { // replace the high bound of the new with the low bound of the old
+              ri2._highConst.assign(ri1._lowConst);
+              ri2._highConst.setInclude(! ri1._lowConst.inclusive());
+            }
+          }
+        }
+      }    
+    }
+  }
+}
+
+void triagens::aql::removeOverlapsIndexOr (IndexOrCondition& ioc) {
+  
+  for (size_t i = 1; i < ioc.size(); i++) {
+    for (size_t j = 0; j < i; j++) {
+      differenceIndexAnd(ioc.at(j), ioc.at(i));
+    }
+  }
+  // remove empty ones
+  for (auto it = ioc.begin(); it < ioc.end(); ) {
+    if (it->empty()) {
+      it = ioc.erase(it);
+    } 
+    else {
+      it++;
+    }
+  }
+}
+

arangod/Aql/RangeInfo.h

@@ -943,6 +943,11 @@ namespace triagens {
     
     void differenceRangeInfos (RangeInfo&, RangeInfo&);
     void differenceIndexOrRangeInfo (IndexOrCondition const*, RangeInfo&);
+
+    bool isContainedIndexAndConditions (IndexAndCondition& and1, IndexAndCondition& and2);
+    void differenceIndexAnd (IndexAndCondition&, IndexAndCondition&);
+    void removeOverlapsIndexOr (IndexOrCondition&);
+    bool areDisjointIndexAndConditions (IndexAndCondition&, IndexAndCondition&);
   }
 }