Revert SearchProgressListener API break for source compatibility

Author: Atri Sharma

server/src/main/java/org/opensearch/action/search/QueryPhaseResultConsumer.java

@@ -287,18 +287,7 @@ private ReduceResult partialReduce(
             SearchShardTarget target = result.getSearchShardTarget();
             processedShards.add(new SearchShard(target.getClusterAlias(), target.getShardId()));
         }
-        // For streaming search with TopDocs, use the new notification method
-        if (hasTopDocs && newTopDocs != null) {
-            progressListener.notifyPartialReduceWithTopDocs(
-                processedShards,
-                topDocsStats.getTotalHits(),
-                newTopDocs,
-                newAggs,
-                numReducePhases
-            );
-        } else {
-            progressListener.notifyPartialReduce(processedShards, topDocsStats.getTotalHits(), newAggs, numReducePhases);
-        }
+        progressListener.notifyPartialReduce(processedShards, topDocsStats.getTotalHits(), newAggs, numReducePhases);
         // we leave the results un-serialized because serializing is slow but we compute the serialized
         // size as an estimate of the memory used by the newly reduced aggregations.
         long serializedSize = hasAggs ? newAggs.getSerializedSize() : 0;
@@ -434,87 +423,6 @@ void consume(QuerySearchResult result, Runnable callback) {
             }
         }
 
