This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 5598ed7f3a [Fix][Transform-V2] Preserve input row kind when extracting
CDC metadata (#12580)
5598ed7f3a is described below
commit 5598ed7f3a88602032a18b617cb21c688aacf965
Author: hutiefang76 <[email protected]>
AuthorDate: Thu Oct 29 10:37:32 2026 +0000
[Fix][Transform-V2] Preserve input row kind when extracting CDC metadata
(#12580)
---
docs/en/transforms/rowkind-extractor.md | 2 +-
docs/zh/transforms/rowkind-extractor.md | 2 +-
.../transform/TestRowKindExtractorTransformIT.java | 7 +
.../test/resources/rowkind_extractor_fanout.conf | 141 +++++++++++++++++++++
.../rowkind/RowKindExtractorTransform.java | 2 +-
.../rowkind/RowKindExtractorTransformTest.java | 40 ++++++
6 files changed, 191 insertions(+), 3 deletions(-)
diff --git a/docs/en/transforms/rowkind-extractor.md
b/docs/en/transforms/rowkind-extractor.md
index 025284db61..40714bf20d 100644
--- a/docs/en/transforms/rowkind-extractor.md
+++ b/docs/en/transforms/rowkind-extractor.md
@@ -4,7 +4,7 @@
## Description
-The RowKindExtractor transform plugin is used to convert CDC (Change Data
Capture) data streams into Append-Only mode while extracting the original
RowKind information as a new field.
+The RowKindExtractor transform plugin is used to convert CDC (Change Data
Capture) data streams into Append-Only mode while extracting the original
RowKind information as a new field. Only the branch handled by this plugin
becomes INSERT, while other branches keep their original CDC RowKind.
**Core Features:**
- Converts all data rows' RowKind to `+I` (INSERT), achieving Append-Only mode
diff --git a/docs/zh/transforms/rowkind-extractor.md
b/docs/zh/transforms/rowkind-extractor.md
index c0622f03e2..04284f68ab 100644
--- a/docs/zh/transforms/rowkind-extractor.md
+++ b/docs/zh/transforms/rowkind-extractor.md
@@ -4,7 +4,7 @@
## 描述
-RowKindExtractor 转换插件用于将 CDC（Change Data Capture）数据流改写为
Append-Only（仅追加）模式，同时把原始的 RowKind 信息提取为一个新的字段。
+RowKindExtractor 转换插件用于将 CDC（Change Data Capture）数据流改写为
Append-Only（仅追加）模式，同时把原始的 RowKind 信息提取为一个新的字段。只有该插件处理的分支会变为 INSERT，其他分支仍保留原始的
CDC RowKind。
**核心功能：**
- 将所有数据行的 RowKind 统一改为 `+I`（INSERT），实现 Append-Only 模式
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestRowKindExtractorTransformIT.java
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestRowKindExtractorTransformIT.java
index a82ff2fb37..7302a10b00 100644
---
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestRowKindExtractorTransformIT.java
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestRowKindExtractorTransformIT.java
@@ -45,4 +45,11 @@ public class TestRowKindExtractorTransformIT extends
TestSuiteBase {
container.executeJob("/rowkind_extractor_transform_case1_multi_table.conf");
Assertions.assertEquals(0, execResult.getExitCode());
}
+
+ @TestTemplate
+ public void testOriginalRowKindPreservedAcrossBranches(TestContainer
container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
container.executeJob("/rowkind_extractor_fanout.conf");
+ Assertions.assertEquals(0, execResult.getExitCode());
+ }
}
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/rowkind_extractor_fanout.conf
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/rowkind_extractor_fanout.conf
new file mode 100644
index 0000000000..99709047e4
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/rowkind_extractor_fanout.conf
@@ -0,0 +1,141 @@
+#
+# 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"
+ row.num = 1
+ schema = {
+ table = "test.events"
+ fields {
+ pk_id = bigint
+ name = string
+ }
+ primaryKey {
+ name = "pk_id"
+ columnNames = [pk_id]
+ }
+ }
+ rows = [
+ {
+ kind = DELETE
+ fields = [1, "A"]
+ }
+ ]
+ }
+}
+
+transform {
+ RowKindExtractor {
+ plugin_input = "source1"
+ plugin_output = "extracted_a"
+ custom_field_name = "original_kind"
+ transform_type = FULL
+ }
+
+ TableRename {
+ plugin_input = "extracted_a"
+ plugin_output = "branch_a"
+ prefix = "a_"
+ }
+
+ RowKindExtractor {
+ plugin_input = "source1"
+ plugin_output = "extracted_b"
+ custom_field_name = "original_kind"
+ transform_type = FULL
+ }
+
+ TableRename {
+ plugin_input = "extracted_b"
+ plugin_output = "branch_b"
+ prefix = "b_"
+ }
+}
+
+sink {
+ Assert {
+ plugin_input = "branch_a"
+ rules = {
+ tables_configs = [
+ {
+ table_path = "test.a_events"
+ row_rules = [
+ {
+ rule_type = MAX_ROW
+ rule_value = 1
+ },
+ {
+ rule_type = MIN_ROW
+ rule_value = 1
+ }
+ ]
+ field_rules = [
+ {
+ field_name = original_kind
+ field_type = string
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ equals_to = "DELETE"
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ }
+
+ Assert {
+ plugin_input = "branch_b"
+ rules = {
+ tables_configs = [
+ {
+ table_path = "test.b_events"
+ row_rules = [
+ {
+ rule_type = MAX_ROW
+ rule_value = 1
+ },
+ {
+ rule_type = MIN_ROW
+ rule_value = 1
+ }
+ ]
+ field_rules = [
+ {
+ field_name = original_kind
+ field_type = string
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ equals_to = "DELETE"
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransform.java
index 58b80c50c1..ce9d98384a 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransform.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransform.java
@@ -54,8 +54,8 @@ public class RowKindExtractorTransform extends
SingleFieldOutputTransform {
@Override
protected SeaTunnelRow transformRow(SeaTunnelRow inputRow) {
Object fieldValue = getOutputFieldValue(new
SeaTunnelRowAccessor(inputRow));
- inputRow.setRowKind(RowKind.INSERT);
SeaTunnelRow outputRow = getRowContainerGenerator().apply(inputRow);
+ outputRow.setRowKind(RowKind.INSERT);
outputRow.setField(getFieldIndex(), fieldValue);
return outputRow;
}
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransformTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransformTest.java
index b958dab7d0..0aa143dac4 100644
---
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransformTest.java
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransformTest.java
@@ -150,4 +150,44 @@ class RowKindExtractorTransformTest {
"SeaTunnelRow{tableId=, kind=+I, fields=[value1, 1,
896657703886127105, 3.1415916, 3.14, DELETE]}",
rowKindExtractorTransform.transformRow(deleteRow).toString());
}
+
+ @Test
+ void testOriginalRowKindIsPreservedForSiblingBranches() {
+ RowKind[] allKinds = RowKind.values();
+ for (RowKindExtractorTransformType transformType :
RowKindExtractorTransformType.values()) {
+ HashMap<String, Object> conf = new HashMap<>();
+ conf.put("transform_type", transformType.name());
+ RowKindExtractorTransform rowKindExtractorTransform =
+ new
RowKindExtractorTransform(ReadonlyConfig.fromMap(conf), catalogTable);
+ rowKindExtractorTransform.initRowContainerGenerator();
+ for (RowKind originalKind : allKinds) {
+ SeaTunnelRow originalRow = inputRow.copy();
+ originalRow.setRowKind(originalKind);
+ originalRow.setTableId("original");
+ HashMap<String, Object> options = new HashMap<>();
+ options.put("marker", "kept");
+ originalRow.setOptions(options);
+
+ SeaTunnelRow outputRow =
rowKindExtractorTransform.map(originalRow);
+
+ Assertions.assertEquals(RowKind.INSERT,
outputRow.getRowKind());
+ String expectedKind =
+ transformType == RowKindExtractorTransformType.SHORT
+ ? originalKind.shortString()
+ : originalKind.name();
+ Assertions.assertEquals(
+ expectedKind,
+
outputRow.getField(rowKindExtractorTransform.getFieldIndex()));
+ Assertions.assertEquals(values.length + 1,
outputRow.getArity());
+ Assertions.assertEquals(values[0], outputRow.getField(0));
+ Assertions.assertEquals("original", outputRow.getTableId());
+ Assertions.assertEquals("kept",
outputRow.getOptionsOrNull().get("marker"));
+
+ Assertions.assertEquals(originalKind,
originalRow.getRowKind());
+ Assertions.assertArrayEquals(values, originalRow.getFields());
+ Assertions.assertEquals("original", originalRow.getTableId());
+ Assertions.assertEquals("kept",
originalRow.getOptionsOrNull().get("marker"));
+ }
+ }
+ }
}
