Separate serialization logic from class

Signed-off-by: Vacha Shah <vachshah@amazon.com>

Author: VachaShah

server/src/main/java/org/opensearch/common/cache/policy/CachedQueryResult.java

@@ -52,12 +52,12 @@ public static void loadQSR(
     ) throws IOException {
         StreamInput in = new NamedWriteableAwareStreamInput(serializedCQR.streamInput(), registry);
         PolicyValues pv = new PolicyValues(in); // Read and discard PolicyValues
-        qsr.readFromWithId(id, in);
+        qsr.getSerializer().readFromWithId(in);
     }
 
     public void writeToNoId(StreamOutput out) throws IOException {
         policyValues.writeTo(out);
-        qsr.writeToNoId(out);
+        qsr.getSerializer().writeToNoId(out);
     }
 
     /**

server/src/main/java/org/opensearch/search/fetch/FetchSearchResult.java

@@ -39,6 +39,7 @@
 import org.opensearch.search.SearchHits;
 import org.opensearch.search.SearchPhaseResult;
 import org.opensearch.search.SearchShardTarget;
+import org.opensearch.search.fetch.subphase.serializer.NativeFetchSearchResultSerializer;
 import org.opensearch.search.internal.ShardSearchContextId;
 import org.opensearch.search.query.QuerySearchResult;
 
@@ -52,16 +53,16 @@
 @PublicApi(since = "1.0.0")
 public final class FetchSearchResult extends SearchPhaseResult {
 
-    private SearchHits hits;
     // client side counter
     private transient int counter;
+    private final NativeFetchSearchResultSerializer serializer = new NativeFetchSearchResultSerializer();
 
     public FetchSearchResult() {}
 
     public FetchSearchResult(StreamInput in) throws IOException {
         super(in);
         contextId = new ShardSearchContextId(in);
-        hits = new SearchHits(in);
+        serializer.readFetchSearchResult(in);
     }
 
     public FetchSearchResult(ShardSearchContextId id, SearchShardTarget shardTarget) {
@@ -81,7 +82,7 @@ public FetchSearchResult fetchResult() {
 
     public void hits(SearchHits hits) {
         assert assertNoSearchTarget(hits);
-        this.hits = hits;
+        serializer.setHits(hits);
     }
 
     private boolean assertNoSearchTarget(SearchHits hits) {
@@ -92,7 +93,7 @@ private boolean assertNoSearchTarget(SearchHits hits) {
     }
 
     public SearchHits hits() {
-        return hits;
+        return serializer.getHits();
     }
 
     public FetchSearchResult initCounter() {
@@ -107,6 +108,6 @@ public int counterGetAndIncrement() {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         contextId.writeTo(out);
-        hits.writeTo(out);
+        serializer.writeFetchSearchResult(out);
     }
 }

server/src/main/java/org/opensearch/search/fetch/subphase/serializer/FetchSearchResultSerializer.java

@@ -0,0 +1,19 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.search.fetch.subphase.serializer;
+
+import java.io.IOException;
+
+public interface FetchSearchResultSerializer<T, V> {
+
+    void readFetchSearchResult(T in) throws IOException;
+
+    void writeFetchSearchResult(V out) throws IOException;
+
+}

server/src/main/java/org/opensearch/search/fetch/subphase/serializer/NativeFetchSearchResultSerializer.java

@@ -0,0 +1,39 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.search.fetch.subphase.serializer;
+
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.search.SearchHits;
+
+import java.io.IOException;
+
+public class NativeFetchSearchResultSerializer implements FetchSearchResultSerializer<StreamInput, StreamOutput> {
+
+    private SearchHits hits;
+
+    @Override
+    public void readFetchSearchResult(StreamInput in) throws IOException {
+        hits = new SearchHits(in);
+    }
+
+    @Override
+    public void writeFetchSearchResult(StreamOutput out) throws IOException {
+        hits.writeTo(out);
+    }
+
+    public SearchHits getHits() {
+        return hits;
+    }
+
+    public void setHits(SearchHits hits) {
+        this.hits = hits;
+    }
+
+}