Added LognHash to early discard values in set and avoid logn searches

Author: ivancea

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/sort/LongTopNSet.java

@@ -10,6 +10,7 @@
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.BinarySearcher;
 import org.elasticsearch.common.util.LongArray;
+import org.elasticsearch.common.util.LongHash;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.search.sort.SortOrder;
@@ -27,15 +28,25 @@ public class LongTopNSet implements Releasable {
 
     private final LongArray values;
     private final LongBinarySearcher searcher;
+    private final LongHash seenValues;
 
     private int count;
 
     public LongTopNSet(BigArrays bigArrays, SortOrder order, int limit) {
         this.order = order;
         this.limit = limit;
         this.count = 0;
-        this.values = bigArrays.newLongArray(limit, false);
-        this.searcher = new LongBinarySearcher(values, order);
+        boolean success = false;
+        try {
+            this.values = bigArrays.newLongArray(limit, false);
+            this.searcher = new LongBinarySearcher(values, order);
+            this.seenValues = new LongHash(limit, 0.05f, bigArrays);
+            success = true;
+        } finally {
+            if (success == false) {
+                close();
+            }
+        }
     }
 
     /**
@@ -58,6 +69,11 @@ public boolean collect(long value) {
             return true;
         }
 
+        if (seenValues.add(value) < 0) {
+            // The value was already added
+            return true;
+        }
+
         int insertionIndex = this.searcher.search(0, count - 1, value);
 
         if (insertionIndex == count - 1) {
@@ -168,6 +184,6 @@ private boolean betterThan(long lhs, long rhs) {
 
     @Override
     public final void close() {
-        Releasables.close(values);
+        Releasables.close(values, seenValues);
     }
 }