Introduce SurrogateExpression (ESQL-1285)

Convert Avg into a SurrogateExpression and introduce dedicated rule
 for handling surrogate AggregateFunction
Remove Avg implementation
Use sum instead of avg in some planning test
Add dataType case for Div operator

Relates ESQL-747

benchmarks/src/main/java/org/elasticsearch/benchmark/compute/operator/AggregatorBenchmark.java

@@ -12,8 +12,6 @@ import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.compute.aggregation.AggregatorFunctionSupplier;
 import org.elasticsearch.compute.aggregation.AggregatorMode;
-import org.elasticsearch.compute.aggregation.AvgDoubleAggregatorFunctionSupplier;
-import org.elasticsearch.compute.aggregation.AvgLongAggregatorFunctionSupplier;
 import org.elasticsearch.compute.aggregation.CountAggregatorFunction;
 import org.elasticsearch.compute.aggregation.CountDistinctDoubleAggregatorFunctionSupplier;
 import org.elasticsearch.compute.aggregation.CountDistinctLongAggregatorFunctionSupplier;
@@ -144,11 +142,6 @@ public class AggregatorBenchmark {
 
     private static AggregatorFunctionSupplier supplier(String op, String dataType, int dataChannel) {
         return switch (op) {
-            case AVG -> switch (dataType) {
-                case LONGS -> new AvgLongAggregatorFunctionSupplier(BIG_ARRAYS, List.of(dataChannel));
-                case DOUBLES -> new AvgDoubleAggregatorFunctionSupplier(BIG_ARRAYS, List.of(dataChannel));
-                default -> throw new IllegalArgumentException("unsupported data type [" + dataType + "]");
-            };
             case COUNT -> CountAggregatorFunction.supplier(BIG_ARRAYS, List.of(dataChannel));
             case COUNT_DISTINCT -> switch (dataType) {
                 case LONGS -> new CountDistinctLongAggregatorFunctionSupplier(BIG_ARRAYS, List.of(dataChannel), 3000);

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/expression/function/EqlFunctionRegistry.java

@@ -68,7 +68,7 @@ public class EqlFunctionRegistry extends FunctionRegistry {
             // Arithmetic
             new FunctionDefinition[] {
                 def(Add.class, Add::new, "add"),
-                def(Div.class, Div::new, "divide"),
+                def(Div.class, (BinaryBuilder<Div>) Div::new, "divide"),
                 def(Mod.class, Mod::new, "modulo"),
                 def(Mul.class, Mul::new, "multiply"),
                 def(ToNumber.class, ToNumber::new, "number"),

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/AvgDoubleAggregatorFunction.java

@@ -1,120 +0,0 @@
-// 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.aggregation;
-
-import java.lang.Integer;
-import java.lang.Override;
-import java.lang.String;
-import java.lang.StringBuilder;
-import java.util.List;
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.AggregatorStateVector;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.data.DoubleVector;
-import org.elasticsearch.compute.data.ElementType;
-import org.elasticsearch.compute.data.IntVector;
-import org.elasticsearch.compute.data.Page;
-import org.elasticsearch.compute.data.Vector;
-
-/**
- * {@link AggregatorFunction} implementation for {@link AvgDoubleAggregator}.
- * This class is generated. Do not edit it.
- */
-public final class AvgDoubleAggregatorFunction implements AggregatorFunction {
-  private final AvgDoubleAggregator.AvgState state;
-
-  private final List<Integer> channels;
-
-  public AvgDoubleAggregatorFunction(List<Integer> channels, AvgDoubleAggregator.AvgState state) {
-    this.channels = channels;
-    this.state = state;
-  }
-
-  public static AvgDoubleAggregatorFunction create(List<Integer> channels) {
-    return new AvgDoubleAggregatorFunction(channels, AvgDoubleAggregator.initSingle());
-  }
-
-  @Override
-  public void addRawInput(Page page) {
-    ElementType type = page.getBlock(channels.get(0)).elementType();
-    if (type == ElementType.NULL) {
-      return;
-    }
-    DoubleBlock block = page.getBlock(channels.get(0));
-    DoubleVector vector = block.asVector();
-    if (vector != null) {
-      addRawVector(vector);
-    } else {
-      addRawBlock(block);
-    }
-  }
-
-  private void addRawVector(DoubleVector vector) {
-    for (int i = 0; i < vector.getPositionCount(); i++) {
-      AvgDoubleAggregator.combine(state, vector.getDouble(i));
-    }
-    AvgDoubleAggregator.combineValueCount(state, vector.getPositionCount());
-  }
-
-  private void addRawBlock(DoubleBlock block) {
-    for (int p = 0; p < block.getPositionCount(); p++) {
-      if (block.isNull(p)) {
-        continue;
-      }
-      int start = block.getFirstValueIndex(p);
-      int end = start + block.getValueCount(p);
-      for (int i = start; i < end; i++) {
-        AvgDoubleAggregator.combine(state, block.getDouble(i));
-      }
-    }
-    AvgDoubleAggregator.combineValueCount(state, block.getTotalValueCount());
-  }
-
-  @Override
-  public void addIntermediateInput(Page page) {
-    Block block = page.getBlock(channels.get(0));
-    Vector vector = block.asVector();
-    if (vector == null || vector instanceof AggregatorStateVector == false) {
-      throw new RuntimeException("expected AggregatorStateBlock, got:" + block);
-    }
-    @SuppressWarnings("unchecked") AggregatorStateVector<AvgDoubleAggregator.AvgState> blobVector = (AggregatorStateVector<AvgDoubleAggregator.AvgState>) vector;
-    // TODO exchange big arrays directly without funny serialization - no more copying
-    BigArrays bigArrays = BigArrays.NON_RECYCLING_INSTANCE;
-    AvgDoubleAggregator.AvgState tmpState = AvgDoubleAggregator.initSingle();
-    for (int i = 0; i < block.getPositionCount(); i++) {
-      blobVector.get(i, tmpState);
-      AvgDoubleAggregator.combineStates(state, tmpState);
-    }
-    tmpState.close();
-  }
-
-  @Override
-  public void evaluateIntermediate(Block[] blocks, int offset) {
-    AggregatorStateVector.Builder<AggregatorStateVector<AvgDoubleAggregator.AvgState>, AvgDoubleAggregator.AvgState> builder =
-        AggregatorStateVector.builderOfAggregatorState(AvgDoubleAggregator.AvgState.class, state.getEstimatedSize());
-    builder.add(state, IntVector.range(0, 1));
-    blocks[offset] = builder.build().asBlock();
-  }
-
-  @Override
-  public void evaluateFinal(Block[] blocks, int offset) {
-    blocks[offset] = AvgDoubleAggregator.evaluateFinal(state);
-  }
-
-  @Override
-  public String toString() {
-    StringBuilder sb = new StringBuilder();
-    sb.append(getClass().getSimpleName()).append("[");
-    sb.append("channels=").append(channels);
-    sb.append("]");
-    return sb.toString();
-  }
-
-  @Override
-  public void close() {
-    state.close();
-  }
-}

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/AvgDoubleAggregatorFunctionSupplier.java

@@ -1,41 +0,0 @@
-// 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.aggregation;
-
-import java.lang.Integer;
-import java.lang.Override;
-import java.lang.String;
-import java.util.List;
-import org.elasticsearch.common.util.BigArrays;
-
-/**
- * {@link AggregatorFunctionSupplier} implementation for {@link AvgDoubleAggregator}.
- * This class is generated. Do not edit it.
- */
-public final class AvgDoubleAggregatorFunctionSupplier implements AggregatorFunctionSupplier {
-  private final BigArrays bigArrays;
-
-  private final List<Integer> channels;
-
-  public AvgDoubleAggregatorFunctionSupplier(BigArrays bigArrays, List<Integer> channels) {
-    this.bigArrays = bigArrays;
-    this.channels = channels;
-  }
-
-  @Override
-  public AvgDoubleAggregatorFunction aggregator() {
-    return AvgDoubleAggregatorFunction.create(channels);
-  }
-
-  @Override
-  public AvgDoubleGroupingAggregatorFunction groupingAggregator() {
-    return AvgDoubleGroupingAggregatorFunction.create(channels, bigArrays);
-  }
-
-  @Override
-  public String describe() {
-    return "avg of doubles";
-  }
-}

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/AvgDoubleGroupingAggregatorFunction.java

@@ -1,182 +0,0 @@
-// 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.aggregation;
-
-import java.lang.Integer;
-import java.lang.Override;
-import java.lang.String;
-import java.lang.StringBuilder;
-import java.util.List;
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.AggregatorStateVector;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.data.DoubleVector;
-import org.elasticsearch.compute.data.IntVector;
-import org.elasticsearch.compute.data.LongBlock;
-import org.elasticsearch.compute.data.LongVector;
-import org.elasticsearch.compute.data.Page;
-import org.elasticsearch.compute.data.Vector;
-
-/**
- * {@link GroupingAggregatorFunction} implementation for {@link AvgDoubleAggregator}.
- * This class is generated. Do not edit it.
- */
-public final class AvgDoubleGroupingAggregatorFunction implements GroupingAggregatorFunction {
-  private final AvgDoubleAggregator.GroupingAvgState state;
-
-  private final List<Integer> channels;
-
-  private final BigArrays bigArrays;
-
-  public AvgDoubleGroupingAggregatorFunction(List<Integer> channels,
-      AvgDoubleAggregator.GroupingAvgState state, BigArrays bigArrays) {
-    this.channels = channels;
-    this.state = state;
-    this.bigArrays = bigArrays;
-  }
-
-  public static AvgDoubleGroupingAggregatorFunction create(List<Integer> channels,
-      BigArrays bigArrays) {
-    return new AvgDoubleGroupingAggregatorFunction(channels, AvgDoubleAggregator.initGrouping(bigArrays), bigArrays);
-  }
-
-  @Override
-  public void addRawInput(LongVector groups, Page page) {
-    DoubleBlock valuesBlock = page.getBlock(channels.get(0));
-    assert groups.getPositionCount() == page.getPositionCount();
-    DoubleVector valuesVector = valuesBlock.asVector();
-    if (valuesVector == null) {
-      addRawInput(groups, valuesBlock);
-    } else {
-      addRawInput(groups, valuesVector);
-    }
-  }
-
-  private void addRawInput(LongVector groups, DoubleBlock values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      int groupId = Math.toIntExact(groups.getLong(position));
-      if (values.isNull(position)) {
-        state.putNull(groupId);
-        continue;
-      }
-      int valuesStart = values.getFirstValueIndex(position);
-      int valuesEnd = valuesStart + values.getValueCount(position);
-      for (int v = valuesStart; v < valuesEnd; v++) {
-        AvgDoubleAggregator.combine(state, groupId, values.getDouble(v));
-      }
-    }
-  }
-
-  private void addRawInput(LongVector groups, DoubleVector values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      int groupId = Math.toIntExact(groups.getLong(position));
-      AvgDoubleAggregator.combine(state, groupId, values.getDouble(position));
-    }
-  }
-
-  @Override
-  public void addRawInput(LongBlock groups, Page page) {
-    DoubleBlock valuesBlock = page.getBlock(channels.get(0));
-    assert groups.getPositionCount() == page.getPositionCount();
-    DoubleVector valuesVector = valuesBlock.asVector();
-    if (valuesVector == null) {
-      addRawInput(groups, valuesBlock);
-    } else {
-      addRawInput(groups, valuesVector);
-    }
-  }
-
-  private void addRawInput(LongBlock groups, DoubleBlock values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      if (groups.isNull(position)) {
-        continue;
-      }
-      int groupStart = groups.getFirstValueIndex(position);
-      int groupEnd = groupStart + groups.getValueCount(position);
-      for (int g = groupStart; g < groupEnd; g++) {
-        int groupId = Math.toIntExact(groups.getLong(g));
-        if (values.isNull(position)) {
-          state.putNull(groupId);
-          continue;
-        }
-        int valuesStart = values.getFirstValueIndex(position);
-        int valuesEnd = valuesStart + values.getValueCount(position);
-        for (int v = valuesStart; v < valuesEnd; v++) {
-          AvgDoubleAggregator.combine(state, groupId, values.getDouble(v));
-        }
-      }
-    }
-  }
-
-  private void addRawInput(LongBlock groups, DoubleVector values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      if (groups.isNull(position)) {
-        continue;
-      }
-      int groupStart = groups.getFirstValueIndex(position);
-      int groupEnd = groupStart + groups.getValueCount(position);
-      for (int g = groupStart; g < groupEnd; g++) {
-        int groupId = Math.toIntExact(groups.getLong(g));
-        AvgDoubleAggregator.combine(state, groupId, values.getDouble(position));
-      }
-    }
-  }
-
-  @Override
-  public void addIntermediateInput(LongVector groupIdVector, Page page) {
-    Block block = page.getBlock(channels.get(0));
-    Vector vector = block.asVector();
-    if (vector == null || vector instanceof AggregatorStateVector == false) {
-      throw new RuntimeException("expected AggregatorStateBlock, got:" + block);
-    }
-    @SuppressWarnings("unchecked") AggregatorStateVector<AvgDoubleAggregator.GroupingAvgState> blobVector = (AggregatorStateVector<AvgDoubleAggregator.GroupingAvgState>) vector;
-    // TODO exchange big arrays directly without funny serialization - no more copying
-    BigArrays bigArrays = BigArrays.NON_RECYCLING_INSTANCE;
-    AvgDoubleAggregator.GroupingAvgState inState = AvgDoubleAggregator.initGrouping(bigArrays);
-    blobVector.get(0, inState);
-    for (int position = 0; position < groupIdVector.getPositionCount(); position++) {
-      int groupId = Math.toIntExact(groupIdVector.getLong(position));
-      AvgDoubleAggregator.combineStates(state, groupId, inState, position);
-    }
-    inState.close();
-  }
-
-  @Override
-  public void addIntermediateRowInput(int groupId, GroupingAggregatorFunction input, int position) {
-    if (input.getClass() != getClass()) {
-      throw new IllegalArgumentException("expected " + getClass() + "; got " + input.getClass());
-    }
-    AvgDoubleAggregator.GroupingAvgState inState = ((AvgDoubleGroupingAggregatorFunction) input).state;
-    AvgDoubleAggregator.combineStates(state, groupId, inState, position);
-  }
-
-  @Override
-  public void evaluateIntermediate(Block[] blocks, int offset, IntVector selected) {
-    AggregatorStateVector.Builder<AggregatorStateVector<AvgDoubleAggregator.GroupingAvgState>, AvgDoubleAggregator.GroupingAvgState> builder =
-        AggregatorStateVector.builderOfAggregatorState(AvgDoubleAggregator.GroupingAvgState.class, state.getEstimatedSize());
-    builder.add(state, selected);
-    blocks[offset] = builder.build().asBlock();
-  }
-
-  @Override
-  public void evaluateFinal(Block[] blocks, int offset, IntVector selected) {
-    blocks[offset] = AvgDoubleAggregator.evaluateFinal(state, selected);
-  }
-
-  @Override
-  public String toString() {
-    StringBuilder sb = new StringBuilder();
-    sb.append(getClass().getSimpleName()).append("[");
-    sb.append("channels=").append(channels);
-    sb.append("]");
-    return sb.toString();
-  }
-
-  @Override
-  public void close() {
-    state.close();
-  }
-}

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/AvgIntAggregatorFunction.java

@@ -1,119 +0,0 @@
-// 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.aggregation;
-
-import java.lang.Integer;
-import java.lang.Override;
-import java.lang.String;
-import java.lang.StringBuilder;
-import java.util.List;
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.AggregatorStateVector;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.ElementType;
-import org.elasticsearch.compute.data.IntBlock;
-import org.elasticsearch.compute.data.IntVector;
-import org.elasticsearch.compute.data.Page;
-import org.elasticsearch.compute.data.Vector;
-
-/**
- * {@link AggregatorFunction} implementation for {@link AvgIntAggregator}.
- * This class is generated. Do not edit it.
- */
-public final class AvgIntAggregatorFunction implements AggregatorFunction {
-  private final AvgLongAggregator.AvgState state;
-
-  private final List<Integer> channels;
-
-  public AvgIntAggregatorFunction(List<Integer> channels, AvgLongAggregator.AvgState state) {
-    this.channels = channels;
-    this.state = state;
-  }
-
-  public static AvgIntAggregatorFunction create(List<Integer> channels) {
-    return new AvgIntAggregatorFunction(channels, AvgIntAggregator.initSingle());
-  }
-
-  @Override
-  public void addRawInput(Page page) {
-    ElementType type = page.getBlock(channels.get(0)).elementType();
-    if (type == ElementType.NULL) {
-      return;
-    }
-    IntBlock block = page.getBlock(channels.get(0));
-    IntVector vector = block.asVector();
-    if (vector != null) {
-      addRawVector(vector);
-    } else {
-      addRawBlock(block);
-    }
-  }
-
-  private void addRawVector(IntVector vector) {
-    for (int i = 0; i < vector.getPositionCount(); i++) {
-      AvgIntAggregator.combine(state, vector.getInt(i));
-    }
-    AvgIntAggregator.combineValueCount(state, vector.getPositionCount());
-  }
-
-  private void addRawBlock(IntBlock block) {
-    for (int p = 0; p < block.getPositionCount(); p++) {
-      if (block.isNull(p)) {
-        continue;
-      }
-      int start = block.getFirstValueIndex(p);
-      int end = start + block.getValueCount(p);
-      for (int i = start; i < end; i++) {
-        AvgIntAggregator.combine(state, block.getInt(i));
-      }
-    }
-    AvgIntAggregator.combineValueCount(state, block.getTotalValueCount());
-  }
-
-  @Override
-  public void addIntermediateInput(Page page) {
-    Block block = page.getBlock(channels.get(0));
-    Vector vector = block.asVector();
-    if (vector == null || vector instanceof AggregatorStateVector == false) {
-      throw new RuntimeException("expected AggregatorStateBlock, got:" + block);
-    }
-    @SuppressWarnings("unchecked") AggregatorStateVector<AvgLongAggregator.AvgState> blobVector = (AggregatorStateVector<AvgLongAggregator.AvgState>) vector;
-    // TODO exchange big arrays directly without funny serialization - no more copying
-    BigArrays bigArrays = BigArrays.NON_RECYCLING_INSTANCE;
-    AvgLongAggregator.AvgState tmpState = AvgIntAggregator.initSingle();
-    for (int i = 0; i < block.getPositionCount(); i++) {
-      blobVector.get(i, tmpState);
-      AvgIntAggregator.combineStates(state, tmpState);
-    }
-    tmpState.close();
-  }
-
-  @Override
-  public void evaluateIntermediate(Block[] blocks, int offset) {
-    AggregatorStateVector.Builder<AggregatorStateVector<AvgLongAggregator.AvgState>, AvgLongAggregator.AvgState> builder =
-        AggregatorStateVector.builderOfAggregatorState(AvgLongAggregator.AvgState.class, state.getEstimatedSize());
-    builder.add(state, IntVector.range(0, 1));
-    blocks[offset] = builder.build().asBlock();
-  }
-
-  @Override
-  public void evaluateFinal(Block[] blocks, int offset) {
-    blocks[offset] = AvgIntAggregator.evaluateFinal(state);
-  }
-
-  @Override
-  public String toString() {
-    StringBuilder sb = new StringBuilder();
-    sb.append(getClass().getSimpleName()).append("[");
-    sb.append("channels=").append(channels);
-    sb.append("]");
-    return sb.toString();
-  }
-
-  @Override
-  public void close() {
-    state.close();
-  }
-}

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/AvgIntAggregatorFunctionSupplier.java

@@ -1,41 +0,0 @@
-// 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.aggregation;
-
-import java.lang.Integer;
-import java.lang.Override;
-import java.lang.String;
-import java.util.List;
-import org.elasticsearch.common.util.BigArrays;
-
-/**
- * {@link AggregatorFunctionSupplier} implementation for {@link AvgIntAggregator}.
- * This class is generated. Do not edit it.
- */
-public final class AvgIntAggregatorFunctionSupplier implements AggregatorFunctionSupplier {
-  private final BigArrays bigArrays;
-
-  private final List<Integer> channels;
-
-  public AvgIntAggregatorFunctionSupplier(BigArrays bigArrays, List<Integer> channels) {
-    this.bigArrays = bigArrays;
-    this.channels = channels;
-  }
-
-  @Override
-  public AvgIntAggregatorFunction aggregator() {
-    return AvgIntAggregatorFunction.create(channels);
-  }
-
-  @Override
-  public AvgIntGroupingAggregatorFunction groupingAggregator() {
-    return AvgIntGroupingAggregatorFunction.create(channels, bigArrays);
-  }
-
-  @Override
-  public String describe() {
-    return "avg of ints";
-  }
-}

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/AvgIntGroupingAggregatorFunction.java

@@ -1,181 +0,0 @@
-// 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.aggregation;
-
-import java.lang.Integer;
-import java.lang.Override;
-import java.lang.String;
-import java.lang.StringBuilder;
-import java.util.List;
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.AggregatorStateVector;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.IntBlock;
-import org.elasticsearch.compute.data.IntVector;
-import org.elasticsearch.compute.data.LongBlock;
-import org.elasticsearch.compute.data.LongVector;
-import org.elasticsearch.compute.data.Page;
-import org.elasticsearch.compute.data.Vector;
-
-/**
- * {@link GroupingAggregatorFunction} implementation for {@link AvgIntAggregator}.
- * This class is generated. Do not edit it.
- */
-public final class AvgIntGroupingAggregatorFunction implements GroupingAggregatorFunction {
-  private final AvgLongAggregator.GroupingAvgState state;
-
-  private final List<Integer> channels;
-
-  private final BigArrays bigArrays;
-
-  public AvgIntGroupingAggregatorFunction(List<Integer> channels,
-      AvgLongAggregator.GroupingAvgState state, BigArrays bigArrays) {
-    this.channels = channels;
-    this.state = state;
-    this.bigArrays = bigArrays;
-  }
-
-  public static AvgIntGroupingAggregatorFunction create(List<Integer> channels,
-      BigArrays bigArrays) {
-    return new AvgIntGroupingAggregatorFunction(channels, AvgIntAggregator.initGrouping(bigArrays), bigArrays);
-  }
-
-  @Override
-  public void addRawInput(LongVector groups, Page page) {
-    IntBlock valuesBlock = page.getBlock(channels.get(0));
-    assert groups.getPositionCount() == page.getPositionCount();
-    IntVector valuesVector = valuesBlock.asVector();
-    if (valuesVector == null) {
-      addRawInput(groups, valuesBlock);
-    } else {
-      addRawInput(groups, valuesVector);
-    }
-  }
-
-  private void addRawInput(LongVector groups, IntBlock values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      int groupId = Math.toIntExact(groups.getLong(position));
-      if (values.isNull(position)) {
-        state.putNull(groupId);
-        continue;
-      }
-      int valuesStart = values.getFirstValueIndex(position);
-      int valuesEnd = valuesStart + values.getValueCount(position);
-      for (int v = valuesStart; v < valuesEnd; v++) {
-        AvgIntAggregator.combine(state, groupId, values.getInt(v));
-      }
-    }
-  }
-
-  private void addRawInput(LongVector groups, IntVector values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      int groupId = Math.toIntExact(groups.getLong(position));
-      AvgIntAggregator.combine(state, groupId, values.getInt(position));
-    }
-  }
-
-  @Override
-  public void addRawInput(LongBlock groups, Page page) {
-    IntBlock valuesBlock = page.getBlock(channels.get(0));
-    assert groups.getPositionCount() == page.getPositionCount();
-    IntVector valuesVector = valuesBlock.asVector();
-    if (valuesVector == null) {
-      addRawInput(groups, valuesBlock);
-    } else {
-      addRawInput(groups, valuesVector);
-    }
-  }
-
-  private void addRawInput(LongBlock groups, IntBlock values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      if (groups.isNull(position)) {
-        continue;
-      }
-      int groupStart = groups.getFirstValueIndex(position);
-      int groupEnd = groupStart + groups.getValueCount(position);
-      for (int g = groupStart; g < groupEnd; g++) {
-        int groupId = Math.toIntExact(groups.getLong(g));
-        if (values.isNull(position)) {
-          state.putNull(groupId);
-          continue;
-        }
-        int valuesStart = values.getFirstValueIndex(position);
-        int valuesEnd = valuesStart + values.getValueCount(position);
-        for (int v = valuesStart; v < valuesEnd; v++) {
-          AvgIntAggregator.combine(state, groupId, values.getInt(v));
-        }
-      }
-    }
-  }
-
-  private void addRawInput(LongBlock groups, IntVector values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      if (groups.isNull(position)) {
-        continue;
-      }
-      int groupStart = groups.getFirstValueIndex(position);
-      int groupEnd = groupStart + groups.getValueCount(position);
-      for (int g = groupStart; g < groupEnd; g++) {
-        int groupId = Math.toIntExact(groups.getLong(g));
-        AvgIntAggregator.combine(state, groupId, values.getInt(position));
-      }
-    }
-  }
-
-  @Override
-  public void addIntermediateInput(LongVector groupIdVector, Page page) {
-    Block block = page.getBlock(channels.get(0));
-    Vector vector = block.asVector();
-    if (vector == null || vector instanceof AggregatorStateVector == false) {
-      throw new RuntimeException("expected AggregatorStateBlock, got:" + block);
-    }
-    @SuppressWarnings("unchecked") AggregatorStateVector<AvgLongAggregator.GroupingAvgState> blobVector = (AggregatorStateVector<AvgLongAggregator.GroupingAvgState>) vector;
-    // TODO exchange big arrays directly without funny serialization - no more copying
-    BigArrays bigArrays = BigArrays.NON_RECYCLING_INSTANCE;
-    AvgLongAggregator.GroupingAvgState inState = AvgIntAggregator.initGrouping(bigArrays);
-    blobVector.get(0, inState);
-    for (int position = 0; position < groupIdVector.getPositionCount(); position++) {
-      int groupId = Math.toIntExact(groupIdVector.getLong(position));
-      AvgIntAggregator.combineStates(state, groupId, inState, position);
-    }
-    inState.close();
-  }
-
-  @Override
-  public void addIntermediateRowInput(int groupId, GroupingAggregatorFunction input, int position) {
-    if (input.getClass() != getClass()) {
-      throw new IllegalArgumentException("expected " + getClass() + "; got " + input.getClass());
-    }
-    AvgLongAggregator.GroupingAvgState inState = ((AvgIntGroupingAggregatorFunction) input).state;
-    AvgIntAggregator.combineStates(state, groupId, inState, position);
-  }
-
-  @Override
-  public void evaluateIntermediate(Block[] blocks, int offset, IntVector selected) {
-    AggregatorStateVector.Builder<AggregatorStateVector<AvgLongAggregator.GroupingAvgState>, AvgLongAggregator.GroupingAvgState> builder =
-        AggregatorStateVector.builderOfAggregatorState(AvgLongAggregator.GroupingAvgState.class, state.getEstimatedSize());
-    builder.add(state, selected);
-    blocks[offset] = builder.build().asBlock();
-  }
-
-  @Override
-  public void evaluateFinal(Block[] blocks, int offset, IntVector selected) {
-    blocks[offset] = AvgIntAggregator.evaluateFinal(state, selected);
-  }
-
-  @Override
-  public String toString() {
-    StringBuilder sb = new StringBuilder();
-    sb.append(getClass().getSimpleName()).append("[");
-    sb.append("channels=").append(channels);
-    sb.append("]");
-    return sb.toString();
-  }
-
-  @Override
-  public void close() {
-    state.close();
-  }
-}

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/AvgLongAggregatorFunction.java

