Skip to content

Enable skiplist based collection using collectRange #20268

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

Open
jainankitk wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Enable skiplist based collection using collectRange #20268

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3edda3b
Enable skiplist based collection using collectRange
jainankitk Dec 17, 2025
e9d3c28
Adding changelog entry
jainankitk Dec 17, 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
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
- Handle custom metadata files in subdirectory-store ([#20157](https://github.com/opensearch-project/OpenSearch/pull/20157))
- Add support for missing proto fields in GRPC FunctionScore and Highlight ([#20169](https://github.com/opensearch-project/OpenSearch/pull/20169))
- Ensure all modules are included in INTEG_TEST testcluster distribution ([#20241](https://github.com/opensearch-project/OpenSearch/pull/20241))
- Enable skiplist based collection using collectRange ([#20268](https://github.com/opensearch-project/OpenSearch/pull/20268))
Copy link
Contributor

@coderabbitai coderabbitai bot Dec 17, 2025

Choose a reason for hiding this comment

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

⚠️ Potential issue | 🟡 Minor

Add hyphen to compound adjective.

Per the static analysis hint, compound adjectives should be hyphenated when they precede a noun: "skiplist-based collection" instead of "skiplist based collection".

Apply this diff:

-- Enable skiplist based collection using collectRange ([#20268](https://github.com/opensearch-project/OpenSearch/pull/20268))
+- Enable skiplist-based collection using collectRange ([#20268](https://github.com/opensearch-project/OpenSearch/pull/20268))
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
- Enable skiplist based collection using collectRange ([#20268](https://github.com/opensearch-project/OpenSearch/pull/20268))
- Enable skiplist-based collection using collectRange ([#20268](https://github.com/opensearch-project/OpenSearch/pull/20268))
🧰 Tools
🪛 LanguageTool

[grammar] ~15-~15: Use a hyphen to join words.
Context: ...penSearch/pull/20241)) - Enable skiplist based collection using collectRange ([#2...

(QB_NEW_EN_HYPHEN)

🤖 Prompt for AI Agents 
In CHANGELOG.md around line 15, the compound adjective "skiplist based
collection" needs a hyphen; change it to "skiplist-based collection" while
preserving the rest of the line (including the PR link and parentheses) exactly
as-is.


### Fixed
- Fix bug of warm index: FullFileCachedIndexInput was closed error ([#20055](https://github.com/opensearch-project/OpenSearch/pull/20055))
Expand Down
4 changes: 2 additions & 2 deletions .../src/main/java/org/opensearch/search/aggregations/matrix/stats/MatrixStatsAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,8 +121,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}

/**
Expand Down
21 changes: 13 additions & 8 deletions server/src/main/java/org/opensearch/search/aggregations/LeafBucketCollector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,24 +131,29 @@ public void collect(DocIdStream stream) throws IOException {
collect(stream, 0);
}

@Override
public void collectRange(int min, int max) throws IOException {
collectRange(min, max, 0);
}

/**
* Collect a range of doc IDs, between {@code min} inclusive and {@code max} exclusive. {@code
* max} is guaranteed to be greater than {@code min}.
* Collect a range of doc IDs into {@code bucket}, between {@code min} inclusive and {@code max} exclusive.
* {@code max} is guaranteed to be greater than {@code min}.
*
* <p>Extending this method is typically useful to take advantage of pre-aggregated data exposed
* in a {@link DocValuesSkipper}.
*
* <p>The default implementation calls {@link #collect(DocIdStream)} on a {@link DocIdStream} that
* matches the given range.
* <p>The default implementation calls {@link #collect(int, long)} on range of doc IDs, between
* {@code min} inclusive and {@code max} exclusive.
*
* @see #collect(int,long)
* @see #collect(int, long)
*/
@Override
public void collectRange(int min, int max) throws IOException {
@ExperimentalApi
public void collectRange(int min, int max, long bucket) throws IOException {
// Different aggregator implementations should override this method even if to just delegate to super for
// helping the performance: when the super call inlines, calls to #collect(int, long) become monomorphic.
for (int docId = min; docId < max; docId++) {
collect(docId, 0);
collect(docId, bucket);
}
}

Expand Down
38 changes: 32 additions & 6 deletions ...c/main/java/org/opensearch/search/aggregations/bucket/HistogramSkiplistLeafCollector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,12 +180,6 @@ public void collect(int doc, long owningBucketOrd) throws IOException {
}
}

@Override
public void collect(DocIdStream stream) throws IOException {
// This will only be called if its the top agg
collect(stream, 0);
}

@Override
public void collect(DocIdStream stream, long owningBucketOrd) throws IOException {
for (;;) {
Expand Down Expand Up @@ -219,6 +213,38 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}
}

@Override
public void collectRange(int min, int max, long bucket) throws IOException {
if (upToInclusive < min) {
advanceSkipper(min, bucket);
}

for (;;) {
int upToExclusive = Integer.min(upToInclusive + 1, max); // Don't collect past max
if (upToExclusive < 0) { // overflow
upToExclusive = Integer.MAX_VALUE;
}

if (upToSameBucket) {
aggregator.incrementBucketDocCount(upToBucketIndex, upToExclusive - min);
if (isSubNoOp == false) {
sub.collectRange(min, upToExclusive, upToBucketIndex);
}
} else {
for (int docId = min; docId < upToExclusive; docId++) {
collect(docId, bucket);
}
}

if (upToExclusive < max) {
min = upToExclusive;
advanceSkipper(min, bucket);
} else {
break;
}
}
}

/**
* Call back for auto date histogram
*
Expand Down
12 changes: 6 additions & 6 deletions ...rc/main/java/org/opensearch/search/aggregations/bucket/composite/CompositeAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -644,8 +644,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down Expand Up @@ -683,8 +683,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down Expand Up @@ -752,8 +752,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down
4 changes: 2 additions & 2 deletions server/src/main/java/org/opensearch/search/aggregations/bucket/filter/FilterAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,8 +90,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down
4 changes: 2 additions & 2 deletions server/src/main/java/org/opensearch/search/aggregations/bucket/filter/FiltersAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,8 +199,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down
8 changes: 4 additions & 4 deletions ...java/org/opensearch/search/aggregations/bucket/histogram/AutoDateHistogramAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -454,8 +454,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}

private void collectValue(int doc, long rounded) throws IOException {
Expand Down Expand Up @@ -748,8 +748,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}

private int collectValue(long owningBucketOrd, int roundingIdx, int doc, long rounded) throws IOException {
Expand Down
8 changes: 4 additions & 4 deletions ...ain/java/org/opensearch/search/aggregations/bucket/histogram/DateHistogramAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,8 +255,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down Expand Up @@ -288,8 +288,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down
4 changes: 2 additions & 2 deletions .../java/org/opensearch/search/aggregations/bucket/histogram/NumericHistogramAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,8 +145,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down
4 changes: 2 additions & 2 deletions server/src/main/java/org/opensearch/search/aggregations/bucket/range/RangeAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -365,8 +365,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}

private int collect(int doc, double value, long owningBucketOrdinal, int lowBound) throws IOException {
Expand Down
28 changes: 14 additions & 14 deletions .../org/opensearch/search/aggregations/bucket/terms/GlobalOrdinalsStringTermsAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,8 +281,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
});
}
Expand All @@ -305,8 +305,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
});
}
Expand Down Expand Up @@ -335,8 +335,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
});
}
Expand All @@ -362,8 +362,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
});
}
Expand Down Expand Up @@ -600,8 +600,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
});
}
Expand All @@ -627,8 +627,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
});
}
Expand Down Expand Up @@ -1243,8 +1243,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down
4 changes: 2 additions & 2 deletions ...rc/main/java/org/opensearch/search/aggregations/bucket/terms/LongRareTermsAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,8 +124,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down
8 changes: 4 additions & 4 deletions ...src/main/java/org/opensearch/search/aggregations/bucket/terms/NumericTermsAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,8 +165,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
});
}
Expand Down Expand Up @@ -733,8 +733,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down
10 changes: 5 additions & 5 deletions server/src/main/java/org/opensearch/search/aggregations/metrics/AvgAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,8 +166,8 @@ public void collect(DocIdStream stream, long bucket) throws IOException {
}

@Override
public void collectRange(int min, int max) throws IOException {
setKahanSummation(0);
public void collectRange(int min, int max, long bucket) throws IOException {
setKahanSummation(bucket);
int count = 0;
for (int docId = min; docId < max; docId++) {
if (values.advanceExact(docId)) {
Expand All @@ -178,9 +178,9 @@ public void collectRange(int min, int max) throws IOException {
}
}
}
counts.increment(0, count);
sums.set(0, kahanSummation.value());
compensations.set(0, kahanSummation.delta());
counts.increment(bucket, count);
sums.set(bucket, kahanSummation.value());
compensations.set(bucket, kahanSummation.delta());
}

private void setKahanSummation(long bucket) {
Expand Down
4 changes: 2 additions & 2 deletions server/src/main/java/org/opensearch/search/aggregations/metrics/ExtendedStatsAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,8 +170,8 @@ public void collect(DocIdStream stream, long owningBucketOrd) throws IOException
}

@Override
public void collectRange(int min, int max) throws IOException {
super.collectRange(min, max);
public void collectRange(int min, int max, long bucket) throws IOException {
super.collectRange(min, max, bucket);
}
};
}
Expand Down
8 changes: 4 additions & 4 deletions server/src/main/java/org/opensearch/search/aggregations/metrics/MaxAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,15 +181,15 @@ public void collect(DocIdStream stream, long bucket) throws IOException {
}

@Override
public void collectRange(int min, int max) throws IOException {
growMaxes(0);
double maximum = maxes.get(0);
public void collectRange(int min, int max, long bucket) throws IOException {
growMaxes(bucket);
double maximum = maxes.get(bucket);
for (int doc = min; doc < max; doc++) {
if (values.advanceExact(doc)) {
maximum = Math.max(maximum, values.doubleValue());
}
}
maxes.set(0, maximum);
maxes.set(bucket, maximum);
}

private void growMaxes(long bucket) {
Expand Down
Loading
Loading