You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by gs...@apache.org on 2026/03/31 15:39:53 UTC
(lucene) branch main updated: Implement method to add all stream elements into a PriorityQueue. (#15823)
This is an automated email from the ASF dual-hosted git repository.
gsmiller pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/lucene.git
The following commit(s) were added to refs/heads/main by this push:
new a68055a36d9 Implement method to add all stream elements into a PriorityQueue. (#15823)
a68055a36d9 is described below
commit a68055a36d933dc0d6099dc62c2660e4a0df9cb3
Author: zhouhui <vs...@163.com>
AuthorDate: Tue Mar 31 23:39:47 2026 +0800
Implement method to add all stream elements into a PriorityQueue. (#15823)
---
lucene/CHANGES.txt | 4 +-
.../lucene/search/DisjunctionMaxBulkScorer.java | 6 +--
.../java/org/apache/lucene/util/PriorityQueue.java | 62 +++++++++++++++------
.../org/apache/lucene/util/TestPriorityQueue.java | 63 ++++++++++++++++++++++
4 files changed, 115 insertions(+), 20 deletions(-)
diff --git a/lucene/CHANGES.txt b/lucene/CHANGES.txt
index 8f892031fc1..635f2fcbb15 100644
--- a/lucene/CHANGES.txt
+++ b/lucene/CHANGES.txt
@@ -89,7 +89,6 @@ New Features
Improvements
---------------------
-
* GITHUB#15704: Replace LinkedList with more efficient data structure. (Renato Haeberli)
* GITHUB#15682: Use ArrayDeque instead of LinkedList in CompoundWordTokenFilterBase.java. (Renato Haeberli)
@@ -258,6 +257,9 @@ New Features
Improvements
---------------------
+* GITHUB#15823: Implement method to add all stream elements into a PriorityQueue.
+ Call PriorityQueue#addAll with mapped stream in DisjunctionMaxBulkScorer's constructor. (Zhou Hui)
+
* GITHUB#15711: DirectoryTaxonomyReader.getBulkPath() now uses getPathFromCache. (Shubham Sharma)
* GITHUB#15757: Better skipper-based numeric sort pruning from the start of segments (Alan Woodward)
diff --git a/lucene/core/src/java/org/apache/lucene/search/DisjunctionMaxBulkScorer.java b/lucene/core/src/java/org/apache/lucene/search/DisjunctionMaxBulkScorer.java
index e6dc7050270..25dd6ab7f3a 100644
--- a/lucene/core/src/java/org/apache/lucene/search/DisjunctionMaxBulkScorer.java
+++ b/lucene/core/src/java/org/apache/lucene/search/DisjunctionMaxBulkScorer.java
@@ -53,9 +53,7 @@ final class DisjunctionMaxBulkScorer extends BulkScorer {
}
this.scorers =
PriorityQueue.usingComparator(scorers.size(), Comparator.comparingInt(b -> b.next));
- for (BulkScorer scorer : scorers) {
- this.scorers.add(new BulkScorerAndNext(scorer));
- }
+ this.scorers.addAll(scorers.stream().map(BulkScorerAndNext::new));
}
@Override
@@ -85,7 +83,7 @@ final class DisjunctionMaxBulkScorer extends BulkScorer {
@Override
public void collect(int doc) throws IOException {
final int delta = doc - windowMin;
- windowMatches.set(doc - windowMin);
+ windowMatches.set(delta);
windowScores[delta] = Math.max(windowScores[delta], scorer.score());
}
},
diff --git a/lucene/core/src/java/org/apache/lucene/util/PriorityQueue.java b/lucene/core/src/java/org/apache/lucene/util/PriorityQueue.java
index ec89bb1b269..c93e12dec74 100644
--- a/lucene/core/src/java/org/apache/lucene/util/PriorityQueue.java
+++ b/lucene/core/src/java/org/apache/lucene/util/PriorityQueue.java
@@ -21,8 +21,11 @@ import java.util.Collection;
import java.util.Comparator;
import java.util.Iterator;
import java.util.NoSuchElementException;
+import java.util.function.Function;
import java.util.function.IntFunction;
+import java.util.function.Predicate;
import java.util.function.Supplier;
+import java.util.stream.Stream;
/**
* A priority queue maintains a partial ordering of its elements such that the least element can
@@ -150,27 +153,56 @@ public class PriorityQueue<T> implements Iterable<T> {
* faster.
*
* <p>If one tries to add more objects than the maxSize passed in the constructor, an {@link
- * ArrayIndexOutOfBoundsException} is thrown.
+ * ArrayIndexOutOfBoundsException} is thrown. Which may result in parts of elements added into the
+ * queue, but the heap is still stay in correct state. In this case, if caller wants to readd or
+ * {@link #updateTop(Object)} with remaining elements, it should skip(continue) consumed elements
+ * with the delta size of queue.
*/
public void addAll(Collection<T> elements) {
- if (this.size + elements.size() > this.maxSize) {
- throw new ArrayIndexOutOfBoundsException(
- "Cannot add "
- + elements.size()
- + " elements to a queue with remaining capacity: "
- + (maxSize - size));
- }
-
// Heap with size S always takes first S elements of the array,
// and thus it's safe to fill array further - no actual non-sentinel value will be overwritten.
- for (T element : elements) {
- this.heap[size + 1] = element;
- this.size++;
+ try {
+ for (T element : elements) {
+ this.heap[size + 1] = element;
+ this.size++;
+ }
+ } finally {
+ // The loop goes down to 1 as heap is 1-based not 0-based.
+ for (int i = (size >>> 1); i >= 1; i--) {
+ downHeap(i);
+ }
}
+ }
- // The loop goes down to 1 as heap is 1-based not 0-based.
- for (int i = (size >>> 1); i >= 1; i--) {
- downHeap(i);
+ /**
+ * Adds all elements of the stream into the queue. This method should be preferred over calling
+ * {@link #add(Object)} in loop if all elements are known in advance as it builds queue faster.
+ *
+ * <p>If one needs to map or filter element in the iteration of elements in this method, call this
+ * method with elements wrapped by {@link Stream#map(Function)} or {@link
+ * Stream#filter(Predicate)}, etc. In these cases, this method should be preferred over calling
+ * {@link #addAll(Collection)}.
+ *
+ * <p>If one tries to add more objects than the maxSize passed in the constructor, an {@link
+ * ArrayIndexOutOfBoundsException} is thrown. Which may result in parts of elements added into the
+ * queue, but the heap is still stay in correct state. In this case, if caller wants to readd or
+ * {@link #updateTop(Object)} with remaining elements, it should use a new stream, and use {@link
+ * Stream#skip(long)} to skip consumed elements with the delta size of queue.
+ */
+ public void addAll(Stream<T> elements) {
+ // Heap with size S always takes first S elements of the array,
+ // and thus it's safe to fill array further - no actual non-sentinel value will be overwritten.
+ try {
+ elements.forEachOrdered(
+ element -> {
+ this.heap[size + 1] = element;
+ this.size++;
+ });
+ } finally {
+ // The loop goes down to 1 as heap is 1-based not 0-based.
+ for (int i = (size >>> 1); i >= 1; i--) {
+ downHeap(i);
+ }
}
}
diff --git a/lucene/core/src/test/org/apache/lucene/util/TestPriorityQueue.java b/lucene/core/src/test/org/apache/lucene/util/TestPriorityQueue.java
index 204d7c0e711..2e1154c519f 100644
--- a/lucene/core/src/test/org/apache/lucene/util/TestPriorityQueue.java
+++ b/lucene/core/src/test/org/apache/lucene/util/TestPriorityQueue.java
@@ -21,6 +21,7 @@ import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import com.carrotsearch.randomizedtesting.generators.RandomNumbers;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
@@ -159,6 +160,64 @@ public class TestPriorityQueue extends LuceneTestCase {
assertThat(pq.top(), equalTo(2));
}
+ public void testAddAllWithStream() {
+ PriorityQueue<Integer> pq = new IntegerQueue(6);
+ List<String> elements = new ArrayList<>();
+ for (String s : Arrays.asList("a", "b", "c", "d", "e", "f")) {
+ if (random().nextBoolean()) {
+ elements.addFirst(s);
+ } else {
+ elements.addLast(s);
+ }
+ }
+
+ pq.addAll(elements.stream().map(String::hashCode));
+ assertEquals("a".hashCode(), pq.top().intValue());
+ }
+
+ public void testAddAllWithStreamNotFitIntoQueue() {
+ PriorityQueue<Integer> pq = new IntegerQueue(10);
+ List<String> elements = new ArrayList<>();
+ for (String s : Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k")) {
+ if (random().nextBoolean()) {
+ elements.addFirst(s);
+ } else {
+ elements.addLast(s);
+ }
+ }
+
+ assertThrows(
+ "Cannot add 11 elements to a queue with remaining capacity: 10",
+ ArrayIndexOutOfBoundsException.class,
+ () -> pq.addAll(elements.stream().map(String::hashCode)));
+ // Partly added.
+ assertEquals(10, pq.size());
+ assertTrue(assertHeap(pq));
+ }
+
+ private boolean assertHeap(PriorityQueue<Integer> pq) {
+ // TODO: Maybe change getHeapArray return type T[].
+ Object[] heapArray = pq.getHeapArray();
+ // The loop goes down to 1 as heap is 1-based not 0-based.
+ for (int i = (heapArray.length >>> 1); i >= 1; i--) {
+ int left = i << 1;
+ int right = left + 1;
+ if (right < heapArray.length) {
+ if ((Integer) heapArray[i] > (Integer) heapArray[right]) {
+ return false;
+ }
+ if ((Integer) heapArray[i] > (Integer) heapArray[left]) {
+ return false;
+ }
+ } else if (left < heapArray.length) {
+ if ((Integer) heapArray[i] > (Integer) heapArray[left]) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
public void testAddAllToEmptyQueue() {
Random random = random();
int size = 10;
@@ -202,10 +261,14 @@ public class TestPriorityQueue extends LuceneTestCase {
pq.add(random.nextInt());
}
+ assertEquals(11, pq.size());
assertThrows(
"Cannot add 11 elements to a queue with remaining capacity: 9",
ArrayIndexOutOfBoundsException.class,
() -> pq.addAll(list));
+ // Partly added.
+ assertEquals(20, pq.size());
+ assertTrue(assertHeap(pq));
}
/** Randomly add some elements, comparing against the reference java.util.PriorityQueue. */