@@ -1,120 +0,0 @@
-// 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.aggregation;
-
-import java.lang.Integer;
-import java.lang.Override;
-import java.lang.String;
-import java.lang.StringBuilder;
-import java.util.List;
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.AggregatorStateVector;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.ElementType;
-import org.elasticsearch.compute.data.IntVector;
-import org.elasticsearch.compute.data.LongBlock;
-import org.elasticsearch.compute.data.LongVector;
-import org.elasticsearch.compute.data.Page;
-import org.elasticsearch.compute.data.Vector;
-
-/**
- * {@link AggregatorFunction} implementation for {@link AvgLongAggregator}.
- * This class is generated. Do not edit it.
- */
-public final class AvgLongAggregatorFunction implements AggregatorFunction {
-  private final AvgLongAggregator.AvgState state;
-
-  private final List<Integer> channels;
-
-  public AvgLongAggregatorFunction(List<Integer> channels, AvgLongAggregator.AvgState state) {
-    this.channels = channels;
-    this.state = state;
-  }
-
-  public static AvgLongAggregatorFunction create(List<Integer> channels) {
-    return new AvgLongAggregatorFunction(channels, AvgLongAggregator.initSingle());
-  }
-
-  @Override
-  public void addRawInput(Page page) {
-    ElementType type = page.getBlock(channels.get(0)).elementType();
-    if (type == ElementType.NULL) {
-      return;
-    }
-    LongBlock block = page.getBlock(channels.get(0));
-    LongVector vector = block.asVector();
-    if (vector != null) {
-      addRawVector(vector);
-    } else {
-      addRawBlock(block);
-    }
-  }
-
-  private void addRawVector(LongVector vector) {
-    for (int i = 0; i < vector.getPositionCount(); i++) {
-      AvgLongAggregator.combine(state, vector.getLong(i));
-    }
-    AvgLongAggregator.combineValueCount(state, vector.getPositionCount());
-  }
-
-  private void addRawBlock(LongBlock block) {
-    for (int p = 0; p < block.getPositionCount(); p++) {
-      if (block.isNull(p)) {
-        continue;
-      }
-      int start = block.getFirstValueIndex(p);
-      int end = start + block.getValueCount(p);
-      for (int i = start; i < end; i++) {
-        AvgLongAggregator.combine(state, block.getLong(i));
-      }
-    }
-    AvgLongAggregator.combineValueCount(state, block.getTotalValueCount());
-  }
-
-  @Override
-  public void addIntermediateInput(Page page) {
-    Block block = page.getBlock(channels.get(0));
-    Vector vector = block.asVector();
-    if (vector == null || vector instanceof AggregatorStateVector == false) {
-      throw new RuntimeException("expected AggregatorStateBlock, got:" + block);
-    }
-    @SuppressWarnings("unchecked") AggregatorStateVector<AvgLongAggregator.AvgState> blobVector = (AggregatorStateVector<AvgLongAggregator.AvgState>) vector;
-    // TODO exchange big arrays directly without funny serialization - no more copying
-    BigArrays bigArrays = BigArrays.NON_RECYCLING_INSTANCE;
-    AvgLongAggregator.AvgState tmpState = AvgLongAggregator.initSingle();
-    for (int i = 0; i < block.getPositionCount(); i++) {
-      blobVector.get(i, tmpState);
-      AvgLongAggregator.combineStates(state, tmpState);
-    }
-    tmpState.close();
-  }
-
-  @Override
-  public void evaluateIntermediate(Block[] blocks, int offset) {
-    AggregatorStateVector.Builder<AggregatorStateVector<AvgLongAggregator.AvgState>, AvgLongAggregator.AvgState> builder =
-        AggregatorStateVector.builderOfAggregatorState(AvgLongAggregator.AvgState.class, state.getEstimatedSize());
-    builder.add(state, IntVector.range(0, 1));
-    blocks[offset] = builder.build().asBlock();
-  }
-
-  @Override
-  public void evaluateFinal(Block[] blocks, int offset) {
-    blocks[offset] = AvgLongAggregator.evaluateFinal(state);
-  }
-
-  @Override
-  public String toString() {
-    StringBuilder sb = new StringBuilder();
-    sb.append(getClass().getSimpleName()).append("[");
-    sb.append("channels=").append(channels);
-    sb.append("]");
-    return sb.toString();
-  }
-
-  @Override
-  public void close() {
-    state.close();
-  }
-}

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/AvgLongAggregatorFunctionSupplier.java