-        /**
-         * Performs a transient, non-destructive reduction of the current state and an optional partial result,
-         * and notifies the listener. This is used for coordinated streaming search.
-         */
-        synchronized void notifySnapshot(QuerySearchResult partial) {
-            checkCancellation();
-            if (hasFailure()) {
-                return;
-            }
-
-            List<TopDocs> topDocsList = new ArrayList<>();
-            List<InternalAggregations> aggsList = new ArrayList<>();
-            List<SearchShard> snapshotShards = new ArrayList<>(emptyResults);
-
-            if (reduceResult != null) {
-                if (reduceResult.reducedTopDocs != null) {
-                    topDocsList.add(reduceResult.reducedTopDocs);
-                }
-                if (reduceResult.reducedAggs != null) {
-                    aggsList.add(reduceResult.reducedAggs);
-                }
-                snapshotShards.addAll(reduceResult.processedShards);
-            }
-
-            for (QuerySearchResult result : buffer) {
-                if (result.hasTopDocs()) {
-                    TopDocs td = result.topDocs().topDocs;
-                    SearchPhaseController.setShardIndex(td, result.getShardIndex());
-                    topDocsList.add(td);
-                }
-                if (result.hasAggs()) {
-                    aggsList.add(result.aggregations().expand());
-                }
-                SearchShardTarget target = result.getSearchShardTarget();
-                snapshotShards.add(new SearchShard(target.getClusterAlias(), target.getShardId()));
-            }
-
-            TotalHits snapshotTotalHits = topDocsStats.getTotalHits();
-
-            if (partial != null) {
-                if (partial.hasTopDocs()) {
-                    TopDocs td = partial.topDocs().topDocs;
-                    SearchPhaseController.setShardIndex(td, partial.getShardIndex());
-                    topDocsList.add(td);
-                    long val = snapshotTotalHits == null ? 0 : snapshotTotalHits.value();
-                    val += td.totalHits.value();
-                    snapshotTotalHits = new TotalHits(
-                        val,
-                        snapshotTotalHits == null ? td.totalHits.relation() : snapshotTotalHits.relation()
-                    );
-                }
-                if (partial.hasAggs()) {
-                    aggsList.add(partial.aggregations().expand());
-                }
-                SearchShardTarget target = partial.getSearchShardTarget();
-                snapshotShards.add(new SearchShard(target.getClusterAlias(), target.getShardId()));
-            }
-
-            TopDocs mergedTopDocs = null;
-            if (hasTopDocs && topDocsList.isEmpty() == false) {
-                mergedTopDocs = SearchPhaseController.mergeTopDocs(topDocsList, topNSize, 0);
-            }
-
-            InternalAggregations mergedAggs = null;
-            if (hasAggs && aggsList.isEmpty() == false) {
-                mergedAggs = InternalAggregations.topLevelReduce(aggsList, aggReduceContextBuilder.forPartialReduction());
-            }
-
-            if (hasTopDocs && mergedTopDocs != null) {
-                progressListener.notifyPartialReduceWithTopDocs(
-                    snapshotShards,
-                    snapshotTotalHits,
-                    mergedTopDocs,
-                    mergedAggs,
-                    numReducePhases
-                );
-            } else {
-                progressListener.notifyPartialReduce(snapshotShards, snapshotTotalHits, mergedAggs, numReducePhases);
-            }
-        }
-
         private synchronized boolean consumeResult(QuerySearchResult result, Runnable callback) {
             if (hasFailure()) {
                 result.consumeAll(); // release memory

server/src/main/java/org/opensearch/action/search/SearchRequest.java

@@ -129,7 +129,7 @@ public class SearchRequest extends ActionRequest implements IndicesRequest.Repla
     private Boolean phaseTook = null;
 
     private boolean streamingScoring = false;
-    private String streamingSearchMode = null; // Will use StreamingSearchMode.SCORED_UNSORTED if null
+    private String streamingSearchMode = null; // Will use StreamingSearchMode.NO_SCORING if null
 
     public SearchRequest() {
         this.localClusterAlias = null;

server/src/main/java/org/opensearch/action/search/SearchResponse.java

@@ -91,11 +91,6 @@ public class SearchResponse extends ActionResponse implements StatusToXContentOb
     private static final ParseField TIMED_OUT = new ParseField("timed_out");
     private static final ParseField TERMINATED_EARLY = new ParseField("terminated_early");
     private static final ParseField NUM_REDUCE_PHASES = new ParseField("num_reduce_phases");
-    
-    // Streaming REST fields
-    private static final ParseField IS_PARTIAL = new ParseField("is_partial");
-    private static final ParseField SEQUENCE_NUMBER = new ParseField("sequence_number");
-    private static final ParseField TOTAL_PARTIALS = new ParseField("total_partials");
 
     private final SearchResponseSections internalResponse;
     private final String scrollId;
@@ -108,11 +103,6 @@ public class SearchResponse extends ActionResponse implements StatusToXContentOb
     private final long tookInMillis;
     private final PhaseTook phaseTook;
 
-    // Fields for streaming responses
-    private boolean isPartial = false;
-    private int sequenceNumber = 0;
-    private int totalPartials = 0;
-
     public SearchResponse(StreamInput in) throws IOException {
         super(in);
         internalResponse = new InternalSearchResponse(in);
@@ -137,12 +127,6 @@ public SearchResponse(StreamInput in) throws IOException {
         }
         skippedShards = in.readVInt();
         pointInTimeId = in.readOptionalString();
-        // Read streaming fields
-        if (in.getVersion().onOrAfter(Version.V_2_15_0)) {
-            isPartial = in.readBoolean();
-            sequenceNumber = in.readVInt();
-            totalPartials = in.readVInt();
-        }
     }
 
     public SearchResponse(
@@ -318,31 +302,6 @@ public String getScrollId() {
         return scrollId;
     }
 
-    // Streaming response methods
-    public boolean isPartial() {
-        return isPartial;
-    }
-
-    public void setPartial(boolean partial) {
-        this.isPartial = partial;
-    }
-
-    public int getSequenceNumber() {
-        return sequenceNumber;
-    }
-
-    public void setSequenceNumber(int sequenceNumber) {
-        this.sequenceNumber = sequenceNumber;
-    }
-
-    public int getTotalPartials() {
-        return totalPartials;
-    }
-
-    public void setTotalPartials(int totalPartials) {
-        this.totalPartials = totalPartials;
-    }
-
     /**
      * Returns the encoded string of the search context that the search request is used to executed
      */
@@ -397,16 +356,6 @@ public XContentBuilder innerToXContent(XContentBuilder builder, Params params) t
             builder.field(NUM_REDUCE_PHASES.getPreferredName(), getNumReducePhases());
         }
 
-        // Write streaming fields if it's a streaming partial
-        // or if we have sequenced streaming responses
-        if (isPartial) {
-            builder.field(IS_PARTIAL.getPreferredName(), isPartial);
-            builder.field(SEQUENCE_NUMBER.getPreferredName(), sequenceNumber);
-            if (totalPartials > 0) {
-                builder.field(TOTAL_PARTIALS.getPreferredName(), totalPartials);
-            }
-        }
-        
         RestActions.buildBroadcastShardsHeader(
             builder,
             params,
@@ -445,9 +394,6 @@ public static SearchResponse innerFromXContent(XContentParser parser) throws IOE
         int skippedShards = 0; // 0 for BWC
         String scrollId = null;
         String searchContextId = null;
-        boolean isPartial = false;
-        int sequenceNumber = 0;
-        int totalPartials = 0;
         List<ShardSearchFailure> failures = new ArrayList<>();
         Clusters clusters = Clusters.EMPTY;
         List<SearchExtBuilder> extBuilders = new ArrayList<>();
@@ -468,12 +414,6 @@ public static SearchResponse innerFromXContent(XContentParser parser) throws IOE
                     terminatedEarly = parser.booleanValue();
                 } else if (NUM_REDUCE_PHASES.match(currentFieldName, parser.getDeprecationHandler())) {
                     numReducePhases = parser.intValue();
-                } else if (IS_PARTIAL.match(currentFieldName, parser.getDeprecationHandler())) {
-                    isPartial = parser.booleanValue();
-                } else if (SEQUENCE_NUMBER.match(currentFieldName, parser.getDeprecationHandler())) {
-                    sequenceNumber = parser.intValue();
-                } else if (TOTAL_PARTIALS.match(currentFieldName, parser.getDeprecationHandler())) {
-                    totalPartials = parser.intValue();
                 } else {
                     parser.skipChildren();
                 }
@@ -614,22 +554,6 @@ public static SearchResponse innerFromXContent(XContentParser parser) throws IOE
             clusters,
             searchContextId
         );
-        SearchResponse response = new SearchResponse(
-            internalResponse,
-            scrollId,
-            totalShards,
-            successfulShards,
-            skippedShards,
-            tookInMillis,
-            phaseTook,
-            failures.toArray(ShardSearchFailure.EMPTY_ARRAY),
-            clusters,
-            searchContextId
-        );
-        response.setPartial(isPartial);
-        response.setSequenceNumber(sequenceNumber);
-        response.setTotalPartials(totalPartials);
-        return response;
     }
 
     @Override
@@ -650,13 +574,6 @@ public void writeTo(StreamOutput out) throws IOException {
         }
         out.writeVInt(skippedShards);
         out.writeOptionalString(pointInTimeId);
-        
-        // Write streaming fields
-        if (out.getVersion().onOrAfter(Version.V_2_15_0)) {
-            out.writeBoolean(isPartial);
-            out.writeVInt(sequenceNumber);
-            out.writeVInt(totalPartials);
-        }
     }
 
     @Override

