[Bug][Transform-V2] Enable regex replacement by default for FieldRename (#10321)

Author: yzeng1618

docs/en/transforms/field-rename.md

@@ -13,7 +13,8 @@ FieldRename transform plugin for rename field name.
 |      convert_case       | string | no       |               | The case conversion type. The options can be `UPPER`, `LOWER`                                                         |
 |         prefix          | string | no       |               | The prefix to be added to the field name                                                                              |
 |         suffix          | string | no       |               | The suffix to be added to the field name                                                                              |
-| replacements_with_regex | array  | no       |               | The array of replacement rules with regex. The replacement rule is a map with `replace_from` and `replace_to` fields. |
+| replacements_with_regex | array  | no       |               | The array of replacement rules. Each rule is a map with `replace_from`, `replace_to`, and optional `is_regex` (default `true`). When `is_regex=false`, `replace_from` is treated as an exact field name (full match). |
+|        specific         | array  | no       |               | Specific rename rules. Each rule is a map with `field_name` and `target_name`. When matched, it will rename the field directly and skip other rename rules. |
 
 ## Examples
 
@@ -73,6 +74,21 @@ sink {
 }
 ```
 
+### Rename specific fields
+
+```
+transform {
+  FieldRename {
+    plugin_input = "input"
+    plugin_output = "output"
+
+    specific = [
+      { field_name = "InvoiceNum", target_name = "invoice_num" }
+    ]
+  }
+}
+```
+
 ### Convert field name to lowercase
 
 ```
@@ -129,4 +145,4 @@ sink {
     data_save_mode = "APPEND_DATA"
   }
 }
-```
+```

docs/zh/transforms/field-rename.md

@@ -13,7 +13,8 @@ FieldRename 用于批量重命名字段名。
 |      convert_case       | string | 否   |        | 字母大小写转换类型，可选 `UPPER`、`LOWER`                                                               |
 |         prefix          | string | 否   |        | 追加到字段名前的前缀                                                                                    |
 |         suffix          | string | 否   |        | 追加到字段名后的后缀                                                                                    |
-| replacements_with_regex | array  | 否   |        | 正则替换规则数组，元素为包含 `replace_from`、`replace_to` 的映射，用于批量替换字段名                    |
+| replacements_with_regex | array  | 否   |        | 替换规则数组，元素为包含 `replace_from`、`replace_to` 以及可选 `is_regex`（默认 `true`）的映射；当 `is_regex=false` 时，`replace_from` 按字段名精确匹配（全匹配） |
+|        specific         | array  | 否   |        | 指定字段重命名规则，元素为包含 `field_name` 和 `target_name` 的映射；命中后会直接重命名并跳过其他规则 |
 
 ## 示例
 
@@ -73,6 +74,21 @@ sink {
 }
 ```
 
+### 指定字段重命名
+
+```
+transform {
+  FieldRename {
+    plugin_input = "input"
+    plugin_output = "output"
+
+    specific = [
+      { field_name = "InvoiceNum", target_name = "invoice_num" }
+    ]
+  }
+}
+```
+
 ### 将字段名转为小写
 
 ```

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestRenameIT.java

@@ -34,4 +34,11 @@ public void testRenameMultiTable(TestContainer container)
                 container.executeJob("/table_field_rename_multi_table.conf");
         Assertions.assertEquals(0, execResult.getExitCode());
     }
