feat(core): emit 4-arg regexp_replace with 'g' flag for SIMPLE patterns

`buildParseRelNode` for `ParseMethod.PATTERNS` lowered through PPL's REPLACE
handler, which always emits Calcite's 3-arg `REGEXP_REPLACE_3`. That works on
the V2 / Calcite path (Calcite's default is replace-all), but the analytics-
engine route converts the call to substrait + DataFusion, and DataFusion's
`regexp_replace` defaults to first-match-only without an explicit "g" flag.

The dashboard test for `source = bank | patterns email mode=label` returned
`<*>@pyrami.com` instead of `<*>@<*>.<*>` because only the first
`[a-zA-Z0-9]+` run was replaced.

Bypass the REPLACE handler for the PATTERNS branch and emit
`REGEXP_REPLACE_PG_4` directly with a constant "g" flag. Same semantics on V2 /
Calcite (Calcite's REGEXP_REPLACE_PG_4 with "g" = replace-all); fixes the
analytics-engine path.

CalcitePPLPatternsTest plan-string expectations updated to match the 4-arg
form. 17/17 unit tests pass. IT result on analytics-engine route:
testSimplePatternLabelMode_NotShowNumberedToken now passes.

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java

@@ -4301,9 +4301,26 @@ private void buildParseRelNode(Parse node, CalcitePlanContext context) {
     }
     List<RexNode> newFields = new ArrayList<>();
     for (String groupCandidate : groupCandidates) {
-      RexNode innerRex =
-          PPLFuncImpTable.INSTANCE.resolve(
-              context.rexBuilder, ParseUtils.BUILTIN_FUNCTION_MAP.get(parseMethod), rexNodeList);
+      RexNode innerRex;
+      if (ParseMethod.PATTERNS.equals(parseMethod)) {
+        // Emit `regexp_replace(field, pattern, replacement, "g")` directly so the replacement
+        // is global (every match replaced). DataFusion's `regexp_replace` defaults to FIRST
+        // match only without the "g" flag — using the 3-arg form via the REPLACE handler
+        // produces `<*>@pyrami.com` instead of `<*>@<*>.<*>` on the analytics-engine route.
+        // Calcite's REGEXP_REPLACE_PG_4 with "g" matches what `replaceAll` does, so V2 /
+        // Calcite-path semantics are preserved.
+        RexNode globalFlag =
+            context.rexBuilder.makeLiteral(
+                "g", context.rexBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR), true);
+        innerRex =
+            context.rexBuilder.makeCall(
+                SqlLibraryOperators.REGEXP_REPLACE_PG_4,
+                ArrayUtils.add(rexNodeList, globalFlag));
+      } else {
+        innerRex =
+            PPLFuncImpTable.INSTANCE.resolve(
+                context.rexBuilder, ParseUtils.BUILTIN_FUNCTION_MAP.get(parseMethod), rexNodeList);
+      }
       if (!ParseMethod.PATTERNS.equals(parseMethod)) {
         newFields.add(
             PPLFuncImpTable.INSTANCE.resolve(

ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLPatternsTest.java

@@ -35,13 +35,13 @@ public void testPatternsLabelMode_NotShowNumberedToken_ForSimplePatternMethod()
     String expectedLogical =
         "LogicalProject(ENAME=[$1], patterns_field=[CASE(SEARCH($1, Sarg['':VARCHAR; NULL AS"
             + " TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1, '[a-zA-Z0-9]+':VARCHAR,"
-            + " '<*>':VARCHAR))])\n"
+            + " '<*>':VARCHAR, 'g':VARCHAR))])\n"
             + "  LogicalTableScan(table=[[scott, EMP]])\n";
     verifyLogical(root, expectedLogical);
 
     String expectedSparkSql =
         "SELECT `ENAME`, CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN '' ELSE"
-            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END `patterns_field`\n"
+            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END `patterns_field`\n"
             + "FROM `scott`.`EMP`";
     verifyPPLToSparkSQL(root, expectedSparkSql);
   }
@@ -56,18 +56,18 @@ public void testPatternsLabelMode_ShowNumberedToken_ForSimplePatternMethod() {
     String expectedLogical =
         "LogicalProject(ENAME=[$1], patterns_field=[SAFE_CAST(ITEM(PATTERN_PARSER(CASE(SEARCH($1,"
             + " Sarg['':VARCHAR; NULL AS TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1,"
-            + " '[a-zA-Z0-9]+':VARCHAR, '<*>':VARCHAR)), $1), 'pattern'))],"
+            + " '[a-zA-Z0-9]+':VARCHAR, '<*>':VARCHAR, 'g':VARCHAR)), $1), 'pattern'))],"
             + " tokens=[SAFE_CAST(ITEM(PATTERN_PARSER(CASE(SEARCH($1, Sarg['':VARCHAR; NULL AS"
             + " TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1, '[a-zA-Z0-9]+':VARCHAR,"
-            + " '<*>':VARCHAR)), $1), 'tokens'))])\n"
+            + " '<*>':VARCHAR, 'g':VARCHAR)), $1), 'tokens'))])\n"
             + "  LogicalTableScan(table=[[scott, EMP]])\n";
     verifyLogical(root, expectedLogical);
 
     String expectedSparkSql =
         "SELECT `ENAME`, TRY_CAST(PATTERN_PARSER(CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN"
-            + " '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END, `ENAME`)['pattern'] AS"
+            + " '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END, `ENAME`)['pattern'] AS"
             + " STRING) `patterns_field`, TRY_CAST(PATTERN_PARSER(CASE WHEN `ENAME` IS NULL OR"
-            + " `ENAME` = '' THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END,"
+            + " `ENAME` = '' THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END,"
             + " `ENAME`)['tokens'] AS MAP< VARCHAR, VARCHAR ARRAY >) `tokens`\n"
             + "FROM `scott`.`EMP`";
     verifyPPLToSparkSQL(root, expectedSparkSql);
@@ -83,18 +83,18 @@ public void testPatternsLabelModeWithCustomPattern_ShowNumberedToken_ForSimplePa
     String expectedLogical =
         "LogicalProject(ENAME=[$1], patterns_field=[SAFE_CAST(ITEM(PATTERN_PARSER(CASE(SEARCH($1,"
             + " Sarg['':VARCHAR; NULL AS TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1,"
-            + " '[A-H]':VARCHAR, '<*>':VARCHAR)), $1), 'pattern'))],"
+            + " '[A-H]':VARCHAR, '<*>':VARCHAR, 'g':VARCHAR)), $1), 'pattern'))],"
             + " tokens=[SAFE_CAST(ITEM(PATTERN_PARSER(CASE(SEARCH($1, Sarg['':VARCHAR; NULL AS"
-            + " TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1, '[A-H]':VARCHAR, '<*>':VARCHAR)),"
+            + " TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1, '[A-H]':VARCHAR, '<*>':VARCHAR, 'g':VARCHAR)),"
             + " $1), 'tokens'))])\n"
             + "  LogicalTableScan(table=[[scott, EMP]])\n";
     verifyLogical(root, expectedLogical);
 
     String expectedSparkSql =
         "SELECT `ENAME`, TRY_CAST(PATTERN_PARSER(CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN"
-            + " '' ELSE REGEXP_REPLACE(`ENAME`, '[A-H]', '<*>') END, `ENAME`)['pattern'] AS STRING)"
+            + " '' ELSE REGEXP_REPLACE(`ENAME`, '[A-H]', '<*>', 'g') END, `ENAME`)['pattern'] AS STRING)"
             + " `patterns_field`, TRY_CAST(PATTERN_PARSER(CASE WHEN `ENAME` IS NULL OR `ENAME` ="
-            + " '' THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[A-H]', '<*>') END, `ENAME`)['tokens'] AS"
+            + " '' THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[A-H]', '<*>', 'g') END, `ENAME`)['tokens'] AS"
             + " MAP< VARCHAR, VARCHAR ARRAY >) `tokens`\n"
             + "FROM `scott`.`EMP`";
     verifyPPLToSparkSQL(root, expectedSparkSql);
@@ -108,13 +108,13 @@ public void testPatternsLabelModeWithCustomField_NotShowNumberedToken_ForSimpleP
 
     String expectedLogical =
         "LogicalProject(ENAME=[$1], upper=[CASE(SEARCH($1, Sarg['':VARCHAR; NULL AS TRUE]:VARCHAR),"
-            + " '':VARCHAR, REGEXP_REPLACE($1, '[A-H]':VARCHAR, '<*>':VARCHAR))])\n"
+            + " '':VARCHAR, REGEXP_REPLACE($1, '[A-H]':VARCHAR, '<*>':VARCHAR, 'g':VARCHAR))])\n"
             + "  LogicalTableScan(table=[[scott, EMP]])\n";
     verifyLogical(root, expectedLogical);
 
     String expectedSparkSql =
         "SELECT `ENAME`, CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN '' ELSE"
-            + " REGEXP_REPLACE(`ENAME`, '[A-H]', '<*>') END `upper`\n"
+            + " REGEXP_REPLACE(`ENAME`, '[A-H]', '<*>', 'g') END `upper`\n"
             + "FROM `scott`.`EMP`";
     verifyPPLToSparkSQL(root, expectedSparkSql);
   }
@@ -130,19 +130,19 @@ public void testPatternsLabelModeWithPartitionBy_ShowNumberedToken_SimplePattern
         "LogicalProject(ENAME=[$1], DEPTNO=[$7],"
             + " patterns_field=[SAFE_CAST(ITEM(PATTERN_PARSER(CASE(SEARCH($1, Sarg['':VARCHAR; NULL"
             + " AS TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1, '[a-zA-Z0-9]+':VARCHAR,"
-            + " '<*>':VARCHAR)), $1), 'pattern'))],"
+            + " '<*>':VARCHAR, 'g':VARCHAR)), $1), 'pattern'))],"
             + " tokens=[SAFE_CAST(ITEM(PATTERN_PARSER(CASE(SEARCH($1, Sarg['':VARCHAR; NULL AS"
             + " TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1, '[a-zA-Z0-9]+':VARCHAR,"
-            + " '<*>':VARCHAR)), $1), 'tokens'))])\n"
+            + " '<*>':VARCHAR, 'g':VARCHAR)), $1), 'tokens'))])\n"
             + "  LogicalTableScan(table=[[scott, EMP]])\n";
     verifyLogical(root, expectedLogical);
 
     String expectedSparkSql =
         "SELECT `ENAME`, `DEPTNO`, TRY_CAST(PATTERN_PARSER(CASE WHEN `ENAME` IS NULL OR `ENAME`"
-            + " = '' THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END,"
+            + " = '' THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END,"
             + " `ENAME`)['pattern'] AS STRING) `patterns_field`, TRY_CAST(PATTERN_PARSER(CASE"
             + " WHEN `ENAME` IS NULL OR `ENAME` = '' THEN '' ELSE REGEXP_REPLACE(`ENAME`,"
-            + " '[a-zA-Z0-9]+', '<*>') END, `ENAME`)['tokens'] AS MAP< VARCHAR, VARCHAR ARRAY >)"
+            + " '[a-zA-Z0-9]+', '<*>', 'g') END, `ENAME`)['tokens'] AS MAP< VARCHAR, VARCHAR ARRAY >)"
             + " `tokens`\n"
             + "FROM `scott`.`EMP`";
     verifyPPLToSparkSQL(root, expectedSparkSql);
@@ -248,18 +248,18 @@ public void testPatternsAggregationMode_NotShowNumberedToken_ForSimplePatternMet
         "LogicalAggregate(group=[{1}], pattern_count=[COUNT($1)], sample_logs=[TAKE($0, $2)])\n"
             + "  LogicalProject(ENAME=[$1], patterns_field=[CASE(SEARCH($1, Sarg['':VARCHAR; NULL"
             + " AS TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1, '[a-zA-Z0-9]+':VARCHAR,"
-            + " '<*>':VARCHAR))], $f9=[10])\n"
+            + " '<*>':VARCHAR, 'g':VARCHAR))], $f9=[10])\n"
             + "    LogicalTableScan(table=[[scott, EMP]])\n";
     verifyLogical(root, expectedLogical);
 
     String expectedSparkSql =
         "SELECT CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN '' ELSE REGEXP_REPLACE(`ENAME`,"
-            + " '[a-zA-Z0-9]+', '<*>') END `patterns_field`, COUNT(CASE WHEN `ENAME` IS NULL OR"
-            + " `ENAME` = '' THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END)"
+            + " '[a-zA-Z0-9]+', '<*>', 'g') END `patterns_field`, COUNT(CASE WHEN `ENAME` IS NULL OR"
+            + " `ENAME` = '' THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END)"
             + " `pattern_count`, `TAKE`(`ENAME`, 10) `sample_logs`\n"
             + "FROM `scott`.`EMP`\n"
             + "GROUP BY CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN '' ELSE"
-            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END";
+            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END";
     verifyPPLToSparkSQL(root, expectedSparkSql);
   }
 
@@ -276,21 +276,21 @@ public void testPatternsAggregationMode_ShowNumberedToken_ForSimplePatternMethod
             + " $2)])\n"
             + "    LogicalProject(ENAME=[$1], patterns_field=[CASE(SEARCH($1, Sarg['':VARCHAR; NULL"
             + " AS TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1, '[a-zA-Z0-9]+':VARCHAR,"
-            + " '<*>':VARCHAR))], $f9=[10])\n"
+            + " '<*>':VARCHAR, 'g':VARCHAR))], $f9=[10])\n"
             + "      LogicalTableScan(table=[[scott, EMP]])\n";
     verifyLogical(root, expectedLogical);
 
     String expectedSparkSql =
         "SELECT TRY_CAST(PATTERN_PARSER(CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN '' ELSE"
-            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END, `TAKE`(`ENAME`, 10))['pattern']"
+            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END, `TAKE`(`ENAME`, 10))['pattern']"
             + " AS STRING) `patterns_field`, COUNT(CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN"
-            + " '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END) `pattern_count`,"
+            + " '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END) `pattern_count`,"
             + " TRY_CAST(PATTERN_PARSER(CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN '' ELSE"
-            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END, `TAKE`(`ENAME`, 10))['tokens']"
+            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END, `TAKE`(`ENAME`, 10))['tokens']"
             + " AS MAP< VARCHAR, VARCHAR ARRAY >) `tokens`, `TAKE`(`ENAME`, 10) `sample_logs`\n"
             + "FROM `scott`.`EMP`\n"
             + "GROUP BY CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN '' ELSE"
