fixing duplicates in dynamic bounds

arangod/Aql/ExecutionBlock.cpp

@@ -1148,10 +1148,8 @@ bool IndexRangeBlock::initRanges () {
                 andCombineIndexOrRIBHigh(rangeInfoOr, rib);
               }
             }
-            if (newCondition != nullptr) {
-              for (size_t i = 0; i < rangeInfoOr->size(); i++) {
-                newCondition->emplace_back(rangeInfoOr->at(i));
-              }
+            if (newCondition != nullptr && !newCondition->empty()) {
+              orCombineIndexOrs(newCondition, rangeInfoOr);
               delete rangeInfoOr;
             } else {
               newCondition = rangeInfoOr;
@@ -1197,6 +1195,36 @@ bool IndexRangeBlock::initRanges () {
   LEAVE_BLOCK;
 }
 
+////////////////////////////////////////////////////////////////////////////////
+/// @brief orCombineIndexOrs: return the lhs by appending the difference of
+/// those RIs in the rhs with those in the lhs into the lhs.
+////////////////////////////////////////////////////////////////////////////////
+
+void IndexRangeBlock::orCombineIndexOrs (IndexOrCondition* lhs, 
+                                         IndexOrCondition* rhs) {
+
+  TRI_ASSERT(lhs != nullptr && !lhs->empty());
+  
+  if (rhs->empty()) {
+    return;
+  }
+  
+  //avoid inserting overlapping conditions
+  for (IndexAndCondition indexAnd: *rhs) {
+    std::vector<RangeInfo> newIndexAnd;
+    for (RangeInfo& ri: indexAnd) {
+      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);
+    }
+  }
+}
+
 // ioc = IndexOrCondition(IndexAndCondition(A && B) || IndexAndCondition(C && D))
 // riv = RangeInfoVec(E || F)
 // combined into

arangod/Aql/RangeInfo.cpp

@@ -518,8 +518,6 @@ std::unordered_set<std::string> RangeInfoMapVec::attributes (std::string const&
 /// @brief orCombineRangeInfoMapVecs: return a new RangeInfoMapVec appending
 /// those RIMs in the right arg (which are not identical to an existing RIM) in
 /// a copy of the left arg.
-///
-/// 
 ////////////////////////////////////////////////////////////////////////////////
 
 RangeInfoMapVec* triagens::aql::orCombineRangeInfoMapVecs (RangeInfoMapVec* lhs, 
@@ -573,6 +571,7 @@ RangeInfoMapVec* triagens::aql::orCombineRangeInfoMapVecs (RangeInfoMapVec* lhs,
   return lhs;
 }
 
+
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief andCombineRangeInfoMaps: insert every RangeInfo in the <rhs> in the
 /// <lhs> and delete the <rhs>
@@ -753,3 +752,18 @@ void RangeInfoMapVec::differenceRangeInfo (RangeInfo& newRi) {
     }
   }
 }
+
+void triagens::aql::differenceIndexOrRangeInfo(IndexOrCondition const* ioc,
+                                               RangeInfo&              newRi) {
+  for (IndexAndCondition iac: *ioc) {
+    for (RangeInfo oldRi: iac) {
+      differenceRangeInfos(oldRi, newRi);
+      if (! newRi.isValid()) { 
+        break;
+      }
+    }
+    if (! newRi.isValid()) { 
+      break;
+    }
+  }
+}

arangod/Aql/RangeInfo.h

@@ -938,6 +938,7 @@ namespace triagens {
 ////////////////////////////////////////////////////////////////////////////////
     
     void differenceRangeInfos (RangeInfo&, RangeInfo&);
+    void differenceIndexOrRangeInfo (IndexOrCondition const*, RangeInfo&);
   }
 }