Adding stream request flag to SearchRequestContext

[harshavamsi]

Description

Adding stream request flag to SearchRequestContext so that other plugins can consume it. Query insights can use this information to emit a dimension on whether search request was powered by streaming

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[coderabbitai]

📝 Walkthrough

Walkthrough

The changes introduce a streaming mode for terms aggregations by adding a streamingEnabled field with version-gated serialization. A new aggregation type name "streaming_terms" is defined and returned when streaming is enabled. The FlushModeResolver automatically enables streaming on eligible terms aggregations during validation.

Changes

Cohort / File(s)	Summary
TermsAggregationBuilder Streaming Support server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregationBuilder.java	Adds streaming mode infrastructure: public constant STREAMING_NAME, setter streamingEnabled(boolean), private field tracking, and updates to serialization (with V_3_5_0 version gating), equality, hashCode, copy constructors, and type resolution logic.
FlushModeResolver Integration server/src/main/java/org/opensearch/search/streaming/FlushModeResolver.java	Introduces side effect in isStreamable to automatically enable streaming on TermsAggregationBuilder instances that pass top-level eligibility checks via casting and streamingEnabled(true) invocation.
FlushModeResolverTests Coverage server/src/test/java/org/opensearch/search/streaming/FlushModeResolverTests.java	Adds four test methods covering streaming behavior: terms aggregation type switching, sub-aggregation handling, non-terms aggregation rejection, and null/empty aggregation handling.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~22 minutes

Suggested labels

v3.5.0

🚥 Pre-merge checks | ✅ 1 | ❌ 2

❌ Failed checks (2 warnings)

Check name	Status	Explanation	Resolution
Title check	⚠️ Warning	The PR title 'Adding stream request flag to SearchRequestContext' does not match the actual changes, which introduce a streaming mode for terms aggregation and modify FlushModeResolver.	Update the title to accurately reflect the main change, such as 'Add streaming mode support to TermsAggregationBuilder' or 'Enable streaming type for terms aggregations'.
Description check	⚠️ Warning	The PR description is vague and generic, with major discrepancies between stated and actual changes. The description mentions 'stream request flag' and 'SearchRequestContext', but the changes involve adding streaming support to TermsAggregationBuilder and FlushModeResolver.	Update the description to clearly explain the actual changes: adding STREAMING_NAME constant and streamingEnabled() method to TermsAggregationBuilder, enabling streaming mode in FlushModeResolver.isStreamable(), and adding corresponding tests. Reference the specific files and classes modified.

✅ Passed checks (1 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 0

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (2)

server/src/main/java/org/opensearch/search/streaming/FlushModeResolver.java (1)

120-132: ⚠️ Potential issue | 🟠 Major

Avoid partial side effects on failed validation.

streamingEnabled(true) is set as you iterate; if a later top-level agg is not streamable, the method returns false but earlier aggs remain mutated. Consider a two-pass approach (validate first, then enable) to keep the method side-effect-free on failure.

Proposed fix (two-pass enablement)

 Collection<AggregationBuilder> topLevelAggs = aggregations.getAggregatorFactories();
 for (AggregationBuilder agg : topLevelAggs) {
     if (!isTopLevelStreamable(agg)) {
         return false;
     }
-    ((TermsAggregationBuilder) agg).streamingEnabled(true);
 }
+for (AggregationBuilder agg : topLevelAggs) {
+    ((TermsAggregationBuilder) agg).streamingEnabled(true);
+}
 return true;

server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregationBuilder.java (1)

70-76: ⚠️ Potential issue | 🟠 Major

Register "streaming_terms" in SearchModule or version-gate the type name change.

getWriteableName() delegates to getType() (AbstractAggregationBuilder:148), which returns "streaming_terms" when streamingEnabled=true. However, SearchModule only registers the aggregation under TermsAggregationBuilder.NAME ("terms"). Deserialization will fail when attempting to read a builder with writeable name "streaming_terms" because the registry has no reader for that name.

Additionally, streamingEnabled is version-gated on serialization (V_3_5_0+), but the type name change is not. In mixed-version clusters, older nodes receiving "streaming_terms" will fail to deserialize since they have no registered reader for this name. Either register "streaming_terms" as an alias in SearchModule, or gate the type name change by version (e.g., only return STREAMING_NAME when sending to nodes that support it).

🧹 Nitpick comments (1)

server/src/test/java/org/opensearch/search/streaming/FlushModeResolverTests.java (1)

11-15: Prefer constants over string literals for agg type names.

Use TermsAggregationBuilder.NAME / STREAMING_NAME to avoid brittle string assertions.

Optional refactor

-        assertEquals("terms", termsAgg.getType());
+        assertEquals(TermsAggregationBuilder.NAME, termsAgg.getType());
...
-        assertEquals("streaming_terms", termsAgg.getType());
+        assertEquals(TermsAggregationBuilder.STREAMING_NAME, termsAgg.getType());
...
-        assertEquals("terms", termsAgg.getType());
+        assertEquals(TermsAggregationBuilder.NAME, termsAgg.getType());
...
-        assertEquals("streaming_terms", termsAgg.getType());
+        assertEquals(TermsAggregationBuilder.STREAMING_NAME, termsAgg.getType());

Also applies to: 70-115

📜 Review details

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between ee403e7 and 1643ee3.

📒 Files selected for processing (3)

• server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregationBuilder.java
• server/src/main/java/org/opensearch/search/streaming/FlushModeResolver.java
• server/src/test/java/org/opensearch/search/streaming/FlushModeResolverTests.java

🧰 Additional context used

🧠 Learnings (1)

📚 Learning: 2026-01-13T17:40:27.167Z

Learnt from: reta
Repo: opensearch-project/OpenSearch PR: 20411
File: server/src/main/java/org/opensearch/index/codec/CodecService.java:112-133
Timestamp: 2026-01-13T17:40:27.167Z
Learning: Avoid capturing or evaluating a supplier (e.g., this::defaultCodec) upfront when passing it to a registry during object construction. If registries may replace defaults during iteration (as in EnginePlugin.getAdditionalCodecs), pass the supplier itself and only resolve it at use time. This ensures dynamic behavior is preserved during initialization and prevents premature binding of defaults in codecs/registry setup. This pattern should apply to similar initialization paths in Java server code where registries may mutate defaults during construction.

Applied to files:

• server/src/main/java/org/opensearch/search/streaming/FlushModeResolver.java
• server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregationBuilder.java

🧬 Code graph analysis (1)

server/src/test/java/org/opensearch/search/streaming/FlushModeResolverTests.java (1)

server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregationBuilder.java (1)

• TermsAggregationBuilder (68-487)

⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (21)

• GitHub Check: gradle-check
• GitHub Check: precommit (21, windows-2025, true)
• GitHub Check: precommit (25, ubuntu-24.04-arm)
• GitHub Check: precommit (25, windows-latest)
• GitHub Check: precommit (21, macos-15)
• GitHub Check: precommit (25, macos-15-intel)
• GitHub Check: precommit (21, ubuntu-24.04-arm)
• GitHub Check: precommit (21, macos-15-intel)
• GitHub Check: precommit (25, macos-15)
• GitHub Check: precommit (21, ubuntu-latest)
• GitHub Check: precommit (21, windows-latest)
• GitHub Check: precommit (25, ubuntu-latest)
• GitHub Check: assemble (21, ubuntu-latest)
• GitHub Check: detect-breaking-change
• GitHub Check: assemble (21, ubuntu-24.04-arm)
• GitHub Check: assemble (25, ubuntu-latest)
• GitHub Check: assemble (25, windows-latest)
• GitHub Check: assemble (25, ubuntu-24.04-arm)
• GitHub Check: assemble (21, macos-15)
• GitHub Check: assemble (21, windows-latest)
• GitHub Check: Analyze (java)

🔇 Additional comments (1)

server/src/main/java/org/opensearch/search/aggregations/bucket/terms/TermsAggregationBuilder.java (1)

147-166: State propagation and equality updates look good.

Streaming state is copied, exposed, and accounted for in equals/hashCode as expected.

Also applies to: 390-396, 453-470

_{✏️ Tip: You can disable this entire section by setting review_details to false in your review settings.}

[rishabhmaurya]

can't we add it to top level agg builder instead of term agg builder to make it more generic?

[github-actions]

❌ Gradle check result for 1643ee3: null

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[harshavamsi]

can't we add it to top level agg builder instead of term agg builder to make it more generic?

makes sense to me, will do

[bowenlan-amzn]

On a high level, streaming could be another dimension to label search request, without combining with term aggregation here.
Example:
A search request labeled with
agg: terms
streaming: true
query: range
...

[harshavamsi]

On a high level, streaming could be another dimension to label search request, without combining with term aggregation here. Example: A search request labeled with agg: terms streaming: true query: range ...

I feel that will pollute the way we are able to categorize requests. Is it possible for request to be terms aggregation and streaming enabled but not use stream search because of flush mode thresholds?

[github-actions]

❌ Gradle check result for 62532d1: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 1fede59: null

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 1fede59: null

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 1fede59: null

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 1fede59: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 1fede59: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 1fede59: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 1fede59: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❕ Gradle check result for 1fede59: UNSTABLE

Please review all flaky tests that succeeded after retry and create an issue if one does not already exist to track the flaky failure.