[ES|QL] add reverse function (#113297) (#114163)

Adds a REVERSE string function

docs/changelog/113297.yaml

@@ -0,0 +1,5 @@
+pr: 113297
+summary: "[ES|QL] add reverse function"
+area: ES|QL
+type: enhancement
+issues: []

docs/reference/esql/functions/description/reverse.asciidoc

@@ -0,0 +1,5 @@
+// This is generated by ESQL's AbstractFunctionTestCase. Do no edit it. See ../README.md for how to regenerate it.
+
+*Description*
+
+Returns a new string representing the input string in reverse order.

docs/reference/esql/functions/examples/reverse.asciidoc

@@ -0,0 +1,22 @@
+// This is generated by ESQL's AbstractFunctionTestCase. Do no edit it. See ../README.md for how to regenerate it.
+
+*Examples*
+
+[source.merge.styled,esql]
+----
+include::{esql-specs}/string.csv-spec[tag=reverse]
+----
+[%header.monospaced.styled,format=dsv,separator=|]
+|===
+include::{esql-specs}/string.csv-spec[tag=reverse-result]
+|===
+`REVERSE` works with unicode, too! It keeps unicode grapheme clusters together during reversal.
+[source.merge.styled,esql]
+----
+include::{esql-specs}/string.csv-spec[tag=reverseEmoji]
+----
+[%header.monospaced.styled,format=dsv,separator=|]
+|===
+include::{esql-specs}/string.csv-spec[tag=reverseEmoji-result]
+|===
+

docs/reference/esql/functions/kibana/definition/reverse.json

@@ -0,0 +1,38 @@
+{
+  "comment" : "This is generated by ESQL's AbstractFunctionTestCase. Do no edit it. See ../README.md for how to regenerate it.",
+  "type" : "eval",
+  "name" : "reverse",
+  "description" : "Returns a new string representing the input string in reverse order.",
+  "signatures" : [
+    {
+      "params" : [
+        {
+          "name" : "str",
+          "type" : "keyword",
+          "optional" : false,
+          "description" : "String expression. If `null`, the function returns `null`."
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "keyword"
+    },
+    {
+      "params" : [
+        {
+          "name" : "str",
+          "type" : "text",
+          "optional" : false,
+          "description" : "String expression. If `null`, the function returns `null`."
+        }
+      ],
+      "variadic" : false,
+      "returnType" : "text"
+    }
+  ],
+  "examples" : [
+    "ROW message = \"Some Text\" | EVAL message_reversed = REVERSE(message);",
+    "ROW bending_arts = \"💧🪨🔥💨\" | EVAL bending_arts_reversed = REVERSE(bending_arts);"
+  ],
+  "preview" : false,
+  "snapshot_only" : false
+}

docs/reference/esql/functions/kibana/docs/reverse.md

@@ -0,0 +1,10 @@
+<!--
+This is generated by ESQL's AbstractFunctionTestCase. Do no edit it. See ../README.md for how to regenerate it.
+-->
+
+### REVERSE
+Returns a new string representing the input string in reverse order.
+
+```
+ROW message = "Some Text" | EVAL message_reversed = REVERSE(message);
+```

docs/reference/esql/functions/layout/reverse.asciidoc

@@ -0,0 +1,15 @@
+// This is generated by ESQL's AbstractFunctionTestCase. Do no edit it. See ../README.md for how to regenerate it.
+
+[discrete]
+[[esql-reverse]]
+=== `REVERSE`
+
+*Syntax*
+
+[.text-center]
+image::esql/functions/signature/reverse.svg[Embedded,opts=inline]
+
+include::../parameters/reverse.asciidoc[]
+include::../description/reverse.asciidoc[]
+include::../types/reverse.asciidoc[]
+include::../examples/reverse.asciidoc[]

docs/reference/esql/functions/parameters/reverse.asciidoc

@@ -0,0 +1,6 @@
+// This is generated by ESQL's AbstractFunctionTestCase. Do no edit it. See ../README.md for how to regenerate it.
+
+*Parameters*
+
+`str`::
+String expression. If `null`, the function returns `null`.

docs/reference/esql/functions/signature/reverse.svg

@@ -0,0 +1 @@
+<svg version="1.1" xmlns:xlink="http://www.w3.org/1999/xlink" xmlns="http://www.w3.org/2000/svg" width="264" height="46" viewbox="0 0 264 46"><defs><style type="text/css">#guide .c{fill:none;stroke:#222222;}#guide .k{fill:#000000;font-family:Roboto Mono,Sans-serif;font-size:20px;}#guide .s{fill:#e4f4ff;stroke:#222222;}#guide .syn{fill:#8D8D8D;font-family:Roboto Mono,Sans-serif;font-size:20px;}</style></defs><path class="c" d="M0 31h5m104 0h10m32 0h10m56 0h10m32 0h5"/><rect class="s" x="5" y="5" width="104" height="36"/><text class="k" x="15" y="31">REVERSE</text><rect class="s" x="119" y="5" width="32" height="36" rx="7"/><text class="syn" x="129" y="31">(</text><rect class="s" x="161" y="5" width="56" height="36" rx="7"/><text class="k" x="171" y="31">str</text><rect class="s" x="227" y="5" width="32" height="36" rx="7"/><text class="syn" x="237" y="31">)</text></svg>

docs/reference/esql/functions/string-functions.asciidoc

@@ -17,6 +17,7 @@
 * <<esql-ltrim>>
 * <<esql-repeat>>
 * <<esql-replace>>
+* <<esql-reverse>>
 * <<esql-right>>
 * <<esql-rtrim>>
 * <<esql-space>>
@@ -38,6 +39,7 @@ include::layout/locate.asciidoc[]
 include::layout/ltrim.asciidoc[]
 include::layout/repeat.asciidoc[]
 include::layout/replace.asciidoc[]
+include::layout/reverse.asciidoc[]
 include::layout/right.asciidoc[]
 include::layout/rtrim.asciidoc[]
 include::layout/space.asciidoc[]

docs/reference/esql/functions/types/reverse.asciidoc

@@ -0,0 +1,10 @@
+// This is generated by ESQL's AbstractFunctionTestCase. Do no edit it. See ../README.md for how to regenerate it.
+
+*Supported types*
+
+[%header.monospaced.styled,format=dsv,separator=|]
+|===
+str | result
+keyword | keyword
+text | text
+|===

server/src/main/java/org/elasticsearch/common/util/ArrayUtils.java

@@ -126,4 +126,20 @@ public class ArrayUtils {
             end--;
         }
     }
+
+    /**
+     * Reverse the {@code length} values on the array starting from {@code offset}.
+     */
+    public static void reverseArray(byte[] array, int offset, int length) {
+        int start = offset;
+        int end = offset + length;
+        while (start < end) {
+            final byte temp = array[start];
+            array[start] = array[end - 1];
+            array[end - 1] = temp;
+            start++;
+            end--;
+        }
+    }
+
 }

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

@@ -69,6 +69,7 @@ double pi()
 "double pow(base:double|integer|long|unsigned_long, exponent:double|integer|long|unsigned_long)"
 "keyword repeat(string:keyword|text, number:integer)"
 "keyword replace(string:keyword|text, regex:keyword|text, newString:keyword|text)"
+"keyword|text reverse(str:keyword|text)"
 "keyword right(string:keyword|text, length:integer)"
 "double|integer|long|unsigned_long round(number:double|integer|long|unsigned_long, ?decimals:integer)"
 "keyword|text rtrim(string:keyword|text)"
@@ -201,6 +202,7 @@ pi            |null                                |null
 pow           |[base, exponent]                    |["double|integer|long|unsigned_long", "double|integer|long|unsigned_long"]                                                        |["Numeric expression for the base. If `null`\, the function returns `null`.", "Numeric expression for the exponent. If `null`\, the function returns `null`."]
 repeat        |[string, number]                    |["keyword|text", integer]                                                                                                         |[String expression., Number times to repeat.]
 replace       |[string, regex, newString]          |["keyword|text", "keyword|text", "keyword|text"]                                                                                  |[String expression., Regular expression., Replacement string.]
+reverse       |str                                 |"keyword|text"                                                                                                                    |String expression. If `null`, the function returns `null`.
 right         |[string, length]                    |["keyword|text", integer]                                                                                                         |[The string from which to returns a substring., The number of characters to return.]
 round         |[number, decimals]                  |["double|integer|long|unsigned_long", integer]                                                                                    |["The numeric value to round. If `null`\, the function returns `null`.", "The number of decimal places to round to. Defaults to 0. If `null`\, the function returns `null`."]
 rtrim         |string                              |"keyword|text"                                                                                                                    |String expression. If `null`, the function returns `null`.
@@ -333,6 +335,7 @@ pi            |Returns {wikipedia}/Pi[Pi], the ratio of a circle's circumference
 pow           |Returns the value of `base` raised to the power of `exponent`.
 repeat        |Returns a string constructed by concatenating `string` with itself the specified `number` of times.
 replace       |The function substitutes in the string `str` any match of the regular expression `regex` with the replacement string `newStr`.
+reverse       |Returns a new string representing the input string in reverse order.
 right         |Return the substring that extracts 'length' chars from 'str' starting from the right.
 round         |Rounds a number to the specified number of decimal places. Defaults to 0, which returns the nearest integer. If the precision is a negative number, rounds to the number of digits left of the decimal point.
 rtrim         |Removes trailing whitespaces from a string.
@@ -467,6 +470,7 @@ pi            |double
 pow           |double                                                                                                                      |[false, false]              |false           |false
 repeat        |keyword                                                                                                                     |[false, false]              |false           |false
 replace       |keyword                                                                                                                     |[false, false, false]       |false           |false
+reverse       |"keyword|text"                                                                                                              |false                       |false           |false
 right         |keyword                                                                                                                     |[false, false]              |false           |false
 round         |"double|integer|long|unsigned_long"                                                                                         |[false, true]               |false           |false
 rtrim         |"keyword|text"                                                                                                              |false                       |false           |false
@@ -544,5 +548,5 @@ required_capability: meta
 meta functions |  stats  a = count(*), b = count(*), c = count(*) |  mv_expand c;
 
 a:long | b:long | c:long
-121    | 121    | 121
+122    | 122    | 122
 ;

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

@@ -1194,6 +1194,113 @@ a:keyword           | upper:keyword         | lower:keyword
 π/2 + a + B + Λ ºC  | Π/2 + A + B + Λ ºC    | π/2 + a + b + λ ºc
 ;
 
+reverse
+required_capability: fn_reverse
+from employees | sort emp_no | eval name_reversed = REVERSE(first_name) | keep emp_no, first_name, name_reversed | limit 1;
+
+emp_no:integer  | first_name:keyword    | name_reversed:keyword
+10001           | Georgi                | igroeG
+;
+
+reverseRow
+required_capability: fn_reverse
+// tag::reverse[]
+ROW message = "Some Text" | EVAL message_reversed = REVERSE(message);
+// end::reverse[]
+
+// tag::reverse-result[]
+message:keyword | message_reversed:keyword
+Some Text       | txeT emoS
+// end::reverse-result[]
+;
+
+reverseEmoji
+required_capability: fn_reverse
+// tag::reverseEmoji[]
+ROW bending_arts = "💧🪨🔥💨" | EVAL bending_arts_reversed = REVERSE(bending_arts);
+// end::reverseEmoji[]
+
+// tag::reverseEmoji-result[]
+bending_arts:keyword | bending_arts_reversed:keyword
+💧🪨🔥💨             | 💨🔥🪨💧
+// end::reverseEmoji-result[]
+;
+
+reverseEmoji2
+required_capability: fn_reverse
+ROW off_on_holiday = "🏠➡️🚌➡️✈️➡️🏝️" | EVAL back_home_again = REVERSE(off_on_holiday);
+
+off_on_holiday:keyword | back_home_again:keyword
+🏠➡️🚌➡️✈️➡️🏝️        | 🏝️➡️✈️➡️🚌➡️🏠
+;
+
+reverseGraphemeClusters
+required_capability: fn_reverse
+ROW message = "áéíóúàèìòùâêîôû😊👍🏽🎉💖कंठाी" | EVAL message_reversed = REVERSE(message);
+
+message:keyword                | message_reversed:keyword
+áéíóúàèìòùâêîôû😊👍🏽🎉💖कंठाी    | ठाीकं💖🎉👍🏽😊ûôîêâùòìèàúóíéá
+;
+
+reverseMultiValue
+required_capability: fn_reverse
+FROM employees | SORT emp_no | EVAL jobs_reversed = REVERSE(job_positions) | KEEP job*, emp_no | LIMIT 5;
+
+warning:Line 1:53: evaluation of [REVERSE(job_positions)] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 1:53: java.lang.IllegalArgumentException: single-value function encountered multi-value
+
+job_positions:keyword                                                   | jobs_reversed:keyword | emp_no:integer
+["Accountant",  "Senior Python Developer"]                              | null                  | 10001
+Senior Team Lead                                                        | daeL maeT roineS      | 10002
+null                                                                    | null                  | 10003
+[Head Human Resources, Reporting Analyst, Support Engineer, Tech Lead]  | null                  | 10004
+null                                                                    | null                  | 10005
+;
+
+reverseNested
+required_capability: fn_reverse
+FROM employees | SORT emp_no | EVAL name_reversed = REVERSE(REVERSE(first_name)), eq = name_reversed == first_name | KEEP first_name, name_reversed, eq, emp_no | LIMIT 5;
+
+first_name:keyword  | name_reversed:keyword | eq:boolean    | emp_no:integer
+Georgi              | Georgi                | true          | 10001
+Bezalel             | Bezalel               | true          | 10002
+Parto               | Parto                 | true          | 10003
+Chirstian           | Chirstian             | true          | 10004
+Kyoichi             | Kyoichi               | true          | 10005
+;
+
+reverseRowNull
+required_capability: fn_reverse
+ROW x = null | EVAL y = REVERSE(x);
+
+x:null      | y:null
+null        | null
+;
+
+
+reverseRowInlineCastWithNull
+required_capability: fn_reverse
+ROW x = 1 | EVAL y = REVERSE((null + 1)::string);
+
+x:integer   | y:string
+1           | null
+;
+
+reverseWithTextFields
+required_capability: fn_reverse
+FROM books 
+| EVAL title_reversed = REVERSE(title), author_reversed_twice = REVERSE(REVERSE(author)), eq = author_reversed_twice == author 
+| KEEP title, title_reversed, author, author_reversed_twice, eq, book_no
+| SORT book_no
+| WHERE book_no IN ("1211", "1463")
+| LIMIT 2;
+
+title:text                                      | title_reversed:text                           | author:text               | author_reversed_twice:text    | eq:boolean    | book_no:keyword
+The brothers Karamazov                          | vozamaraK srehtorb ehT                        | Fyodor Dostoevsky         | Fyodor Dostoevsky             | true          | 1211
+Realms of Tolkien: Images of Middle-earth       | htrae-elddiM fo segamI :neikloT fo smlaeR     | J. R. R. Tolkien          | J. R. R. Tolkien              | true          | 1463
+;
+
+
 values
 required_capability: agg_values
 

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

@@ -0,0 +1,111 @@
+// 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.string;
+
+import java.lang.IllegalArgumentException;
+import java.lang.Override;
+import java.lang.String;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.compute.data.Block;
+import org.elasticsearch.compute.data.BytesRefBlock;
+import org.elasticsearch.compute.data.BytesRefVector;
+import org.elasticsearch.compute.data.Page;
+import org.elasticsearch.compute.operator.DriverContext;
+import org.elasticsearch.compute.operator.EvalOperator;
+import org.elasticsearch.core.Releasables;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.expression.function.Warnings;
+
+/**
+ * {@link EvalOperator.ExpressionEvaluator} implementation for {@link Reverse}.
+ * This class is generated. Do not edit it.
+ */
+public final class ReverseEvaluator implements EvalOperator.ExpressionEvaluator {
+  private final Warnings warnings;
+
+  private final EvalOperator.ExpressionEvaluator val;
+
+  private final DriverContext driverContext;
+
+  public ReverseEvaluator(Source source, EvalOperator.ExpressionEvaluator val,
+      DriverContext driverContext) {
+    this.val = val;
+    this.driverContext = driverContext;
+    this.warnings = Warnings.createWarnings(driverContext.warningsMode(), source);
+  }
+
+  @Override
+  public Block eval(Page page) {
+    try (BytesRefBlock valBlock = (BytesRefBlock) val.eval(page)) {
+      BytesRefVector valVector = valBlock.asVector();
+      if (valVector == null) {
+        return eval(page.getPositionCount(), valBlock);
+      }
+      return eval(page.getPositionCount(), valVector).asBlock();
+    }
+  }
+
+  public BytesRefBlock eval(int positionCount, BytesRefBlock valBlock) {
+    try(BytesRefBlock.Builder result = driverContext.blockFactory().newBytesRefBlockBuilder(positionCount)) {
+      BytesRef valScratch = new BytesRef();
+      position: for (int p = 0; p < positionCount; p++) {
+        if (valBlock.isNull(p)) {
+          result.appendNull();
+          continue position;
+        }
+        if (valBlock.getValueCount(p) != 1) {
+          if (valBlock.getValueCount(p) > 1) {
+            warnings.registerException(new IllegalArgumentException("single-value function encountered multi-value"));
+          }
+          result.appendNull();
+          continue position;
+        }
+        result.appendBytesRef(Reverse.process(valBlock.getBytesRef(valBlock.getFirstValueIndex(p), valScratch)));
+      }
+      return result.build();
+    }
+  }
+
+  public BytesRefVector eval(int positionCount, BytesRefVector valVector) {
+    try(BytesRefVector.Builder result = driverContext.blockFactory().newBytesRefVectorBuilder(positionCount)) {
+      BytesRef valScratch = new BytesRef();
+      position: for (int p = 0; p < positionCount; p++) {
+        result.appendBytesRef(Reverse.process(valVector.getBytesRef(p, valScratch)));
+      }
+      return result.build();
+    }
+  }
+
+  @Override
+  public String toString() {
+    return "ReverseEvaluator[" + "val=" + val + "]";
+  }
+
+  @Override
+  public void close() {
+    Releasables.closeExpectNoException(val);
+  }
+
+  static class Factory implements EvalOperator.ExpressionEvaluator.Factory {
+    private final Source source;
+
+    private final EvalOperator.ExpressionEvaluator.Factory val;
+
+    public Factory(Source source, EvalOperator.ExpressionEvaluator.Factory val) {
+      this.source = source;
+      this.val = val;
+    }
+
+    @Override
+    public ReverseEvaluator get(DriverContext context) {
+      return new ReverseEvaluator(source, val.get(context), context);
+    }
+
+    @Override
+    public String toString() {
+      return "ReverseEvaluator[" + "val=" + val + "]";
+    }
+  }
+}

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

@@ -27,6 +27,11 @@ import java.util.Set;
  */
 public class EsqlCapabilities {
     public enum Cap {
+        /**
+         * Support for function {@code REVERSE}.
+         */
+        FN_REVERSE,
+
         /**
          * Support for function {@code CBRT}. Done in #108574.
          */

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

@@ -124,6 +124,7 @@ import org.elasticsearch.xpack.esql.expression.function.scalar.string.Locate;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.RTrim;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Repeat;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Replace;
+import org.elasticsearch.xpack.esql.expression.function.scalar.string.Reverse;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Right;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Space;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Split;
@@ -300,22 +301,23 @@ public class EsqlFunctionRegistry {
                 def(Tau.class, Tau::new, "tau") },
             // string
             new FunctionDefinition[] {
-                def(Length.class, Length::new, "length"),
-                def(Substring.class, Substring::new, "substring"),
                 def(Concat.class, Concat::new, "concat"),
+                def(EndsWith.class, EndsWith::new, "ends_with"),
                 def(LTrim.class, LTrim::new, "ltrim"),
-                def(RTrim.class, RTrim::new, "rtrim"),
-                def(Trim.class, Trim::new, "trim"),
                 def(Left.class, Left::new, "left"),
+                def(Length.class, Length::new, "length"),
+                def(Locate.class, Locate::new, "locate"),
+                def(RTrim.class, RTrim::new, "rtrim"),
+                def(Repeat.class, Repeat::new, "repeat"),
                 def(Replace.class, Replace::new, "replace"),
+                def(Reverse.class, Reverse::new, "reverse"),
                 def(Right.class, Right::new, "right"),
+                def(Space.class, Space::new, "space"),
                 def(StartsWith.class, StartsWith::new, "starts_with"),
-                def(EndsWith.class, EndsWith::new, "ends_with"),
+                def(Substring.class, Substring::new, "substring"),
                 def(ToLower.class, ToLower::new, "to_lower"),
                 def(ToUpper.class, ToUpper::new, "to_upper"),
-                def(Locate.class, Locate::new, "locate"),
-                def(Repeat.class, Repeat::new, "repeat"),
-                def(Space.class, Space::new, "space") },
+                def(Trim.class, Trim::new, "trim") },
             // date
             new FunctionDefinition[] {
                 def(DateDiff.class, DateDiff::new, "date_diff"),

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

@@ -43,6 +43,7 @@ import org.elasticsearch.xpack.esql.expression.function.scalar.string.Left;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Locate;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Repeat;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Replace;
+import org.elasticsearch.xpack.esql.expression.function.scalar.string.Reverse;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Right;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.Split;
 import org.elasticsearch.xpack.esql.expression.function.scalar.string.StartsWith;
@@ -100,6 +101,7 @@ public abstract class EsqlScalarFunction extends ScalarFunction implements Evalu
         entries.add(Right.ENTRY);
         entries.add(Repeat.ENTRY);
         entries.add(Replace.ENTRY);
+        entries.add(Reverse.ENTRY);
         entries.add(Round.ENTRY);
         entries.add(Split.ENTRY);
         entries.add(Substring.ENTRY);

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

@@ -151,6 +151,8 @@
  *              <li>{@code docs/reference/esql/functions/parameters/myfunction.asciidoc}</li>
  *              <li>{@code docs/reference/esql/functions/signature/myfunction.svg}</li>
  *              <li>{@code docs/reference/esql/functions/types/myfunction.asciidoc}</li>
+ *              <li>{@code docs/reference/esql/functions/kibana/definition/myfunction.json}</li>
+ *              <li>{@code docs/reference/esql/functions/kibana/docs/myfunction.asciidoc}</li>
  *         </ul>
  *
  *         Make sure to commit them. Add a reference to the
@@ -194,6 +196,9 @@
  *         for your function. Now add something like {@code required_capability: my_function}
  *         to all of your csv-spec tests. Run those csv-spec tests as integration tests to double
  *         check that they run on the main branch.
+ *         <br><br>
+ *         **Note:** you may notice tests gated based on Elasticsearch version. This was the old way
+ *         of doing things. Now, we use specific capabilities for each function.
  *     </li>
  *     <li>
  *         Open the PR. The subject and description of the PR are important because those'll turn
@@ -201,7 +206,7 @@
  *         happy. But functions don't need an essay.
  *     </li>
  *     <li>
- *         Add the {@code >enhancement} and {@code :Query Languages/ES|QL} tags if you are able.
+ *         Add the {@code >enhancement} and {@code :Analytics/ES|QL} tags if you are able.
  *         Request a review if you can, probably from one of the folks that github proposes to you.
  *     </li>
  *     <li>

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

@@ -0,0 +1,140 @@
+/*
+ * 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.string;
+
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.lucene.BytesRefs;
+import org.elasticsearch.compute.ann.Evaluator;
+import org.elasticsearch.compute.operator.EvalOperator.ExpressionEvaluator;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.expression.function.Example;
+import org.elasticsearch.xpack.esql.expression.function.FunctionInfo;
+import org.elasticsearch.xpack.esql.expression.function.Param;
+import org.elasticsearch.xpack.esql.expression.function.scalar.UnaryScalarFunction;
+
+import java.io.IOException;
+import java.text.BreakIterator;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+
+import static org.elasticsearch.common.util.ArrayUtils.reverseArray;
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.DEFAULT;
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isString;
+
+/**
+ * Function that reverses a string.
+ */
+public class Reverse extends UnaryScalarFunction {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(Expression.class, "Reverse", Reverse::new);
+
+    @FunctionInfo(
+        returnType = { "keyword", "text" },
+        description = "Returns a new string representing the input string in reverse order.",
+        examples = {
+            @Example(file = "string", tag = "reverse"),
+            @Example(
+                file = "string",
+                tag = "reverseEmoji",
+                description = "`REVERSE` works with unicode, too! It keeps unicode grapheme clusters together during reversal."
+            ) }
+    )
+    public Reverse(
+        Source source,
+        @Param(
+            name = "str",
+            type = { "keyword", "text" },
+            description = "String expression. If `null`, the function returns `null`."
+        ) Expression field
+    ) {
+        super(source, field);
+    }
+
+    private Reverse(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        if (childrenResolved() == false) {
+            return new TypeResolution("Unresolved children");
+        }
+
+        return isString(field, sourceText(), DEFAULT);
+    }
+
+    /**
+     * Reverses a unicode string, keeping grapheme clusters together
+     * @param str
+     * @return
+     */
+    public static String reverseStringWithUnicodeCharacters(String str) {
+        BreakIterator boundary = BreakIterator.getCharacterInstance(Locale.ROOT);
+        boundary.setText(str);
+
+        List<String> characters = new ArrayList<>();
+        int start = boundary.first();
+        for (int end = boundary.next(); end != BreakIterator.DONE; start = end, end = boundary.next()) {
+            characters.add(str.substring(start, end));
+        }
+
+        StringBuilder reversed = new StringBuilder(str.length());
+        for (int i = characters.size() - 1; i >= 0; i--) {
+            reversed.append(characters.get(i));
+        }
+
+        return reversed.toString();
+    }
+
+    private static boolean isOneByteUTF8(BytesRef ref) {
+        int end = ref.offset + ref.length;
+        for (int i = ref.offset; i < end; i++) {
+            if (ref.bytes[i] < 0) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Evaluator
+    static BytesRef process(BytesRef val) {
+        if (isOneByteUTF8(val)) {
+            // this is the fast path. we know we can just reverse the bytes.
+            BytesRef reversed = BytesRef.deepCopyOf(val);
+            reverseArray(reversed.bytes, reversed.offset, reversed.length);
+            return reversed;
+        }
+        return BytesRefs.toBytesRef(reverseStringWithUnicodeCharacters(val.utf8ToString()));
+    }
+
+    @Override
+    public ExpressionEvaluator.Factory toEvaluator(ToEvaluator toEvaluator) {
+        var fieldEvaluator = toEvaluator.apply(field);
+        return new ReverseEvaluator.Factory(source(), fieldEvaluator);
+    }
+
+    @Override
+    public Expression replaceChildren(List<Expression> newChildren) {
+        assert newChildren.size() == 1;
+        return new Reverse(source(), newChildren.get(0));
+    }
+
+    @Override
+    protected NodeInfo<? extends Expression> info() {
+        return NodeInfo.create(this, Reverse::new, field);
+    }
+}

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ReverseSerializationTests.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.function.scalar.string;
+
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.expression.AbstractUnaryScalarSerializationTests;
+
+public class ReverseSerializationTests extends AbstractUnaryScalarSerializationTests<Reverse> {
+    @Override
+    protected Reverse create(Source source, Expression child) {
+        return new Reverse(source, child);
+    }
+}

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ReverseTests.java

@@ -0,0 +1,65 @@
+/*
+ * 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.string;
+
+import com.carrotsearch.randomizedtesting.annotations.Name;
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.lucene.BytesRefs;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.expression.function.AbstractScalarFunctionTestCase;
+import org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class ReverseTests extends AbstractScalarFunctionTestCase {
+    public ReverseTests(@Name("TestCase") Supplier<TestCaseSupplier.TestCase> testCaseSupplier) {
+        this.testCase = testCaseSupplier.get();
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() {
+        List<TestCaseSupplier> suppliers = new ArrayList<>();
+
+        for (DataType stringType : new DataType[] { DataType.KEYWORD, DataType.TEXT }) {
+            for (var supplier : TestCaseSupplier.stringCases(stringType)) {
+                suppliers.add(makeSupplier(supplier));
+            }
+        }
+
+        return parameterSuppliersFromTypedData(suppliers);
+    }
+
+    @Override
+    protected Expression build(Source source, List<Expression> args) {
+        return new Reverse(source, args.get(0));
+    }
+
+    private static TestCaseSupplier makeSupplier(TestCaseSupplier.TypedDataSupplier fieldSupplier) {
+        return new TestCaseSupplier(fieldSupplier.name(), List.of(fieldSupplier.type()), () -> {
+            var fieldTypedData = fieldSupplier.get();
+            String expectedToString = "ReverseEvaluator[val=Attribute[channel=0]]";
+            String value = BytesRefs.toString(fieldTypedData.data());
+            String expectedValue = Reverse.reverseStringWithUnicodeCharacters(value);
+
+            return new TestCaseSupplier.TestCase(
+                List.of(fieldTypedData),
+                expectedToString,
+                fieldSupplier.type(),
+                equalTo(new BytesRef(expectedValue))
+            );
+        });
+    }
+}

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/80_text.yml

@@ -3,7 +3,7 @@ setup:
   - requires:
       cluster_features: ["gte_v8.11.0"]
       reason: "ESQL is available in 8.11+"
-      test_runner_features: allowed_warnings_regex
+      test_runner_features: [allowed_warnings_regex, capabilities]
 
   - do:
       indices.create:
@@ -385,8 +385,31 @@ setup:
   - length: { values: 2 }
   - match: { values.0: [ [ "foo", "bar" ] ] }
   - match: { values.1: [ "baz" ] }
+---
+"reverse text":
+  - requires:
+      capabilities:
+        - method: POST
+          path: /_query
+          parameters: [method, path, parameters, capabilities]
+          capabilities: [fn_reverse]
+      reason: "reverse not yet added"
+  - do:
+      allowed_warnings_regex:
+        - "No limit defined, adding default limit of \\[.*\\]"
+      esql.query:
+        body:
+          query: 'FROM test | SORT name | EVAL job_reversed = REVERSE(job), tag_reversed = REVERSE(tag) | KEEP job_reversed, tag_reversed'
+
+  - match: { columns.0.name: "job_reversed" }
+  - match: { columns.0.type: "text" }
 
+  - match: { columns.1.name: "tag_reversed" }
+  - match: { columns.1.type: "text" }
 
+  - length: { values: 2 }
+  - match: { values.0: [ "rotceriD TI", "rab oof" ] }
+  - match: { values.1: [ "tsilaicepS lloryaP", "zab" ] }
 ---
 "stats text with raw":
   - do: