Add "REVERSE" / "REPEAT" / "RIGHT" / "LEFT" functions (#7334)

* Add "REVERSE" / "REPEAT" / "RIGHT" / "LEFT" functions

* Fix ImportOrder

* Use RuntimeException instead of OutOfMemoryError according to "Effective Java"

* Simplify

* Patch suggestions

core/src/main/java/org/apache/druid/java/util/common/StringUtils.java

@@ -28,6 +28,7 @@ import java.net.URLEncoder;
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.Base64;
 import java.util.IllegalFormatException;
 import java.util.Locale;
@@ -360,4 +361,54 @@ public class StringUtils
   {
     return BASE64_DECODER.decode(input);
   }
+
+  /**
+   * Returns a string whose value is the concatenation of the
+   * string {@code s} repeated {@code count} times.
+   * <p>
+   * If count or length is zero then the empty string is returned.
+   * <p>
+   * This method may be used to create space padding for
+   * formatting text or zero padding for formatting numbers.
+   *
+   * @param count number of times to repeat
+   *
+   * @return A string composed of this string repeated
+   * {@code count} times or the empty string if count
+   * or length is zero.
+   *
+   * @throws IllegalArgumentException if the {@code count} is negative.
+   * @link https://bugs.openjdk.java.net/browse/JDK-8197594
+   */
+  public static String repeat(String s, int count)
+  {
+    if (count < 0) {
+      throw new IllegalArgumentException("count is negative, " + count);
+    }
+    if (count == 1) {
+      return s;
+    }
+    byte[] value = s.getBytes(StandardCharsets.UTF_8);
+    final int len = value.length;
+    if (len == 0 || count == 0) {
+      return "";
+    }
+    if (len == 1) {
+      final byte[] single = new byte[count];
+      Arrays.fill(single, value[0]);
+      return new String(single, StandardCharsets.UTF_8);
+    }
+    if (Integer.MAX_VALUE / count < len) {
+      throw new RuntimeException("The produced string is too large.");
+    }
+    final int limit = len * count;
+    final byte[] multiple = new byte[limit];
+    System.arraycopy(value, 0, multiple, 0, len);
+    int copied = len;
+    for (; copied < limit - copied; copied <<= 1) {
+      System.arraycopy(multiple, 0, multiple, copied, copied);
+    }
+    System.arraycopy(multiple, 0, multiple, copied, limit - copied);
+    return new String(multiple, StandardCharsets.UTF_8);
+  }
 }

core/src/main/java/org/apache/druid/math/expr/Function.java

@@ -122,6 +122,23 @@ interface Function
     }
   }
 
+  abstract class DoubleParamString extends DoubleParam
+  {
+    @Override
+    protected final ExprEval eval(ExprEval x, ExprEval y)
+    {
+      if (x.type() != ExprType.STRING || y.type() != ExprType.LONG) {
+        throw new IAE(
+            "Function[%s] needs a string as first argument and an integer as second argument",
+            name()
+        );
+      }
+      return eval(x.asString(), y.asInt());
+    }
+
+    protected abstract ExprEval eval(String x, int y);
+  }
+
   class ParseLong implements Function
   {
     @Override
@@ -326,7 +343,6 @@ interface Function
     }
   }
 
-
   class Div extends DoubleParamMath
   {
     @Override
@@ -1126,6 +1142,49 @@ interface Function
     }
   }
 
+  class RightFunc extends DoubleParamString
+  {
+    @Override
+    public String name()
+    {
+      return "right";
+    }
+
+    @Override
+    protected ExprEval eval(String x, int y)
+    {
+      if (y < 0) {
+        throw new IAE(
+            "Function[%s] needs a postive integer as second argument",
+            name()
+        );
+      }
+      int len = x.length();
+      return ExprEval.of(y < len ? x.substring(len - y) : x);
+    }
+  }
+
+  class LeftFunc extends DoubleParamString
+  {
+    @Override
+    public String name()
+    {
+      return "left";
+    }
+
+    @Override
+    protected ExprEval eval(String x, int y)
+    {
+      if (y < 0) {
+        throw new IAE(
+            "Function[%s] needs a postive integer as second argument",
+            name()
+        );
+      }
+      return ExprEval.of(y < x.length() ? x.substring(0, y) : x);
+    }
+  }
+
   class ReplaceFunc implements Function
   {
     @Override
@@ -1197,6 +1256,43 @@ interface Function
     }
   }
 
