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 16:21:26 UTC
(lucene) 01/02: 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 branch_10x
in repository https://gitbox.apache.org/repos/asf/lucene.git
commit 8a23ceebc16f0b6aaee46e5e7bcb7684b237e634
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 | 3 ++
.../lucene/search/DisjunctionMaxBulkScorer.java | 6 +--
.../java/org/apache/lucene/util/PriorityQueue.java | 63 ++++++++++++++++------
.../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 a1c69c0d507..ccd54d78283 100644
--- a/lucene/CHANGES.txt
+++ b/lucene/CHANGES.txt
@@ -40,6 +40,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#15757: Better skipper-based numeric sort pruning from the start of segments (Alan Woodward)
* GITHUB#15753: Add Case Insensitive makeChar and makeString to Automata. (John Wagster)
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 78184b48eda..464b32b72e7 100644
--- a/lucene/core/src/java/org/apache/lucene/search/DisjunctionMaxBulkScorer.java
+++ b/lucene/core/src/java/org/apache/lucene/search/DisjunctionMaxBulkScorer.java
@@ -57,9 +57,7 @@ final class DisjunctionMaxBulkScorer extends BulkScorer {
return a.next < b.next;
}
};
- for (BulkScorer scorer : scorers) {
- this.scorers.add(new BulkScorerAndNext(scorer));
- }
+ this.scorers.addAll(scorers.stream().map(BulkScorerAndNext::new));
}
@Override
@@ -89,7 +87,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 ad4350316c8..b25ca98c0c2 100644
--- a/lucene/core/src/java/org/apache/lucene/util/PriorityQueue.java
+++ b/lucene/core/src/java/org/apache/lucene/util/PriorityQueue.java
@@ -19,7 +19,10 @@ package org.apache.lucene.util;
import java.util.Collection;
import java.util.Iterator;
import java.util.NoSuchElementException;
+import java.util.function.Function;
+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
@@ -114,28 +117,56 @@ public abstract 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.
- Iterator<T> iterator = elements.iterator();
- while (iterator.hasNext()) {
- this.heap[size + 1] = iterator.next();
- 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 dc7f2feb9e0..2769d1e2aa6 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 static org.hamcrest.Matchers.lessThanOrEqualTo;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -167,6 +168,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;
@@ -210,10 +269,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));
}
public void testRemovalsAndInsertions() {