@@ -1,41 +0,0 @@
-// 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.aggregation;
-
-import java.lang.Integer;
-import java.lang.Override;
-import java.lang.String;
-import java.util.List;
-import org.elasticsearch.common.util.BigArrays;
-
-/**
- * {@link AggregatorFunctionSupplier} implementation for {@link AvgLongAggregator}.
- * This class is generated. Do not edit it.
- */
-public final class AvgLongAggregatorFunctionSupplier implements AggregatorFunctionSupplier {
-  private final BigArrays bigArrays;
-
-  private final List<Integer> channels;
-
-  public AvgLongAggregatorFunctionSupplier(BigArrays bigArrays, List<Integer> channels) {
-    this.bigArrays = bigArrays;
-    this.channels = channels;
-  }
-
-  @Override
-  public AvgLongAggregatorFunction aggregator() {
-    return AvgLongAggregatorFunction.create(channels);
-  }
-
-  @Override
-  public AvgLongGroupingAggregatorFunction groupingAggregator() {
-    return AvgLongGroupingAggregatorFunction.create(channels, bigArrays);
-  }
-
-  @Override
-  public String describe() {
-    return "avg of longs";
-  }
-}

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/AvgLongGroupingAggregatorFunction.java

@@ -1,180 +0,0 @@
-// 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.aggregation;
-
-import java.lang.Integer;
-import java.lang.Override;
-import java.lang.String;
-import java.lang.StringBuilder;
-import java.util.List;
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.AggregatorStateVector;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.IntVector;
-import org.elasticsearch.compute.data.LongBlock;
-import org.elasticsearch.compute.data.LongVector;
-import org.elasticsearch.compute.data.Page;
-import org.elasticsearch.compute.data.Vector;
-
-/**
- * {@link GroupingAggregatorFunction} implementation for {@link AvgLongAggregator}.
- * This class is generated. Do not edit it.
- */
-public final class AvgLongGroupingAggregatorFunction implements GroupingAggregatorFunction {
-  private final AvgLongAggregator.GroupingAvgState state;
-
-  private final List<Integer> channels;
-
-  private final BigArrays bigArrays;
-
-  public AvgLongGroupingAggregatorFunction(List<Integer> channels,
-      AvgLongAggregator.GroupingAvgState state, BigArrays bigArrays) {
-    this.channels = channels;
-    this.state = state;
-    this.bigArrays = bigArrays;
-  }
-
-  public static AvgLongGroupingAggregatorFunction create(List<Integer> channels,
-      BigArrays bigArrays) {
-    return new AvgLongGroupingAggregatorFunction(channels, AvgLongAggregator.initGrouping(bigArrays), bigArrays);
-  }
-
-  @Override
-  public void addRawInput(LongVector groups, Page page) {
-    LongBlock valuesBlock = page.getBlock(channels.get(0));
-    assert groups.getPositionCount() == page.getPositionCount();
-    LongVector valuesVector = valuesBlock.asVector();
-    if (valuesVector == null) {
-      addRawInput(groups, valuesBlock);
-    } else {
-      addRawInput(groups, valuesVector);
-    }
-  }
-
-  private void addRawInput(LongVector groups, LongBlock values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      int groupId = Math.toIntExact(groups.getLong(position));
-      if (values.isNull(position)) {
-        state.putNull(groupId);
-        continue;
-      }
-      int valuesStart = values.getFirstValueIndex(position);
-      int valuesEnd = valuesStart + values.getValueCount(position);
-      for (int v = valuesStart; v < valuesEnd; v++) {
-        AvgLongAggregator.combine(state, groupId, values.getLong(v));
-      }
-    }
-  }
-
-  private void addRawInput(LongVector groups, LongVector values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      int groupId = Math.toIntExact(groups.getLong(position));
-      AvgLongAggregator.combine(state, groupId, values.getLong(position));
-    }
-  }
-
-  @Override
-  public void addRawInput(LongBlock groups, Page page) {
-    LongBlock valuesBlock = page.getBlock(channels.get(0));
-    assert groups.getPositionCount() == page.getPositionCount();
-    LongVector valuesVector = valuesBlock.asVector();
-    if (valuesVector == null) {
-      addRawInput(groups, valuesBlock);
-    } else {
-      addRawInput(groups, valuesVector);
-    }
-  }
-
-  private void addRawInput(LongBlock groups, LongBlock values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      if (groups.isNull(position)) {
-        continue;
-      }
-      int groupStart = groups.getFirstValueIndex(position);
-      int groupEnd = groupStart + groups.getValueCount(position);
-      for (int g = groupStart; g < groupEnd; g++) {
-        int groupId = Math.toIntExact(groups.getLong(g));
-        if (values.isNull(position)) {
-          state.putNull(groupId);
-          continue;
-        }
-        int valuesStart = values.getFirstValueIndex(position);
-        int valuesEnd = valuesStart + values.getValueCount(position);
-        for (int v = valuesStart; v < valuesEnd; v++) {
-          AvgLongAggregator.combine(state, groupId, values.getLong(v));
-        }
-      }
-    }
-  }
-
-  private void addRawInput(LongBlock groups, LongVector values) {
-    for (int position = 0; position < groups.getPositionCount(); position++) {
-      if (groups.isNull(position)) {
-        continue;
-      }
-      int groupStart = groups.getFirstValueIndex(position);
-      int groupEnd = groupStart + groups.getValueCount(position);
-      for (int g = groupStart; g < groupEnd; g++) {
-        int groupId = Math.toIntExact(groups.getLong(g));
-        AvgLongAggregator.combine(state, groupId, values.getLong(position));
-      }
-    }
-  }
-
-  @Override
-  public void addIntermediateInput(LongVector groupIdVector, Page page) {
-    Block block = page.getBlock(channels.get(0));
-    Vector vector = block.asVector();
-    if (vector == null || vector instanceof AggregatorStateVector == false) {
-      throw new RuntimeException("expected AggregatorStateBlock, got:" + block);
-    }
-    @SuppressWarnings("unchecked") AggregatorStateVector<AvgLongAggregator.GroupingAvgState> blobVector = (AggregatorStateVector<AvgLongAggregator.GroupingAvgState>) vector;
-    // TODO exchange big arrays directly without funny serialization - no more copying
-    BigArrays bigArrays = BigArrays.NON_RECYCLING_INSTANCE;
-    AvgLongAggregator.GroupingAvgState inState = AvgLongAggregator.initGrouping(bigArrays);
-    blobVector.get(0, inState);
-    for (int position = 0; position < groupIdVector.getPositionCount(); position++) {
-      int groupId = Math.toIntExact(groupIdVector.getLong(position));
-      AvgLongAggregator.combineStates(state, groupId, inState, position);
-    }
-    inState.close();
-  }
-
-  @Override
-  public void addIntermediateRowInput(int groupId, GroupingAggregatorFunction input, int position) {
-    if (input.getClass() != getClass()) {
-      throw new IllegalArgumentException("expected " + getClass() + "; got " + input.getClass());
-    }
-    AvgLongAggregator.GroupingAvgState inState = ((AvgLongGroupingAggregatorFunction) input).state;
-    AvgLongAggregator.combineStates(state, groupId, inState, position);
-  }
-
-  @Override
-  public void evaluateIntermediate(Block[] blocks, int offset, IntVector selected) {
-    AggregatorStateVector.Builder<AggregatorStateVector<AvgLongAggregator.GroupingAvgState>, AvgLongAggregator.GroupingAvgState> builder =
-        AggregatorStateVector.builderOfAggregatorState(AvgLongAggregator.GroupingAvgState.class, state.getEstimatedSize());
-    builder.add(state, selected);
-    blocks[offset] = builder.build().asBlock();
-  }
-
-  @Override
-  public void evaluateFinal(Block[] blocks, int offset, IntVector selected) {
-    blocks[offset] = AvgLongAggregator.evaluateFinal(state, selected);
-  }
-
-  @Override
-  public String toString() {
-    StringBuilder sb = new StringBuilder();
-    sb.append(getClass().getSimpleName()).append("[");
-    sb.append("channels=").append(channels);
-    sb.append("]");
-    return sb.toString();
-  }
-
-  @Override
-  public void close() {
-    state.close();
-  }
-}

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/AvgDoubleAggregator.java

