This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 0bccf8b275 [Feature][Transform-V2] Add TextChunk transform plugin
(#11414)
0bccf8b275 is described below
commit 0bccf8b275b9be12920d7cda630aeeee0e1ebfe8
Author: Shuai Liu <[email protected]>
AuthorDate: Fri Jul 24 18:30:10 2026 +0800
[Feature][Transform-V2] Add TextChunk transform plugin (#11414)
---
docs/en/transforms/text-chunk.md | 164 ++++++
docs/sidebars.js | 1 +
docs/zh/transforms/text-chunk.md | 147 +++++
.../connector/TransformSpecificationCheckTest.java | 2 +-
.../seatunnel/e2e/transform/TestTextChunkIT.java | 51 ++
.../src/test/resources/text_chunk_transform.conf | 98 ++++
.../text_chunk_transform_multi_table.conf | 136 +++++
.../resources/text_chunk_transform_overlap.conf | 105 ++++
.../chunk/TextChunkMultiCatalogTransform.java | 52 ++
.../transform/chunk/TextChunkTransform.java | 257 +++++++++
.../transform/chunk/TextChunkTransformConfig.java | 110 ++++
.../transform/chunk/TextChunkTransformFactory.java | 146 +++++
.../seatunnel/transform/chunk/TextChunker.java | 179 ++++++
.../chunk/TextChunkTransformFactoryTest.java | 248 ++++++++
.../transform/chunk/TextChunkTransformTest.java | 636 +++++++++++++++++++++
.../seatunnel/transform/chunk/TextChunkerTest.java | 142 +++++
16 files changed, 2473 insertions(+), 1 deletion(-)
diff --git a/docs/en/transforms/text-chunk.md b/docs/en/transforms/text-chunk.md
new file mode 100644
index 0000000000..815da88b36
--- /dev/null
+++ b/docs/en/transforms/text-chunk.md
@@ -0,0 +1,164 @@
+# TextChunk
+
+> TextChunk transform plugin
+
+## Description
+
+Splits a long text field into multiple smaller chunks, emitting **one row per
chunk** (a 1-row → N-rows
+transform). Each output row keeps **all of the original source fields**
unchanged, and appends two columns: the
+chunk text and its 0-based sequence index within the source document.
+
+## Options
+
+| name | type | required | default value |
+|-------------------|--------|----------|---------------|
+| text_field | string | yes | |
+| output_field | string | no | chunk |
+| chunk_index_field | string | no | chunk_index |
+| chunk_size | int | no | 1000 |
+| overlap_size | int | no | 0 |
+| separators | array | no | ["\n\n", "\n", "。", "！", "？", ". ",
" "] |
+| skip_empty_text | boolean| no | true |
+
+### text_field [string]
+
+The source text field to split into chunks.
+
+### output_field [string]
+
+The name of the output column holding each chunk (type `STRING`). If a column
with this name already
+exists it is reused, otherwise a new column is appended. Defaults to `chunk`,
but **must not** match a
+column that takes part in a primary key or a unique key.
+
+### chunk_index_field [string]
+
+The name of the output column holding the chunk sequence index within a
document (type `INT`,
+0-based). If a column with this name already exists it is reused, otherwise a
new column is appended.
+Defaults to `chunk_index`, but **must not** match a column that takes part in
a primary key or a
+unique key.
+
+### chunk_size [int]
+
+Maximum length of each chunk, counted in **Unicode code points** — a
supplementary character such as
+an emoji counts as one. (A combining sequence or a ZWJ-joined emoji still
counts as several code
+points and may be split between them.) Must be greater than `0`. Defaults to
`1000`.
+
+### overlap_size [int]
+
+Overlap length between adjacent chunks, counted in **Unicode code points**.
+Carrying a little context from the end
+of one chunk into the start of the next helps preserve meaning across chunk
boundaries. The overlap is
+made up of **whole trailing pieces** (complete words/sentences), so it never
begins in the middle of a
+word; `overlap_size` is therefore an upper bound — the actual overlap rounds
down to whole pieces, and
+is empty when even the last single piece is longer than the budget. (In the
empty-`separators`
+fixed-size fallback there are no pieces, so the overlap is a plain
`overlap_size`-code-point window and
+is not rounded to a word boundary.) Must satisfy `0 <= overlap_size <
chunk_size`. Defaults to `0`.
+
+> Note: the number of chunks grows roughly as `chunk_size / (chunk_size -
overlap_size)`, so an
+> `overlap_size` close to `chunk_size` multiplies the output row count (and
memory) dramatically —
+> e.g. `chunk_size = 1000` with `overlap_size = 999` produces about one chunk
per code point. Keep the
+> overlap well below `chunk_size`.
+
+### separators [array]
+
+Separators tried in priority order to avoid cutting mid-sentence: the text is
split by the first
+separator, and any piece still longer than `chunk_size` is further split by
the next separator, and so
+on. If the list is left empty, the transform falls back to fixed-size
splitting. Defaults to
+`["\n\n", "\n", "。", "！", "？", ". ", " "]`.
+
+### skip_empty_text [boolean]
+
+Controls how a row whose `text_field` is `null` or empty is handled. When
`true` the row is dropped;
+when `false` the row is passed through (`output_field` set to `null`,
`chunk_index_field` set to `0`).
+Defaults to `true`.
+
+### common options [string]
+
+Transform plugin common parameters, please refer to [Transform
Plugin](common-options/common-options.md) for details.
+
+## Behavior
+
+- A `null` or empty text value produces **no** output rows for that input row
when `skip_empty_text = true` (the default); set `skip_empty_text = false` to
pass such a row through instead (`chunk = null`, `chunk_index = 0`).
+- Each produced chunk is at most `chunk_size` Unicode code points long (the
carried overlap counts toward the budget); a surrogate pair (e.g. an emoji)
counts as one code point and is never split.
+- Separators are retained: each separator stays attached to the end of the
piece it follows, so merging pieces to fill a chunk keeps the spaces/newlines
between them (words split on `" "` are re-joined as `a b`, not `ab`) and runs
of consecutive separators (e.g. blank lines) are preserved. With `overlap_size
= 0` the chunks concatenate back to the original text.
+- The rows produced from one input row all carry the same source key (e.g. the
same `id`). To keep it unique, the transform appends `chunk_index_field` to the
primary key and to every unique key.
+- Because of that key extension, `output_field` and `chunk_index_field` must
not reuse a column that is part of a primary or unique key: overwriting such a
key column with per-chunk values would let chunks from different source rows
collide in key-aware sinks, so the transform rejects that configuration with an
error. Reusing a non-key column (e.g. writing the chunk back into the source
text column) is still allowed.
+
+## Example
+
+The data read from source is a table like this:
+
+| id | content |
+|----|-----------------------------------------------|
+| 1 | SeaTunnel is a data integration platform. ... |
+
+We want to split the `content` field into overlapping chunks, so we can add a
`TextChunk` transform
+like this:
+
+```
+transform {
+ TextChunk {
+ plugin_input = "fake"
+ plugin_output = "fake1"
+ text_field = "content"
+ output_field = "chunk"
+ chunk_index_field = "chunk_index"
+ chunk_size = 200
+ overlap_size = 20
+ }
+}
+```
+
+Then each input row is expanded into one row per chunk in result table
`fake1`, keeping the original
+fields and adding `chunk` / `chunk_index`:
+
+| id | content | chunk
| chunk_index |
+|----|-----------------------------------------------|------------------------|-------------|
+| 1 | SeaTunnel is a data integration platform. ... | SeaTunnel is a data...
| 0 |
+| 1 | SeaTunnel is a data integration platform. ... | ...platform that ...
| 1 |
+
+## Job Config Example
+
+```
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ plugin_output = "fake"
+ row.num = 5
+ schema = {
+ fields {
+ id = "int"
+ content = "string"
+ }
+ }
+ }
+}
+
+transform {
+ TextChunk {
+ plugin_input = "fake"
+ plugin_output = "fake1"
+ text_field = "content"
+ output_field = "chunk"
+ chunk_index_field = "chunk_index"
+ chunk_size = 200
+ overlap_size = 20
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "fake1"
+ }
+}
+```
+
+## Changelog
+
+### new version
+
+- Add TextChunk Transform Plugin
diff --git a/docs/sidebars.js b/docs/sidebars.js
index 81e307bd34..217c4c993e 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -215,6 +215,7 @@ const sidebars = {
"transforms/table-filter",
"transforms/table-merge",
"transforms/table-rename",
+ "transforms/text-chunk",
"transforms/transform-multi-table",
"transforms/multi-table-transform-and-join-boundary"
]
diff --git a/docs/zh/transforms/text-chunk.md b/docs/zh/transforms/text-chunk.md
new file mode 100644
index 0000000000..2d03448eb3
--- /dev/null
+++ b/docs/zh/transforms/text-chunk.md
@@ -0,0 +1,147 @@
+# TextChunk
+
+> 文本分块转换插件
+
+## 描述
+
+将一个长文本字段切分成多个较小的块（chunk），**每个块输出为一行**（1 行 → N 行的转换）。
每个输出行都会**原样保留全部源字段**，并追加两列：块文本，以及该块在源文档内的序号（从 0 开始）。
+
+## 属性
+
+| 名称 | 类型 | 是否必须 | 默认值 |
+|-------------------|--------|------|-----|
+| text_field | string | yes | |
+| output_field | string | no | chunk |
+| chunk_index_field | string | no | chunk_index |
+| chunk_size | int | no | 1000 |
+| overlap_size | int | no | 0 |
+| separators | array | no | ["\n\n", "\n", "。", "！", "？", ". ", " "]
|
+| skip_empty_text | boolean| no | true |
+
+### text_field [string]
+
+需要切分成块的源文本字段。
+
+### output_field [string]
+
+存放每个块的输出列名（类型 `STRING`）。若已存在同名列则复用该列，否则追加新列。默认值为 `chunk`，但**不能**与主键或唯一键所包含的列同名。
+
+### chunk_index_field [string]
+
+存放块在文档内序号的输出列名（类型 `INT`，从 0 开始）。若已存在同名列则复用该列，否则追加新列。默认值为
`chunk_index`，但**不能**与主键或唯一键所包含的列同名。
+
+### chunk_size [int]
+
+每个块的最大长度，以 **Unicode 码点（code point）** 计——一个增补字符（如 emoji）计为一个码点。（组合字符或 ZWJ 连字
emoji 仍算作
+多个码点，可能在码点之间被切分。）必须大于 `0`。默认值为 `1000`。
+
+### overlap_size [int]
+
+相邻块之间的重叠长度，以 **Unicode 码点（code point）**
计。把前一个块末尾的少量上下文带入下一个块开头，有助于在块边界处保留语义。重叠由
+**若干个完整的片段**（完整的词/句）组成，因此绝不会从单词中间开始；`overlap_size` 是一个上界——实际重叠会向下取整到完整片段，
+当连最后一个片段都超过该预算时，则不携带任何重叠。（当 `separators` 留空、回退为固定长度切分时不存在"片段"，此时重叠为
+`overlap_size` 个码点的定长窗口，不会向词边界取整。）必须满足 `0 <= overlap_size < chunk_size`。默认值为
`0`。
+
+> 注意：块的数量大致按 `chunk_size / (chunk_size - overlap_size)` 增长，因此当 `overlap_size`
接近
+> `chunk_size` 时，输出行数（以及内存占用）会急剧膨胀——例如 `chunk_size = 1000` 且 `overlap_size =
999`
+> 时，几乎每个码点都会产生一个块。请让 overlap 明显小于 `chunk_size`。
+
+### separators [array]
+
+按优先级顺序尝试的分隔符，用于避免从句子中间切断：先用第一个分隔符切分，任何仍长于 `chunk_size` 的片段再用
+下一个分隔符继续切分，依此类推。若该列表留空，则回退为固定长度切分。默认值为
+`["\n\n", "\n", "。", "！", "？", ". ", " "]`。
+
+### skip_empty_text [boolean]
+
+控制 `text_field` 为 `null` 或空字符串时的行处理逻辑。为 `true` 时该行被丢弃，为 `false` 时该行透传
+（`output_field` 置为 `null`、`chunk_index_field` 置为 `0`），默认值为 `true`。
+
+### common options [string]
+
+转换插件的常见参数, 请参考 [Transform Plugin](common-options/common-options.md) 了解详情。
+
+## 行为说明
+
+- 当 `skip_empty_text = true`（默认）时，文本值为 `null` 或空字符串的输入行**不产生**任何输出行；设为
`skip_empty_text = false` 则该行会被透传（`chunk = null`、`chunk_index = 0`）。
+- 每个产生的块长度不超过 `chunk_size` 个 Unicode 码点（携带的 overlap 也计入该上限）；代理对（例如
emoji）计为一个码点，不会被劈开。
+- 分隔符会被保留：每个分隔符附着在它前面那一片的末尾，因此合并多片以填满一个块时，片与片之间的空格/换行会被保留（按 `" "` 切分的单词会重新拼成
`a b` 而不是 `ab`），连续的分隔符（例如空行）也会被保留。当 `overlap_size = 0` 时，各块拼接后可还原为原始文本。
+- 同一输入行产生的多行都带有相同的源键（例如相同的 `id`）。为让它仍然唯一，转换会把 `chunk_index_field` 追加到主键和每个唯一键上。
+- 正因为存在上述键扩展，`output_field` 与 `chunk_index_field`
不能复用主键或唯一键所包含的列：用每块的值覆盖这样的键列，会导致不同源行的块在依赖键的 sink
中碰撞，因此转换会对这类配置直接报错拒绝。复用非键列（例如把块写回源文本列）仍然允许。
+
+## 示例
+
+源端数据读取的表格如下：
+
+| id | content |
+|----|-----------------------------------------------|
+| 1 | SeaTunnel is a data integration platform. ... |
+
+我们想要将 `content` 字段切分成带重叠的块，可以像这样添加 `TextChunk` 转换：
+
+```
+transform {
+ TextChunk {
+ plugin_input = "fake"
+ plugin_output = "fake1"
+ text_field = "content"
+ output_field = "chunk"
+ chunk_index_field = "chunk_index"
+ chunk_size = 200
+ overlap_size = 20
+ }
+}
+```
+
+那么每个输入行会被展开为每块一行，写入结果表 `fake1`，并保留原始字段、追加 `chunk` / `chunk_index`：
+
+| id | content | chunk
| chunk_index |
+|----|-----------------------------------------------|------------------------|-------------|
+| 1 | SeaTunnel is a data integration platform. ... | SeaTunnel is a data...
| 0 |
+| 1 | SeaTunnel is a data integration platform. ... | ...platform that ...
| 1 |
+
+## 作业配置示例
+
+```
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ plugin_output = "fake"
+ row.num = 5
+ schema = {
+ fields {
+ id = "int"
+ content = "string"
+ }
+ }
+ }
+}
+
+transform {
+ TextChunk {
+ plugin_input = "fake"
+ plugin_output = "fake1"
+ text_field = "content"
+ output_field = "chunk"
+ chunk_index_field = "chunk_index"
+ chunk_size = 200
+ overlap_size = 20
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "fake1"
+ }
+}
+```
+
+## 更新日志
+
+### 新版本
+
+- 添加文本分块（TextChunk）转换插件
diff --git
a/seatunnel-dist/src/test/java/org/apache/seatunnel/api/connector/TransformSpecificationCheckTest.java
b/seatunnel-dist/src/test/java/org/apache/seatunnel/api/connector/TransformSpecificationCheckTest.java
index d963d315df..78bb6c0164 100644
---
a/seatunnel-dist/src/test/java/org/apache/seatunnel/api/connector/TransformSpecificationCheckTest.java
+++
b/seatunnel-dist/src/test/java/org/apache/seatunnel/api/connector/TransformSpecificationCheckTest.java
@@ -43,7 +43,7 @@ class TransformSpecificationCheckTest {
FactoryUtil.discoverFactories(
Thread.currentThread().getContextClassLoader(),
TableTransformFactory.class);
- Assertions.assertEquals(22, factories.size());
+ Assertions.assertEquals(23, factories.size());
}
@Test
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestTextChunkIT.java
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestTextChunkIT.java
new file mode 100644
index 0000000000..8a3cc6d1e8
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestTextChunkIT.java
@@ -0,0 +1,51 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.e2e.transform;
+
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+
+import java.io.IOException;
+
+public class TestTextChunkIT extends TestSuiteBase {
+
+ @TestTemplate
+ public void testTextChunk(TestContainer container) throws IOException,
InterruptedException {
+ Container.ExecResult execResult =
container.executeJob("/text_chunk_transform.conf");
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ }
+
+ @TestTemplate
+ public void testTextChunkMultiTable(TestContainer container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
+ container.executeJob("/text_chunk_transform_multi_table.conf");
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ }
+
+ @TestTemplate
+ public void testTextChunkOverlapWithSeparatorsAndNonBmp(TestContainer
container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
+ container.executeJob("/text_chunk_transform_overlap.conf");
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/text_chunk_transform.conf
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/text_chunk_transform.conf
new file mode 100644
index 0000000000..35b00da93d
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/text_chunk_transform.conf
@@ -0,0 +1,98 @@
+#
+# 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 = "fake"
+ schema = {
+ fields {
+ id = "int"
+ content = "string"
+ }
+ }
+ rows = [
+ {fields = [1, "abcabcabc"], kind = INSERT}
+ ]
+ }
+}
+
+transform {
+ TextChunk {
+ plugin_input = "fake"
+ plugin_output = "fake1"
+ text_field = "content"
+ output_field = "chunk"
+ chunk_index_field = "chunk_index"
+ chunk_size = 3
+ overlap_size = 0
+ separators = []
+ }
+}
+
+sink {
+ Assert {
+ plugin_input = "fake1"
+ rules = {
+ row_rules = [
+ {
+ rule_type = MIN_ROW
+ rule_value = 3
+ },
+ {
+ rule_type = MAX_ROW
+ rule_value = 3
+ }
+ ]
+ field_rules = [
+ {
+ field_name = "id"
+ field_type = "int"
+ field_value = [
+ {equals_to = 1}
+ ]
+ },
+ {
+ field_name = "content"
+ field_type = "string"
+ field_value = [
+ {equals_to = "abcabcabc"}
+ ]
+ },
+ {
+ field_name = "chunk"
+ field_type = "string"
+ field_value = [
+ {equals_to = "abc"}
+ ]
+ },
+ {
+ field_name = "chunk_index"
+ field_type = "int"
+ field_value = [
+ {rule_type = MIN, rule_value = 0},
+ {rule_type = MAX, rule_value = 2}
+ ]
+ }
+ ]
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/text_chunk_transform_multi_table.conf
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/text_chunk_transform_multi_table.conf
new file mode 100644
index 0000000000..368141843d
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/text_chunk_transform_multi_table.conf
@@ -0,0 +1,136 @@
+#
+# 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 = "fake"
+ tables_configs = [
+ {
+ schema = {
+ table = "test.doc_a"
+ columns = [
+ {name = "id", type = "int"},
+ {name = "content", type = "string"}
+ ]
+ }
+ rows = [
+ {fields = [1, "xxxx"], kind = INSERT}
+ ]
+ },
+ {
+ schema = {
+ table = "test.doc_b"
+ columns = [
+ {name = "id", type = "int"},
+ {name = "content", type = "string"}
+ ]
+ }
+ rows = [
+ {fields = [2, "yy yy yy "], kind = INSERT}
+ ]
+ }
+ ]
+ }
+}
+
+transform {
+ TextChunk {
+ plugin_input = "fake"
+ plugin_output = "fake1"
+ text_field = "content"
+ chunk_size = 3
+ overlap_size = 0
+ table_transform = [
+ {
+ table_path = "test.doc_a"
+ text_field = "content"
+ chunk_size = 2
+ overlap_size = 0
+ separators = []
+ }
+ ]
+ }
+}
+
+sink {
+ Assert {
+ plugin_input = "fake1"
+ rules = {
+ tables_configs = [
+ {
+ table_path = "test.doc_a"
+ row_rules = [
+ {rule_type = MIN_ROW, rule_value = 2},
+ {rule_type = MAX_ROW, rule_value = 2}
+ ]
+ field_rules = [
+ {
+ field_name = "id"
+ field_type = "int"
+ field_value = [{equals_to = 1}]
+ },
+ {
+ field_name = "chunk"
+ field_type = "string"
+ field_value = [{equals_to = "xx"}]
+ },
+ {
+ field_name = "chunk_index"
+ field_type = "int"
+ field_value = [
+ {rule_type = MIN, rule_value = 0},
+ {rule_type = MAX, rule_value = 1}
+ ]
+ }
+ ]
+ },
+ {
+ table_path = "test.doc_b"
+ row_rules = [
+ {rule_type = MIN_ROW, rule_value = 3},
+ {rule_type = MAX_ROW, rule_value = 3}
+ ]
+ field_rules = [
+ {
+ field_name = "id"
+ field_type = "int"
+ field_value = [{equals_to = 2}]
+ },
+ {
+ field_name = "chunk"
+ field_type = "string"
+ field_value = [{equals_to = "yy "}]
+ },
+ {
+ field_name = "chunk_index"
+ field_type = "int"
+ field_value = [
+ {rule_type = MIN, rule_value = 0},
+ {rule_type = MAX, rule_value = 2}
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/text_chunk_transform_overlap.conf
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/text_chunk_transform_overlap.conf
new file mode 100644
index 0000000000..822a8df3e9
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/text_chunk_transform_overlap.conf
@@ -0,0 +1,105 @@
+#
+# 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.
+#
+
+# Exercises the separator-priority + greedy-merge + whole-piece overlap path
end-to-end, with a
+# non-BMP (surrogate-pair) payload, so the code paths the docs call fragile
all cross the engine
+# serialization boundary. Input is six "😀 " pieces (each 2 Unicode code
points); with
+# separators=[" "], chunk_size=4 and overlap_size=2 (code points) every chunk
is "😀 😀 " (4 code
+# points / 6 UTF-16 code units) and one whole "😀 " piece is carried forward as
overlap, yielding
+# exactly 5 chunks.
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ plugin_output = "fake"
+ schema = {
+ fields {
+ id = "int"
+ content = "string"
+ }
+ }
+ rows = [
+ {fields = [1, "😀 😀 😀 😀 😀 😀 "], kind = INSERT}
+ ]
+ }
+}
+
+transform {
+ TextChunk {
+ plugin_input = "fake"
+ plugin_output = "fake1"
+ text_field = "content"
+ output_field = "chunk"
+ chunk_index_field = "chunk_index"
+ chunk_size = 4
+ overlap_size = 2
+ separators = [" "]
+ }
+}
+
+sink {
+ Assert {
+ plugin_input = "fake1"
+ rules = {
+ row_rules = [
+ {
+ rule_type = MIN_ROW
+ rule_value = 5
+ },
+ {
+ rule_type = MAX_ROW
+ rule_value = 5
+ }
+ ]
+ field_rules = [
+ {
+ field_name = "id"
+ field_type = "int"
+ field_value = [
+ {equals_to = 1}
+ ]
+ },
+ {
+ field_name = "content"
+ field_type = "string"
+ field_value = [
+ {equals_to = "😀 😀 😀 😀 😀 😀 "}
+ ]
+ },
+ {
+ field_name = "chunk"
+ field_type = "string"
+ field_value = [
+ {equals_to = "😀 😀 "}
+ ]
+ },
+ {
+ field_name = "chunk_index"
+ field_type = "int"
+ field_value = [
+ {rule_type = MIN, rule_value = 0},
+ {rule_type = MAX, rule_value = 4}
+ ]
+ }
+ ]
+ }
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkMultiCatalogTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkMultiCatalogTransform.java
new file mode 100644
index 0000000000..d512ff82f8
--- /dev/null
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkMultiCatalogTransform.java
@@ -0,0 +1,52 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.transform.chunk;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.transform.SeaTunnelFlatMapTransform;
+import org.apache.seatunnel.api.transform.SeaTunnelTransform;
+import
org.apache.seatunnel.transform.common.AbstractMultiCatalogFlatMapTransform;
+import org.apache.seatunnel.transform.common.IdentityFlatMapTransform;
+
+import java.util.List;
+
+public class TextChunkMultiCatalogTransform extends
AbstractMultiCatalogFlatMapTransform {
+
+ public TextChunkMultiCatalogTransform(
+ List<CatalogTable> inputCatalogTables, ReadonlyConfig config) {
+ super(inputCatalogTables, config);
+ }
+
+ @Override
+ public String getPluginName() {
+ return TextChunkTransform.PLUGIN_NAME;
+ }
+
+ @Override
+ protected SeaTunnelFlatMapTransform<SeaTunnelRow> buildTransform(
+ CatalogTable inputCatalogTable, ReadonlyConfig config) {
+ return new TextChunkTransform(TextChunkTransformConfig.of(config),
inputCatalogTable);
+ }
+
+ @Override
+ protected SeaTunnelTransform<SeaTunnelRow>
createIdentityTransform(CatalogTable catalogTable) {
+ return new IdentityFlatMapTransform(catalogTable);
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkTransform.java
new file mode 100644
index 0000000000..8566d9be1b
--- /dev/null
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkTransform.java
@@ -0,0 +1,257 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.transform.chunk;
+
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.ConstraintKey;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import
org.apache.seatunnel.api.table.schema.handler.AlterTableSchemaEventHandler;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import
org.apache.seatunnel.transform.common.AbstractCatalogSupportFlatMapTransform;
+import org.apache.seatunnel.transform.exception.TransformCommonError;
+
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class TextChunkTransform extends AbstractCatalogSupportFlatMapTransform
{
+
+ public static final String PLUGIN_NAME = "TextChunk";
+
+ private static final double OVERLAP_WARN_RATIO = 0.5;
+
+ private final TextChunkTransformConfig config;
+ private int textFieldIndex;
+
+ private int outputFieldCount;
+ private int chunkFieldIndex;
+ private int chunkIndexFieldIndex;
+
+ public TextChunkTransform(
+ @NonNull TextChunkTransformConfig config, @NonNull CatalogTable
inputCatalogTable) {
+ super(inputCatalogTable);
+ this.config = config;
+ this.textFieldIndex = resolveTextFieldIndex();
+ if (config.getOverlapSize() >= config.getChunkSize() *
OVERLAP_WARN_RATIO) {
+ log.warn(
+ "Configured overlap_size={} exceeds {}% of chunk_size={},
which may cause "
+ + "row-count and memory amplification. Consider a
smaller overlap_size.",
+ config.getOverlapSize(),
+ (int) (OVERLAP_WARN_RATIO * 100),
+ config.getChunkSize());
+ }
+ getProducedCatalogTable();
+ }
+
+ @Override
+ public String getPluginName() {
+ return PLUGIN_NAME;
+ }
+
+ @Override
+ public SchemaChangeEvent mapSchemaChangeEvent(SchemaChangeEvent event) {
+ if (event instanceof AlterTableEvent) {
+ TableSchema newSchema =
+ new AlterTableSchemaEventHandler()
+ .reset(inputCatalogTable.getTableSchema())
+ .apply(event);
+ CatalogTable newInputCatalogTable =
+ CatalogTable.of(
+ inputCatalogTable.getTableId(),
+ newSchema,
+ inputCatalogTable.getOptions(),
+ inputCatalogTable.getPartitionKeys(),
+ inputCatalogTable.getComment(),
+ inputCatalogTable.getTableId().getCatalogName(),
+ inputCatalogTable.getMetadataSchema());
+ setInputCatalogTable(newInputCatalogTable);
+ }
+ return event;
+ }
+
+ @Override
+ public void setInputCatalogTable(@NonNull CatalogTable inputCatalogTable) {
+ super.setInputCatalogTable(inputCatalogTable);
+ this.textFieldIndex = resolveTextFieldIndex();
+ getProducedCatalogTable();
+ }
+
+ private int resolveTextFieldIndex() {
+ SeaTunnelRowType rowType =
inputCatalogTable.getTableSchema().toPhysicalRowDataType();
+ try {
+ return rowType.indexOf(config.getTextField());
+ } catch (IllegalArgumentException e) {
+ throw TransformCommonError.cannotFindInputFieldError(
+ getPluginName(), config.getTextField());
+ }
+ }
+
+ @Override
+ protected List<SeaTunnelRow> transformRow(SeaTunnelRow inputRow) {
+ Object value = inputRow.getField(textFieldIndex);
+ if (value == null || value.toString().isEmpty()) {
+ if (config.isSkipEmptyText()) {
+ log.debug(
+ "Dropping row because text_field '{}' is null or empty
(skip_empty_text=true)",
+ config.getTextField());
+ return Collections.emptyList();
+ }
+ return Collections.singletonList(buildOutputRow(inputRow, null,
0));
+ }
+ List<String> chunks =
+ TextChunker.split(
+ value.toString(),
+ config.getSeparators(),
+ config.getChunkSize(),
+ config.getOverlapSize());
+
+ List<SeaTunnelRow> outputRows = new ArrayList<>(chunks.size());
+ for (int i = 0; i < chunks.size(); i++) {
+ outputRows.add(buildOutputRow(inputRow, chunks.get(i), i));
+ }
+ return outputRows;
+ }
+
+ private SeaTunnelRow buildOutputRow(SeaTunnelRow inputRow, String chunk,
int chunkIndex) {
+ Object[] fields = Arrays.copyOf(inputRow.getFields(),
outputFieldCount);
+ fields[chunkFieldIndex] = chunk;
+ fields[chunkIndexFieldIndex] = chunkIndex;
+ SeaTunnelRow outputRow = new SeaTunnelRow(fields);
+ outputRow.setTableId(inputRow.getTableId());
+ outputRow.setRowKind(inputRow.getRowKind());
+ outputRow.setOptions(inputRow.getOptions());
+ return outputRow;
+ }
+
+ @Override
+ protected TableSchema transformTableSchema() {
+ List<Column> columns =
+ inputCatalogTable.getTableSchema().getColumns().stream()
+ .map(Column::copy)
+ .collect(Collectors.toList());
+
+ this.chunkFieldIndex =
+ addOrReplace(columns, config.getOutputField(),
BasicType.STRING_TYPE, true);
+ this.chunkIndexFieldIndex =
+ addOrReplace(columns, config.getChunkIndexField(),
BasicType.INT_TYPE, false);
+ this.outputFieldCount = columns.size();
+
+ TableSchema.Builder builder = TableSchema.builder().columns(columns);
+
+ PrimaryKey primaryKey =
inputCatalogTable.getTableSchema().getPrimaryKey();
+ if (primaryKey != null) {
+ builder.primaryKey(extendPrimaryKeyWithChunkIndex(primaryKey));
+ }
+
+ List<ConstraintKey> outputConstraintKeys =
+ inputCatalogTable.getTableSchema().getConstraintKeys().stream()
+ .map(
+ key ->
+ key.getConstraintType()
+ ==
ConstraintKey.ConstraintType.UNIQUE_KEY
+ ?
extendUniqueConstraintKeyWithChunkIndex(key)
+ : key.copy())
+ .collect(Collectors.toList());
+ builder.constraintKey(outputConstraintKeys);
+
+ return builder.build();
+ }
+
+ private Set<String> keyColumnNames() {
+ Set<String> keyColumns = new LinkedHashSet<>();
+ PrimaryKey primaryKey =
inputCatalogTable.getTableSchema().getPrimaryKey();
+ if (primaryKey != null && primaryKey.getColumnNames() != null) {
+ keyColumns.addAll(primaryKey.getColumnNames());
+ }
+ for (ConstraintKey constraintKey :
inputCatalogTable.getTableSchema().getConstraintKeys()) {
+ if (constraintKey.getConstraintType() ==
ConstraintKey.ConstraintType.UNIQUE_KEY) {
+ constraintKey
+ .getColumnNames()
+ .forEach(column ->
keyColumns.add(column.getColumnName()));
+ }
+ }
+ return keyColumns;
+ }
+
+ private PrimaryKey extendPrimaryKeyWithChunkIndex(PrimaryKey primaryKey) {
+ List<String> columnNames = new
ArrayList<>(primaryKey.getColumnNames());
+ if (!Boolean.TRUE.equals(primaryKey.getEnableAutoId())
+ && !columnNames.contains(config.getChunkIndexField())) {
+ columnNames.add(config.getChunkIndexField());
+ }
+ return PrimaryKey.of(primaryKey.getPrimaryKey(), columnNames,
primaryKey.getEnableAutoId());
+ }
+
+ private ConstraintKey
extendUniqueConstraintKeyWithChunkIndex(ConstraintKey uniqueKey) {
+ List<ConstraintKey.ConstraintKeyColumn> keyColumns =
+ uniqueKey.getColumnNames().stream()
+ .map(ConstraintKey.ConstraintKeyColumn::copy)
+ .collect(Collectors.toList());
+ boolean present =
+ keyColumns.stream()
+ .anyMatch(c ->
c.getColumnName().equals(config.getChunkIndexField()));
+ if (!present) {
+ keyColumns.add(
+ ConstraintKey.ConstraintKeyColumn.of(
+ config.getChunkIndexField(),
ConstraintKey.ColumnSortType.ASC));
+ }
+ return ConstraintKey.of(
+ uniqueKey.getConstraintType(), uniqueKey.getConstraintName(),
keyColumns);
+ }
+
+ @Override
+ protected TableIdentifier transformTableIdentifier() {
+ return inputCatalogTable.getTableId().copy();
+ }
+
+ private int addOrReplace(
+ List<Column> columns, String name, SeaTunnelDataType<?> type,
boolean nullable) {
+ for (int i = 0; i < columns.size(); i++) {
+ if (columns.get(i).getName().equals(name)) {
+ if (keyColumnNames().contains(name)) {
+ throw TransformCommonError.validationFailed(
+ String.format(
+ "%s transform: output/index field '%s'
must not reuse a "
+ + "primary/unique key column",
+ getPluginName(), name));
+ }
+ columns.set(i, columns.get(i).copy(type));
+ return i;
+ }
+ }
+ columns.add(PhysicalColumn.of(name, type, (Long) null, nullable, null,
""));
+ return columns.size() - 1;
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkTransformConfig.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkTransformConfig.java
new file mode 100644
index 0000000000..aaaf2fbff4
--- /dev/null
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkTransformConfig.java
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.transform.chunk;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.List;
+
+@Getter
+@Setter
+public class TextChunkTransformConfig implements Serializable {
+
+ public static final Option<String> TEXT_FIELD =
+ Options.key("text_field")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The source text field to split into
chunks");
+
+ public static final Option<String> OUTPUT_FIELD =
+ Options.key("output_field")
+ .stringType()
+ .defaultValue("chunk")
+ .withDescription(
+ "The output field name holding each chunk
(STRING). Must not match an "
+ + "existing primary or unique key column");
+
+ public static final Option<String> CHUNK_INDEX_FIELD =
+ Options.key("chunk_index_field")
+ .stringType()
+ .defaultValue("chunk_index")
+ .withDescription(
+ "The output field name holding the chunk sequence
index within a "
+ + "document (INT, 0-based). Must not match
an existing primary "
+ + "or unique key column");
+
+ public static final Option<Integer> CHUNK_SIZE =
+ Options.key("chunk_size")
+ .intType()
+ .defaultValue(1000)
+ .withDescription(
+ "Maximum length of each chunk, counted in Unicode
code points");
+
+ public static final Option<Integer> OVERLAP_SIZE =
+ Options.key("overlap_size")
+ .intType()
+ .defaultValue(0)
+ .withDescription(
+ "Overlap length between adjacent chunks, counted
in Unicode code "
+ + "points. Must be less than chunk_size");
+
+ public static final Option<List<String>> SEPARATORS =
+ Options.key("separators")
+ .listType()
+ .defaultValue(Arrays.asList("\n\n", "\n", "。", "！", "？",
". ", " "))
+ .withDescription(
+ "Separators tried in priority order to avoid
cutting mid-sentence. "
+ + "If left empty, falls back to fixed-size
splitting");
+
+ public static final Option<Boolean> SKIP_EMPTY_TEXT =
+ Options.key("skip_empty_text")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "When true (default), a row whose text_field is
null or empty "
+ + "produces no output rows (the row is
dropped). When false, "
+ + "such a row is passed through unchanged
with output_field set "
+ + "to null and chunk_index_field set to
0.");
+
+ private String textField;
+ private String outputField;
+ private String chunkIndexField;
+ private int chunkSize;
+ private int overlapSize;
+ private List<String> separators;
+ private boolean skipEmptyText;
+
+ public static TextChunkTransformConfig of(ReadonlyConfig config) {
+ TextChunkTransformConfig c = new TextChunkTransformConfig();
+ c.setTextField(config.get(TEXT_FIELD));
+ c.setOutputField(config.get(OUTPUT_FIELD));
+ c.setChunkIndexField(config.get(CHUNK_INDEX_FIELD));
+ c.setChunkSize(config.get(CHUNK_SIZE));
+ c.setOverlapSize(config.get(OVERLAP_SIZE));
+ c.setSeparators(config.get(SEPARATORS));
+ c.setSkipEmptyText(config.get(SKIP_EMPTY_TEXT));
+ return c;
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkTransformFactory.java
new file mode 100644
index 0000000000..53f9d13632
--- /dev/null
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunkTransformFactory.java
@@ -0,0 +1,146 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.transform.chunk;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.Conditions;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import org.apache.seatunnel.api.table.connector.TableTransform;
+import org.apache.seatunnel.api.table.factory.Factory;
+import org.apache.seatunnel.api.table.factory.TableTransformFactory;
+import org.apache.seatunnel.api.table.factory.TableTransformFactoryContext;
+import org.apache.seatunnel.transform.common.TransformCommonOptions;
+
+import com.google.auto.service.AutoService;
+
+import java.util.List;
+import java.util.Map;
+
+@AutoService(Factory.class)
+public class TextChunkTransformFactory implements TableTransformFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return TextChunkTransform.PLUGIN_NAME;
+ }
+
+ @Override
+ public OptionRule optionRule() {
+ return OptionRule.builder()
+ .required(
+ TextChunkTransformConfig.TEXT_FIELD,
+
Conditions.notBlank(TextChunkTransformConfig.TEXT_FIELD),
+ // Anchored on TEXT_FIELD (the only
absolutely-required option) on purpose:
+ // a value constraint only runs when its head option
is required or
+ // explicitly present, so anchoring on the optional
output_field/
+ // chunk_index_field would skip the check when the
user relies on their
+ // defaults -- exactly the default-collision case this
validator must catch.
+ Conditions.extension(
+ TextChunkTransformConfig.TEXT_FIELD,
+ new OutputFieldNamesValidator()))
+ .optional(
+ TextChunkTransformConfig.OUTPUT_FIELD,
+ TextChunkTransformConfig.CHUNK_INDEX_FIELD,
+ TextChunkTransformConfig.SEPARATORS,
+ TextChunkTransformConfig.SKIP_EMPTY_TEXT)
+ .optional(
+ TextChunkTransformConfig.CHUNK_SIZE,
+
Conditions.greaterThan(TextChunkTransformConfig.CHUNK_SIZE, 0))
+ .optional(
+ TextChunkTransformConfig.OVERLAP_SIZE,
+
Conditions.greaterOrEqual(TextChunkTransformConfig.OVERLAP_SIZE, 0),
+ Conditions.lessThanField(
+ TextChunkTransformConfig.OVERLAP_SIZE,
+ TextChunkTransformConfig.CHUNK_SIZE))
+ .optional(
+ TransformCommonOptions.MULTI_TABLES,
+ Conditions.extension(
+ TransformCommonOptions.MULTI_TABLES,
+ new TableTransformRulesValidator()))
+ .optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .build();
+ }
+
+ @Override
+ public TableTransform createTransform(TableTransformFactoryContext
context) {
+ return () ->
+ new TextChunkMultiCatalogTransform(
+ context.getCatalogTables(), context.getOptions());
+ }
+
+ static void validate(TextChunkTransformConfig config) {
+ if (config.getTextField() == null ||
config.getTextField().trim().isEmpty()) {
+ throw new OptionValidationException("text_field is required, but
was not configured");
+ }
+ if (config.getChunkSize() <= 0) {
+ throw new OptionValidationException(
+ "chunk_size must be > 0, but was " +
config.getChunkSize());
+ }
+ if (config.getOverlapSize() < 0 || config.getOverlapSize() >=
config.getChunkSize()) {
+ throw new OptionValidationException(
+ "overlap_size must satisfy 0 <= overlap_size < chunk_size,
but was "
+ + config.getOverlapSize()
+ + " (chunk_size="
+ + config.getChunkSize()
+ + ")");
+ }
+ if (config.getOutputField().equals(config.getChunkIndexField())) {
+ throw new OptionValidationException(
+ "output_field and chunk_index_field must be different, but
both are '"
+ + config.getOutputField()
+ + "'");
+ }
+ }
+
+ static class OutputFieldNamesValidator implements
ConditionExtension<String> {
+
+ @Override
+ public String description() {
+ return "output_field and chunk_index_field must be different";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String textField) {
+ return !config.get(TextChunkTransformConfig.OUTPUT_FIELD)
+
.equals(config.get(TextChunkTransformConfig.CHUNK_INDEX_FIELD));
+ }
+ }
+
+ static class TableTransformRulesValidator
+ implements ConditionExtension<List<Map<String, Object>>> {
+
+ @Override
+ public String description() {
+ return "each table_transform entry must satisfy chunk_size > 0, "
+ + "0 <= overlap_size < chunk_size, and output_field !=
chunk_index_field";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, List<Map<String,
Object>> tables) {
+ if (tables == null) {
+ return true;
+ }
+ for (Map<String, Object> table : tables) {
+
validate(TextChunkTransformConfig.of(ReadonlyConfig.fromMap(table)));
+ }
+ return true;
+ }
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunker.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunker.java
new file mode 100644
index 0000000000..1db491c122
--- /dev/null
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/chunk/TextChunker.java
@@ -0,0 +1,179 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.transform.chunk;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Pure text-chunking algorithm.
+ *
+ * <ul>
+ * <li>null/empty text produces no chunks;
+ * <li>empty {@code separators} falls back to fixed-size splitting with
overlap;
+ * <li>otherwise splits recursively by separators (priority order), then
greedily merges the
+ * pieces up to {@code chunkSize}.
+ * </ul>
+ *
+ * <p>All lengths ({@code chunkSize} and {@code overlapSize}) are measured in
<b>Unicode code
+ * points</b>, not UTF-16 code units, so a supplementary character (e.g. an
emoji) counts as one and
+ * is never split across chunks.
+ *
+ * <p>Guarantees:
+ *
+ * <ul>
+ * <li>each chunk holds {@code <= chunkSize} code points (carried overlap
counts toward the
+ * budget);
+ * <li>in the separator path, overlap is composed of whole trailing pieces,
so it never starts
+ * mid-word; {@code overlapSize} is an upper bound and rounds down to
whole pieces. In the
+ * empty-separators fixed-size fallback there are no pieces, so overlap
is a plain window of
+ * {@code overlapSize} code points and is not rounded;
+ * <li>separators are retained on the piece they follow, so with {@code
overlapSize == 0} the
+ * chunks concatenate back to the original text;
+ * <li>boundaries align to whole Unicode code points, so a surrogate pair
(e.g. an emoji) is never
+ * split.
+ * </ul>
+ */
+public final class TextChunker {
+
+ private TextChunker() {}
+
+ public static List<String> split(
+ String text, List<String> separators, int chunkSize, int
overlapSize) {
+ if (text == null || text.isEmpty()) {
+ return Collections.emptyList();
+ }
+ if (separators == null || separators.isEmpty()) {
+ return fixedSize(text, chunkSize, overlapSize);
+ }
+ List<String> segments = recursiveSplit(text, separators, 0, chunkSize);
+ return mergeWithOverlap(segments, chunkSize, overlapSize);
+ }
+
+ private static List<String> recursiveSplit(
+ String text, List<String> separators, int index, int chunkSize) {
+ List<String> result = new ArrayList<>();
+ String separator = separators.get(index);
+ List<String> parts =
+ separator.isEmpty()
+ ? Collections.singletonList(text)
+ : splitByLiteral(text, separator);
+ for (String part : parts) {
+ if (part.isEmpty()) {
+ continue;
+ }
+ if (codePointLength(part) <= chunkSize) {
+ result.add(part);
+ } else if (index + 1 < separators.size()) {
+ result.addAll(recursiveSplit(part, separators, index + 1,
chunkSize));
+ } else {
+ result.addAll(fixedSize(part, chunkSize, 0));
+ }
+ }
+ return result;
+ }
+
+ private static List<String> splitByLiteral(String text, String separator) {
+ List<String> parts = new ArrayList<>();
+ int start = 0;
+ int idx;
+ while ((idx = text.indexOf(separator, start)) >= 0) {
+ int next = idx + separator.length();
+ parts.add(text.substring(start, next));
+ start = next;
+ }
+ if (start < text.length()) {
+ parts.add(text.substring(start));
+ }
+ return parts;
+ }
+
+ private static List<String> mergeWithOverlap(
+ List<String> segments, int chunkSize, int overlapSize) {
+ List<String> chunks = new ArrayList<>();
+ List<String> current = new ArrayList<>();
+ int currentLen = 0;
+ for (String segment : segments) {
+ int segmentLen = codePointLength(segment);
+ if (currentLen > 0 && currentLen + segmentLen > chunkSize) {
+ chunks.add(String.join("", current));
+ current =
+ trailingSegmentsWithin(
+ current, Math.min(overlapSize, chunkSize -
segmentLen));
+ currentLen = 0;
+ for (String s : current) {
+ currentLen += codePointLength(s);
+ }
+ }
+ current.add(segment);
+ currentLen += segmentLen;
+ }
+ if (currentLen > 0) {
+ chunks.add(String.join("", current));
+ }
+ return chunks;
+ }
+
+ /**
+ * Returns the longest suffix of {@code segments} whose combined length
does not exceed {@code
+ * budget}.
+ */
+ private static List<String> trailingSegmentsWithin(List<String> segments,
int budget) {
+ int total = 0;
+ int from = segments.size();
+ for (int i = segments.size() - 1; i >= 0; i--) {
+ int len = codePointLength(segments.get(i));
+ if (total + len > budget) {
+ break;
+ }
+ total += len;
+ from = i;
+ }
+ return new ArrayList<>(segments.subList(from, segments.size()));
+ }
+
+ private static List<String> fixedSize(String text, int chunkSize, int
overlapSize) {
+ List<String> chunks = new ArrayList<>();
+ int len = text.length();
+ int stepCodePoints = Math.max(1, chunkSize - overlapSize);
+ int start = 0;
+ while (start < len) {
+ int end = advanceCodePoints(text, start, chunkSize);
+ chunks.add(text.substring(start, end));
+ if (end >= len) {
+ break;
+ }
+ start = advanceCodePoints(text, start, stepCodePoints);
+ }
+ return chunks;
+ }
+
+ private static int codePointLength(String s) {
+ return s.codePointCount(0, s.length());
+ }
+
+ private static int advanceCodePoints(String text, int start, int
codePoints) {
+ int index = start;
+ int len = text.length();
+ for (int c = 0; c < codePoints && index < len; c++) {
+ index += Character.charCount(text.codePointAt(index));
+ }
+ return index;
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/chunk/TextChunkTransformFactoryTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/chunk/TextChunkTransformFactoryTest.java
new file mode 100644
index 0000000000..39ac5c7d13
--- /dev/null
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/chunk/TextChunkTransformFactoryTest.java
@@ -0,0 +1,248 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.transform.chunk;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.connector.TableTransform;
+import org.apache.seatunnel.api.table.factory.TableTransformFactoryContext;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.api.transform.SeaTunnelTransform;
+import org.apache.seatunnel.transform.common.TransformCommonOptions;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class TextChunkTransformFactoryTest {
+
+ @Test
+ public void testFactoryIdentifierAndOptionRule() {
+ TextChunkTransformFactory factory = new TextChunkTransformFactory();
+ Assertions.assertEquals(TextChunkTransform.PLUGIN_NAME,
factory.factoryIdentifier());
+ Assertions.assertNotNull(factory.optionRule());
+ }
+
+ @Test
+ public void testCreateTransformReturnsMultiCatalogTransform() {
+ TextChunkTransformFactory factory = new TextChunkTransformFactory();
+
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id", "content"},
+ new SeaTunnelDataType[] {BasicType.INT_TYPE,
BasicType.STRING_TYPE});
+ CatalogTable catalogTable =
+ CatalogTableUtil.getCatalogTable("schema", "default", null,
"test", rowType);
+ List<CatalogTable> tables = Collections.singletonList(catalogTable);
+
+ ReadonlyConfig config =
+ ReadonlyConfig.fromMap(
+ Collections.singletonMap(
+ TextChunkTransformConfig.TEXT_FIELD.key(),
"content"));
+
+ TableTransformFactoryContext context =
+ new TableTransformFactoryContext(
+ tables, config,
Thread.currentThread().getContextClassLoader());
+
+ TableTransform<?> tableTransform = factory.createTransform(context);
+ Assertions.assertNotNull(tableTransform);
+
+ SeaTunnelTransform<?> inner = tableTransform.createTransform();
+ Assertions.assertNotNull(inner);
+ Assertions.assertTrue(inner instanceof TextChunkMultiCatalogTransform);
+ }
+
+ @Test
+ public void testValidConfigWithTextField() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ }
+
+ @Test
+ public void testMissingTextFieldFails() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("text_field"),
ex.getMessage());
+ }
+
+ @Test
+ public void testBlankTextFieldFailsAtCheckTime() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("text_field"),
ex.getMessage());
+ }
+
+ @Test
+ public void testNonPositiveChunkSizeFails() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ cfg.put(TextChunkTransformConfig.CHUNK_SIZE.key(), 0);
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("chunk_size"),
ex.getMessage());
+ }
+
+ @Test
+ public void testNegativeOverlapFails() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ cfg.put(TextChunkTransformConfig.OVERLAP_SIZE.key(), -1);
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("overlap_size"),
ex.getMessage());
+ }
+
+ @Test
+ public void testOverlapNotLessThanChunkSizeFails() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ cfg.put(TextChunkTransformConfig.CHUNK_SIZE.key(), 100);
+ cfg.put(TextChunkTransformConfig.OVERLAP_SIZE.key(), 100);
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("overlap_size"),
ex.getMessage());
+ }
+
+ @Test
+ public void testTopLevelDuplicateOutputAndIndexFieldFails() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ cfg.put(TextChunkTransformConfig.OUTPUT_FIELD.key(), "same");
+ cfg.put(TextChunkTransformConfig.CHUNK_INDEX_FIELD.key(), "same");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("output_field and chunk_index_field
must be different"),
+ ex.getMessage());
+ }
+
+ @Test
+ public void testTopLevelIndexFieldCollidingWithDefaultOutputFails() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ cfg.put(TextChunkTransformConfig.CHUNK_INDEX_FIELD.key(), "chunk");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("output_field and chunk_index_field
must be different"),
+ ex.getMessage());
+ }
+
+ @Test
+ public void testInvalidPerTableRuleFailsAtCheckTime() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> badTable = new HashMap<>();
+ badTable.put(TransformCommonOptions.TABLE_PATH.key(), "db.tbl");
+ badTable.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ badTable.put(TextChunkTransformConfig.CHUNK_SIZE.key(), 100);
+ badTable.put(TextChunkTransformConfig.OVERLAP_SIZE.key(), 100); //
overlap >= chunk_size
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ cfg.put(TransformCommonOptions.MULTI_TABLES.key(),
Arrays.asList(badTable));
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("overlap_size"),
ex.getMessage());
+ }
+
+ @Test
+ public void testPerTableMissingTextFieldFailsAtCheckTime() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> table = new HashMap<>();
+ table.put(TransformCommonOptions.TABLE_PATH.key(), "db.tbl");
+ table.put(TextChunkTransformConfig.CHUNK_SIZE.key(), 100); //
text_field omitted
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content"); //
top-level present
+ cfg.put(TransformCommonOptions.MULTI_TABLES.key(),
Arrays.asList(table));
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("text_field"),
ex.getMessage());
+ }
+
+ @Test
+ public void testPerTableBlankTextFieldFailsAtCheckTime() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> table = new HashMap<>();
+ table.put(TransformCommonOptions.TABLE_PATH.key(), "db.tbl");
+ table.put(TextChunkTransformConfig.TEXT_FIELD.key(), " ");
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ cfg.put(TransformCommonOptions.MULTI_TABLES.key(),
Arrays.asList(table));
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("text_field"),
ex.getMessage());
+ }
+
+ @Test
+ public void testValidPerTableRulePasses() {
+ OptionRule rule = new TextChunkTransformFactory().optionRule();
+ Map<String, Object> goodTable = new HashMap<>();
+ goodTable.put(TransformCommonOptions.TABLE_PATH.key(), "db.tbl");
+ goodTable.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ goodTable.put(TextChunkTransformConfig.CHUNK_SIZE.key(), 100);
+ goodTable.put(TextChunkTransformConfig.OVERLAP_SIZE.key(), 20);
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ cfg.put(TransformCommonOptions.MULTI_TABLES.key(),
Arrays.asList(goodTable));
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(rule));
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/chunk/TextChunkTransformTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/chunk/TextChunkTransformTest.java
new file mode 100644
index 0000000000..ec88982f69
--- /dev/null
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/chunk/TextChunkTransformTest.java
@@ -0,0 +1,636 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.transform.chunk;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.ConstraintKey;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.RowKind;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+import org.apache.seatunnel.transform.common.TransformCommonOptions;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+class TextChunkTransformTest {
+
+ private CatalogTable catalogTable;
+
+ @BeforeEach
+ void setUp() {
+ catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of("default", "default", "default",
"docs"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.INT_TYPE, 0L,
true, "", ""))
+ .column(
+ PhysicalColumn.of(
+ "content",
+ BasicType.STRING_TYPE,
+ 1000L,
+ true,
+ "",
+ ""))
+ .build(),
+ new HashMap<>(),
+ Collections.emptyList(),
+ "");
+ }
+
+ private TextChunkTransform newTransform() {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ configMap.put(TextChunkTransformConfig.CHUNK_SIZE.key(), 4);
+ configMap.put(TextChunkTransformConfig.OVERLAP_SIZE.key(), 0);
+ // empty separators -> deterministic fixed-size chunking
+ configMap.put(TextChunkTransformConfig.SEPARATORS.key(),
Collections.emptyList());
+ ReadonlyConfig config = ReadonlyConfig.fromMap(configMap);
+ return new TextChunkTransform(TextChunkTransformConfig.of(config),
catalogTable);
+ }
+
+ @Test
+ void producedSchemaAppendsChunkColumns() {
+ CatalogTable output = newTransform().getProducedCatalogTable();
+ TableSchema schema = output.getTableSchema();
+
+ // original two columns + chunk + chunk_index
+ Assertions.assertEquals(4, schema.getColumns().size());
+ Assertions.assertEquals(BasicType.STRING_TYPE,
schema.getColumn("chunk").getDataType());
+ Assertions.assertEquals(BasicType.INT_TYPE,
schema.getColumn("chunk_index").getDataType());
+ // original columns preserved
+ Assertions.assertNotNull(schema.getColumn("id"));
+ Assertions.assertNotNull(schema.getColumn("content"));
+ }
+
+ @Test
+ void oneRowExpandsToNRowsPreservingSourceFields() {
+ TextChunkTransform transform = newTransform();
+ SeaTunnelRowType outputType =
+
transform.getProducedCatalogTable().getTableSchema().toPhysicalRowDataType();
+ int chunkIdx = outputType.indexOf("chunk");
+ int chunkSeqIdx = outputType.indexOf("chunk_index");
+
+ SeaTunnelRow input = new SeaTunnelRow(new Object[] {7, "abcdefghij"});
+ input.setTableId("docs");
+ input.setRowKind(RowKind.UPDATE_AFTER);
+ // per-row metadata (e.g. stain-trace payload) that must survive onto
every chunk row
+ input.getOptions().put("trace", "payload");
+
+ List<SeaTunnelRow> out = transform.flatMap(input);
+
+ // "abcdefghij" (10 chars) with chunk_size=4 -> ["abcd","efgh","ij"]
+ Assertions.assertEquals(3, out.size());
+ String[] expectedChunks = {"abcd", "efgh", "ij"};
+ for (int i = 0; i < out.size(); i++) {
+ SeaTunnelRow row = out.get(i);
+ // source fields preserved on every chunk row
+ Assertions.assertEquals(7, row.getField(0));
+ Assertions.assertEquals("abcdefghij", row.getField(1));
+ // appended chunk + 0-based index
+ Assertions.assertEquals(expectedChunks[i], row.getField(chunkIdx));
+ Assertions.assertEquals(i, row.getField(chunkSeqIdx));
+ // tableId, rowKind and options (e.g. stain-trace payload) carried
through
+ Assertions.assertEquals("docs", row.getTableId());
+ Assertions.assertEquals(RowKind.UPDATE_AFTER, row.getRowKind());
+ Assertions.assertEquals("payload", row.getOptions().get("trace"));
+ }
+ }
+
+ @Test
+ void nullOrEmptyTextProducesNoRows() {
+ TextChunkTransform transform = newTransform();
+
+ SeaTunnelRow nullText = new SeaTunnelRow(new Object[] {1, null});
+ SeaTunnelRow emptyText = new SeaTunnelRow(new Object[] {2, ""});
+
+ Assertions.assertTrue(transform.flatMap(nullText).isEmpty());
+ Assertions.assertTrue(transform.flatMap(emptyText).isEmpty());
+ }
+
+ @Test
+ void skipEmptyTextFalsePassesRowThroughInsteadOfDropping() {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ configMap.put(TextChunkTransformConfig.CHUNK_SIZE.key(), 4);
+ configMap.put(TextChunkTransformConfig.OVERLAP_SIZE.key(), 0);
+ configMap.put(TextChunkTransformConfig.SEPARATORS.key(),
Collections.emptyList());
+ configMap.put(TextChunkTransformConfig.SKIP_EMPTY_TEXT.key(), false);
+ TextChunkTransform transform =
+ new TextChunkTransform(
+
TextChunkTransformConfig.of(ReadonlyConfig.fromMap(configMap)),
+ catalogTable);
+
+ SeaTunnelRowType outputType =
+
transform.getProducedCatalogTable().getTableSchema().toPhysicalRowDataType();
+ int chunkIdx = outputType.indexOf("chunk");
+ int chunkSeqIdx = outputType.indexOf("chunk_index");
+
+ for (Object emptyValue : new Object[] {null, ""}) {
+ SeaTunnelRow input = new SeaTunnelRow(new Object[] {1,
emptyValue});
+ input.setTableId("docs");
+ input.setRowKind(RowKind.INSERT);
+
+ List<SeaTunnelRow> out = transform.flatMap(input);
+
+ // one passthrough row: source fields kept, chunk = null,
chunk_index = 0
+ Assertions.assertEquals(1, out.size());
+ SeaTunnelRow row = out.get(0);
+ Assertions.assertEquals(1, row.getField(0));
+ Assertions.assertNull(row.getField(chunkIdx));
+ Assertions.assertEquals(0, row.getField(chunkSeqIdx));
+ Assertions.assertEquals("docs", row.getTableId());
+ }
+ }
+
+ @Test
+ void producedSchemaExtendsPrimaryKeyAndUniqueKeyWithChunkIndex() {
+ CatalogTable withKeys =
+ CatalogTable.of(
+ TableIdentifier.of("default", "default", "default",
"docs"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.INT_TYPE, 0L,
false, "", ""))
+ .column(
+ PhysicalColumn.of(
+ "content",
+ BasicType.STRING_TYPE,
+ 1000L,
+ true,
+ "",
+ ""))
+ .primaryKey(PrimaryKey.of("pk_id",
Collections.singletonList("id")))
+ .constraintKey(
+ Arrays.asList(
+ ConstraintKey.of(
+
ConstraintKey.ConstraintType.UNIQUE_KEY,
+ "uk_content",
+
Collections.singletonList(
+
ConstraintKey.ConstraintKeyColumn
+ .of(
+
"content",
+
ConstraintKey
+
.ColumnSortType
+
.ASC))),
+ ConstraintKey.of(
+
ConstraintKey.ConstraintType.INDEX_KEY,
+ "idx_content",
+
Collections.singletonList(
+
ConstraintKey.ConstraintKeyColumn
+ .of(
+
"content",
+
ConstraintKey
+
.ColumnSortType
+
.ASC)))))
+ .build(),
+ new HashMap<>(),
+ Collections.emptyList(),
+ "");
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ TextChunkTransform transform =
+ new TextChunkTransform(
+
TextChunkTransformConfig.of(ReadonlyConfig.fromMap(configMap)), withKeys);
+
+ TableSchema schema =
transform.getProducedCatalogTable().getTableSchema();
+
+ // the chunk index column joins the keys below, so it must be
non-nullable
+ Assertions.assertFalse(schema.getColumn("chunk_index").isNullable());
+
+ // the primary key is extended
+ Assertions.assertNotNull(schema.getPrimaryKey());
+ Assertions.assertEquals(
+ Arrays.asList("id", "chunk_index"),
schema.getPrimaryKey().getColumnNames());
+
+ Map<String, ConstraintKey> keysByName =
+ schema.getConstraintKeys().stream()
+
.collect(Collectors.toMap(ConstraintKey::getConstraintName, k -> k));
+ Assertions.assertEquals(2, keysByName.size());
+
+ // the unique key is extended
+ List<String> ukColumns =
+ keysByName.get("uk_content").getColumnNames().stream()
+ .map(ConstraintKey.ConstraintKeyColumn::getColumnName)
+ .collect(Collectors.toList());
+ Assertions.assertEquals(Arrays.asList("content", "chunk_index"),
ukColumns);
+
+ // the non-unique index is copied unchanged
+ List<String> idxColumns =
+ keysByName.get("idx_content").getColumnNames().stream()
+ .map(ConstraintKey.ConstraintKeyColumn::getColumnName)
+ .collect(Collectors.toList());
+ Assertions.assertEquals(Collections.singletonList("content"),
idxColumns);
+ }
+
+ @Test
+ void autoIdPrimaryKeyIsNotExtended() {
+ // A sink-auto-generated primary key (enableAutoId) already yields a
fresh unique value
+ // per output row, so it must be kept as-is rather than extended with
chunk_index.
+ CatalogTable withAutoId =
+ CatalogTable.of(
+ TableIdentifier.of("default", "default", "default",
"docs"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.LONG_TYPE, 0L,
false, "", ""))
+ .column(
+ PhysicalColumn.of(
+ "content",
+ BasicType.STRING_TYPE,
+ 1000L,
+ true,
+ "",
+ ""))
+ .primaryKey(
+ PrimaryKey.of(
+ "pk_id",
Collections.singletonList("id"), true))
+ .build(),
+ new HashMap<>(),
+ Collections.emptyList(),
+ "");
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ TextChunkTransform transform =
+ new TextChunkTransform(
+
TextChunkTransformConfig.of(ReadonlyConfig.fromMap(configMap)), withAutoId);
+
+ TableSchema schema =
transform.getProducedCatalogTable().getTableSchema();
+ Assertions.assertNotNull(schema.getPrimaryKey());
+ Assertions.assertEquals(
+ Collections.singletonList("id"),
schema.getPrimaryKey().getColumnNames());
+ Assertions.assertEquals(Boolean.TRUE,
schema.getPrimaryKey().getEnableAutoId());
+ }
+
+ @Test
+ void chunkIndexFieldCollidingWithPrimaryKeyIsRejected() {
+ CatalogTable withPk =
+ CatalogTable.of(
+ TableIdentifier.of("default", "default", "default",
"docs"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.INT_TYPE, 0L,
false, "", ""))
+ .column(
+ PhysicalColumn.of(
+ "content",
+ BasicType.STRING_TYPE,
+ 1000L,
+ true,
+ "",
+ ""))
+ .primaryKey(PrimaryKey.of("pk_id",
Collections.singletonList("id")))
+ .build(),
+ new HashMap<>(),
+ Collections.emptyList(),
+ "");
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ configMap.put(TextChunkTransformConfig.CHUNK_INDEX_FIELD.key(), "id");
+
+ SeaTunnelRuntimeException ex =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () ->
+ new TextChunkTransform(
+ TextChunkTransformConfig.of(
+
ReadonlyConfig.fromMap(configMap)),
+ withPk));
+ Assertions.assertTrue(ex.getMessage().contains("id"));
+ }
+
+ @Test
+ void outputFieldCollidingWithUniqueKeyIsRejected() {
+ // output_field = "doc_hash" would overwrite a unique-key column with
per-chunk text.
+ CatalogTable withUniqueKey =
+ CatalogTable.of(
+ TableIdentifier.of("default", "default", "default",
"docs"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.INT_TYPE, 0L,
false, "", ""))
+ .column(
+ PhysicalColumn.of(
+ "doc_hash",
+ BasicType.STRING_TYPE,
+ 64L,
+ true,
+ "",
+ ""))
+ .column(
+ PhysicalColumn.of(
+ "content",
+ BasicType.STRING_TYPE,
+ 1000L,
+ true,
+ "",
+ ""))
+ .constraintKey(
+ Collections.singletonList(
+ ConstraintKey.of(
+
ConstraintKey.ConstraintType.UNIQUE_KEY,
+ "uk_doc_hash",
+
Collections.singletonList(
+
ConstraintKey.ConstraintKeyColumn
+ .of(
+
"doc_hash",
+
ConstraintKey
+
.ColumnSortType
+
.ASC)))))
+ .build(),
+ new HashMap<>(),
+ Collections.emptyList(),
+ "");
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ configMap.put(TextChunkTransformConfig.OUTPUT_FIELD.key(), "doc_hash");
+
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () ->
+ new TextChunkTransform(
+
TextChunkTransformConfig.of(ReadonlyConfig.fromMap(configMap)),
+ withUniqueKey));
+ }
+
+ @Test
+ void reusingNonKeyColumnAsOutputFieldIsAllowed() {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(TextChunkTransformConfig.TEXT_FIELD.key(), "content");
+ configMap.put(TextChunkTransformConfig.OUTPUT_FIELD.key(), "content");
+ TextChunkTransform transform =
+ new TextChunkTransform(
+
TextChunkTransformConfig.of(ReadonlyConfig.fromMap(configMap)),
+ catalogTable);
+
+ TableSchema schema =
transform.getProducedCatalogTable().getTableSchema();
+ // "content" is reused (not duplicated): original 2 columns +
chunk_index only
+ Assertions.assertEquals(3, schema.getColumns().size());
+ Assertions.assertNotNull(schema.getColumn("content"));
+ Assertions.assertNotNull(schema.getColumn("chunk_index"));
+ }
+
+ @Test
+ void setInputCatalogTableRefreshesTextFieldIndexAfterSchemaChange() {
+ // Original schema: [id, content] -> text_field "content" is at index
1.
+ TextChunkTransform transform = newTransform();
+
+ // A schema-change event inserts a "created_at" column before
"content", shifting the text
+ CatalogTable evolved =
+ CatalogTable.of(
+ TableIdentifier.of("default", "default", "default",
"docs"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.INT_TYPE, 0L,
true, "", ""))
+ .column(
+ PhysicalColumn.of(
+ "created_at",
+ BasicType.LONG_TYPE,
+ 0L,
+ true,
+ "",
+ ""))
+ .column(
+ PhysicalColumn.of(
+ "content",
+ BasicType.STRING_TYPE,
+ 1000L,
+ true,
+ "",
+ ""))
+ .build(),
+ new HashMap<>(),
+ Collections.emptyList(),
+ "");
+ transform.setInputCatalogTable(evolved);
+
+ SeaTunnelRowType outputType =
+
transform.getProducedCatalogTable().getTableSchema().toPhysicalRowDataType();
+ int chunkIdx = outputType.indexOf("chunk");
+ int chunkSeqIdx = outputType.indexOf("chunk_index");
+
+ SeaTunnelRow input = new SeaTunnelRow(new Object[] {7, 123456789L,
"abcdefghij"});
+ input.setTableId("docs");
+
+ List<SeaTunnelRow> out = transform.flatMap(input);
+
+ Assertions.assertEquals(3, out.size());
+ String[] expectedChunks = {"abcd", "efgh", "ij"};
+ for (int i = 0; i < out.size(); i++) {
+ SeaTunnelRow row = out.get(i);
+ Assertions.assertEquals(7, row.getField(0));
+ Assertions.assertEquals(123456789L, row.getField(1));
+ Assertions.assertEquals("abcdefghij", row.getField(2));
+ Assertions.assertEquals(expectedChunks[i], row.getField(chunkIdx));
+ Assertions.assertEquals(i, row.getField(chunkSeqIdx));
+ }
+ }
+
+ @Test
+ void mapSchemaChangeEventRefreshesTextFieldIndexOnDirectAlter() {
+ // Original schema: [id, content] -> text_field "content" is at index
1.
+ TextChunkTransform transform = newTransform();
+
+ // TextChunk sitting first after the source receives the ALTER
directly via
+ // mapSchemaChangeEvent (not setInputCatalogTable). Dropping "id"
shifts "content" to index
+ // 0.
+ AlterTableDropColumnEvent dropId =
+ new AlterTableDropColumnEvent(catalogTable.getTableId(), "id");
+ transform.mapSchemaChangeEvent(dropId);
+
+ SeaTunnelRowType outputType =
+
transform.getProducedCatalogTable().getTableSchema().toPhysicalRowDataType();
+ Assertions.assertEquals(3, outputType.getTotalFields());
+ int chunkIdx = outputType.indexOf("chunk");
+ int chunkSeqIdx = outputType.indexOf("chunk_index");
+
+ SeaTunnelRow input = new SeaTunnelRow(new Object[] {"abcdefghij"});
+ input.setTableId("docs");
+
+ List<SeaTunnelRow> out = transform.flatMap(input);
+
+ Assertions.assertEquals(3, out.size());
+ String[] expectedChunks = {"abcd", "efgh", "ij"};
+ for (int i = 0; i < out.size(); i++) {
+ SeaTunnelRow row = out.get(i);
+ Assertions.assertEquals("abcdefghij", row.getField(0));
+ Assertions.assertEquals(expectedChunks[i], row.getField(chunkIdx));
+ Assertions.assertEquals(i, row.getField(chunkSeqIdx));
+ }
+ }
+
+ @Test
+ void
mapSchemaChangeEventAddColumnFirstShiftsTextFieldAndKeepsProducedOrdering() {
+ // Original schema: [id, content] -> text_field "content" is at index
1.
+ TextChunkTransform transform = newTransform();
+
+ // An ADD COLUMN ALTER inserts "created_at" at the FRONT, shifting
"content" to index 2.
+ // This is the ordering-sensitive case: chunk/chunk_index are always
appended, so the
+ // produced schema must stay [<input cols in new order>, chunk,
chunk_index] and every
+ // emitted row must line up with it.
+ Column createdAt = PhysicalColumn.of("created_at",
BasicType.LONG_TYPE, 0L, true, "", "");
+ AlterTableAddColumnEvent addFirst =
+ AlterTableAddColumnEvent.addFirst(catalogTable.getTableId(),
createdAt);
+ transform.mapSchemaChangeEvent(addFirst);
+
+ // produced column ordering: added input column, then the originals,
then the appended chunk
+ // columns -- nothing reordered.
+ List<String> producedColumns =
+
transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+ .map(Column::getName)
+ .collect(Collectors.toList());
+ Assertions.assertEquals(
+ Arrays.asList("created_at", "id", "content", "chunk",
"chunk_index"),
+ producedColumns);
+
+ SeaTunnelRowType outputType =
+
transform.getProducedCatalogTable().getTableSchema().toPhysicalRowDataType();
+ int chunkIdx = outputType.indexOf("chunk");
+ int chunkSeqIdx = outputType.indexOf("chunk_index");
+
+ SeaTunnelRow input = new SeaTunnelRow(new Object[] {123456789L, 7,
"abcdefghij"});
+ input.setTableId("docs");
+
+ List<SeaTunnelRow> out = transform.flatMap(input);
+
+ // text is now read from index 2; source fields (including the newly
added one) survive on
+ // every chunk row in their new order.
+ Assertions.assertEquals(3, out.size());
+ String[] expectedChunks = {"abcd", "efgh", "ij"};
+ for (int i = 0; i < out.size(); i++) {
+ SeaTunnelRow row = out.get(i);
+ Assertions.assertEquals(123456789L, row.getField(0));
+ Assertions.assertEquals(7, row.getField(1));
+ Assertions.assertEquals("abcdefghij", row.getField(2));
+ Assertions.assertEquals(expectedChunks[i], row.getField(chunkIdx));
+ Assertions.assertEquals(i, row.getField(chunkSeqIdx));
+ }
+ }
+
+ @Test
+ void setInputCatalogTableRejectsSchemaThatDropsTextField() {
+ TextChunkTransform transform = newTransform();
+
+ CatalogTable withoutText =
+ CatalogTable.of(
+ TableIdentifier.of("default", "default", "default",
"docs"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.INT_TYPE, 0L,
true, "", ""))
+ .build(),
+ new HashMap<>(),
+ Collections.emptyList(),
+ "");
+
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class, () ->
transform.setInputCatalogTable(withoutText));
+ }
+
+ @Test
+ void
multiCatalogMapSchemaChangeEventAddColumnPropagatesToInnerAndSetsChangeAfter() {
+ String tablePath = catalogTable.getTableId().toTablePath().toString();
+
+ Map<String, Object> tableTransform = new HashMap<>();
+ tableTransform.put(TransformCommonOptions.TABLE_PATH.key(), tablePath);
+ tableTransform.put(TextChunkTransformConfig.TEXT_FIELD.key(),
"content");
+ tableTransform.put(TextChunkTransformConfig.CHUNK_SIZE.key(), 4);
+ tableTransform.put(TextChunkTransformConfig.OVERLAP_SIZE.key(), 0);
+ tableTransform.put(TextChunkTransformConfig.SEPARATORS.key(),
Collections.emptyList());
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(
+ TransformCommonOptions.MULTI_TABLES.key(),
+ Collections.singletonList(tableTransform));
+
+ TextChunkMultiCatalogTransform multi =
+ new TextChunkMultiCatalogTransform(
+ Collections.singletonList(catalogTable),
ReadonlyConfig.fromMap(configMap));
+
+ // ADD COLUMN "created_at" at the FRONT, shifting text_field "content"
from index 1 to 2.
+ Column createdAt = PhysicalColumn.of("created_at",
BasicType.LONG_TYPE, 0L, true, "", "");
+ AlterTableAddColumnEvent addFirst =
+ AlterTableAddColumnEvent.addFirst(catalogTable.getTableId(),
createdAt);
+ SchemaChangeEvent mapped = multi.mapSchemaChangeEvent(addFirst);
+
+ // (3) the wrapper stamps its actual produced layout onto changeAfter
for downstream/sink.
+ List<String> expectedProduced =
+ Arrays.asList("created_at", "id", "content", "chunk",
"chunk_index");
+ Assertions.assertNotNull(mapped.getChangeAfter());
+ Assertions.assertEquals(
+ expectedProduced,
+ mapped.getChangeAfter().getTableSchema().getColumns().stream()
+ .map(Column::getName)
+ .collect(Collectors.toList()));
+
+ // (2) the wrapper's own produced schema reflects the new column order
too.
+ Assertions.assertEquals(
+ expectedProduced,
+
multi.getProducedCatalogTable().getTableSchema().getColumns().stream()
+ .map(Column::getName)
+ .collect(Collectors.toList()));
+
+ // (1) rows in the new layout chunk correctly: text is read from its
shifted index and every
+ // source field (including the newly added one) survives on each chunk
row.
+ SeaTunnelRowType outputType =
+
multi.getProducedCatalogTable().getTableSchema().toPhysicalRowDataType();
+ int chunkIdx = outputType.indexOf("chunk");
+ int chunkSeqIdx = outputType.indexOf("chunk_index");
+
+ SeaTunnelRow input = new SeaTunnelRow(new Object[] {123456789L, 7,
"abcdefghij"});
+ input.setTableId(tablePath);
+
+ List<SeaTunnelRow> out = multi.flatMap(input);
+
+ Assertions.assertEquals(3, out.size());
+ String[] expectedChunks = {"abcd", "efgh", "ij"};
+ for (int i = 0; i < out.size(); i++) {
+ SeaTunnelRow row = out.get(i);
+ Assertions.assertEquals(123456789L, row.getField(0));
+ Assertions.assertEquals(7, row.getField(1));
+ Assertions.assertEquals("abcdefghij", row.getField(2));
+ Assertions.assertEquals(expectedChunks[i], row.getField(chunkIdx));
+ Assertions.assertEquals(i, row.getField(chunkSeqIdx));
+ }
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/chunk/TextChunkerTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/chunk/TextChunkerTest.java
new file mode 100644
index 0000000000..6d4fe58856
--- /dev/null
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/chunk/TextChunkerTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.transform.chunk;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+class TextChunkerTest {
+
+ private static final List<String> SEPARATORS = Arrays.asList("\n\n", "\n",
" ");
+
+ @Test
+ void nullOrEmptyProducesNoChunks() {
+ Assertions.assertTrue(TextChunker.split(null, SEPARATORS, 100,
0).isEmpty());
+ Assertions.assertTrue(TextChunker.split("", SEPARATORS, 100,
0).isEmpty());
+ }
+
+ @Test
+ void shortTextStaysSingleChunk() {
+ List<String> chunks = TextChunker.split("hello world", SEPARATORS,
100, 0);
+ Assertions.assertEquals(Collections.singletonList("hello world"),
chunks);
+ }
+
+ @Test
+ void everyChunkWithinChunkSize() {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < 50; i++) {
+ sb.append("para-").append(i).append("\n\n");
+ }
+ List<String> chunks = TextChunker.split(sb.toString(), SEPARATORS, 40,
0);
+ Assertions.assertFalse(chunks.isEmpty());
+ chunks.forEach(c -> Assertions.assertTrue(codePoints(c) <= 40, "chunk
too long: " + c));
+ }
+
+ @Test
+ void emptySeparatorsFallsBackToFixedSize() {
+ List<String> chunks = TextChunker.split("abcdefghij",
Collections.emptyList(), 4, 0);
+ Assertions.assertEquals(Arrays.asList("abcd", "efgh", "ij"), chunks);
+ }
+
+ @Test
+ void fixedSizeHonorsOverlap() {
+ List<String> chunks = TextChunker.split("abcdefghij",
Collections.emptyList(), 4, 1);
+ Assertions.assertEquals(Arrays.asList("abcd", "defg", "ghij"), chunks);
+ }
+
+ @Test
+ void longUnbreakableSegmentIsHardSplit() {
+ // single 25-code-point token, no usable separator -> hard-split to <=
chunkSize
+ String token = "abcdefghijklmnopqrstuvwxy";
+ List<String> chunks = TextChunker.split(token, SEPARATORS, 10, 0);
+ chunks.forEach(c -> Assertions.assertTrue(codePoints(c) <= 10));
+ Assertions.assertEquals(token, String.join("", chunks));
+ }
+
+ @Test
+ void overlapCarriesWholeWordsNotCharacterFragments() {
+ List<String> chunks = TextChunker.split("aa bb cc dd", Arrays.asList("
"), 6, 3);
+ Assertions.assertEquals(Arrays.asList("aa bb ", "bb cc ", "cc dd"),
chunks);
+ Assertions.assertTrue(chunks.get(1).startsWith("bb "));
+ }
+
+ @Test
+ void overlapDegradesToEmptyWhenAWholeWordExceedsBudget() {
+ List<String> chunks = TextChunker.split("aa bb cc dd", Arrays.asList("
"), 5, 2);
+ Assertions.assertEquals(Arrays.asList("aa ", "bb ", "cc dd"), chunks);
+ }
+
+ @Test
+ void separatorsAndBlankLinesAreRetainedSoChunkingIsLossless() {
+ List<String> chunks = TextChunker.split("aa bb\n\ncc dd",
Arrays.asList("\n\n", " "), 6, 0);
+ Assertions.assertEquals(Arrays.asList("aa ", "bb\n\n", "cc dd"),
chunks);
+ Assertions.assertEquals("aa bb\n\ncc dd", String.join("", chunks));
+ }
+
+ @Test
+ void fixedSizeOverlapDoesNotBreakSurrogatePairs() {
+ String emoji = "😀";
+ String text = emoji + emoji + emoji; // 3 code points, 6 chars
+ // chunk_size=2, overlap=1 (code points): the window steps by one
whole code point.
+ List<String> chunks = TextChunker.split(text, Collections.emptyList(),
2, 1);
+ Assertions.assertEquals(Arrays.asList(emoji + emoji, emoji + emoji),
chunks);
+ }
+
+ @Test
+ void chunkSizeIsCountedInCodePointsNotCodeUnits() {
+ String emoji = "😀";
+ String text = emoji + emoji + emoji;
+ List<String> chunks = TextChunker.split(text, Collections.emptyList(),
1, 0);
+ Assertions.assertEquals(Arrays.asList(emoji, emoji, emoji), chunks);
+ }
+
+ @Test
+ void mergeMeasuresChunkSizeInCodePoints() {
+ String text = "😀 😀 😀 😀";
+ List<String> chunks = TextChunker.split(text, Arrays.asList(" "), 3,
0);
+ chunks.forEach(
+ c ->
+ Assertions.assertTrue(
+ codePoints(c) <= 3, "chunk exceeds 3 code
points: " + c));
+ chunks.forEach(TextChunkerTest::assertNoDanglingSurrogate);
+ Assertions.assertEquals(text, String.join("", chunks));
+ }
+
+ private static int codePoints(String s) {
+ return s.codePointCount(0, s.length());
+ }
+
+ private static void assertNoDanglingSurrogate(String s) {
+ for (int i = 0; i < s.length(); i++) {
+ char ch = s.charAt(i);
+ if (Character.isHighSurrogate(ch)) {
+ Assertions.assertTrue(
+ i + 1 < s.length() &&
Character.isLowSurrogate(s.charAt(i + 1)),
+ "dangling high surrogate in chunk: " + s);
+ i++;
+ } else {
+ Assertions.assertFalse(
+ Character.isLowSurrogate(ch), "dangling low surrogate
in chunk: " + s);
+ }
+ }
+ }
+}