-            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END";
+            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END";
     verifyPPLToSparkSQL(root, expectedSparkSql);
   }
 
@@ -307,22 +307,22 @@ public void testPatternsAggregationModeWithGroupBy_ShowNumberedToken_ForSimplePa
             + " $3)])\n"
             + "    LogicalProject(ENAME=[$1], DEPTNO=[$7], patterns_field=[CASE(SEARCH($1,"
             + " Sarg['':VARCHAR; NULL AS TRUE]:VARCHAR), '':VARCHAR, REGEXP_REPLACE($1,"
-            + " '[a-zA-Z0-9]+':VARCHAR, '<*>':VARCHAR))], $f9=[10])\n"
+            + " '[a-zA-Z0-9]+':VARCHAR, '<*>':VARCHAR, 'g':VARCHAR))], $f9=[10])\n"
             + "      LogicalTableScan(table=[[scott, EMP]])\n";
     verifyLogical(root, expectedLogical);
 
     String expectedSparkSql =
         "SELECT `DEPTNO`, TRY_CAST(PATTERN_PARSER(CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN ''"
-            + " ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END, `TAKE`(`ENAME`,"
+            + " ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END, `TAKE`(`ENAME`,"
             + " 10))['pattern'] AS STRING) `patterns_field`, COUNT(CASE WHEN `ENAME` IS NULL OR"
-            + " `ENAME` = '' THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END)"
+            + " `ENAME` = '' THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END)"
             + " `pattern_count`, TRY_CAST(PATTERN_PARSER(CASE WHEN `ENAME` IS NULL OR `ENAME` = ''"
-            + " THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END, `TAKE`(`ENAME`,"
+            + " THEN '' ELSE REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END, `TAKE`(`ENAME`,"
             + " 10))['tokens'] AS MAP< VARCHAR, VARCHAR ARRAY >) `tokens`, `TAKE`(`ENAME`, 10)"
             + " `sample_logs`\n"
             + "FROM `scott`.`EMP`\n"
             + "GROUP BY `DEPTNO`, CASE WHEN `ENAME` IS NULL OR `ENAME` = '' THEN '' ELSE"
-            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>') END";
+            + " REGEXP_REPLACE(`ENAME`, '[a-zA-Z0-9]+', '<*>', 'g') END";
     verifyPPLToSparkSQL(root, expectedSparkSql);
   }