[Analytics Backend / DataFusion] Wire PPL replace command + replace()/regexp_replace() functions

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java

@@ -67,6 +67,8 @@ public enum ScalarFunction {
      */
     CONCAT(Category.STRING, SqlKind.OTHER_FUNCTION, SqlStdOperatorTable.CONCAT),
     CHAR_LENGTH(Category.STRING, SqlKind.OTHER_FUNCTION),
+    REPLACE(Category.STRING, SqlKind.OTHER_FUNCTION),
+    REGEXP_REPLACE(Category.STRING, SqlKind.OTHER_FUNCTION),
 
     // ── Math ─────────────────────────────────────────────────────────
     PLUS(Category.MATH, SqlKind.PLUS),

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java

@@ -114,6 +114,8 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP
         ScalarFunction.IN,
         ScalarFunction.LIKE,
         ScalarFunction.REGEXP_CONTAINS,
+        ScalarFunction.REPLACE,
+        ScalarFunction.REGEXP_REPLACE,
         ScalarFunction.PLUS,
         ScalarFunction.MINUS,
         ScalarFunction.TIMES,
@@ -203,6 +205,7 @@ public Map<ScalarFunction, ScalarFunctionAdapter> scalarFunctionAdapters() {
                     Map.entry(ScalarFunction.DIVIDE, new StdOperatorRewriteAdapter("DIVIDE", SqlStdOperatorTable.DIVIDE)),
                     Map.entry(ScalarFunction.LIKE, new LikeAdapter()),
                     Map.entry(ScalarFunction.MOD, new StdOperatorRewriteAdapter("MOD", SqlStdOperatorTable.MOD)),
+                    Map.entry(ScalarFunction.REGEXP_REPLACE, new RegexpReplaceAdapter()),
                     Map.entry(ScalarFunction.SARG_PREDICATE, new SargAdapter()),
                     Map.entry(ScalarFunction.TIMESTAMP, new TimestampFunctionAdapter()),
                     Map.entry(ScalarFunction.UNIX_TIMESTAMP, new UnixTimestampAdapter()),

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java

@@ -26,6 +26,7 @@
 import org.apache.calcite.schema.ColumnStrategy;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.fun.SqlLibraryOperators;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -89,6 +90,11 @@ public class DataFusionFragmentConvertor implements FragmentConvertor {
      *       resolved by DataFusion's substrait consumer to {@code Operator::RegexMatch}, the same
      *       binary operator that backs PostgreSQL's {@code ~} regex match). Lowering target for PPL
      *       {@code regex} command and {@code regexp_match()} function.</li>
+     *   <li>{@link SqlStdOperatorTable#REPLACE} → {@code replace} (literal string replacement;
+     *       lowering target for PPL `replace` command on non-wildcard patterns).</li>
+     *   <li>{@link SqlLibraryOperators#REGEXP_REPLACE_3} → {@code regexp_replace} (regex string
+     *       replacement; lowering target for PPL `replace` command on wildcard patterns and for
+     *       PPL `replace()` / `regexp_replace()` functions in `eval`).</li>
      * </ul>
      */
     private static final List<FunctionMappings.Sig> ADDITIONAL_SCALAR_SIGS = List.of(
@@ -98,7 +104,9 @@ public class DataFusionFragmentConvertor implements FragmentConvertor {
         FunctionMappings.s(SqlLibraryOperators.DATE_PART, "date_part"),
         FunctionMappings.s(ConvertTzAdapter.LOCAL_CONVERT_TZ_OP, "convert_tz"),
         FunctionMappings.s(UnixTimestampAdapter.LOCAL_TO_UNIXTIME_OP, "to_unixtime"),
-        FunctionMappings.s(SqlLibraryOperators.REGEXP_CONTAINS, "regex_match")
+        FunctionMappings.s(SqlLibraryOperators.REGEXP_CONTAINS, "regex_match"),
+        FunctionMappings.s(SqlStdOperatorTable.REPLACE, "replace"),
+        FunctionMappings.s(SqlLibraryOperators.REGEXP_REPLACE_3, "regexp_replace")
     );
 
     private final SimpleExtension.ExtensionCollection extensions;

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/RegexpReplaceAdapter.java

@@ -0,0 +1,190 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.opensearch.analytics.spi.FieldStorageInfo;
+import org.opensearch.analytics.spi.ScalarFunctionAdapter;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Rewrites the pattern and replacement operands of {@code REGEXP_REPLACE} from Java syntax
+ * to a Rust-{@code regex}-crate-compatible form. Two transforms:
+ *
+ * <ol>
+ *   <li><b>Pattern</b>: expand {@code \Q…\E} quoted-literal blocks to per-char escaped
+ *       sequences. The SQL plugin's {@code WildcardUtils.convertWildcardPatternToRegex()}
+ *       emits Java {@link java.util.regex.Pattern} {@code \Q…\E} syntax (e.g.
+ *       {@code ^\Q\E(.*?)\QBOARDS\E$}). Rust's {@code regex} crate (used by DataFusion)
+ *       rejects {@code \Q…\E} with {@code unrecognized escape sequence}.</li>
+ *   <li><b>Replacement</b>: wrap bare {@code $N} backreferences in braces ({@code ${N}}).
+ *       Rust's regex replacement parser greedily extends {@code $N} into the longest
+ *       valid identifier — so {@code $1_$2} is parsed as a reference to the (non-existent)
+ *       group named {@code 1_} followed by {@code $2}, yielding empty + group-2's value.
+ *       Java's {@link java.util.regex.Matcher#replaceAll} stops at the first non-digit, so
+ *       {@code $1_$2} means group-1 + literal underscore + group-2. Wrapping every numeric
+ *       backreference in braces gives Rust the unambiguous form regardless of what
+ *       follows.</li>
+ * </ol>
+ *
+ * <p>Both rewrites preserve semantics — they're syntactic normalizations, not behavior
+ * changes. Calls without {@code \Q} in the pattern AND without bare {@code $N} in the
+ * replacement pass through unchanged.
+ *
+ * <p>Pattern faithful to {@link java.util.regex.Pattern} semantics: an unterminated
+ * {@code \Q} (no closing {@code \E}) quotes through end-of-string. Replacement preserves
+ * existing {@code ${…}} braces and the {@code $$} literal-dollar escape.
+ *
+ * @opensearch.internal
+ */
+class RegexpReplaceAdapter implements ScalarFunctionAdapter {
+
+    /** Standard regex metacharacters that must be backslash-escaped to match literally. */
+    private static final String REGEX_METACHARS = ".\\+*?^$()[]{}|/";
+
+    @Override
+    public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelOptCluster cluster) {
+        // REGEXP_REPLACE_3 has signature (input, pattern, replacement) — exactly 3 operands.
+        if (original.getOperands().size() != 3) {
+            return original;
+        }
+        RexNode patternOperand = original.getOperands().get(1);
+        RexNode replacementOperand = original.getOperands().get(2);
+
+        String rewrittenPattern = null;
+        if (patternOperand instanceof RexLiteral patternLiteral) {
+            String pattern = patternLiteral.getValueAs(String.class);
+            if (pattern != null && pattern.contains("\\Q")) {
+                String rewritten = unquoteJavaRegex(pattern);
+                if (!pattern.equals(rewritten)) {
+                    rewrittenPattern = rewritten;
+                }
+            }
+        }
+
+        String rewrittenReplacement = null;
+        if (replacementOperand instanceof RexLiteral replacementLiteral) {
+            String replacement = replacementLiteral.getValueAs(String.class);
+            if (replacement != null && replacement.indexOf('$') >= 0) {
+                String rewritten = braceBackreferences(replacement);
+                if (!replacement.equals(rewritten)) {
+                    rewrittenReplacement = rewritten;
+                }
+            }
+        }
+
+        if (rewrittenPattern == null && rewrittenReplacement == null) {
+            return original;
+        }
+
+        RexBuilder rexBuilder = cluster.getRexBuilder();
+        // makeLiteral(String) infers a CHAR type sized to the rewritten string. Reusing the
+        // original literal's type would right-pad to the OLD length (e.g. CHAR(23) → 8 trailing
+        // spaces after a 15-char rewrite), corrupting the value at runtime.
+        List<RexNode> newOperands = new ArrayList<>(3);
+        newOperands.add(original.getOperands().get(0));
+        newOperands.add(rewrittenPattern != null ? rexBuilder.makeLiteral(rewrittenPattern) : patternOperand);
+        newOperands.add(rewrittenReplacement != null ? rexBuilder.makeLiteral(rewrittenReplacement) : replacementOperand);
+        return rexBuilder.makeCall(original.getType(), original.getOperator(), newOperands);
+    }
+
+    /**
+     * Wrap every numeric backreference {@code $N} in the input with braces ({@code ${N}}).
+     * Preserves {@code $$} (literal dollar) and existing {@code ${…}} braced groups.
+     *
+     * <p>Why: Rust's regex replacement parser uses identifier-greedy matching — {@code $1_}
+     * is a named-group reference where the name is {@code 1_}. Java's parser stops at the
+     * first non-digit, so {@code $1_} means group 1 followed by literal underscore. Wrapping
+     * in braces gives Rust the unambiguous form: {@code ${1}} is always group 1, regardless
+     * of what follows.
+     *
+     * <p>Visible for unit testing.
+     */
+    static String braceBackreferences(String replacement) {
+        StringBuilder out = new StringBuilder(replacement.length());
+        int i = 0;
+        while (i < replacement.length()) {
+            char c = replacement.charAt(i);
+            if (c == '$' && i + 1 < replacement.length()) {
+                char next = replacement.charAt(i + 1);
+                if (next == '$') {
+                    // Literal dollar — pass through both characters unchanged.
+                    out.append("$$");
+                    i += 2;
+                    continue;
+                }
+                if (next == '{') {
+                    // Already braced — copy through to (and including) the closing '}'.
+                    int closeIdx = replacement.indexOf('}', i + 2);
+                    if (closeIdx == -1) {
+                        // Malformed — leave the rest verbatim.
+                        out.append(replacement, i, replacement.length());
+                        return out.toString();
+                    }
+                    out.append(replacement, i, closeIdx + 1);
+                    i = closeIdx + 1;
+                    continue;
+                }
+                if (Character.isDigit(next)) {
+                    // Bare $N — wrap in braces so Rust doesn't consume following identifier
+                    // characters (letters, digits, underscores) as part of the group name.
+                    int j = i + 1;
+                    while (j < replacement.length() && Character.isDigit(replacement.charAt(j))) {
+                        j++;
+                    }
+                    out.append("${").append(replacement, i + 1, j).append("}");
+                    i = j;
+                    continue;
+                }
+            }
+            out.append(c);
+            i++;
+        }
+        return out.toString();
+    }
+
+    /**
+     * Replace each {@code \Q…\E} block in the input with a per-char escaped equivalent.
+     * Characters inside the block that are regex metacharacters get prefixed with {@code \};
+     * other characters pass through. Faithfully handles unterminated {@code \Q} (runs to end).
+     *
+     * <p>Visible for unit testing — the rewrite logic is the substantive part of this adapter.
+     */
+    static String unquoteJavaRegex(String regex) {
+        StringBuilder out = new StringBuilder(regex.length());
+        int i = 0;
+        while (i < regex.length()) {
+            // Look for \Q at position i (literal backslash + Q in the source string).
+            if (i + 1 < regex.length() && regex.charAt(i) == '\\' && regex.charAt(i + 1) == 'Q') {
+                int contentStart = i + 2;
+                int closeIdx = regex.indexOf("\\E", contentStart);
+                int contentEnd = (closeIdx == -1) ? regex.length() : closeIdx;
+                for (int j = contentStart; j < contentEnd; j++) {
+                    char c = regex.charAt(j);
+                    if (REGEX_METACHARS.indexOf(c) >= 0) {
+                        out.append('\\');
+                    }
+                    out.append(c);
+                }
+                // Skip past \E (or off the end if unterminated).
+                i = (closeIdx == -1) ? regex.length() : closeIdx + 2;
+            } else {
+                out.append(regex.charAt(i));
+                i++;
+            }
+        }
+        return out.toString();
+    }
+}

sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml

@@ -59,3 +59,53 @@ scalar_functions:
           - value: "string"
             name: "pattern"
         return: boolean
+  - name: replace
+    description: >-
+      Literal string replacement — replace every occurrence of `search` in `input`
+      with `replacement`. Lowering target for PPL's `replace` command on
+      non-wildcard patterns (Calcite `SqlStdOperatorTable.REPLACE`).
+      datafusion-substrait resolves the extension name "replace" to DataFusion's
+      native `replace` UDF (datafusion-functions/src/string/replace.rs).
+    impls:
+      - args:
+          - value: "varchar<L1>"
+            name: "input"
+          - value: "varchar<L2>"
+            name: "search"
+          - value: "varchar<L3>"
+            name: "replacement"
+        return: "varchar<L1>"
+      - args:
+          - value: "string"
+            name: "input"
+          - value: "string"
+            name: "search"
+          - value: "string"
+            name: "replacement"
+        return: string
+  - name: regexp_replace
+    description: >-
+      Regex string replacement — replace every match of `pattern` in `input`
+      with `replacement`. Lowering target for PPL's `replace` command on
+      wildcard patterns (after `*` → regex conversion) and for the PPL
+      `replace()` / `regexp_replace()` functions in `eval`. Calcite emits
+      `SqlLibraryOperators.REGEXP_REPLACE_3`. datafusion-substrait resolves
+      the extension name "regexp_replace" to DataFusion's native `regexp_replace`
+      UDF (datafusion-functions/src/regex/regexpreplace.rs).
+    impls:
+      - args:
+          - value: "varchar<L1>"
+            name: "input"
+          - value: "varchar<L2>"
+            name: "pattern"
+          - value: "varchar<L3>"
+            name: "replacement"
+        return: "varchar<L1>"
+      - args:
+          - value: "string"
+            name: "input"
+          - value: "string"
+            name: "pattern"
+          - value: "string"
+            name: "replacement"
+        return: string