+  class ReverseFunc extends SingleParam
+  {
+    @Override
+    public String name()
+    {
+      return "reverse";
+    }
+
+    @Override
+    protected ExprEval eval(ExprEval param)
+    {
+      if (param.type() != ExprType.STRING) {
+        throw new IAE(
+            "Function[%s] needs a string argument",
+            name()
+        );
+      }
+      final String arg = param.asString();
+      return ExprEval.of(arg == null ? NullHandling.defaultStringValue() : new StringBuilder(arg).reverse().toString());
+    }
+  }
+
+  class RepeatFunc extends DoubleParamString
+  {
+    @Override
+    public String name()
+    {
+      return "repeat";
+    }
+
+    @Override
+    protected ExprEval eval(String x, int y)
+    {
+      return ExprEval.of(y < 1 ? NullHandling.defaultStringValue() : StringUtils.repeat(x, y));
+    }
+  }
+
   class IsNullFunc implements Function
   {
     @Override

core/src/test/java/org/apache/druid/java/util/common/StringUtilsTest.java

@@ -20,7 +20,9 @@
 package org.apache.druid.java.util.common;
 
 import org.junit.Assert;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 import java.io.UnsupportedEncodingException;
 import java.nio.BufferUnderflowException;
@@ -31,6 +33,9 @@ import java.nio.ByteBuffer;
  */
 public class StringUtilsTest
 {
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
   @Test
   public void fromUtf8ConversionTest() throws UnsupportedEncodingException
   {
@@ -160,4 +165,20 @@ public class StringUtilsTest
     Assert.assertEquals(s2, "fff%2Bggg");
     Assert.assertEquals("fff+ggg", StringUtils.urlDecode(s2));
   }
+
+  @Test
+  public void testRepeat()
+  {
+    Assert.assertEquals("", StringUtils.repeat("foo", 0));
+    Assert.assertEquals("foo", StringUtils.repeat("foo", 1));
+    Assert.assertEquals("foofoofoo", StringUtils.repeat("foo", 3));
+
+    Assert.assertEquals("", StringUtils.repeat("", 0));
+    Assert.assertEquals("", StringUtils.repeat("", 1));
+    Assert.assertEquals("", StringUtils.repeat("", 3));
+
+    expectedException.expect(IllegalArgumentException.class);
+    expectedException.expectMessage("count is negative, -1");
+    Assert.assertEquals("", StringUtils.repeat("foo", -1));
+  }
 }

docs/content/misc/math-expr.md

@@ -74,6 +74,8 @@ The following built-in functions are available.
 |regexp_extract|regexp_extract(expr, pattern[, index]) applies a regular expression pattern and extracts a capture group index, or null if there is no match. If index is unspecified or zero, returns the substring that matched the pattern.|
 |replace|replace(expr, pattern, replacement) replaces pattern with replacement|
 |substring|substring(expr, index, length) behaves like java.lang.String's substring|
+|right|right(expr, length) returns the rightmost length characters from a string|
+|left|left(expr, length) returns the leftmost length characters from a string|
 |strlen|strlen(expr) returns length of a string in UTF-16 code units|
 |strpos|strpos(haystack, needle[, fromIndex]) returns the position of the needle within the haystack, with indexes starting from 0. The search will begin at fromIndex, or 0 if fromIndex is not specified. If the needle is not found then the function returns -1.|
 |trim|trim(expr[, chars]) remove leading and trailing characters from `expr` if they are present in `chars`. `chars` defaults to ' ' (space) if not provided.|
@@ -81,6 +83,8 @@ The following built-in functions are available.
 |rtrim|rtrim(expr[, chars]) remove trailing characters from `expr` if they are present in `chars`. `chars` defaults to ' ' (space) if not provided.|
 |lower|lower(expr) converts a string to lowercase|
 |upper|upper(expr) converts a string to uppercase|
+|reverse|reverse(expr) reverses a string|
+|repeat|repeat(expr, N) repeats a string N times|
 
 ## Time functions
 

docs/content/querying/sql.md

@@ -187,12 +187,16 @@ String functions accept strings, and return a type appropriate to the function.
 |`REPLACE(expr, pattern, replacement)`|Replaces pattern with replacement in expr, and returns the result.|
 |`STRPOS(haystack, needle)`|Returns the index of needle within haystack, with indexes starting from 1. If the needle is not found, returns 0.|
 |`SUBSTRING(expr, index, [length])`|Returns a substring of expr starting at index, with a max length, both measured in UTF-16 code units.|
+|`RIGHT(expr, [length])`|Returns the rightmost length characters from expr.|
+|`LEFT(expr, [length])`|Returns the leftmost length characters from expr.|
 |`SUBSTR(expr, index, [length])`|Synonym for SUBSTRING.|
 |`TRIM([BOTH \| LEADING \| TRAILING] [<chars> FROM] expr)`|Returns expr with characters removed from the leading, trailing, or both ends of "expr" if they are in "chars". If "chars" is not provided, it defaults to " " (a space). If the directional argument is not provided, it defaults to "BOTH".|
 |`BTRIM(expr[, chars])`|Alternate form of `TRIM(BOTH <chars> FROM <expr>`).|
 |`LTRIM(expr[, chars])`|Alternate form of `TRIM(LEADING <chars> FROM <expr>`).|
 |`RTRIM(expr[, chars])`|Alternate form of `TRIM(TRAILING <chars> FROM <expr>`).|
 |`UPPER(expr)`|Returns expr in all uppercase.|
+|`REVERSE(expr)`|Reverses expr.|
+|`REPEAT(expr, [N])`|Repeats expr N times|
 
 ### Time functions
 

sql/src/main/java/org/apache/druid/sql/calcite/expression/builtin/LeftOperatorConversion.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.sql.calcite.expression.builtin;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.type.SqlTypeFamily;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.druid.sql.calcite.expression.DruidExpression;
+import org.apache.druid.sql.calcite.expression.Expressions;
+import org.apache.druid.sql.calcite.expression.OperatorConversions;
+import org.apache.druid.sql.calcite.expression.SqlOperatorConversion;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.table.RowSignature;
+
+public class LeftOperatorConversion implements SqlOperatorConversion
+{
+  private static final SqlFunction SQL_FUNCTION = OperatorConversions
+      .operatorBuilder("LEFT")
+      .operandTypes(SqlTypeFamily.CHARACTER, SqlTypeFamily.INTEGER)
+      .functionCategory(SqlFunctionCategory.STRING)
+      .returnType(SqlTypeName.VARCHAR)
+      .build();
+
+  @Override
+  public SqlOperator calciteOperator()
+  {
+    return SQL_FUNCTION;
+  }
+
+  @Override
+  public DruidExpression toDruidExpression(
+      final PlannerContext plannerContext,
+      final RowSignature rowSignature,
+      final RexNode rexNode
+  )
+  {
+    final RexCall call = (RexCall) rexNode;
+    final DruidExpression input = Expressions.toDruidExpression(
+        plannerContext,
+        rowSignature,
+        call.getOperands().get(0)
+    );
+    if (input == null) {
+      return null;
+    }
+    if (call.getOperands().size() != 2) {
+      return null;
+    }
+    return OperatorConversions.convertCall(
+        plannerContext,
+        rowSignature,
+        rexNode,
+        druidExpressions -> DruidExpression.of(
+            null,
+            DruidExpression.functionCall("left", druidExpressions)
+        )
+    );
+  }
+}

sql/src/main/java/org/apache/druid/sql/calcite/expression/builtin/RepeatOperatorConversion.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.sql.calcite.expression.builtin;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.type.SqlTypeFamily;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.druid.sql.calcite.expression.DruidExpression;
+import org.apache.druid.sql.calcite.expression.Expressions;
+import org.apache.druid.sql.calcite.expression.OperatorConversions;
+import org.apache.druid.sql.calcite.expression.SqlOperatorConversion;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.table.RowSignature;
+
+public class RepeatOperatorConversion implements SqlOperatorConversion
+{
+  private static final SqlFunction SQL_FUNCTION = OperatorConversions
+      .operatorBuilder("REPEAT")
+      .operandTypes(SqlTypeFamily.CHARACTER, SqlTypeFamily.INTEGER)
+      .functionCategory(SqlFunctionCategory.STRING)
+      .returnType(SqlTypeName.VARCHAR)
+      .build();
+
+  @Override
+  public SqlOperator calciteOperator()
+  {
+    return SQL_FUNCTION;
+  }
+
+  @Override
+  public DruidExpression toDruidExpression(
+      final PlannerContext plannerContext,
+      final RowSignature rowSignature,
+      final RexNode rexNode
+  )
+  {
+    final RexCall call = (RexCall) rexNode;
+    final DruidExpression input = Expressions.toDruidExpression(
+        plannerContext,
+        rowSignature,
+        call.getOperands().get(0)
+    );
+    if (input == null) {
+      return null;
+    }
+    if (call.getOperands().size() != 2) {
+      return null;
+    }
+    return OperatorConversions.convertCall(
+        plannerContext,
+        rowSignature,
+        rexNode,
+        druidExpressions -> DruidExpression.of(
+            null,
+            DruidExpression.functionCall("repeat", druidExpressions)
+        )
+    );
+  }
+}

