atlas: handle OR with multiple matches (#1184)

Update query index to automatically dedup results to handle
the case of OR clauses where both sides match the same data
point. Before, the same match would be returned for each OR
clause that matched a given data point. If the consumer did
not handle that case, then they could aggregate the same
value multiple times leading to incorrect results.

This will add some overhead as the set needs to be maintained,
but should improve correctness for general usage.

Author: brharrington

spectator-reg-atlas/src/main/java/com/netflix/spectator/atlas/impl/QueryIndex.java

@@ -23,6 +23,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Deque;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -63,6 +64,27 @@ public Cache<String, List<QueryIndex<V>>> get() {
     }
   }
 
+  /**
+   * Dedup values as they are consumed. This is used to avoid processing the same result
+   * multiple times in the case of OR clauses where multiple match the same data point.
+   */
+  private static class DedupConsumer<T> implements Consumer<T> {
+
+    private final Consumer<T> consumer;
+    private final Set<T> alreadySeen;
+
+    DedupConsumer(Consumer<T> consumer) {
+      this.consumer = consumer;
+      this.alreadySeen = new HashSet<>();
+    }
+
+    @Override public void accept(T t) {
+      if (alreadySeen.add(t)) {
+        consumer.accept(t);
+      }
+    }
+  }
+
   /**
    * Return a new instance of an index that is empty. The default caching behavior will be
    * used.
@@ -365,7 +387,7 @@ public List<T> findMatches(Id id) {
    *     Function to invoke for values associated with a query that matches the id.
    */
   public void forEachMatch(Id id, Consumer<T> consumer) {
-    forEachMatch(id, 0, consumer);
+    forEachMatch(id, 0, new DedupConsumer<>(consumer));
   }
 
   @SuppressWarnings("PMD.NPathComplexity")
@@ -473,6 +495,10 @@ public List<T> findMatches(Function<String, String> tags) {
    *     Function to invoke for values associated with a query that matches the id.
    */
   public void forEachMatch(Function<String, String> tags, Consumer<T> consumer) {
+    forEachMatchImpl(tags, new DedupConsumer<>(consumer));
+  }
+
+  private void forEachMatchImpl(Function<String, String> tags, Consumer<T> consumer) {
     // Matches for this level
     matches.forEach(consumer);
 

spectator-reg-atlas/src/test/java/com/netflix/spectator/atlas/impl/QueryIndexTest.java

@@ -316,6 +316,14 @@ public void inClauseExpansion() {
     assertEquals(list(query), idx, id1);
   }
 
+  @Test
+  public void orMultiMatch() {
+    Query q = Parser.parseQuery("name,a,:eq,name,a,:re,:or");
+    QueryIndex<Query> idx = QueryIndex.newInstance(cacheSupplier);
+    idx.add(q, q);
+    assertEquals(list(q), idx, id("a"));
+  }
+
   @Test
   public void manyQueries() {
     // CpuUsage for all instances