This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12580-c6185fb3c3263f6b20fd974d653098181d4b4344 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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")); + } + } + } }
