Deduplicate the list of names when deserializing InternalTopMetrics (e…

…lastic#116298) use deduplication infrastructure to deduplicate the names of metrics in InternalTopMetrics.
alexey-ivanov-es · Nov 28, 2024 · 38d754e · 38d754e
1 parent 9d93eb6
commit 38d754e
Show file tree

Hide file tree

Showing 3 changed files with 39 additions and 9 deletions.
diff --git a/server/src/test/java/org/elasticsearch/common/lucene/LuceneTests.java b/server/src/test/java/org/elasticsearch/common/lucene/LuceneTests.java
@@ -50,7 +50,10 @@
 import org.apache.lucene.tests.store.MockDirectoryWrapper;
 import org.apache.lucene.util.Bits;
 import org.apache.lucene.util.BytesRef;
-import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.fielddata.IndexFieldData;
@@ -74,7 +77,6 @@
 import static org.hamcrest.Matchers.equalTo;
 
 public class LuceneTests extends ESTestCase {
-    private static final NamedWriteableRegistry EMPTY_REGISTRY = new NamedWriteableRegistry(Collections.emptyList());
 
     public void testCleanIndex() throws IOException {
         MockDirectoryWrapper dir = newMockDirectory();
@@ -551,7 +553,6 @@ public void testSortFieldSerialization() throws IOException {
         Tuple<SortField, SortField> sortFieldTuple = randomSortField();
         SortField deserialized = copyInstance(
             sortFieldTuple.v1(),
-            EMPTY_REGISTRY,
             Lucene::writeSortField,
             Lucene::readSortField,
             TransportVersionUtils.randomVersion(random())
@@ -563,14 +564,25 @@ public void testSortValueSerialization() throws IOException {
         Object sortValue = randomSortValue();
         Object deserialized = copyInstance(
             sortValue,
-            EMPTY_REGISTRY,
             Lucene::writeSortValue,
             Lucene::readSortValue,
             TransportVersionUtils.randomVersion(random())
         );
         assertEquals(sortValue, deserialized);
     }
 
+    private static <T> T copyInstance(T original, Writeable.Writer<T> writer, Writeable.Reader<T> reader, TransportVersion version)
+        throws IOException {
+        try (BytesStreamOutput output = new BytesStreamOutput()) {
+            output.setTransportVersion(version);
+            writer.write(output, original);
+            try (StreamInput in = output.bytes().streamInput()) {
+                in.setTransportVersion(version);
+                return reader.read(in);
+            }
+        }
+    }
+
     public static Object randomSortValue() {
         return switch (randomIntBetween(0, 9)) {
             case 0 -> null;

diff --git a/test/framework/src/main/java/org/elasticsearch/test/ESTestCase.java b/test/framework/src/main/java/org/elasticsearch/test/ESTestCase.java
@@ -65,6 +65,7 @@
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.bytes.CompositeBytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.io.stream.DelayableWriteable;
 import org.elasticsearch.common.io.stream.NamedWriteable;
 import org.elasticsearch.common.io.stream.NamedWriteableAwareStreamInput;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
@@ -1849,7 +1850,7 @@ public static <C extends NamedWriteable, T extends C> C copyNamedWriteable(
         );
     }
 
-    protected static <T> T copyInstance(
+    protected static <T extends Writeable> T copyInstance(
         T original,
         NamedWriteableRegistry namedWriteableRegistry,
         Writeable.Writer<T> writer,
@@ -1859,9 +1860,20 @@ protected static <T> T copyInstance(
         try (BytesStreamOutput output = new BytesStreamOutput()) {
             output.setTransportVersion(version);
             writer.write(output, original);
-            try (StreamInput in = new NamedWriteableAwareStreamInput(output.bytes().streamInput(), namedWriteableRegistry)) {
-                in.setTransportVersion(version);
-                return reader.read(in);
+            if (randomBoolean()) {
+                try (StreamInput in = new NamedWriteableAwareStreamInput(output.bytes().streamInput(), namedWriteableRegistry)) {
+                    in.setTransportVersion(version);
+                    return reader.read(in);
+                }
+            } else {
+                BytesReference bytesReference = output.copyBytes();
+                output.reset();
+                output.writeBytesReference(bytesReference);
+                try (StreamInput in = new NamedWriteableAwareStreamInput(output.bytes().streamInput(), namedWriteableRegistry)) {
+                    in.setTransportVersion(version);
+                    DelayableWriteable<T> delayableWriteable = DelayableWriteable.delayed(reader, in);
+                    return delayableWriteable.expand();
+                }
             }
         }
     }

diff --git a/...lytics/src/main/java/org/elasticsearch/xpack/analytics/topmetrics/InternalTopMetrics.java b/...lytics/src/main/java/org/elasticsearch/xpack/analytics/topmetrics/InternalTopMetrics.java
@@ -7,6 +7,7 @@
 package org.elasticsearch.xpack.analytics.topmetrics;
 
 import org.apache.lucene.util.PriorityQueue;
+import org.elasticsearch.common.io.stream.DelayableWriteable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -68,7 +69,12 @@ static InternalTopMetrics buildEmptyAggregation(String name, List<String> metric
     public InternalTopMetrics(StreamInput in) throws IOException {
         super(in);
         sortOrder = SortOrder.readFromStream(in);
-        metricNames = in.readStringCollectionAsList();
+        final List<String> metricNames = in.readStringCollectionAsList();
+        if (in instanceof DelayableWriteable.Deduplicator bo) {
+            this.metricNames = bo.deduplicate(metricNames);
+        } else {
+            this.metricNames = metricNames;
+        }
         size = in.readVInt();
         topMetrics = in.readCollectionAsList(TopMetric::new);
     }