sql/src/main/java/org/apache/druid/sql/calcite/expression/builtin/ReverseOperatorConversion.java

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.sql.calcite.expression.builtin;
+
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.type.SqlTypeFamily;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.druid.sql.calcite.expression.DruidExpression;
+import org.apache.druid.sql.calcite.expression.OperatorConversions;
+import org.apache.druid.sql.calcite.expression.SqlOperatorConversion;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.table.RowSignature;
+
+public class ReverseOperatorConversion implements SqlOperatorConversion
+{
+  private static final SqlFunction SQL_FUNCTION = OperatorConversions
+      .operatorBuilder("REVERSE")
+      .operandTypes(SqlTypeFamily.CHARACTER)
+      .functionCategory(SqlFunctionCategory.STRING)
+      .returnType(SqlTypeName.VARCHAR)
+      .build();
+
+  @Override
+  public SqlOperator calciteOperator()
+  {
+    return SQL_FUNCTION;
+  }
+
+  @Override
+  public DruidExpression toDruidExpression(
+      final PlannerContext plannerContext,
+      final RowSignature rowSignature,
+      final RexNode rexNode
+  )
+  {
+    return OperatorConversions.convertCall(
+        plannerContext,
+        rowSignature,
+        rexNode,
+        druidExpressions -> DruidExpression.of(
+            null,
+            DruidExpression.functionCall("reverse", druidExpressions)
+        )
+    );
+  }
+}

sql/src/main/java/org/apache/druid/sql/calcite/expression/builtin/RightOperatorConversion.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.sql.calcite.expression.builtin;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.type.SqlTypeFamily;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.druid.sql.calcite.expression.DruidExpression;
+import org.apache.druid.sql.calcite.expression.Expressions;
+import org.apache.druid.sql.calcite.expression.OperatorConversions;
+import org.apache.druid.sql.calcite.expression.SqlOperatorConversion;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.table.RowSignature;
+
+public class RightOperatorConversion implements SqlOperatorConversion
+{
+  private static final SqlFunction SQL_FUNCTION = OperatorConversions
+      .operatorBuilder("RIGHT")
+      .operandTypes(SqlTypeFamily.CHARACTER, SqlTypeFamily.INTEGER)
+      .functionCategory(SqlFunctionCategory.STRING)
+      .returnType(SqlTypeName.VARCHAR)
+      .build();
+
+  @Override
+  public SqlOperator calciteOperator()
+  {
+    return SQL_FUNCTION;
+  }
+
+  @Override
+  public DruidExpression toDruidExpression(
+      final PlannerContext plannerContext,
+      final RowSignature rowSignature,
+      final RexNode rexNode
+  )
+  {
+    final RexCall call = (RexCall) rexNode;
+    final DruidExpression input = Expressions.toDruidExpression(
+        plannerContext,
+        rowSignature,
+        call.getOperands().get(0)
+    );
+    if (input == null) {
+      return null;
+    }
+    if (call.getOperands().size() != 2) {
+      return null;
+    }
+    return OperatorConversions.convertCall(
+        plannerContext,
+        rowSignature,
+        rexNode,
+        druidExpressions -> DruidExpression.of(
+            null,
+            DruidExpression.functionCall("right", druidExpressions)
+        )
+    );
+  }
+}

sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidOperatorTable.java

@@ -54,6 +54,7 @@ import org.apache.druid.sql.calcite.expression.builtin.DateTruncOperatorConversi
 import org.apache.druid.sql.calcite.expression.builtin.ExtractOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.FloorOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.LTrimOperatorConversion;
+import org.apache.druid.sql.calcite.expression.builtin.LeftOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.LikeOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.MillisToTimestampOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.ParseLongOperatorConversion;
@@ -61,6 +62,9 @@ import org.apache.druid.sql.calcite.expression.builtin.PositionOperatorConversio
 import org.apache.druid.sql.calcite.expression.builtin.RTrimOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.RegexpExtractOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.ReinterpretOperatorConversion;
