Merge branch 'chunked_fetch_phase' of github.com:drempapis/elasticsearch into chunked_fetch_phase

Author: drempapis

server/src/main/java/org/elasticsearch/search/fetch/FetchPhaseDocsIterator.java

@@ -224,11 +224,7 @@ public final IterateResult iterate(
                     hit.decRef();
                 }
 
-                lastChunk = new SearchHits(
-                    lastHitsArray,
-                    new TotalHits(lastHitsArray.length, TotalHits.Relation.EQUAL_TO),
-                    Float.NaN
-                );
+                lastChunk = new SearchHits(lastHitsArray, new TotalHits(lastHitsArray.length, TotalHits.Relation.EQUAL_TO), Float.NaN);
                 chunkBuffer.clear();
             }
         } catch (SearchTimeoutException e) {
@@ -279,11 +275,7 @@ private static CompletableFuture<Void> sendChunk(
 
         SearchHits chunkHits = null;
         try {
-            chunkHits = new SearchHits(
-                hitsArray,
-                new TotalHits(hitsArray.length, TotalHits.Relation.EQUAL_TO),
-                maxScore
-            );
+            chunkHits = new SearchHits(hitsArray, new TotalHits(hitsArray.length, TotalHits.Relation.EQUAL_TO), maxScore);
             final SearchHits finalChunkHits = chunkHits;
 
             FetchPhaseResponseChunk chunk = new FetchPhaseResponseChunk(
@@ -298,18 +290,15 @@ private static CompletableFuture<Void> sendChunk(
             );
 
             // Send the chunk - coordinator will take ownership of the hits
-            writer.writeResponseChunk(chunk, ActionListener.wrap(
-                ack -> {
-                    // Coordinator now owns the hits, decRef to release local reference
-                    finalChunkHits.decRef();
-                    future.complete(null);
-                },
-                ex -> {
-                    // Failed to send - we still own the hits, must clean up
-                    finalChunkHits.decRef();
-                    future.completeExceptionally(ex);
-                }
-            ));
+            writer.writeResponseChunk(chunk, ActionListener.wrap(ack -> {
+                // Coordinator now owns the hits, decRef to release local reference
+                finalChunkHits.decRef();
+                future.complete(null);
+            }, ex -> {
+                // Failed to send - we still own the hits, must clean up
+                finalChunkHits.decRef();
+                future.completeExceptionally(ex);
+            }));
         } catch (Exception e) {
             future.completeExceptionally(e);
             // If chunk creation failed after SearchHits was created, clean up

server/src/main/java/org/elasticsearch/search/fetch/chunk/FetchPhaseResponseStream.java

@@ -103,13 +103,13 @@ void writeChunk(FetchPhaseResponseChunk chunk, Releasable releasable) {
 
             if (logger.isTraceEnabled()) {
                 logger.info(
-                        "Received chunk [{}] docs for shard [{}]: [{}/{}] hits accumulated, [{}] breaker bytes, used breaker bytes [{}]",
-                        chunk.hits() == null ? 0 : chunk.hits().getHits().length,
-                        shardIndex,
-                        queue.size(),
-                        expectedDocs,
-                        totalBreakerBytes.get(),
-                        circuitBreaker.getUsed()
+                    "Received chunk [{}] docs for shard [{}]: [{}/{}] hits accumulated, [{}] breaker bytes, used breaker bytes [{}]",
+                    chunk.hits() == null ? 0 : chunk.hits().getHits().length,
+                    shardIndex,
+                    queue.size(),
+                    expectedDocs,
+                    totalBreakerBytes.get(),
+                    circuitBreaker.getUsed()
                 );
             }
             success = true;
@@ -159,9 +159,9 @@ FetchSearchResult buildFinalResult(ShardSearchContextId ctxId, SearchShardTarget
         ownershipTransferred = true;
 
         SearchHits searchHits = new SearchHits(
-                orderedHits.toArray(SearchHit[]::new),
-                new TotalHits(orderedHits.size(), TotalHits.Relation.EQUAL_TO),
-                maxScore
+            orderedHits.toArray(SearchHit[]::new),
+            new TotalHits(orderedHits.size(), TotalHits.Relation.EQUAL_TO),
+            maxScore
         );
 
         FetchSearchResult result = new FetchSearchResult(ctxId, shardTarget);
@@ -201,10 +201,10 @@ int getCurrentQueueSize() {
     protected void closeInternal() {
         if (logger.isTraceEnabled()) {
             logger.info(
-                    "Closing response stream for shard [{}], releasing [{}] hits, [{}] breaker bytes",
-                    shardIndex,
-                    queue.size(),
-                    totalBreakerBytes.get()
+                "Closing response stream for shard [{}], releasing [{}] hits, [{}] breaker bytes",
+                shardIndex,
+                queue.size(),
+                totalBreakerBytes.get()
             );
         }
 
@@ -220,10 +220,10 @@ protected void closeInternal() {
             circuitBreaker.addWithoutBreaking(-totalBreakerBytes.get());
             if (logger.isTraceEnabled()) {
                 logger.info(
-                        "Released [{}] breaker bytes for shard [{}], used breaker bytes [{}]",
-                        totalBreakerBytes.get(),
-                        shardIndex,
-                        circuitBreaker.getUsed()
+                    "Released [{}] breaker bytes for shard [{}], used breaker bytes [{}]",
+                    totalBreakerBytes.get(),
+                    shardIndex,
+                    circuitBreaker.getUsed()
                 );
             }
             totalBreakerBytes.set(0);

server/src/test/java/org/elasticsearch/action/search/KnnSearchSingleNodeTests.java

@@ -9,7 +9,6 @@
 
 package org.elasticsearch.action.search;
 
-
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.settings.Settings;