Skip to content

ES|QL: Pass fix size instead of maxPageSize to LuceneTopNOperator scorer #135767

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Oct 2, 2025
Merged

ES|QL: Pass fix size instead of maxPageSize to LuceneTopNOperator scorer #135767

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
add472e
Pass max doc to scorer instead of maxPageSize
ioanatia Oct 1, 2025
123a4f6
Merge remote-tracking branch 'elasticsearch/main' into bulk_scorer_fix
ioanatia Oct 2, 2025
fb17f8b
Score docs in fixed sized batches
ioanatia Oct 2, 2025
d403f49
Fix test - I have no idea why this works
ioanatia Oct 2, 2025
8009eba
Update docs/changelog/135767.yaml
ioanatia Oct 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion ...esql/compute/src/main/java/org/elasticsearch/compute/lucene/LuceneTopNSourceOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,9 @@ public String describe() {
}
}

// We use the same value as the INITIAL_INTERVAL from CancellableBulkScorer
private static final int NUM_DOCS_INTERVAL = 1 << 12;

private final CircuitBreaker breaker;
private final List<SortBuilder<?>> sorts;
private final long estimatedPerRowSortSize;
Expand Down Expand Up @@ -213,7 +216,7 @@ private Page collect() throws IOException {
perShardCollector = newPerShardCollector(scorer.shardContext(), sorts, needsScore, limit);
}
var leafCollector = perShardCollector.getLeafCollector(scorer.leafReaderContext());
scorer.scoreNextRange(leafCollector, scorer.leafReaderContext().reader().getLiveDocs(), maxPageSize);
scorer.scoreNextRange(leafCollector, scorer.leafReaderContext().reader().getLiveDocs(), NUM_DOCS_INTERVAL);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need to lock this to the max of the range?

It looks like CancellableBulkScorer makes this bigger and bigger with time. But I think this is good and we can get it in and iterate.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We do something similar here, at least we try to avoid overflows:

elasticsearch/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/lucene/LuceneOperator.java

Lines 241 to 246 in 7cbc236

void scoreNextRange(LeafCollector collector, Bits acceptDocs, int numDocs) throws IOException {
assert isDone() == false : "scorer is exhausted";
// avoid overflow and limit the range
numDocs = Math.min(maxPosition - position, numDocs);
assert numDocs > 0 : "scorer was exhausted";
position = bulkScorer.score(collector, acceptDocs, position, Math.min(maxPosition, position + numDocs));

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah. got it.

} catch (CollectionTerminatedException cte) {
// Lucene terminated early the collection (doing topN for an index that's sorted and the topN uses the same sorting)
scorer.markAsDone();
Expand Down
2 changes: 1 addition & 1 deletion ...ql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlActionTaskIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -511,7 +511,7 @@ public void testTaskContentsForTopNQuery() throws Exception {
try {
getTasksStarting();
logger.info("unblocking script");
scriptPermits.release(pageSize());
scriptPermits.release(numberOfDocs() + pageSize());
List<TaskInfo> tasks = getTasksRunning();
String sortStatus = """
[{"pause_me":{"order":"asc","missing":"_last","unmapped_type":"long"}}]""";
Expand Down