+import org.apache.druid.sql.calcite.expression.builtin.RepeatOperatorConversion;
+import org.apache.druid.sql.calcite.expression.builtin.ReverseOperatorConversion;
+import org.apache.druid.sql.calcite.expression.builtin.RightOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.StringFormatOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.StrposOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.SubstringOperatorConversion;
@@ -180,6 +184,10 @@ public class DruidOperatorTable implements SqlOperatorTable
           .add(new StringFormatOperatorConversion())
           .add(new StrposOperatorConversion())
           .add(new SubstringOperatorConversion())
+          .add(new RightOperatorConversion())
+          .add(new LeftOperatorConversion())
+          .add(new ReverseOperatorConversion())
+          .add(new RepeatOperatorConversion())
           .add(new AliasedOperatorConversion(new SubstringOperatorConversion(), "SUBSTR"))
           .add(new ConcatOperatorConversion())
           .add(new TextcatOperatorConversion())

sql/src/test/java/org/apache/druid/sql/calcite/expression/ExpressionsTest.java

@@ -36,13 +36,18 @@ import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.IAE;
 import org.apache.druid.math.expr.ExprEval;
 import org.apache.druid.math.expr.Parser;
 import org.apache.druid.query.extraction.RegexDimExtractionFn;
 import org.apache.druid.segment.column.ValueType;
 import org.apache.druid.sql.calcite.expression.builtin.DateTruncOperatorConversion;
+import org.apache.druid.sql.calcite.expression.builtin.LeftOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.ParseLongOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.RegexpExtractOperatorConversion;
+import org.apache.druid.sql.calcite.expression.builtin.RepeatOperatorConversion;
+import org.apache.druid.sql.calcite.expression.builtin.ReverseOperatorConversion;
+import org.apache.druid.sql.calcite.expression.builtin.RightOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.StringFormatOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.StrposOperatorConversion;
 import org.apache.druid.sql.calcite.expression.builtin.TimeExtractOperatorConversion;
@@ -61,7 +66,9 @@ import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.joda.time.Period;
 import org.junit.Assert;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 import java.math.BigDecimal;
 import java.util.Map;
@@ -69,6 +76,9 @@ import java.util.Map;
 public class ExpressionsTest extends CalciteTestBase
 {
 
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
   private final PlannerContext plannerContext = PlannerContext.create(
       CalciteTests.createOperatorTable(),
       CalciteTests.createExprMacroTable(),
@@ -895,6 +905,292 @@ public class ExpressionsTest extends CalciteTestBase
     );
   }
 
