[ES|QL] Fix aggregate_metric_double sorting and mv_expand issues

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/AggregateMetricDoubleBlockBuilder.java

@@ -59,7 +59,7 @@ protected int elementSize() {
     }
 
     @Override
-    public Block.Builder copyFrom(Block b, int beginInclusive, int endExclusive) {
+    public AggregateMetricDoubleBlockBuilder copyFrom(Block b, int beginInclusive, int endExclusive) {
         Block minBlock;
         Block maxBlock;
         Block sumBlock;
@@ -84,7 +84,7 @@ public Block.Builder copyFrom(Block b, int beginInclusive, int endExclusive) {
     }
 
     @Override
-    public AbstractBlockBuilder appendNull() {
+    public AggregateMetricDoubleBlockBuilder appendNull() {
         minBuilder.appendNull();
         maxBuilder.appendNull();
         sumBuilder.appendNull();
@@ -93,7 +93,7 @@ public AbstractBlockBuilder appendNull() {
     }
 
     @Override
-    public Block.Builder mvOrdering(Block.MvOrdering mvOrdering) {
+    public AggregateMetricDoubleBlockBuilder mvOrdering(Block.MvOrdering mvOrdering) {
         minBuilder.mvOrdering(mvOrdering);
         maxBuilder.mvOrdering(mvOrdering);
         sumBuilder.mvOrdering(mvOrdering);
@@ -102,7 +102,7 @@ public Block.Builder mvOrdering(Block.MvOrdering mvOrdering) {
     }
 
     @Override
-    public Block build() {
+    public AggregateMetricDoubleBlock build() {
         DoubleBlock minBlock = null;
         DoubleBlock maxBlock = null;
         DoubleBlock sumBlock = null;

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/BlockFactory.java

@@ -436,7 +436,7 @@ public AggregateMetricDoubleBlockBuilder newAggregateMetricDoubleBlockBuilder(in
         return new AggregateMetricDoubleBlockBuilder(estimatedSize, this);
     }
 
-    public final Block newConstantAggregateMetricDoubleBlock(
+    public final AggregateMetricDoubleBlock newConstantAggregateMetricDoubleBlock(
         AggregateMetricDoubleBlockBuilder.AggregateMetricDoubleLiteral value,
         int positions
     ) {

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/data/AggregateMetricDoubleBlockEqualityTests.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.compute.data;
+
+import org.elasticsearch.compute.test.ComputeTestCase;
+import org.elasticsearch.compute.test.TestBlockFactory;
+import org.elasticsearch.core.Releasables;
+
+import java.util.List;
+
+public class AggregateMetricDoubleBlockEqualityTests extends ComputeTestCase {
+
+    static final BlockFactory blockFactory = TestBlockFactory.getNonBreakingInstance();
+
+    // TODO: Add additional tests
+
+    public void testEmptyBlock() {
+        // all these "empty" vectors should be equivalent
+        var partialMetricBuilder = blockFactory.newAggregateMetricDoubleBlockBuilder(0);
+        for (var subBuilder : List.of(
+            partialMetricBuilder.min(),
+            partialMetricBuilder.max(),
+            partialMetricBuilder.sum(),
+            partialMetricBuilder.count()
+        )) {
+            if (randomBoolean()) {
+                subBuilder.appendNull();
+            } else {
+                if (subBuilder instanceof DoubleBlockBuilder doubleBlockBuilder) {
+                    doubleBlockBuilder.appendDouble(0.0);
+                } else if (subBuilder instanceof IntBlockBuilder intBlockBuilder) {
+                    intBlockBuilder.appendInt(0);
+                }
+            }
+        }
+
+        List<AggregateMetricDoubleBlock> blocks = List.of(
+            blockFactory.newAggregateMetricDoubleBlockBuilder(0).build(),
+            blockFactory.newAggregateMetricDoubleBlockBuilder(0).appendNull().build().filter(),
+            partialMetricBuilder.build().filter(),
+            blockFactory.newConstantAggregateMetricDoubleBlock(
+                new AggregateMetricDoubleBlockBuilder.AggregateMetricDoubleLiteral(0.0, 0.0, 0.0, 0),
+                0
+            ).filter(),
+            (ConstantNullBlock) blockFactory.newConstantNullBlock(0)
+        );
+
+        assertAllEquals(blocks);
+        Releasables.close(blocks);
+    }
+
+    public void testSimpleBlockWithManyNulls() {
+        int positions = randomIntBetween(1, 256);
+        boolean grow = randomBoolean();
+        AggregateMetricDoubleBlockBuilder builder1 = blockFactory.newAggregateMetricDoubleBlockBuilder(grow ? 0 : positions);
+        AggregateMetricDoubleBlockBuilder builder2 = blockFactory.newAggregateMetricDoubleBlockBuilder(grow ? 0 : positions);
+        ConstantNullBlock.Builder builder3 = new ConstantNullBlock.Builder(blockFactory);
+        for (int p = 0; p < positions; p++) {
+            builder1.appendNull();
+            builder2.appendNull();
+            builder3.appendNull();
+        }
+        AggregateMetricDoubleBlock block1 = builder1.build();
+        AggregateMetricDoubleBlock block2 = builder2.build();
+        Block block3 = builder3.build();
+        assertEquals(positions, block1.getPositionCount());
+        assertTrue(block1.mayHaveNulls());
+        assertTrue(block1.isNull(0));
+
+        List<Block> blocks = List.of(block1, block2, block3);
+        assertAllEquals(blocks);
+    }
+
+    static void assertAllEquals(List<?> objs) {
+        for (Object obj1 : objs) {
+            for (Object obj2 : objs) {
+                assertEquals(obj1, obj2);
+                assertEquals(obj2, obj1);
+                // equal objects must generate the same hash code
+                assertEquals(obj1.hashCode(), obj2.hashCode());
+            }
+        }
+    }
+}