@@ -1,287 +0,0 @@
-/*
- * 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.aggregation;
-
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.common.util.DoubleArray;
-import org.elasticsearch.common.util.LongArray;
-import org.elasticsearch.compute.ann.Aggregator;
-import org.elasticsearch.compute.ann.GroupingAggregator;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.data.IntVector;
-import org.elasticsearch.core.Releasables;
-import org.elasticsearch.search.aggregations.metrics.CompensatedSum;
-
-import java.lang.invoke.MethodHandles;
-import java.lang.invoke.VarHandle;
-import java.nio.ByteOrder;
-import java.util.Objects;
-
-@Aggregator
-@GroupingAggregator
-class AvgDoubleAggregator {
-    public static AvgState initSingle() {
-        return new AvgState();
-    }
-
-    public static void combine(AvgState current, double v) {
-        current.add(v);
-    }
-
-    public static void combineStates(AvgState current, AvgState state) {
-        current.add(state.value(), state.delta());
-        current.count += state.count;
-    }
-
-    public static Block evaluateFinal(AvgState state) {
-        if (state.count == 0) {
-            return Block.constantNullBlock(1);
-        }
-        double result = state.value() / state.count;
-        return DoubleBlock.newConstantBlockWith(result, 1);
-    }
-
-    public static GroupingAvgState initGrouping(BigArrays bigArrays) {
-        return new GroupingAvgState(bigArrays);
-    }
-
-    public static void combineValueCount(AvgState current, int positions) {
-        current.count += positions;
-    }
-
-    public static void combine(GroupingAvgState current, int groupId, double v) {
-        current.add(v, groupId);
-    }
-
-    public static void combineStates(GroupingAvgState current, int currentGroupId, GroupingAvgState state, int statePosition) {
-        current.add(state.values.get(statePosition), state.deltas.get(statePosition), currentGroupId, state.counts.get(statePosition));
-    }
-
-    public static Block evaluateFinal(GroupingAvgState state, IntVector selected) {
-        DoubleBlock.Builder builder = DoubleBlock.newBlockBuilder(selected.getPositionCount());
-        for (int i = 0; i < selected.getPositionCount(); i++) {
-            int group = selected.getInt(i);
-            final long count = state.counts.get(group);
-            if (count > 0) {
-                builder.appendDouble(state.values.get(group) / count);
-            } else {
-                assert state.values.get(group) == 0.0;
-                builder.appendNull();
-            }
-        }
-        return builder.build();
-    }
-
-    // @SerializedSize(value = Double.BYTES + Double.BYTES + Long.BYTES)
-    static class AvgState extends CompensatedSum implements AggregatorState<AvgState> {
-
-        private long count;
-
-        private final AvgDoubleAggregator.AvgStateSerializer serializer;
-
-        AvgState() {
-            this(0, 0, 0);
-        }
-
-        AvgState(double value, double delta, long count) {
-            super(value, delta);
-            this.count = count;
-            this.serializer = new AvgDoubleAggregator.AvgStateSerializer();
-        }
-
-        @Override
-        public long getEstimatedSize() {
-            return AvgStateSerializer.BYTES_SIZE;
-        }
-
-        @Override
-        public void close() {}
-
-        @Override
-        public AggregatorStateSerializer<AvgState> serializer() {
-            return serializer;
-        }
-    }
-
-    // @SerializedSize(value = Double.BYTES + Double.BYTES + Long.BYTES)
-    static class AvgStateSerializer implements AggregatorStateSerializer<AvgDoubleAggregator.AvgState> {
-
-        // record Shape (double value, double delta, long count) {}
-
-        static final int BYTES_SIZE = Double.BYTES + Double.BYTES + Long.BYTES;
-
-        @Override
-        public int size() {
-            return BYTES_SIZE;
-        }
-
-        private static final VarHandle doubleHandle = MethodHandles.byteArrayViewVarHandle(double[].class, ByteOrder.BIG_ENDIAN);
-        private static final VarHandle longHandle = MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.BIG_ENDIAN);
-
-        @Override
-        public int serialize(AvgDoubleAggregator.AvgState value, byte[] ba, int offset, IntVector selected) {
-            assert selected.getPositionCount() == 1;
-            assert selected.getInt(0) == 0;
-            doubleHandle.set(ba, offset, value.value());
-            doubleHandle.set(ba, offset + 8, value.delta());
-            longHandle.set(ba, offset + 16, value.count);
-            return BYTES_SIZE; // number of bytes written
-        }
-
-        // sets the state in value
-        @Override
-        public void deserialize(AvgDoubleAggregator.AvgState value, byte[] ba, int offset) {
-            Objects.requireNonNull(value);
-            double kvalue = (double) doubleHandle.get(ba, offset);
-            double kdelta = (double) doubleHandle.get(ba, offset + 8);
-            long count = (long) longHandle.get(ba, offset + 16);
-
-            value.reset(kvalue, kdelta);
-            value.count = count;
-        }
-    }
-
-    static class GroupingAvgState implements AggregatorState<GroupingAvgState> {
-        private final BigArrays bigArrays;
-        static final long BYTES_SIZE = Double.BYTES + Double.BYTES + Long.BYTES;
-
-        DoubleArray values;
-        DoubleArray deltas;
-        LongArray counts;
-
-        // total number of groups; <= values.length
-        int largestGroupId;
-
-        private final GroupingAvgStateSerializer serializer;
-
-        GroupingAvgState(BigArrays bigArrays) {
-            this.bigArrays = bigArrays;
-            boolean success = false;
-            try {
-                this.values = bigArrays.newDoubleArray(1);
-                this.deltas = bigArrays.newDoubleArray(1);
-                this.counts = bigArrays.newLongArray(1);
-                success = true;
-            } finally {
-                if (success == false) {
-                    close();
-                }
-            }
-            this.serializer = new GroupingAvgStateSerializer();
-        }
-
-        void add(double valueToAdd, int groupId) {
-            add(valueToAdd, 0d, groupId, 1);
-        }
-
-        void add(double valueToAdd, double deltaToAdd, int groupId, long increment) {
-            ensureCapacity(groupId);
-            add(valueToAdd, deltaToAdd, groupId);
-            counts.increment(groupId, increment);
-        }
-
-        void add(double valueToAdd, double deltaToAdd, int position) {
-            // If the value is Inf or NaN, just add it to the running tally to "convert" to
-            // Inf/NaN. This keeps the behavior bwc from before kahan summing
-            if (Double.isFinite(valueToAdd) == false) {
-                values.increment(position, valueToAdd);
-                return;
-            }
-
-            double value = values.get(position);
-            if (Double.isFinite(value) == false) {
-                // It isn't going to get any more infinite.
-                return;
-            }
-            double delta = deltas.get(position);
-            double correctedSum = valueToAdd + (delta + deltaToAdd);
-            double updatedValue = value + correctedSum;
-            deltas.set(position, correctedSum - (updatedValue - value));
-            values.set(position, updatedValue);
-        }
-
-        void putNull(int position) {
-            // counts = 0 is for nulls
-            ensureCapacity(position);
-        }
-
-        private void ensureCapacity(int groupId) {
-            if (groupId > largestGroupId) {
-                largestGroupId = groupId;
-                values = bigArrays.grow(values, groupId + 1);
-                deltas = bigArrays.grow(deltas, groupId + 1);
-                counts = bigArrays.grow(counts, groupId + 1);
-            }
-        }
-
-        @Override
-        public long getEstimatedSize() {
-            return Long.BYTES + (largestGroupId + 1) * BYTES_SIZE;
-        }
-
-        @Override
-        public AggregatorStateSerializer<GroupingAvgState> serializer() {
-            return serializer;
-        }
-
-        @Override
-        public void close() {
-            Releasables.close(values, deltas, counts);
-        }
-    }
-
-    // @SerializedSize(value = Double.BYTES + Double.BYTES + Long.BYTES)
-    static class GroupingAvgStateSerializer implements AggregatorStateSerializer<GroupingAvgState> {
-
-        // record Shape (double value, double delta, long count) {}
-
-        static final int BYTES_SIZE = Double.BYTES + Double.BYTES + Long.BYTES;
-
-        @Override
-        public int size() {
-            return BYTES_SIZE;
-        }
-
-        private static final VarHandle doubleHandle = MethodHandles.byteArrayViewVarHandle(double[].class, ByteOrder.BIG_ENDIAN);
-        private static final VarHandle longHandle = MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.BIG_ENDIAN);
-
-        @Override
-        public int serialize(GroupingAvgState state, byte[] ba, int offset, IntVector selected) {
-            longHandle.set(ba, offset, selected.getPositionCount());
-            offset += Long.BYTES;
-            for (int i = 0; i < selected.getPositionCount(); i++) {
-                int group = selected.getInt(i);
-                doubleHandle.set(ba, offset, state.values.get(group));
-                doubleHandle.set(ba, offset + 8, state.deltas.get(group));
-                longHandle.set(ba, offset + 16, state.counts.get(group));
-                offset += BYTES_SIZE;
-            }
-            return 8 + (BYTES_SIZE * selected.getPositionCount()); // number of bytes written
-        }
-
-        // sets the state in value
-        @Override
-        public void deserialize(GroupingAvgState state, byte[] ba, int offset) {
-            Objects.requireNonNull(state);
-            int positions = (int) (long) longHandle.get(ba, offset);
-            // TODO replace deserialization with direct passing - no more non_recycling_instance then
-            state.values = BigArrays.NON_RECYCLING_INSTANCE.grow(state.values, positions);
-            state.deltas = BigArrays.NON_RECYCLING_INSTANCE.grow(state.deltas, positions);
-            state.counts = BigArrays.NON_RECYCLING_INSTANCE.grow(state.counts, positions);
-            offset += 8;
-            for (int i = 0; i < positions; i++) {
-                state.values.set(i, (double) doubleHandle.get(ba, offset));
-                state.deltas.set(i, (double) doubleHandle.get(ba, offset + 8));
-                state.counts.set(i, (long) longHandle.get(ba, offset + 16));
-                offset += BYTES_SIZE;
-            }
-            state.largestGroupId = positions - 1;
-        }
-    }
-}

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/AvgIntAggregator.java

@@ -1,73 +0,0 @@
-/*
- * 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.aggregation;
-
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.aggregation.AvgLongAggregator.AvgState;
-import org.elasticsearch.compute.aggregation.AvgLongAggregator.GroupingAvgState;
-import org.elasticsearch.compute.ann.Aggregator;
-import org.elasticsearch.compute.ann.GroupingAggregator;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.data.IntVector;
-
-@Aggregator
-@GroupingAggregator
-class AvgIntAggregator {
-    public static AvgState initSingle() {
-        return new AvgState();
-    }
-
-    public static void combine(AvgState current, int v) {
-        current.value = Math.addExact(current.value, v);
-    }
-
-    public static void combineValueCount(AvgState current, int positions) {
-        current.count += positions;
-    }
-
-    public static void combineStates(AvgState current, AvgState state) {
-        current.value = Math.addExact(current.value, state.value);
-        current.count += state.count;
-    }
-
-    public static Block evaluateFinal(AvgState state) {
-        if (state.count == 0) {
-            return Block.constantNullBlock(1);
-        }
-        double result = ((double) state.value) / state.count;
-        return DoubleBlock.newConstantBlockWith(result, 1);
-    }
-
-    public static GroupingAvgState initGrouping(BigArrays bigArrays) {
-        return new GroupingAvgState(bigArrays);
-    }
-
-    public static void combine(GroupingAvgState current, int groupId, int v) {
-        current.add(v, groupId, 1);
-    }
-
-    public static void combineStates(GroupingAvgState current, int currentGroupId, GroupingAvgState state, int statePosition) {
-        current.add(state.values.get(statePosition), currentGroupId, state.counts.get(statePosition));
-    }
-
-    public static Block evaluateFinal(GroupingAvgState state, IntVector selected) {
-        DoubleBlock.Builder builder = DoubleBlock.newBlockBuilder(selected.getPositionCount());
-        for (int i = 0; i < selected.getPositionCount(); i++) {
-            int group = selected.getInt(i);
-            final long count = state.counts.get(group);
-            if (count > 0) {
-                builder.appendDouble((double) state.values.get(group) / count);
-            } else {
-                assert state.values.get(group) == 0;
-                builder.appendNull();
-            }
-        }
-        return builder.build();
-    }
-}

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/AvgLongAggregator.java

@@ -1,246 +0,0 @@
-/*
- * 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.aggregation;
-
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.common.util.LongArray;
-import org.elasticsearch.compute.ann.Aggregator;
-import org.elasticsearch.compute.ann.GroupingAggregator;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.data.IntVector;
-import org.elasticsearch.core.Releasables;
-
-import java.lang.invoke.MethodHandles;
-import java.lang.invoke.VarHandle;
-import java.nio.ByteOrder;
-import java.util.Objects;
-
-@Aggregator
-@GroupingAggregator
-class AvgLongAggregator {
-    public static AvgState initSingle() {
-        return new AvgState();
-    }
-
-    public static void combine(AvgState current, long v) {
-        current.value = Math.addExact(current.value, v);
-    }
-
-    public static void combineValueCount(AvgState current, int positions) {
-        current.count += positions;
-    }
-
-    public static void combineStates(AvgState current, AvgState state) {
-        current.value = Math.addExact(current.value, state.value);
-        current.count += state.count;
-    }
-
-    public static Block evaluateFinal(AvgState state) {
-        if (state.count == 0) {
-            return Block.constantNullBlock(1);
-        }
-        double result = ((double) state.value) / state.count;
-        return DoubleBlock.newConstantBlockWith(result, 1);
-    }
-
-    public static GroupingAvgState initGrouping(BigArrays bigArrays) {
-        return new GroupingAvgState(bigArrays);
-    }
-
-    public static void combine(GroupingAvgState current, int groupId, long v) {
-        current.add(v, groupId, 1);
-    }
-
-    public static void combineStates(GroupingAvgState current, int currentGroupId, GroupingAvgState state, int statePosition) {
-        current.add(state.values.get(statePosition), currentGroupId, state.counts.get(statePosition));
-    }
-
-    public static Block evaluateFinal(GroupingAvgState state, IntVector selected) {
-        DoubleBlock.Builder builder = DoubleBlock.newBlockBuilder(selected.getPositionCount());
-        for (int i = 0; i < selected.getPositionCount(); i++) {
-            int group = selected.getInt(i);
-            final long count = state.counts.get(group);
-            if (count > 0) {
-                builder.appendDouble((double) state.values.get(group) / count);
-            } else {
-                assert state.values.get(group) == 0;
-                builder.appendNull();
-            }
-        }
-        return builder.build();
-    }
-
-    static class AvgState implements AggregatorState<AvgLongAggregator.AvgState> {
-
-        long value;
-        long count;
-
-        private final AvgLongAggregator.AvgStateSerializer serializer;
-
-        AvgState() {
-            this(0, 0);
-        }
-
-        AvgState(long value, long count) {
-            this.value = value;
-            this.count = count;
-            this.serializer = new AvgLongAggregator.AvgStateSerializer();
-        }
-
-        @Override
-        public long getEstimatedSize() {
-            return AvgLongAggregator.AvgStateSerializer.BYTES_SIZE;
-        }
-
-        @Override
-        public void close() {}
-
-        @Override
-        public AggregatorStateSerializer<AvgLongAggregator.AvgState> serializer() {
-            return serializer;
-        }
-    }
-
-    // @SerializedSize(value = Long.BYTES + Long.BYTES)
-    static class AvgStateSerializer implements AggregatorStateSerializer<AvgLongAggregator.AvgState> {
-        static final int BYTES_SIZE = Long.BYTES + Long.BYTES;
-
-        @Override
-        public int size() {
-            return BYTES_SIZE;
-        }
-
-        private static final VarHandle longHandle = MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.BIG_ENDIAN);
-
-        @Override
-        public int serialize(AvgLongAggregator.AvgState value, byte[] ba, int offset, IntVector selected) {
-            assert selected.getPositionCount() == 1;
-            assert selected.getInt(0) == 0;
-            longHandle.set(ba, offset, value.value);
-            longHandle.set(ba, offset + 8, value.count);
-            return BYTES_SIZE; // number of bytes written
-        }
-
-        // sets the state in value
-        @Override
-        public void deserialize(AvgLongAggregator.AvgState value, byte[] ba, int offset) {
-            Objects.requireNonNull(value);
-            long kvalue = (long) longHandle.get(ba, offset);
-            long count = (long) longHandle.get(ba, offset + 8);
-
-            value.value = kvalue;
-            value.count = count;
-        }
-    }
-
-    static class GroupingAvgState implements AggregatorState<GroupingAvgState> {
-        private final BigArrays bigArrays;
-
-        LongArray values;
-        LongArray counts;
-
-        // total number of groups; <= values.length
-        int largestGroupId;
-
-        private final GroupingAvgStateSerializer serializer;
-
-        GroupingAvgState(BigArrays bigArrays) {
-            this.bigArrays = bigArrays;
-            boolean success = false;
-            try {
-                this.values = bigArrays.newLongArray(1);
-                this.counts = bigArrays.newLongArray(1);
-                success = true;
-            } finally {
-                if (success == false) {
-                    close();
-                }
-            }
-            this.serializer = new GroupingAvgStateSerializer();
-        }
-
-        void add(long valueToAdd, int groupId, long increment) {
-            ensureCapacity(groupId);
-            values.set(groupId, Math.addExact(values.get(groupId), valueToAdd));
-            counts.increment(groupId, increment);
-        }
-
-        void putNull(int position) {
-            ensureCapacity(position);
-        }
-
-        private void ensureCapacity(int groupId) {
-            if (groupId > largestGroupId) {
-                largestGroupId = groupId;
-                values = bigArrays.grow(values, groupId + 1);
-                counts = bigArrays.grow(counts, groupId + 1);
-            }
-        }
-
-        @Override
-        public long getEstimatedSize() {
-            return Long.BYTES + (largestGroupId + 1) * AvgStateSerializer.BYTES_SIZE;
-        }
-
-        @Override
-        public AggregatorStateSerializer<GroupingAvgState> serializer() {
-            return serializer;
-        }
-
-        @Override
-        public void close() {
-            Releasables.close(values, counts);
-        }
-    }
-
-    // @SerializedSize(value = Double.BYTES + Double.BYTES + Long.BYTES)
-    static class GroupingAvgStateSerializer implements AggregatorStateSerializer<GroupingAvgState> {
-
-        // record Shape (double value, double delta, long count) {}
-
-        static final int BYTES_SIZE = Long.BYTES + Long.BYTES;
-
-        @Override
-        public int size() {
-            return BYTES_SIZE;
-        }
-
-        private static final VarHandle longHandle = MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.BIG_ENDIAN);
-
-        @Override
-        public int serialize(GroupingAvgState state, byte[] ba, int offset, IntVector selected) {
-            longHandle.set(ba, offset, selected.getPositionCount());
-            offset += 8;
-            for (int i = 0; i < selected.getPositionCount(); i++) {
-                int group = selected.getInt(i);
-                longHandle.set(ba, offset, state.values.get(group));
-                longHandle.set(ba, offset + 8, state.counts.get(group));
-                offset += BYTES_SIZE;
-            }
-            return 8 + (BYTES_SIZE * selected.getPositionCount()); // number of bytes written
-        }
-
-        // sets the state in value
-        @Override
-        public void deserialize(GroupingAvgState state, byte[] ba, int offset) {
-            Objects.requireNonNull(state);
-            int positions = (int) (long) longHandle.get(ba, offset);
-            // TODO replace deserialization with direct passing - no more non_recycling_instance then
-            state.values = BigArrays.NON_RECYCLING_INSTANCE.grow(state.values, positions);
-            state.counts = BigArrays.NON_RECYCLING_INSTANCE.grow(state.counts, positions);
-            offset += 8;
-            for (int i = 0; i < positions; i++) {
-                state.values.set(i, (long) longHandle.get(ba, offset));
-                state.counts.set(i, (long) longHandle.get(ba, offset + 8));
-                offset += BYTES_SIZE;
-            }
-            state.largestGroupId = positions - 1;
-        }
-    }
-}

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AvgDoubleAggregatorFunctionTests.java

@@ -1,43 +0,0 @@
-/*
- * 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.aggregation;
-
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.operator.SequenceDoubleBlockSourceOperator;
-import org.elasticsearch.compute.operator.SourceOperator;
-import org.elasticsearch.test.ESTestCase;
-
-import java.util.List;
-import java.util.stream.LongStream;
-
-import static org.hamcrest.Matchers.closeTo;
-
-public class AvgDoubleAggregatorFunctionTests extends AggregatorFunctionTestCase {
-    @Override
-    protected SourceOperator simpleInput(int size) {
-        return new SequenceDoubleBlockSourceOperator(LongStream.range(0, size).mapToDouble(l -> ESTestCase.randomDouble()));
-    }
-
-    @Override
-    protected AggregatorFunctionSupplier aggregatorFunction(BigArrays bigArrays, List<Integer> inputChannels) {
-        return new AvgDoubleAggregatorFunctionSupplier(bigArrays, inputChannels);
-    }
-
-    @Override
-    protected String expectedDescriptionOfAggregator() {
-        return "avg of doubles";
-    }
-
-    @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        double avg = input.stream().flatMapToDouble(b -> allDoubles(b)).average().getAsDouble();
-        assertThat(((DoubleBlock) result).getDouble(0), closeTo(avg, .0001));
-    }
-}

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AvgDoubleGroupingAggregatorFunctionTests.java

@@ -1,56 +0,0 @@
-/*
- * 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.aggregation;
-
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.data.Page;
-import org.elasticsearch.compute.operator.LongDoubleTupleBlockSourceOperator;
-import org.elasticsearch.compute.operator.SourceOperator;
-import org.elasticsearch.core.Tuple;
-import org.elasticsearch.search.aggregations.metrics.CompensatedSum;
-
-import java.util.List;
-import java.util.stream.LongStream;
-
-import static org.hamcrest.Matchers.closeTo;
-import static org.hamcrest.Matchers.equalTo;
-
-public class AvgDoubleGroupingAggregatorFunctionTests extends GroupingAggregatorFunctionTestCase {
-    @Override
-    protected SourceOperator simpleInput(int size) {
-        return new LongDoubleTupleBlockSourceOperator(
-            LongStream.range(0, size).mapToObj(l -> Tuple.tuple(randomLongBetween(0, 4), randomDouble()))
-        );
-    }
-
-    @Override
-    protected AggregatorFunctionSupplier aggregatorFunction(BigArrays bigArrays, List<Integer> inputChannels) {
-        return new AvgDoubleAggregatorFunctionSupplier(bigArrays, inputChannels);
-    }
-
-    @Override
-    protected String expectedDescriptionOfAggregator() {
-        return "avg of doubles";
-    }
-
-    @Override
-    protected void assertSimpleGroup(List<Page> input, Block result, int position, long group) {
-        CompensatedSum sum = new CompensatedSum();
-        input.stream().flatMapToDouble(p -> allDoubles(p, group)).forEach(sum::add);
-        long count = input.stream().flatMapToDouble(p -> allDoubles(p, group)).count();
-        if (count == 0) {
-            // If all values are null we'll have a count of 0. So we'll be null.
-            assertThat(result.isNull(position), equalTo(true));
-            return;
-        }
-        assertThat(result.isNull(position), equalTo(false));
-        assertThat(((DoubleBlock) result).getDouble(position), closeTo(sum.value() / count, 0.001));
-    }
-}

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AvgIntAggregatorFunctionTests.java

@@ -1,44 +0,0 @@
-/*
- * 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.aggregation;
-
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.operator.SequenceIntBlockSourceOperator;
-import org.elasticsearch.compute.operator.SourceOperator;
-
-import java.util.List;
-import java.util.stream.LongStream;
-
-import static org.hamcrest.Matchers.equalTo;
-
-public class AvgIntAggregatorFunctionTests extends AggregatorFunctionTestCase {
-    @Override
-    protected SourceOperator simpleInput(int size) {
-        int max = between(1, (int) Math.min(Integer.MAX_VALUE, Long.MAX_VALUE / size));
-        return new SequenceIntBlockSourceOperator(LongStream.range(0, size).mapToInt(l -> between(-max, max)));
-    }
-
-    @Override
-    protected AggregatorFunctionSupplier aggregatorFunction(BigArrays bigArrays, List<Integer> inputChannels) {
-        return new AvgIntAggregatorFunctionSupplier(bigArrays, inputChannels);
-    }
-
-    @Override
-    protected String expectedDescriptionOfAggregator() {
-        return "avg of ints";
-    }
-
-    @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        long sum = input.stream().flatMapToInt(b -> allInts(b)).asLongStream().sum();
-        long count = input.stream().flatMapToInt(b -> allInts(b)).count();
-        assertThat(((DoubleBlock) result).getDouble(0), equalTo(((double) sum) / count));
-    }
-}

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AvgIntGroupingAggregatorFunctionTests.java

@@ -1,54 +0,0 @@
-/*
- * 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.aggregation;
-
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.data.Page;
-import org.elasticsearch.compute.operator.LongIntBlockSourceOperator;
-import org.elasticsearch.compute.operator.SourceOperator;
-import org.elasticsearch.core.Tuple;
-
-import java.util.List;
-import java.util.stream.LongStream;
-
-import static org.hamcrest.Matchers.closeTo;
-import static org.hamcrest.Matchers.equalTo;
-
-public class AvgIntGroupingAggregatorFunctionTests extends GroupingAggregatorFunctionTestCase {
-    @Override
-    protected AggregatorFunctionSupplier aggregatorFunction(BigArrays bigArrays, List<Integer> inputChannels) {
-        return new AvgIntAggregatorFunctionSupplier(bigArrays, inputChannels);
-    }
-
-    @Override
-    protected String expectedDescriptionOfAggregator() {
-        return "avg of ints";
-    }
-
-    @Override
-    protected SourceOperator simpleInput(int size) {
-        int max = between(1, (int) Math.min(Integer.MAX_VALUE, Long.MAX_VALUE / size));
-        return new LongIntBlockSourceOperator(
-            LongStream.range(0, size).mapToObj(l -> Tuple.tuple(randomLongBetween(0, 4), between(-max, max)))
-        );
-    }
-
-    @Override
-    public void assertSimpleGroup(List<Page> input, Block result, int position, long group) {
-        double sum = input.stream().flatMapToInt(p -> allInts(p, group)).asLongStream().sum();
-        long count = input.stream().flatMapToInt(p -> allInts(p, group)).count();
-        if (count == 0) {
-            // If all values are null we'll have a count of 0. So we'll be null.
-            assertThat(result.isNull(position), equalTo(true));
-            return;
-        }
-        assertThat(((DoubleBlock) result).getDouble(position), closeTo(sum / count, 0.001));
-    }
-}

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AvgLongAggregatorFunctionTests.java

@@ -1,64 +0,0 @@
-/*
- * 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.aggregation;
-
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.operator.Driver;
-import org.elasticsearch.compute.operator.DriverContext;
-import org.elasticsearch.compute.operator.PageConsumerOperator;
-import org.elasticsearch.compute.operator.SequenceLongBlockSourceOperator;
-import org.elasticsearch.compute.operator.SourceOperator;
-
-import java.util.List;
-import java.util.stream.LongStream;
-
-import static org.hamcrest.Matchers.equalTo;
-
-public class AvgLongAggregatorFunctionTests extends AggregatorFunctionTestCase {
-    @Override
-    protected SourceOperator simpleInput(int size) {
-        long max = randomLongBetween(1, Long.MAX_VALUE / size);
-        return new SequenceLongBlockSourceOperator(LongStream.range(0, size).map(l -> randomLongBetween(-max, max)));
-    }
-
-    @Override
-    protected AggregatorFunctionSupplier aggregatorFunction(BigArrays bigArrays, List<Integer> inputChannels) {
-        return new AvgLongAggregatorFunctionSupplier(bigArrays, inputChannels);
-    }
-
-    @Override
-    protected String expectedDescriptionOfAggregator() {
-        return "avg of longs";
-    }
-
-    @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        long sum = input.stream().flatMapToLong(b -> allLongs(b)).sum();
-        long count = input.stream().flatMapToLong(b -> allLongs(b)).count();
-        assertThat(((DoubleBlock) result).getDouble(0), equalTo(((double) sum) / count));
-    }
-
-    public void testOverflowFails() {
-        DriverContext driverContext = new DriverContext();
-        try (
-            Driver d = new Driver(
-                driverContext,
-                new SequenceLongBlockSourceOperator(LongStream.of(Long.MAX_VALUE - 1, 2)),
-                List.of(simple(nonBreakingBigArrays()).get(driverContext)),
-                new PageConsumerOperator(page -> fail("shouldn't have made it this far")),
-                () -> {}
-            )
-        ) {
-            Exception e = expectThrows(ArithmeticException.class, d::run);
-            assertThat(e.getMessage(), equalTo("long overflow"));
-            assertDriverContext(driverContext);
-        }
-    }
-}

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AvgLongGroupingAggregatorFunctionTests.java

@@ -1,54 +0,0 @@
-/*
- * 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.aggregation;
-
-import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.DoubleBlock;
-import org.elasticsearch.compute.data.Page;
-import org.elasticsearch.compute.operator.SourceOperator;
-import org.elasticsearch.compute.operator.TupleBlockSourceOperator;
-import org.elasticsearch.core.Tuple;
-
-import java.util.List;
-import java.util.stream.LongStream;
-
-import static org.hamcrest.Matchers.closeTo;
-import static org.hamcrest.Matchers.equalTo;
-
-public class AvgLongGroupingAggregatorFunctionTests extends GroupingAggregatorFunctionTestCase {
-    @Override
-    protected AggregatorFunctionSupplier aggregatorFunction(BigArrays bigArrays, List<Integer> inputChannels) {
-        return new AvgLongAggregatorFunctionSupplier(bigArrays, inputChannels);
-    }
-
-    @Override
-    protected String expectedDescriptionOfAggregator() {
-        return "avg of longs";
-    }
-
-    @Override
-    protected SourceOperator simpleInput(int size) {
-        long max = randomLongBetween(1, Long.MAX_VALUE / size);
-        return new TupleBlockSourceOperator(
-            LongStream.range(0, size).mapToObj(l -> Tuple.tuple(randomLongBetween(0, 4), randomLongBetween(-max, max)))
-        );
-    }
-
-    @Override
-    public void assertSimpleGroup(List<Page> input, Block result, int position, long group) {
-        double sum = input.stream().flatMapToLong(p -> allLongs(p, group)).sum();
-        long count = input.stream().flatMapToLong(p -> allLongs(p, group)).count();
-        if (count == 0) {
-            // If all values are null we'll have a count of 0. So we'll be null.
-            assertThat(result.isNull(position), equalTo(true));
-            return;
-        }
-        assertThat(((DoubleBlock) result).getDouble(position), closeTo(sum / count, 0.001));
-    }
-}

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinLongAggregatorFunctionTests.java

@@ -36,7 +36,7 @@ public class MinLongAggregatorFunctionTests extends AggregatorFunctionTestCase {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    public void assertSimpleOutput(List<Block> input, Block result) {
         long min = input.stream().flatMapToLong(b -> allLongs(b)).min().getAsLong();
         assertThat(((LongBlock) result).getLong(0), equalTo(min));
     }

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumLongAggregatorFunctionTests.java

@@ -43,7 +43,7 @@ public class SumLongAggregatorFunctionTests extends AggregatorFunctionTestCase {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    public void assertSimpleOutput(List<Block> input, Block result) {
         long sum = input.stream().flatMapToLong(b -> allLongs(b)).sum();
         assertThat(((LongBlock) result).getLong(0), equalTo(sum));
     }

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumLongGroupingAggregatorFunctionTests.java

@@ -40,7 +40,7 @@ public class SumLongGroupingAggregatorFunctionTests extends GroupingAggregatorFu
     }
 
     @Override
-    protected void assertSimpleGroup(List<Page> input, Block result, int position, long group) {
+    public void assertSimpleGroup(List<Page> input, Block result, int position, long group) {
         long sum = input.stream().flatMapToLong(p -> allLongs(p, group)).sum();
         assertThat(((LongBlock) result).getLong(position), equalTo(sum));
     }

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

@@ -9,7 +9,7 @@ package org.elasticsearch.compute.data;
 
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.compute.aggregation.AvgLongAggregatorFunction;
+import org.elasticsearch.compute.aggregation.SumLongAggregatorFunction;
 import org.elasticsearch.test.EqualsHashCodeTestUtils;
 
 import java.io.IOException;
@@ -103,7 +103,7 @@ public class BlockSerializationTests extends SerializationTestCase {
         Page page = new Page(new LongArrayVector(new long[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 }, 10).asBlock());
         var bigArrays = BigArrays.NON_RECYCLING_INSTANCE;
         var params = new Object[] {};
-        var function = AvgLongAggregatorFunction.create(List.of(0));
+        var function = SumLongAggregatorFunction.create(List.of(0));
         function.addRawInput(page);
         Block[] blocks = new Block[1];
         function.evaluateIntermediate(blocks, 0);
@@ -112,11 +112,11 @@ public class BlockSerializationTests extends SerializationTestCase {
         Block deserBlock = serializeDeserializeBlock(origBlock);
         EqualsHashCodeTestUtils.checkEqualsAndHashCode(origBlock, unused -> deserBlock);
 
-        var finalAggregator = AvgLongAggregatorFunction.create(List.of(0));
+        var finalAggregator = SumLongAggregatorFunction.create(List.of(0));
         finalAggregator.addIntermediateInput(new Page(deserBlock));
         Block[] finalBlocks = new Block[1];
         finalAggregator.evaluateFinal(finalBlocks, 0);
-        DoubleBlock finalBlock = (DoubleBlock) finalBlocks[0];
-        assertThat(finalBlock.getDouble(0), is(5.5));
+        var finalBlock = (LongBlock) finalBlocks[0];
+        assertThat(finalBlock.getLong(0), is(55L));
     }
 }

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/AggregationOperatorTests.java

@@ -10,10 +10,10 @@ package org.elasticsearch.compute.operator;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.compute.aggregation.AggregatorMode;
-import org.elasticsearch.compute.aggregation.AvgLongAggregatorFunctionSupplier;
-import org.elasticsearch.compute.aggregation.AvgLongAggregatorFunctionTests;
 import org.elasticsearch.compute.aggregation.MaxLongAggregatorFunctionSupplier;
 import org.elasticsearch.compute.aggregation.MaxLongAggregatorFunctionTests;
+import org.elasticsearch.compute.aggregation.SumLongAggregatorFunctionSupplier;
+import org.elasticsearch.compute.aggregation.SumLongAggregatorFunctionTests;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.Page;
 
@@ -35,7 +35,7 @@ public class AggregationOperatorTests extends ForkingOperatorTestCase {
         int maxChannel = mode.isInputPartial() ? 1 : 0;
         return new AggregationOperator.AggregationOperatorFactory(
             List.of(
-                new AvgLongAggregatorFunctionSupplier(bigArrays, List.of(0)).aggregatorFactory(mode),
+                new SumLongAggregatorFunctionSupplier(bigArrays, List.of(0)).aggregatorFactory(mode),
                 new MaxLongAggregatorFunctionSupplier(bigArrays, List.of(maxChannel)).aggregatorFactory(mode)
             ),
             mode
@@ -44,13 +44,13 @@ public class AggregationOperatorTests extends ForkingOperatorTestCase {
 
     @Override
     protected String expectedDescriptionOfSimple() {
-        return "AggregationOperator[mode = SINGLE, aggs = avg of longs, max of longs]";
+        return "AggregationOperator[mode = SINGLE, aggs = sum of longs, max of longs]";
     }
 
     @Override
     protected String expectedToStringOfSimple() {
         return "AggregationOperator[aggregators=["
-            + "Aggregator[aggregatorFunction=AvgLongAggregatorFunction[channels=[0]], mode=SINGLE], "
+            + "Aggregator[aggregatorFunction=SumLongAggregatorFunction[channels=[0]], mode=SINGLE], "
             + "Aggregator[aggregatorFunction=MaxLongAggregatorFunction[channels=[0]], mode=SINGLE]]]";
     }
 
@@ -60,12 +60,12 @@ public class AggregationOperatorTests extends ForkingOperatorTestCase {
         assertThat(results.get(0).getBlockCount(), equalTo(2));
         assertThat(results.get(0).getPositionCount(), equalTo(1));
 
-        AvgLongAggregatorFunctionTests avg = new AvgLongAggregatorFunctionTests();
+        SumLongAggregatorFunctionTests sum = new SumLongAggregatorFunctionTests();
         MaxLongAggregatorFunctionTests max = new MaxLongAggregatorFunctionTests();
 
-        Block avgs = results.get(0).getBlock(0);
+        Block sums = results.get(0).getBlock(0);
         Block maxs = results.get(0).getBlock(1);
-        avg.assertSimpleOutput(input.stream().map(p -> p.<Block>getBlock(0)).toList(), avgs);
+        sum.assertSimpleOutput(input.stream().map(p -> p.<Block>getBlock(0)).toList(), sums);
         max.assertSimpleOutput(input.stream().map(p -> p.<Block>getBlock(0)).toList(), maxs);
     }
 

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/HashAggregationOperatorTests.java

@@ -10,10 +10,10 @@ package org.elasticsearch.compute.operator;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.compute.aggregation.AggregatorMode;
-import org.elasticsearch.compute.aggregation.AvgLongAggregatorFunctionSupplier;
-import org.elasticsearch.compute.aggregation.AvgLongGroupingAggregatorFunctionTests;
 import org.elasticsearch.compute.aggregation.MaxLongAggregatorFunctionSupplier;
 import org.elasticsearch.compute.aggregation.MaxLongGroupingAggregatorFunctionTests;
+import org.elasticsearch.compute.aggregation.SumLongAggregatorFunctionSupplier;
+import org.elasticsearch.compute.aggregation.SumLongGroupingAggregatorFunctionTests;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.ElementType;
 import org.elasticsearch.compute.data.LongBlock;
@@ -39,7 +39,7 @@ public class HashAggregationOperatorTests extends ForkingOperatorTestCase {
         return new HashAggregationOperator.HashAggregationOperatorFactory(
             List.of(new HashAggregationOperator.GroupSpec(0, ElementType.LONG)),
             List.of(
-                new AvgLongAggregatorFunctionSupplier(bigArrays, List.of(1)).groupingAggregatorFactory(mode),
+                new SumLongAggregatorFunctionSupplier(bigArrays, List.of(1)).groupingAggregatorFactory(mode),
                 new MaxLongAggregatorFunctionSupplier(bigArrays, List.of(maxChannel)).groupingAggregatorFactory(mode)
             ),
             bigArrays
@@ -48,13 +48,13 @@ public class HashAggregationOperatorTests extends ForkingOperatorTestCase {
 
     @Override
     protected String expectedDescriptionOfSimple() {
-        return "HashAggregationOperator[mode = <not-needed>, aggs = avg of longs, max of longs]";
+        return "HashAggregationOperator[mode = <not-needed>, aggs = sum of longs, max of longs]";
     }
 
     @Override
     protected String expectedToStringOfSimple() {
         return "HashAggregationOperator[blockHash=LongBlockHash{channel=0, entries=0}, aggregators=["
-            + "GroupingAggregator[aggregatorFunction=AvgLongGroupingAggregatorFunction[channels=[1]], mode=SINGLE], "
+            + "GroupingAggregator[aggregatorFunction=SumLongGroupingAggregatorFunction[channels=[1]], mode=SINGLE], "
             + "GroupingAggregator[aggregatorFunction=MaxLongGroupingAggregatorFunction[channels=[1]], mode=SINGLE]]]";
     }
 
@@ -64,15 +64,15 @@ public class HashAggregationOperatorTests extends ForkingOperatorTestCase {
         assertThat(results.get(0).getBlockCount(), equalTo(3));
         assertThat(results.get(0).getPositionCount(), equalTo(5));
 
-        AvgLongGroupingAggregatorFunctionTests avg = new AvgLongGroupingAggregatorFunctionTests();
+        SumLongGroupingAggregatorFunctionTests sum = new SumLongGroupingAggregatorFunctionTests();
         MaxLongGroupingAggregatorFunctionTests max = new MaxLongGroupingAggregatorFunctionTests();
 
         LongBlock groups = results.get(0).getBlock(0);
-        Block avgs = results.get(0).getBlock(1);
+        Block sums = results.get(0).getBlock(1);
         Block maxs = results.get(0).getBlock(2);
         for (int i = 0; i < 5; i++) {
             long group = groups.getLong(i);
-            avg.assertSimpleGroup(input, avgs, i, group);
+            sum.assertSimpleGroup(input, sums, i, group);
             max.assertSimpleGroup(input, maxs, i, group);
         }
     }

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlActionIT.java

@@ -176,6 +176,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         assertEquals(expectedValues, actualValues);
     }
 
+    @AwaitsFix(bugUrl = "1306")
     public void testFromGroupingByNumericFieldWithNulls() {
         for (int i = 0; i < 5; i++) {
             client().prepareBulk()
@@ -245,6 +246,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         assertThat(actualGroups, equalTo(expectedGroups));
     }
 
+    @AwaitsFix(bugUrl = "1306")
     public void testFromStatsGroupingByKeywordWithNulls() {
         for (int i = 0; i < 5; i++) {
             client().prepareBulk()

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/SurrogateExpression.java

@@ -0,0 +1,19 @@
+/*
+ * 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.xpack.esql.expression;
+
+import org.elasticsearch.xpack.ql.expression.Expression;
+
+/**
+ * Interface signaling to the planner that the declaring expression
+ * has to be replaced by a different form: e.g. avg = sum / count
+ */
+public interface SurrogateExpression {
+
+    Expression surrogate();
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Avg.java

@@ -9,18 +9,18 @@ package org.elasticsearch.xpack.esql.expression.function.aggregate;
 
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.compute.aggregation.AggregatorFunctionSupplier;
-import org.elasticsearch.compute.aggregation.AvgDoubleAggregatorFunctionSupplier;
-import org.elasticsearch.compute.aggregation.AvgIntAggregatorFunctionSupplier;
-import org.elasticsearch.compute.aggregation.AvgLongAggregatorFunctionSupplier;
 import org.elasticsearch.compute.ann.Experimental;
+import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
+import org.elasticsearch.xpack.esql.expression.SurrogateExpression;
 import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.expression.predicate.operator.arithmetic.Div;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
 
 import java.util.List;
 
 @Experimental
-public class Avg extends NumericAggregate {
+public class Avg extends NumericAggregate implements SurrogateExpression {
 
     public Avg(Source source, Expression field) {
         super(source, field);
@@ -36,18 +36,24 @@ public class Avg extends NumericAggregate {
         return new Avg(source(), newChildren.get(0));
     }
 
+    public Expression surrogate() {
+        var s = source();
+        var field = field();
+        return new Div(s, new Sum(s, field), new Count(s, field), dataType());
+    }
+
     @Override
     protected AggregatorFunctionSupplier longSupplier(BigArrays bigArrays, List<Integer> inputChannels) {
-        return new AvgLongAggregatorFunctionSupplier(bigArrays, inputChannels);
+        throw new EsqlIllegalArgumentException("unsupported operation");
     }
 
     @Override
     protected AggregatorFunctionSupplier intSupplier(BigArrays bigArrays, List<Integer> inputChannels) {
-        return new AvgIntAggregatorFunctionSupplier(bigArrays, inputChannels);
+        throw new EsqlIllegalArgumentException("unsupported operation");
     }
 
     @Override
     protected AggregatorFunctionSupplier doubleSupplier(BigArrays bigArrays, List<Integer> inputChannels) {
-        return new AvgDoubleAggregatorFunctionSupplier(bigArrays, inputChannels);
+        throw new EsqlIllegalArgumentException("unsupported operation");
     }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizer.java

@@ -10,6 +10,7 @@ package org.elasticsearch.xpack.esql.optimizer;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.compute.data.BlockUtils;
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
+import org.elasticsearch.xpack.esql.expression.SurrogateExpression;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Count;
 import org.elasticsearch.xpack.esql.expression.function.scalar.conditional.IsNull;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.In;
@@ -17,6 +18,7 @@ import org.elasticsearch.xpack.esql.plan.logical.Enrich;
 import org.elasticsearch.xpack.esql.plan.logical.Eval;
 import org.elasticsearch.xpack.esql.plan.logical.RegexExtract;
 import org.elasticsearch.xpack.esql.plan.logical.TopN;
+import org.elasticsearch.xpack.esql.plan.logical.local.EsqlProject;
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalRelation;
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalSupplier;
 import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
@@ -58,7 +60,9 @@ import org.elasticsearch.xpack.ql.util.CollectionUtils;
 
 import java.time.ZoneId;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Predicate;
 
 import static java.util.Arrays.asList;
@@ -78,6 +82,8 @@ public class LogicalPlanOptimizer extends RuleExecutor<LogicalPlan> {
     }
 
     protected static List<Batch<LogicalPlan>> rules() {
+        var substitutions = new Batch<>("Substitutions", Limiter.ONCE, new SubstituteSurrogates());
+
         var operators = new Batch<>(
             "Operator Optimization",
             new CombineProjections(),
@@ -112,7 +118,88 @@ public class LogicalPlanOptimizer extends RuleExecutor<LogicalPlan> {
         var cleanup = new Batch<>("Clean Up", new ReplaceLimitAndSortAsTopN());
         var label = new Batch<>("Set as Optimized", Limiter.ONCE, new SetAsOptimized());
 
-        return asList(operators, skip, cleanup, label);
+        return asList(substitutions, operators, skip, cleanup, label);
+    }
+
+    // TODO: currently this rule only works for aggregate functions (AVG)
+    static class SubstituteSurrogates extends OptimizerRules.OptimizerRule<Aggregate> {
+
+        SubstituteSurrogates() {
+            super(OptimizerRules.TransformDirection.UP);
+        }
+
+        @Override
+        protected LogicalPlan rule(Aggregate aggregate) {
+            var aggs = aggregate.aggregates();
+            List<NamedExpression> newAggs = new ArrayList<>(aggs.size());
+            // existing aggregate and their respective attributes
+            Map<AggregateFunction, Attribute> aggFuncToAttr = new HashMap<>();
+            // surrogate functions eval
+            List<NamedExpression> transientEval = new ArrayList<>();
+            boolean changed = false;
+
+            // first pass to check existing aggregates (to avoid duplication and alias waste)
+            for (NamedExpression agg : aggs) {
+                if (agg instanceof Alias a && a.child() instanceof AggregateFunction af && af instanceof SurrogateExpression == false) {
+                    aggFuncToAttr.put(af, a.toAttribute());
+                }
+            }
+
+            // 0. check list of surrogate expressions
+            for (NamedExpression agg : aggs) {
+                Expression e = agg instanceof Alias a ? a.child() : agg;
+                if (e instanceof SurrogateExpression sf) {
+                    changed = true;
+                    Expression s = sf.surrogate();
+                    // 1. collect all aggregate functions from the expression
+                    var surrogateWithRefs = s.transformUp(AggregateFunction.class, af -> {
+                        // 2. check if they are already use otherwise add them to the Aggregate with some made-up aliases
+                        // 3. replace them inside the expression using the given alias
+                        var attr = aggFuncToAttr.get(af);
+                        // the agg doesn't exist in the Aggregate, create an alias for it and save its attribute
+                        if (attr == null) {
+                            var temporaryName = temporaryName(agg, af);
+                            // create a synthetic alias (so it doesn't clash with a user defined name)
+                            var newAlias = new Alias(agg.source(), temporaryName, null, af, null, true);
+                            attr = newAlias.toAttribute();
+                            aggFuncToAttr.put(af, attr);
+                            newAggs.add(newAlias);
+                        }
+                        return attr;
+                    });
+                    // 4. move the expression as an eval using the original alias
+                    // check however if the expression requires an eval in the first place
+                    if (surrogateWithRefs instanceof AggregateFunction == false) {
+                        // copy the original alias id so that other nodes using it down stream (e.g. eval referring to the original agg)
+                        // don't have to updated
+                        var aliased = new Alias(agg.source(), agg.name(), null, surrogateWithRefs, agg.toAttribute().id());
+                        transientEval.add(aliased);
+                    }
+                } else {
+                    newAggs.add(agg);
+                }
+            }
+
+            LogicalPlan plan = aggregate;
+            if (changed) {
+                var source = aggregate.source();
+                plan = new Aggregate(aggregate.source(), aggregate.child(), aggregate.groupings(), newAggs);
+                // 5. force the initial projection in place
+                if (transientEval.size() > 0) {
+                    plan = new Eval(source, plan, transientEval);
+                    // project away transient fields and re-enforce the original order using references (not copies) to the original aggs
+                    // this works since the replaced aliases have their nameId copied to avoid having to update all references (which has
+                    // a cascading effect)
+                    plan = new EsqlProject(source, plan, Expressions.asAttributes(aggs));
+                }
+            }
+
+            return plan;
+        }
+
+        private static String temporaryName(NamedExpression agg, AggregateFunction af) {
+            return "__" + agg.name() + "_" + af.functionName() + "@" + Integer.toHexString(af.hashCode());
+        }
     }
 
     static class ConvertStringToByteRef extends OptimizerRules.OptimizerExpressionRule<Literal> {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/ArithmeticMapper.java

@@ -76,7 +76,12 @@ abstract class ArithmeticMapper<T extends ArithmeticOperation> extends EvalMappe
     @Override
     protected final Supplier<EvalOperator.ExpressionEvaluator> map(ArithmeticOperation op, Layout layout) {
         if (op.left().dataType().isNumeric()) {
-            DataType type = EsqlDataTypeRegistry.INSTANCE.commonType(op.left().dataType(), op.right().dataType());
+            DataType type = null;
+            if (op instanceof Div div) {
+                type = div.dataType();
+            } else {
+                type = EsqlDataTypeRegistry.INSTANCE.commonType(op.left().dataType(), op.right().dataType());
+            }
             if (type == DataTypes.INTEGER) {
                 return castToEvaluator(op, layout, DataTypes.INTEGER, ints);
             }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java

@@ -17,6 +17,8 @@ import org.elasticsearch.xpack.esql.analysis.EnrichResolution;
 import org.elasticsearch.xpack.esql.analysis.Verifier;
 import org.elasticsearch.xpack.esql.enrich.EnrichPolicyResolution;
 import org.elasticsearch.xpack.esql.expression.function.EsqlFunctionRegistry;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.Count;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.Sum;
 import org.elasticsearch.xpack.esql.expression.function.scalar.conditional.IsNull;
 import org.elasticsearch.xpack.esql.expression.function.scalar.date.DateFormat;
 import org.elasticsearch.xpack.esql.expression.function.scalar.date.DateParse;
@@ -45,7 +47,6 @@ import org.elasticsearch.xpack.ql.expression.NamedExpression;
 import org.elasticsearch.xpack.ql.expression.Nullability;
 import org.elasticsearch.xpack.ql.expression.Order;
 import org.elasticsearch.xpack.ql.expression.ReferenceAttribute;
-import org.elasticsearch.xpack.ql.expression.function.aggregate.Count;
 import org.elasticsearch.xpack.ql.expression.predicate.logical.And;
 import org.elasticsearch.xpack.ql.expression.predicate.logical.Not;
 import org.elasticsearch.xpack.ql.expression.predicate.logical.Or;
@@ -100,6 +101,7 @@ import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.startsWith;
 
 //@TestLogging(value = "org.elasticsearch.xpack.esql:TRACE", reason = "debug")
 public class LogicalPlanOptimizerTests extends ESTestCase {
@@ -209,12 +211,13 @@ public class LogicalPlanOptimizerTests extends ESTestCase {
     public void testCombineProjectionWithAggregation() {
         var plan = plan("""
             from test
-            | stats avg(salary) by last_name, first_name
+            | stats s = sum(salary) by last_name, first_name
+            | keep s, last_name, first_name
             """);
 
         var limit = as(plan, Limit.class);
         var agg = as(limit.child(), Aggregate.class);
-        assertThat(Expressions.names(agg.aggregates()), contains("avg(salary)", "last_name", "first_name"));
+        assertThat(Expressions.names(agg.aggregates()), contains("s", "last_name", "first_name"));
         assertThat(Expressions.names(agg.groupings()), contains("last_name", "first_name"));
     }
 
@@ -345,7 +348,7 @@ public class LogicalPlanOptimizerTests extends ESTestCase {
         EsRelation relation = relation();
         GreaterThan conditionA = greaterThanOf(getFieldAttribute("a"), ONE);
         LessThan conditionB = lessThanOf(getFieldAttribute("b"), TWO);
-        GreaterThanOrEqual aggregateCondition = greaterThanOrEqualOf(new Count(EMPTY, ONE, false), THREE);
+        GreaterThanOrEqual aggregateCondition = greaterThanOrEqualOf(new Count(EMPTY, ONE), THREE);
 
         Filter fa = new Filter(EMPTY, relation, conditionA);
         // invalid aggregate but that's fine cause its properties are not used by this rule
@@ -709,7 +712,7 @@ public class LogicalPlanOptimizerTests extends ESTestCase {
             from test
             | sort emp_no
             | where emp_no > 10
-            | stats x = avg(salary) by first_name""");
+            | stats x = sum(salary) by first_name""");
 
         var limit = as(plan, Limit.class);
         var stats = as(limit.child(), Aggregate.class);
@@ -722,7 +725,7 @@ public class LogicalPlanOptimizerTests extends ESTestCase {
             from test
             | sort emp_no
             | limit 100
-            | stats x = avg(salary) by first_name""");
+            | stats x = sum(salary) by first_name""");
 
         var limit = as(plan, Limit.class);
         var stats = as(limit.child(), Aggregate.class);
@@ -1083,6 +1086,103 @@ public class LogicalPlanOptimizerTests extends ESTestCase {
         as(topN.child(), Enrich.class);
     }
 
+    /**
+     * Expects
+     * EsqlProject[[a{r}#3, last_name{f}#9]]
+     * \_Eval[[__a_SUM_123{r}#12 / __a_COUNT_150{r}#13 AS a]]
+     *   \_Limit[10000[INTEGER]]
+     *     \_Aggregate[[last_name{f}#9],[SUM(salary{f}#10) AS __a_SUM_123, COUNT(salary{f}#10) AS __a_COUNT_150, last_nam
+     * e{f}#9]]
+     *       \_EsRelation[test][_meta_field{f}#11, emp_no{f}#5, first_name{f}#6, !g..]
+     */
+    public void testSimpleAvgReplacement() {
+        var plan = plan("""
+              from test
+            | stats a = avg(salary) by last_name
+            """);
+
+        var project = as(plan, Project.class);
+        assertThat(Expressions.names(project.projections()), contains("a", "last_name"));
+        var eval = as(project.child(), Eval.class);
+        var f = eval.fields();
+        assertThat(f, hasSize(1));
+        assertThat(f.get(0).name(), is("a"));
+        var limit = as(eval.child(), Limit.class);
+        var agg = as(limit.child(), Aggregate.class);
+        var aggs = agg.aggregates();
+        var a = as(aggs.get(0), Alias.class);
+        assertThat(a.name(), startsWith("__a_SUM@"));
+        var sum = as(a.child(), Sum.class);
+
+        a = as(aggs.get(1), Alias.class);
+        assertThat(a.name(), startsWith("__a_COUNT@"));
+        var count = as(a.child(), Count.class);
+
+        assertThat(Expressions.names(agg.groupings()), contains("last_name"));
+    }
+
+    /**
+     * Expects
+     * EsqlProject[[a{r}#3, c{r}#6, s{r}#9, last_name{f}#15]]
+     * \_Eval[[s{r}#9 / c{r}#6 AS a]]
+     *   \_Limit[10000[INTEGER]]
+     *     \_Aggregate[[last_name{f}#15],[COUNT(salary{f}#16) AS c, SUM(salary{f}#16) AS s, last_name{f}#15]]
+     *       \_EsRelation[test][_meta_field{f}#17, emp_no{f}#11, first_name{f}#12, ..]
+     */
+    public void testClashingAggAvgReplacement() {
+        var plan = plan("""
+            from test
+            | stats a = avg(salary), c = count(salary), s = sum(salary) by last_name
+            """);
+
+        assertThat(Expressions.names(plan.output()), contains("a", "c", "s", "last_name"));
+        var project = as(plan, EsqlProject.class);
+        var eval = as(project.child(), Eval.class);
+        var f = eval.fields();
+        assertThat(f, hasSize(1));
+        assertThat(f.get(0).name(), is("a"));
+        var limit = as(eval.child(), Limit.class);
+        var agg = as(limit.child(), Aggregate.class);
+        var aggs = agg.aggregates();
+        assertThat(Expressions.names(aggs), contains("c", "s", "last_name"));
+    }
+
+    /**
+     * Expects
+     * EsqlProject[[a{r}#3, c{r}#6, s{r}#9, last_name{f}#15]]
+     * \_Eval[[s{r}#9 / __a_COUNT@xxx{r}#18 AS a]]
+     *   \_Limit[10000[INTEGER]]
+     *     \_Aggregate[[last_name{f}#15],[COUNT(salary{f}#16) AS __a_COUNT@xxx, COUNT(languages{f}#14) AS c, SUM(salary{f}#16) AS
+     *  s, last_name{f}#15]]
+     *       \_EsRelation[test][_meta_field{f}#17, emp_no{f}#11, first_name{f}#12, ..]
+     */
+    public void testSemiClashingAvgReplacement() {
+        var plan = plan("""
+            from test
+            | stats a = avg(salary), c = count(languages), s = sum(salary) by last_name
+            """);
+
+        var project = as(plan, Project.class);
+        assertThat(Expressions.names(project.projections()), contains("a", "c", "s", "last_name"));
+        var eval = as(project.child(), Eval.class);
+        var f = eval.fields();
+        assertThat(f, hasSize(1));
+        assertThat(f.get(0).name(), is("a"));
+        var limit = as(eval.child(), Limit.class);
+        var agg = as(limit.child(), Aggregate.class);
+        var aggs = agg.aggregates();
+        var a = as(aggs.get(0), Alias.class);
+        assertThat(a.name(), startsWith("__a_COUNT@"));
+        var sum = as(a.child(), Count.class);
+
+        a = as(aggs.get(1), Alias.class);
+        assertThat(a.name(), is("c"));
+        var count = as(a.child(), Count.class);
+
+        a = as(aggs.get(2), Alias.class);
+        assertThat(a.name(), is("s"));
+    }
+
     private LogicalPlan optimizedPlan(String query) {
         return logicalOptimizer.optimize(analyzer.analyze(parser.createStatement(query)));
     }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/PhysicalPlanOptimizerTests.java

@@ -188,7 +188,7 @@ public class PhysicalPlanOptimizerTests extends ESTestCase {
             from test
             | where round(emp_no) > 10
             | eval c = salary
-            | stats x = avg(c)
+            | stats x = sum(c)
             """);
 
         var optimized = optimizedPlan(plan);
@@ -213,7 +213,7 @@ public class PhysicalPlanOptimizerTests extends ESTestCase {
             from test
             | where round(emp_no) > 10
             | eval c = first_name
-            | stats x = avg(salary)
+            | stats x = sum(salary)
             """);
 
         var optimized = optimizedPlan(plan);
@@ -256,7 +256,7 @@ public class PhysicalPlanOptimizerTests extends ESTestCase {
             | limit 10
             | where round(emp_no) > 10
             | eval c = first_name
-            | stats x = avg(salary)
+            | stats x = sum(salary)
             """);
 
         var optimized = optimizedPlan(plan);
@@ -346,7 +346,7 @@ public class PhysicalPlanOptimizerTests extends ESTestCase {
     public void testExtractorsOverridingFields() {
         var plan = physicalPlan("""
             from test
-            | stats emp_no = avg(emp_no)
+            | stats emp_no = sum(emp_no)
             """);
 
         var optimized = optimizedPlan(plan);
@@ -362,7 +362,7 @@ public class PhysicalPlanOptimizerTests extends ESTestCase {
     public void testDoNotExtractGroupingFields() {
         var plan = physicalPlan("""
             from test
-            | stats x = avg(salary) by first_name
+            | stats x = sum(salary) by first_name
             """);
 
         var optimized = optimizedPlan(plan);
@@ -428,7 +428,7 @@ public class PhysicalPlanOptimizerTests extends ESTestCase {
     public void testQueryWithAggregation() {
         var plan = physicalPlan("""
             from test
-            | stats avg(emp_no)
+            | stats sum(emp_no)
             """);
 
         var optimized = optimizedPlan(plan);
@@ -444,8 +444,8 @@ public class PhysicalPlanOptimizerTests extends ESTestCase {
     public void testQueryWithAggAndEval() {
         var plan = physicalPlan("""
             from test
-            | stats avg_emp = avg(emp_no)
-            | eval x = avg_emp + 7
+            | stats agg_emp = sum(emp_no)
+            | eval x = agg_emp + 7
             """);
 
         var optimized = optimizedPlan(plan);

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/predicate/operator/arithmetic/Div.java

@@ -17,23 +17,33 @@ import org.elasticsearch.xpack.ql.type.DataTypeConverter;
  */
 public class Div extends ArithmeticOperation implements BinaryComparisonInversible {
 
+    private DataType dataType;
+
     public Div(Source source, Expression left, Expression right) {
+        this(source, left, right, null);
+    }
+
+    public Div(Source source, Expression left, Expression right, DataType dataType) {
         super(source, left, right, DefaultBinaryArithmeticOperation.DIV);
+        this.dataType = dataType;
     }
 
     @Override
     protected NodeInfo<Div> info() {
-        return NodeInfo.create(this, Div::new, left(), right());
+        return NodeInfo.create(this, Div::new, left(), right(), dataType);
     }
 
     @Override
     protected Div replaceChildren(Expression newLeft, Expression newRight) {
-        return new Div(source(), newLeft, newRight);
+        return new Div(source(), newLeft, newRight, dataType);
     }
 
     @Override
     public DataType dataType() {
-        return DataTypeConverter.commonType(left().dataType(), right().dataType());
+        if (dataType == null) {
+            dataType = DataTypeConverter.commonType(left().dataType(), right().dataType());
+        }
+        return dataType;
     }
 
     @Override