+  @Test
+  public void testReverse()
+  {
+    testExpression(
+        rexBuilder.makeCall(
+            new ReverseOperatorConversion().calciteOperator(),
+            inputRef("s")
+        ),
+        DruidExpression.fromExpression("reverse(\"s\")"),
+        "oof"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new ReverseOperatorConversion().calciteOperator(),
+            inputRef("spacey")
+        ),
+        DruidExpression.fromExpression("reverse(\"spacey\")"),
+        "  ereht yeh  "
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new ReverseOperatorConversion().calciteOperator(),
+            inputRef("tstr")
+        ),
+        DruidExpression.fromExpression("reverse(\"tstr\")"),
+        "60:50:40 30-20-0002"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new ReverseOperatorConversion().calciteOperator(),
+            inputRef("dstr")
+        ),
+        DruidExpression.fromExpression("reverse(\"dstr\")"),
+        "30-20-0002"
+    );
+  }
+
+  @Test
+  public void testAbnormalReverseWithWrongType()
+  {
+    expectedException.expect(IAE.class);
+    expectedException.expectMessage("Function[reverse] needs a string argument");
+
+    testExpression(
+        rexBuilder.makeCall(
+            new ReverseOperatorConversion().calciteOperator(),
+            inputRef("a")
+        ),
+        DruidExpression.fromExpression("reverse(\"a\")"),
+        null
+    );
+  }
+
+  @Test
+  public void testRight()
+  {
+    testExpression(
+        rexBuilder.makeCall(
+            new RightOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(1)
+        ),
+        DruidExpression.fromExpression("right(\"s\",1)"),
+        "o"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new RightOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(2)
+        ),
+        DruidExpression.fromExpression("right(\"s\",2)"),
+        "oo"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new RightOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(3)
+        ),
+        DruidExpression.fromExpression("right(\"s\",3)"),
+        "foo"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new RightOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(4)
+        ),
+        DruidExpression.fromExpression("right(\"s\",4)"),
+        "foo"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new RightOperatorConversion().calciteOperator(),
+            inputRef("tstr"),
+            integerLiteral(5)
+        ),
+        DruidExpression.fromExpression("right(\"tstr\",5)"),
+        "05:06"
+    );
+  }
+
+  @Test
+  public void testAbnormalRightWithNegativeNumber()
+  {
+    expectedException.expect(IAE.class);
+    expectedException.expectMessage("Function[right] needs a postive integer as second argument");
+
+    testExpression(
+        rexBuilder.makeCall(
+            new RightOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(-1)
+        ),
+        DruidExpression.fromExpression("right(\"s\",-1)"),
+        null
+    );
+  }
+
+  @Test
+  public void testAbnormalRightWithWrongType()
+  {
+    expectedException.expect(IAE.class);
+    expectedException.expectMessage("Function[right] needs a string as first argument "
+                                    + "and an integer as second argument");
+
+    testExpression(
+        rexBuilder.makeCall(
+            new RightOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            inputRef("s")
+        ),
+        DruidExpression.fromExpression("right(\"s\",\"s\")"),
+        null
+    );
+  }
+
+  @Test
+  public void testLeft()
+  {
+    testExpression(
+        rexBuilder.makeCall(
+            new LeftOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(1)
+        ),
+        DruidExpression.fromExpression("left(\"s\",1)"),
+        "f"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new LeftOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(2)
+        ),
+        DruidExpression.fromExpression("left(\"s\",2)"),
+        "fo"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new LeftOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(3)
+        ),
+        DruidExpression.fromExpression("left(\"s\",3)"),
+        "foo"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new LeftOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(4)
+        ),
+        DruidExpression.fromExpression("left(\"s\",4)"),
+        "foo"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new LeftOperatorConversion().calciteOperator(),
+            inputRef("tstr"),
+            integerLiteral(10)
+        ),
+        DruidExpression.fromExpression("left(\"tstr\",10)"),
+        "2000-02-03"
+    );
+  }
+
+  @Test
+  public void testAbnormalLeftWithNegativeNumber()
+  {
+    expectedException.expect(IAE.class);
+    expectedException.expectMessage("Function[left] needs a postive integer as second argument");
+
+    testExpression(
+        rexBuilder.makeCall(
+            new LeftOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(-1)
+        ),
+        DruidExpression.fromExpression("left(\"s\",-1)"),
+        null
+    );
+  }
+
+  @Test
+  public void testAbnormalLeftWithWrongType()
+  {
+    expectedException.expect(IAE.class);
+    expectedException.expectMessage("Function[left] needs a string as first argument "
+                                    + "and an integer as second argument");
+
+    testExpression(
+        rexBuilder.makeCall(
+            new LeftOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            inputRef("s")
+        ),
+        DruidExpression.fromExpression("left(\"s\",\"s\")"),
+        null
+    );
+  }
+
+  @Test
+  public void testRepeat()
+  {
+    testExpression(
+        rexBuilder.makeCall(
+            new RepeatOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(1)
+        ),
+        DruidExpression.fromExpression("repeat(\"s\",1)"),
+        "foo"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new RepeatOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(3)
+        ),
+        DruidExpression.fromExpression("repeat(\"s\",3)"),
+        "foofoofoo"
+    );
+
+    testExpression(
+        rexBuilder.makeCall(
+            new RepeatOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            integerLiteral(-1)
+        ),
+        DruidExpression.fromExpression("repeat(\"s\",-1)"),
+        null
+    );
+  }
+
+  @Test
+  public void testAbnormalRepeatWithWrongType()
+  {
+    expectedException.expect(IAE.class);
+    expectedException.expectMessage("Function[repeat] needs a string as first argument "
+                                    + "and an integer as second argument");
+
+    testExpression(
+        rexBuilder.makeCall(
+            new RepeatOperatorConversion().calciteOperator(),
+            inputRef("s"),
+            inputRef("s")
+        ),
+        DruidExpression.fromExpression("repeat(\"s\",\"s\")"),
+        null
+    );
+  }
+
   private RexNode inputRef(final String columnName)
   {
     final int columnNumber = rowSignature.getRowOrder().indexOf(columnName);