introduce AggregateMetricDoubleVector?

Author: limotova

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

@@ -205,4 +205,69 @@ public TransportVersion getMinimalSupportedVersion() {
         }
 
     }
+
+    public static class AggregateMetricDoubleVectorBuilder extends AbstractBlockBuilder {
+        private final DoubleBlockBuilder valuesBuilder;
+
+        public AggregateMetricDoubleVectorBuilder(int estimatedSize, BlockFactory blockFactory) {
+            super(blockFactory);
+            valuesBuilder = new DoubleBlockBuilder(estimatedSize, blockFactory);
+        }
+
+        @Override
+        protected int valuesLength() {
+            throw new UnsupportedOperationException("Not available on aggregate_metric_double");
+        }
+
+        @Override
+        protected void growValuesArray(int newSize) {
+            throw new UnsupportedOperationException("Not available on aggregate_metric_double");
+        }
+
+        @Override
+        protected int elementSize() {
+            throw new UnsupportedOperationException("Not available on aggregate_metric_double");
+        }
+
+        @Override
+        public Block.Builder copyFrom(Block block, int beginInclusive, int endExclusive) {
+            // TODO
+            return null;
+        }
+
+        @Override
+        public Block.Builder mvOrdering(Block.MvOrdering mvOrdering) {
+            // TODO
+            return null;
+        }
+
+        public void appendValue(double value) {
+            valuesBuilder.appendDouble(value);
+        }
+
+        @Override
+        public Block build() {
+            Block[] blocks = new Block[4];
+            Block block = null;
+            ConstantIntVector countVector = null;
+            boolean success = false;
+            try {
+                finish();
+                block = valuesBuilder.build();
+                countVector = new ConstantIntVector(1, block.getPositionCount(), blockFactory);
+                blocks[Metric.MIN.getIndex()] = block;
+                blocks[Metric.MAX.getIndex()] = block;
+                blocks[Metric.SUM.getIndex()] = block;
+                blocks[Metric.COUNT.getIndex()] = countVector.asBlock();
+                CompositeBlock compositeBlock = new CompositeBlock(blocks);
+                success = true;
+                return compositeBlock;
+            } finally {
+                if (success == false) {
+                    Releasables.closeExpectNoException(block, countVector);
+                }
+            }
+        }
+
+    }
 }

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

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

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/ToAggregateMetricDouble.java

@@ -12,9 +12,11 @@
 import org.elasticsearch.compute.data.AggregateMetricDoubleBlockBuilder;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.DoubleVector;
 import org.elasticsearch.compute.data.IntBlock;
 import org.elasticsearch.compute.data.LongBlock;
 import org.elasticsearch.compute.data.Page;
+import org.elasticsearch.compute.data.Vector;
 import org.elasticsearch.compute.operator.DriverContext;
 import org.elasticsearch.compute.operator.EvalOperator;
 import org.elasticsearch.core.Releasables;
@@ -130,40 +132,60 @@ public EvalOperator.ExpressionEvaluator get(DriverContext context) {
             final EvalOperator.ExpressionEvaluator eval = fieldEvaluator.get(context);
 
             return new EvalOperator.ExpressionEvaluator() {
+                private Block evalBlock(Block block) {
+                    int positionCount = block.getPositionCount();
+                    DoubleBlock doubleBlock = (DoubleBlock) block;
+                    try (
+                        AggregateMetricDoubleBlockBuilder result = context.blockFactory()
+                            .newAggregateMetricDoubleBlockBuilder(positionCount)
+                    ) {
+                        CompensatedSum sum = new CompensatedSum();
+                        for (int p = 0; p < positionCount; p++) {
+                            int valueCount = doubleBlock.getValueCount(p);
+                            int start = doubleBlock.getFirstValueIndex(p);
+                            int end = start + valueCount;
+                            if (valueCount == 0) {
+                                result.appendNull();
+                                continue;
+                            }
+                            double min = Double.POSITIVE_INFINITY;
+                            double max = Double.NEGATIVE_INFINITY;
+                            for (int i = start; i < end; i++) {
+                                double current = doubleBlock.getDouble(i);
+                                min = Math.min(min, current);
+                                max = Math.max(max, current);
+                                sum.add(current);
+                            }
+                            result.min().appendDouble(min);
+                            result.max().appendDouble(max);
+                            result.sum().appendDouble(sum.value());
+                            result.count().appendInt(valueCount);
+                            sum.reset(0, 0);
+                        }
+                        return result.build();
+                    }
+                }
+
+                private Block evalVector(Vector vector) {
+                    int positionCount = vector.getPositionCount();
+                    DoubleVector doubleVector = (DoubleVector) vector;
+                    try (
+                        AggregateMetricDoubleBlockBuilder.AggregateMetricDoubleVectorBuilder builder = context.blockFactory()
+                            .newAggregateMetricDoubleVectorBuilder(positionCount)
+                    ) {
+                        for (int p = 0; p < positionCount; p++) {
+                            double value = doubleVector.getDouble(p);
+                            builder.appendValue(value);
+                        }
+                        return builder.build();
+                    }
+                }
+
                 @Override
                 public Block eval(Page page) {
                     try (Block block = eval.eval(page)) {
-                        int positionCount = block.getPositionCount();
-                        DoubleBlock doubleBlock = (DoubleBlock) block;
-                        try (
-                            AggregateMetricDoubleBlockBuilder result = context.blockFactory()
-                                .newAggregateMetricDoubleBlockBuilder(positionCount)
-                        ) {
-                            CompensatedSum sum = new CompensatedSum();
-                            for (int p = 0; p < positionCount; p++) {
-                                int valueCount = doubleBlock.getValueCount(p);
-                                int start = doubleBlock.getFirstValueIndex(p);
-                                int end = start + valueCount;
-                                if (valueCount == 0) {
-                                    result.appendNull();
-                                    continue;
-                                }
-                                double min = Double.POSITIVE_INFINITY;
-                                double max = Double.NEGATIVE_INFINITY;
-                                for (int i = start; i < end; i++) {
-                                    double current = doubleBlock.getDouble(i);
-                                    min = Math.min(min, current);
-                                    max = Math.max(max, current);
-                                    sum.add(current);
-                                }
-                                result.min().appendDouble(min);
-                                result.max().appendDouble(max);
-                                result.sum().appendDouble(sum.value());
-                                result.count().appendInt(valueCount);
-                                sum.reset(0, 0);
-                            }
-                            return result.build();
-                        }
+                        Vector vector = block.asVector();
+                        return vector == null ? evalBlock(block) : evalVector(vector);
                     }
                 }