ES|QL: Add PRESENT ES|QL function

Optimize AggregatorFunction

Part of #131069

Author: leontyevdv

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

@@ -15,7 +15,6 @@
 import org.elasticsearch.compute.operator.DriverContext;
 
 import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 public class PresentAggregatorFunction implements AggregatorFunction {
     public static AggregatorFunctionSupplier supplier() {
@@ -55,14 +54,15 @@ public static List<IntermediateStateDesc> intermediateStateDesc() {
         return INTERMEDIATE_STATE_DESC;
     }
 
-    private final AtomicBoolean state;
     private final List<Integer> channels;
 
+    private boolean state;
+
     public static PresentAggregatorFunction create(List<Integer> inputChannels) {
-        return new PresentAggregatorFunction(inputChannels, new AtomicBoolean(false));
+        return new PresentAggregatorFunction(inputChannels, false);
     }
 
-    private PresentAggregatorFunction(List<Integer> channels, AtomicBoolean state) {
+    private PresentAggregatorFunction(List<Integer> channels, boolean state) {
         this.channels = channels;
         this.state = state;
     }
@@ -78,17 +78,10 @@ private int blockIndex() {
 
     @Override
     public void addRawInput(Page page, BooleanVector mask) {
+        if (mask.isConstant() && mask.getBoolean(0) == false) return;
+
         Block block = page.getBlock(blockIndex());
-        boolean present;
-        if (mask.isConstant()) {
-            if (mask.getBoolean(0) == false) {
-                return;
-            }
-            present = block.getTotalValueCount() > 0;
-        } else {
-            present = presentMasked(block, mask);
-        }
-        this.state.set(present);
+        this.state = mask.isConstant() ? block.getTotalValueCount() > 0 : presentMasked(block, mask);
     }
 
     private boolean presentMasked(Block block, BooleanVector mask) {
@@ -112,7 +105,7 @@ public void addIntermediateInput(Page page) {
         BooleanVector present = page.<BooleanBlock>getBlock(channels.get(0)).asVector();
         assert present.getPositionCount() == 1;
         if (present.getBoolean(0)) {
-            state.set(true);
+            this.state = true;
         }
     }
 
@@ -123,7 +116,7 @@ public void evaluateIntermediate(Block[] blocks, int offset, DriverContext drive
 
     @Override
     public void evaluateFinal(Block[] blocks, int offset, DriverContext driverContext) {
-        blocks[offset] = driverContext.blockFactory().newConstantBooleanBlockWith(state.get(), 1);
+        blocks[offset] = driverContext.blockFactory().newConstantBooleanBlockWith(state, 1);
     }
 
     @Override

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/PresentSerializationTests.java

@@ -0,0 +1,24 @@
+/*
+ * 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.function.aggregate;
+
+import org.elasticsearch.xpack.esql.expression.AbstractExpressionSerializationTests;
+
+import java.io.IOException;
+
+public class PresentSerializationTests extends AbstractExpressionSerializationTests<Present> {
+    @Override
+    protected Present createTestInstance() {
+        return new Present(randomSource(), randomChild());
+    }
+
+    @Override
+    protected Present mutateInstance(Present instance) throws IOException {
+        return new Present(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
+    }
+}