server/src/main/java/org/opensearch/action/search/StreamQueryPhaseResultConsumer.java

@@ -10,43 +10,19 @@
 
 import org.opensearch.core.common.breaker.CircuitBreaker;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
-import org.opensearch.search.SearchPhaseResult;
-import org.opensearch.search.query.QuerySearchResult;
-import org.opensearch.search.query.StreamingSearchMode;
 
 import java.util.concurrent.Executor;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 
 /**
  * Query phase result consumer for streaming search.
- * Supports progressive batch reduction with configurable scoring modes.
- *
- * Batch reduction frequency is controlled by per-mode multipliers:
- * - NO_SCORING: Immediate reduction (batch size = 1) for fastest
- * time-to-first-byte
- * - SCORED_UNSORTED: Small batches (minBatchReduceSize * 2)
- * - SCORED_SORTED: Larger batches (minBatchReduceSize * 10)
- *
- * These multipliers are applied to the base batch reduce size (typically 5) to
- * determine
- * how many shard results are accumulated before triggering a partial reduction.
- * Lower values
- * mean more frequent reductions and faster streaming, but higher coordinator
- * CPU usage.
+ * Supports progressive batch reduction with a fixed batch policy suitable for unsorted streaming.
  *
  * @opensearch.internal
  */
 public class StreamQueryPhaseResultConsumer extends QueryPhaseResultConsumer {
 
-    private final StreamingSearchMode scoringMode;
-    private int resultsReceived = 0;
 
-    // TTFB tracking for demonstrating fetch phase timing
-    private long queryStartTime = System.currentTimeMillis();
-    private long firstBatchReadyForFetchTime = -1;
-    private boolean firstBatchReadyForFetch = false;
-    private final AtomicInteger batchesReduced = new AtomicInteger(0);
 
     /**
      * Creates a streaming query phase result consumer.
@@ -71,80 +47,20 @@ public StreamQueryPhaseResultConsumer(
             expectedResultSize,
             onPartialMergeFailure
         );
-
-        // Initialize scoring mode from request
-        String mode = request.getStreamingSearchMode();
-        this.scoringMode = (mode != null) ? StreamingSearchMode.fromString(mode) : StreamingSearchMode.SCORED_SORTED;
     }
 
     /**
-     * Controls partial reduction frequency based on scoring mode.
+     * Controls partial reduction frequency.
+     * With NO_SCORING streaming, we reduce immediately for fastest TTFB.
      *
      * @param requestBatchedReduceSize request batch size
      * @param minBatchReduceSize       minimum batch size
      */
     @Override
     int getBatchReduceSize(int requestBatchedReduceSize, int minBatchReduceSize) {
-        // Handle null during construction (parent constructor calls this before our
-        // constructor body runs)
-        if (scoringMode == null) {
-            return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * 10);
-        }
-
-        switch (scoringMode) {
-            case NO_SCORING:
-                // Reduce immediately for fastest TTFB
-                return Math.min(requestBatchedReduceSize, 1);
-            case SCORED_UNSORTED:
-                // Small batches for quick emission without sorting overhead
-                return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * 2);
-            case SCORED_SORTED:
-                // Higher batch size to collect more results before reducing (sorting is
-                // expensive)
-                return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * 10);
-            default:
-                return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * 10);
-        }
-    }
-
-    /**
-     * Consume streaming results with frequency-based emission
-     */
-
-    @Override
-    public void consumeResult(SearchPhaseResult result, Runnable next) {
-        // Handle partial results (shard progress updates)
-        if (result.queryResult() != null && result.queryResult().isPartial()) {
-            // Coordinated snapshot: merge this partial result with current final results
-            // for immediate emission without saving the partial result in the reducer state.
-            QuerySearchResult queryResult = result.queryResult();
-            pendingReduces.notifySnapshot(queryResult);
-
-            // Continue the pipeline
-            next.run();
-            return;
-        }
-
-        // Handle final shard results
-        super.consumeResult(result, () -> {
-            // Once a final result is added to the reducer (via super.consumeResult),
-            // trigger a snapshot emission to show the new global state immediately.
-            pendingReduces.notifySnapshot(null);
-            next.run();
-        });
+        // Reduce immediately for fastest TTFB
+        return Math.min(requestBatchedReduceSize, 1);
     }
+}
 
-    /**
-     * Get TTFB metrics for benchmarking
-     */
-    public long getTimeToFirstBatch() {
-        if (firstBatchReadyForFetchTime > 0) {
-            return firstBatchReadyForFetchTime - queryStartTime;
-        }
-        return -1;
-    }
 
-    public boolean isFirstBatchReady() {
-        return firstBatchReadyForFetch;
-    }
-}