[8.x] Implement delay() function for debug purposes (#115882) (#116452)

* Implement delay() function for debug purposes (#115882)

* Implement delay() function for debug purposes

(cherry picked from commit f2a5295)

* Fix compile error

Author: smalyshev

x-pack/plugin/esql/qa/testFixtures/src/main/resources/delay.csv-spec

@@ -0,0 +1,7 @@
+DELAY
+required_capability: delay_debug_fn
+row a = 1 | where delay(1ms);
+
+a:integer
+1
+;

x-pack/plugin/esql/src/main/generated/org/elasticsearch/xpack/esql/expression/function/scalar/util/DelayEvaluator.java

@@ -469,6 +469,7 @@ public enum Cap {
          */
         ADD_LIMIT_INSIDE_MV_EXPAND,
 
+        DELAY_DEBUG_FN(Build.current().isSnapshot()),
         /**
          * WIP on Join planning
          * - Introduce BinaryPlan and co

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -136,6 +136,7 @@
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.ToLower;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.ToUpper;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Trim;
+import org.elasticsearch.xpack.esql.expression.function.scalar.util.Delay;
 import org.elasticsearch.xpack.esql.session.Configuration;
 
 import java.lang.reflect.Constructor;
@@ -399,6 +400,9 @@ private FunctionDefinition[][] functions() {
     private static FunctionDefinition[][] snapshotFunctions() {
         return new FunctionDefinition[][] {
             new FunctionDefinition[] {
+                // The delay() function is for debug/snapshot environments only and should never be enabled in a non-snapshot build.
+                // This is an experimental function and can be removed without notice.
+                def(Delay.class, Delay::new, "delay"),
                 def(Categorize.class, Categorize::new, "categorize"),
                 def(Rate.class, Rate::withUnresolvedTimestamp, "rate") } };
     }

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

@@ -62,6 +62,7 @@
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Space;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Trim;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.WildcardLike;
+import org.elasticsearch.xpack.esql.expression.function.scalar.util.Delay;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic.Neg;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 
@@ -123,6 +124,7 @@ public static List<NamedWriteableRegistry.Entry> getNamedWriteables() {
         entries.add(ToVersion.ENTRY);
         entries.add(Trim.ENTRY);
         entries.add(WildcardLike.ENTRY);
+        entries.add(Delay.ENTRY);
         entries.addAll(AbstractMultivalueFunction.getNamedWriteables());
         return entries;
     }

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

@@ -0,0 +1,121 @@
+/*
+ * 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.scalar.util;
+
+import org.elasticsearch.Build;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.compute.ann.Evaluator;
+import org.elasticsearch.compute.ann.Fixed;
+import org.elasticsearch.compute.operator.EvalOperator.ExpressionEvaluator;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.expression.Nullability;
+import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.evaluator.mapper.EvaluatorMapper;
+import org.elasticsearch.xpack.esql.expression.function.Param;
+import org.elasticsearch.xpack.esql.expression.function.scalar.UnaryScalarFunction;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.List;
+
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.FIRST;
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isType;
+
+/**
+ * Slowdown function - for debug purposes only.
+ * Syntax: WAIT(ms) - will sleep for ms milliseconds.
+ */
+public class Delay extends UnaryScalarFunction {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(Expression.class, "Delay", Delay::new);
+
+    public Delay(Source source, @Param(name = "ms", type = { "time_duration" }, description = "For how long") Expression ms) {
+        super(source, ms);
+    }
+
+    private Delay(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public Expression replaceChildren(List<Expression> newChildren) {
+        return new Delay(source(), newChildren.get(0));
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    public DataType dataType() {
+        return DataType.BOOLEAN;
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        if (childrenResolved() == false) {
+            return new TypeResolution("Unresolved children");
+        }
+
+        return isType(field(), t -> t == DataType.TIME_DURATION, sourceText(), FIRST, "time_duration");
+    }
+
+    @Override
+    public Nullability nullable() {
+        return Nullability.FALSE;
+    }
+
+    @Override
+    protected NodeInfo<? extends Expression> info() {
+        return NodeInfo.create(this, Delay::new, field());
+    }
+
+    @Override
+    public boolean foldable() {
+        return false;
+    }
+
+    @Override
+    public Object fold() {
+        return null;
+    }
+
+    private long msValue() {
+        if (field().foldable() == false) {
+            throw new IllegalArgumentException("function [" + sourceText() + "] has invalid argument [" + field().sourceText() + "]");
+        }
+        var ms = field().fold();
+        if (ms instanceof Duration duration) {
+            return duration.toMillis();
+        }
+        return ((Number) ms).longValue();
+    }
+
+    @Override
+    public ExpressionEvaluator.Factory toEvaluator(EvaluatorMapper.ToEvaluator toEvaluator) {
+        return new DelayEvaluator.Factory(source(), msValue());
+    }
+
+    @Evaluator
+    static boolean process(@Fixed long ms) {
+        // Only activate in snapshot builds
+        if (Build.current().isSnapshot()) {
+            try {
+                Thread.sleep(ms);
+            } catch (InterruptedException e) {
+                return true;
+            }
+        } else {
+            throw new IllegalArgumentException("Delay function is only available in snapshot builds");
+        }
+        return true;
+    }
+}

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

@@ -91,7 +91,7 @@ setup:
   - match: {esql.functions.cos: $functions_cos}
   - gt: {esql.functions.to_long: $functions_to_long}
   - match: {esql.functions.coalesce: $functions_coalesce}
-  - length: {esql.functions: 118} # check the "sister" test below for a likely update to the same esql.functions length check
+  - length: {esql.functions: 119} # check the "sister" test below for a likely update to the same esql.functions length check
 
 ---
 "Basic ESQL usage output (telemetry) non-snapshot version":