[NSE-872] implement replace function

native-sql-engine/core/src/main/scala/com/intel/oap/expression/ColumnarExpressionConverter.scala

@@ -306,6 +306,21 @@ object ColumnarExpressionConverter extends Logging {
             convertBoundRefToAttrRef = convertBoundRefToAttrRef),
           expr
         )
+      case sr: StringReplace =>
+        check_if_no_calculation = false
+        logInfo(s"${expr.getClass} ${expr} is supported, no_cal is $check_if_no_calculation.")
+        ColumnarTernaryOperator.create(
+          replaceWithColumnarExpression(
+            sr.srcExpr,
+            attributeSeq,
+            convertBoundRefToAttrRef = convertBoundRefToAttrRef),
+          replaceWithColumnarExpression(
+            sr.searchExpr,
+            convertBoundRefToAttrRef = convertBoundRefToAttrRef),
+          replaceWithColumnarExpression(
+            sr.replaceExpr,
+            convertBoundRefToAttrRef = convertBoundRefToAttrRef),
+          expr)
       case u: UnaryExpression =>
         logInfo(s"${expr.getClass} ${expr} is supported, no_cal is $check_if_no_calculation.")
         if (!u.isInstanceOf[CheckOverflow] || !u.child.isInstanceOf[Divide]) {
@@ -424,6 +439,10 @@ object ColumnarExpressionConverter extends Logging {
           regexp.regexp) || containsSubquery(regexp.rep) || containsSubquery(regexp.pos)
       case substrIndex: ColumnarSubstringIndex =>
         substrIndex.children.map(containsSubquery).exists(_ == true)
+      case sr: StringReplace =>
+        containsSubquery(sr.srcExpr) ||
+          containsSubquery(sr.searchExpr) ||
+          containsSubquery(sr.replaceExpr)
       case expr =>
         throw new UnsupportedOperationException(
           s" --> ${expr.getClass} | ${expr} is not currently supported.")

native-sql-engine/core/src/main/scala/com/intel/oap/expression/ColumnarTernaryOperator.scala

@@ -223,6 +223,42 @@ class ColumnarSubstringIndex(strExpr: Expression, delimExpr: Expression,
   }
 }
 
+class ColumnarStringReplace(
+    srcExpr: Expression,
+    searchExpr: Expression,
+    replaceExpr: Expression)
+  extends StringReplace(srcExpr, searchExpr, replaceExpr) with ColumnarExpression {
+
+  buildCheck()
+  def buildCheck(): Unit = {
+    val unsupportedDataType =
+      Seq(srcExpr.dataType, searchExpr.dataType, replaceExpr.dataType)
+        .filterNot(_ == StringType)
+    if (unsupportedDataType.nonEmpty) {
+      throw new UnsupportedOperationException(
+        s"${unsupportedDataType.mkString(",")} is not supported in ColumnarStringReplace.")
+    }
+  }
+
+  override def doColumnarCodeGen(args: java.lang.Object)
+  : (TreeNode, ArrowType) = {
+    val (srcNode, _): (TreeNode, ArrowType) =
+      srcExpr.asInstanceOf[ColumnarExpression].doColumnarCodeGen(args)
+    val (searchNode, _): (TreeNode, ArrowType) =
+      searchExpr.asInstanceOf[ColumnarExpression].doColumnarCodeGen(args)
+    val (replaceNode, _): (TreeNode, ArrowType) =
+      replaceExpr.asInstanceOf[ColumnarExpression].doColumnarCodeGen(args)
+
+    val resultType = new ArrowType.Utf8()
+    val funcNode =
+      TreeBuilder.makeFunction(
+        "replace",
+        Lists.newArrayList(srcNode, searchNode, replaceNode),
+        resultType)
+    (funcNode, resultType)
+  }
+}
+
 object ColumnarTernaryOperator {
 
   def create(src: Expression, arg1: Expression, arg2: Expression,
@@ -240,6 +276,8 @@ object ColumnarTernaryOperator {
       new ColumnarRegExpExtract(src, arg1, arg2, re)
     case substrIndex: SubstringIndex =>
       new ColumnarSubstringIndex(src, arg1, arg2, substrIndex)
+    case _: StringReplace =>
+      new ColumnarStringReplace(src, arg1, arg2)
     case other =>
       throw new UnsupportedOperationException(s"not currently supported: $other.")
   }