(seatunnel) branch dev updated: [Fix][Transform-V2] Prevent input row mutation in ReplaceTransform (#10941)

[doyeon]

This is an automated email from the ASF dual-hosted git repository.

dybyte pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git


The following commit(s) were added to refs/heads/dev by this push:
     new c780843b8b [Fix][Transform-V2] Prevent input row mutation in 
ReplaceTransform (#10941)
c780843b8b is described below

commit c780843b8bf63fd95322ab0ac01c0cfb2ae05cee
Author: Doyeon Kim <132787602+dyb...@users.noreply.github.com>
AuthorDate: Mon May 25 05:40:19 2026 +0900

    [Fix][Transform-V2] Prevent input row mutation in ReplaceTransform (#10941)
---
 .../transform/replace/ReplaceTransform.java        |  8 ++++--
 .../transform/replace/ReplaceTransformTest.java    | 33 ++++++++++++++++++++++
 2 files changed, 38 insertions(+), 3 deletions(-)

diff --git 
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransform.java
 
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransform.java
index 2a2d6d64f5..0228e1f0ec 100644
--- 
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransform.java
+++ 
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransform.java
@@ -123,14 +123,16 @@ public class ReplaceTransform extends 
AbstractCatalogSupportMapTransform {
 
     @Override
     protected SeaTunnelRow transformRow(SeaTunnelRow inputRow) {
+        SeaTunnelRow outputRow = inputRow.copy();
+
         for (int index : replaceFieldIndexes) {
-            Object value = inputRow.getField(index);
+            Object value = outputRow.getField(index);
             if (value == null) {
                 continue;
             }
-            inputRow.setField(index, applyReplacement(value.toString()));
+            outputRow.setField(index, applyReplacement(value.toString()));
         }
-        return inputRow;
+        return outputRow;
     }
 
     private String applyReplacement(String value) {
diff --git 
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/replace/ReplaceTransformTest.java
 
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/replace/ReplaceTransformTest.java
index 7ff23bf9a5..c716627c95 100644
--- 
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/replace/ReplaceTransformTest.java
+++ 
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/replace/ReplaceTransformTest.java
@@ -268,4 +268,37 @@ class ReplaceTransformTest {
         Assertions.assertEquals("abcNUMdef456", output.getField(1));
         Assertions.assertEquals("xyzNUMuvw012", output.getField(2));
     }
+
+    @Test
+    void testReplaceDoesNotMutateInputRowForFanOut() {
+        Map<String, Object> phoneReplaceConfig = new HashMap<>();
+        phoneReplaceConfig.put("replace_field", "title");
+        phoneReplaceConfig.put(ReplaceTransformConfig.KEY_PATTERN.key(), "1");
+        phoneReplaceConfig.put(ReplaceTransformConfig.KEY_REPLACEMENT.key(), 
"a");
+
+        Map<String, Object> nameReplaceConfig = new HashMap<>();
+        nameReplaceConfig.put("replace_field", "name");
+        nameReplaceConfig.put(ReplaceTransformConfig.KEY_PATTERN.key(), 
"before");
+        nameReplaceConfig.put(ReplaceTransformConfig.KEY_REPLACEMENT.key(), 
"after");
+
+        ReplaceTransform phoneReplaceTransform =
+                new 
ReplaceTransform(ReadonlyConfig.fromMap(phoneReplaceConfig), catalogTable);
+        ReplaceTransform nameReplaceTransform =
+                new 
ReplaceTransform(ReadonlyConfig.fromMap(nameReplaceConfig), catalogTable);
+
+        SeaTunnelRow input = new SeaTunnelRow(new Object[] {1, "before name", 
"17111999384"});
+
+        SeaTunnelRow phoneOutput = phoneReplaceTransform.transformRow(input);
+        SeaTunnelRow nameOutput = nameReplaceTransform.transformRow(input);
+
+        Assertions.assertEquals("a7aaa999384", phoneOutput.getField(2));
+        Assertions.assertEquals("after name", nameOutput.getField(1));
+
+        // The sibling transform branch should still see the original title 
value.
+        Assertions.assertEquals("17111999384", nameOutput.getField(2));
+
+        // ReplaceTransform should not mutate the shared input row.
+        Assertions.assertEquals("before name", input.getField(1));
+        Assertions.assertEquals("17111999384", input.getField(2));
+    }
 }