+
+    @TestTemplate
+    public void testFieldRenameRegexDefault(TestContainer container)
+            throws IOException, InterruptedException {
+        Container.ExecResult execResult = container.executeJob("/field_rename_regex_default.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+    }
 }

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/field_rename_regex_default.conf

@@ -0,0 +1,100 @@
+#
+# 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.
+#
+
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
+source {
+  FakeSource {
+    plugin_output = "source1"
+
+    tables_configs = [
+      {
+        row.num = 1
+        schema = {
+          table = "test.regex"
+          columns = [
+            {
+              name = "InvoiceNum"
+              type = "bigint"
+            },
+            {
+              name = "VendorID"
+              type = "string"
+            }
+          ]
+        }
+      }
+    ]
+  }
+}
+
+transform {
+  FieldRename {
+    plugin_input = "source1"
+    plugin_output = "transform1"
+
+    convert_case = "LOWER"
+    # intentionally omit is_regex to verify default behavior
+    replacements_with_regex = [
+      {
+        replace_from = "(?<=[a-z0-9])(?=[A-Z])"
+        replace_to = "_"
+      }
+    ]
+  }
+}
+
+sink {
+  Assert {
+    plugin_input = "transform1"
+
+    rules =
+      {
+        tables_configs = [
+          {
+            table_path = "test.regex"
+            row_rules = [
+              {
+                rule_type = MAX_ROW
+                rule_value = 1
+              },
+              {
+                rule_type = MIN_ROW
+                rule_value = 1
+              }
+            ],
+            catalog_table_rule {
+              table_path = "test.regex"
+              column_rule = [
+                {
+                  name = "invoice_num"
+                  type = "bigint"
+                },
+                {
+                  name = "vendor_id"
+                  type = "string"
+                }
+              ]
+            }
+          }
+        ]
+      }
+  }
+}

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rename/FieldRenameConfig.java

@@ -114,7 +114,7 @@ public static class ReplacementsWithRegex implements Serializable {
         private String replaceTo;
 
         @JsonAlias("is_regex")
-        private Boolean isRegex;
+        private Boolean isRegex = true;
     }
 
     public static FieldRenameConfig of(ReadonlyConfig config) {

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rename/FieldRenameTransform.java

@@ -143,7 +143,7 @@ public String convertName(String name) {
                 String replacement = replacementsWithRegex.getReplaceFrom();
                 if (StringUtils.isNotEmpty(replacement)) {
                     Map<Integer, Integer> matched = new LinkedHashMap<>();
-                    if (BooleanUtils.isNotTrue(isRegex)) {
+                    if (BooleanUtils.isFalse(isRegex)) {
                         if (StringUtils.equals(replacement, name)) {
                             matched.put(0, name.length());
                         }

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/rename/FieldRenameTransformTest.java

@@ -240,4 +240,38 @@ public void testRename() {
         Assertions.assertEquals("f5", outputChangeEvent.getColumn().getName());
         Assertions.assertEquals("f5", outputDropEvent.getColumn());
     }
+
+    @Test
+    public void testRegexReplacementEnabledByDefault() {
+        FieldRenameConfig.ReplacementsWithRegex rule =
+                new FieldRenameConfig.ReplacementsWithRegex();
+        rule.setReplaceFrom("(?<=[a-z0-9])(?=[A-Z])");
+        rule.setReplaceTo("_");
+
+        FieldRenameConfig config =
+                new FieldRenameConfig()
+                        .setConvertCase(ConvertCase.LOWER)
+                        .setReplacementsWithRegex(Collections.singletonList(rule));
+        FieldRenameTransform transform = new FieldRenameTransform(config, DEFAULT_TABLE);
+
+        Assertions.assertEquals("invoice_num", transform.convertName("InvoiceNum"));
+        Assertions.assertEquals("vendor_id", transform.convertName("VendorID"));
+    }
+
+    @Test
+    public void testRegexReplacementCanBeDisabled() {
+        FieldRenameConfig.ReplacementsWithRegex rule =
+                new FieldRenameConfig.ReplacementsWithRegex();
+        rule.setReplaceFrom("(?<=[a-z0-9])(?=[A-Z])");
+        rule.setReplaceTo("_");
+        rule.setIsRegex(false);
+
+        FieldRenameConfig config =
+                new FieldRenameConfig()
+                        .setConvertCase(ConvertCase.LOWER)
+                        .setReplacementsWithRegex(Collections.singletonList(rule));
+        FieldRenameTransform transform = new FieldRenameTransform(config, DEFAULT_TABLE);
+
+        Assertions.assertEquals("invoicenum", transform.convertName("InvoiceNum"));
+    }
 }