This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12861-7ec58eb6c375ad90b1f5c17858c77708a6f331d6 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 9b65d6634b7951250afac20e875e128ece5ad6eb Author: JeremyXin <[email protected]> AuthorDate: Wed Oct 21 11:51:42 2026 +0000 [Fix][Connector-File] Fix child_ids output inconsistent with the expected result (#12861) Co-authored-by: JeremyXin <--golbal> --- docs/en/connectors/source/CosFile.md | 2 +- docs/en/connectors/source/FtpFile.md | 2 +- docs/en/connectors/source/HdfsFile.md | 2 +- docs/en/connectors/source/Hive.md | 2 +- docs/en/connectors/source/LocalFile.md | 2 +- docs/en/connectors/source/ObsFile.md | 2 +- docs/en/connectors/source/OssFile.md | 2 +- docs/en/connectors/source/OssJindoFile.md | 2 +- docs/en/connectors/source/S3File.md | 2 +- docs/en/connectors/source/SftpFile.md | 2 +- .../introduction/concepts/incompatible-changes.md | 4 + docs/zh/connectors/source/CosFile.md | 2 +- docs/zh/connectors/source/FtpFile.md | 2 +- docs/zh/connectors/source/HdfsFile.md | 2 +- docs/zh/connectors/source/Hive.md | 2 +- docs/zh/connectors/source/LocalFile.md | 2 +- docs/zh/connectors/source/ObsFile.md | 2 +- docs/zh/connectors/source/OssFile.md | 2 +- docs/zh/connectors/source/OssJindoFile.md | 2 +- docs/zh/connectors/source/S3File.md | 2 +- docs/zh/connectors/source/SftpFile.md | 2 +- .../introduction/concepts/incompatible-changes.md | 4 + .../file/source/reader/MarkdownReadStrategy.java | 33 +++++++- .../seatunnel/file/source/BaseFileSourceTest.java | 5 ++ .../source/reader/MarkdownReadStrategyTest.java | 97 ++++++++++++++++++++-- .../e2e/connector/file/local/LocalFileIT.java | 2 +- ...local_file_markdown_rag_metadata_to_assert.conf | 81 +++++++++++++++++- 27 files changed, 235 insertions(+), 31 deletions(-) diff --git a/docs/en/connectors/source/CosFile.md b/docs/en/connectors/source/CosFile.md index ae4a92d3c5..2d8671338a 100644 --- a/docs/en/connectors/source/CosFile.md +++ b/docs/en/connectors/source/CosFile.md @@ -199,7 +199,7 @@ Each extracted element is converted to a document-element row with the following - `page_number`: Page number (default: 1) - `position_index`: Position index within the document - `parent_id`: ID of the parent element -- `child_ids`: Comma-separated list of child element IDs +- `child_ids`: Direct child element IDs as an `ARRAY<STRING>`; an empty array for a leaf element When either `markdown_rag_metadata_enabled` or `pdf_rag_metadata_enabled` is set to `true`, SeaTunnel appends the following RAG metadata fields after `child_ids` for the corresponding file type: - `source_uri`: Source file path or URI diff --git a/docs/en/connectors/source/FtpFile.md b/docs/en/connectors/source/FtpFile.md index ab53f543de..49bbe38f39 100644 --- a/docs/en/connectors/source/FtpFile.md +++ b/docs/en/connectors/source/FtpFile.md @@ -301,7 +301,7 @@ Each extracted element is converted to a document-element row with the following - `page_number`: Page number (default: 1) - `position_index`: Position index within the document - `parent_id`: ID of the parent element -- `child_ids`: Comma-separated list of child element IDs +- `child_ids`: Direct child element IDs as an `ARRAY<STRING>`; an empty array for a leaf element When either `markdown_rag_metadata_enabled` or `pdf_rag_metadata_enabled` is set to `true`, SeaTunnel appends the following RAG metadata fields after `child_ids` for the corresponding file type: - `source_uri`: Source file path or URI diff --git a/docs/en/connectors/source/HdfsFile.md b/docs/en/connectors/source/HdfsFile.md index df272f0aa8..5e77640827 100644 --- a/docs/en/connectors/source/HdfsFile.md +++ b/docs/en/connectors/source/HdfsFile.md @@ -126,7 +126,7 @@ Each extracted element is converted to a document-element row with the following - `page_number`: Page number (default: 1) - `position_index`: Position index within the document - `parent_id`: ID of the parent element -- `child_ids`: Comma-separated list of child element IDs +- `child_ids`: Direct child element IDs as an `ARRAY<STRING>`; an empty array for a leaf element When either `markdown_rag_metadata_enabled` or `pdf_rag_metadata_enabled` is set to `true`, SeaTunnel appends the following RAG metadata fields after `child_ids` for the corresponding file type: - `source_uri`: Source file path or URI diff --git a/docs/en/connectors/source/Hive.md b/docs/en/connectors/source/Hive.md index 5f4257589a..3af44f03dd 100644 --- a/docs/en/connectors/source/Hive.md +++ b/docs/en/connectors/source/Hive.md @@ -22,7 +22,7 @@ When using markdown format, SeaTunnel can parse markdown files stored in Hive ta - `page_number`: Page number (default: 1) - `position_index`: Position index within the document - `parent_id`: ID of the parent element -- `child_ids`: Comma-separated list of child element IDs +- `child_ids`: Direct child element IDs as an `ARRAY<STRING>`; an empty array for a leaf element Note: Markdown format only supports reading, not writing. diff --git a/docs/en/connectors/source/LocalFile.md b/docs/en/connectors/source/LocalFile.md index 31885568e9..995a20447d 100644 --- a/docs/en/connectors/source/LocalFile.md +++ b/docs/en/connectors/source/LocalFile.md @@ -213,7 +213,7 @@ Each extracted element is converted to a document-element row with the following - `page_number`: Page number (default: 1) - `position_index`: Position index within the document - `parent_id`: ID of the parent element -- `child_ids`: Comma-separated list of child element IDs +- `child_ids` (`ARRAY<STRING>`): Direct child element IDs; an empty array for a leaf element. When either `markdown_rag_metadata_enabled` or `pdf_rag_metadata_enabled` is set to `true`, SeaTunnel appends the following RAG metadata fields after `child_ids` for the corresponding file type: - `source_uri`: Source file path or URI diff --git a/docs/en/connectors/source/ObsFile.md b/docs/en/connectors/source/ObsFile.md index a27053cdd6..b0db487279 100644 --- a/docs/en/connectors/source/ObsFile.md +++ b/docs/en/connectors/source/ObsFile.md @@ -247,7 +247,7 @@ schema { > - `page_number`: Page number (default: 1) > - `position_index`: Position index within the document > - `parent_id`: ID of the parent element -> - `child_ids`: Comma-separated list of child element IDs +> - `child_ids`: Direct child element IDs as an `ARRAY<STRING>`; an empty array for a leaf element > > When either `markdown_rag_metadata_enabled` or `pdf_rag_metadata_enabled` is > set to `true`, SeaTunnel appends the following RAG metadata fields after > `child_ids` for the corresponding file type: > - `source_uri`: Source file path or URI diff --git a/docs/en/connectors/source/OssFile.md b/docs/en/connectors/source/OssFile.md index 2e3b8357cc..ad60d4b6dd 100644 --- a/docs/en/connectors/source/OssFile.md +++ b/docs/en/connectors/source/OssFile.md @@ -255,7 +255,7 @@ Each extracted element is converted to a document-element row with the following - `page_number`: Page number (default: 1) - `position_index`: Position index within the document - `parent_id`: ID of the parent element -- `child_ids`: Comma-separated list of child element IDs +- `child_ids`: Direct child element IDs as an `ARRAY<STRING>`; an empty array for a leaf element When either `markdown_rag_metadata_enabled` or `pdf_rag_metadata_enabled` is set to `true`, SeaTunnel appends the following RAG metadata fields after `child_ids` for the corresponding file type: - `source_uri`: Source file path or URI diff --git a/docs/en/connectors/source/OssJindoFile.md b/docs/en/connectors/source/OssJindoFile.md index af07373807..d8adb90a5f 100644 --- a/docs/en/connectors/source/OssJindoFile.md +++ b/docs/en/connectors/source/OssJindoFile.md @@ -204,7 +204,7 @@ Each extracted element is converted to a document-element row with the following - `page_number`: Page number (default: 1) - `position_index`: Position index within the document - `parent_id`: ID of the parent element -- `child_ids`: Comma-separated list of child element IDs +- `child_ids`: Direct child element IDs as an `ARRAY<STRING>`; an empty array for a leaf element When either `markdown_rag_metadata_enabled` or `pdf_rag_metadata_enabled` is set to `true`, SeaTunnel appends the following RAG metadata fields after `child_ids` for the corresponding file type: - `source_uri`: Source file path or URI diff --git a/docs/en/connectors/source/S3File.md b/docs/en/connectors/source/S3File.md index 9a253ac423..cdff12e9c2 100644 --- a/docs/en/connectors/source/S3File.md +++ b/docs/en/connectors/source/S3File.md @@ -305,7 +305,7 @@ Each extracted element is converted to a document-element row with the following - `page_number`: Page number (default: 1) - `position_index`: Position index within the document - `parent_id`: ID of the parent element -- `child_ids`: Comma-separated list of child element IDs +- `child_ids`: Direct child element IDs as an `ARRAY<STRING>`; an empty array for a leaf element When either `markdown_rag_metadata_enabled` or `pdf_rag_metadata_enabled` is set to `true`, SeaTunnel appends the following RAG metadata fields after `child_ids` for the corresponding file type: - `source_uri`: Source file path or URI diff --git a/docs/en/connectors/source/SftpFile.md b/docs/en/connectors/source/SftpFile.md index fc6f426032..eb04505c68 100644 --- a/docs/en/connectors/source/SftpFile.md +++ b/docs/en/connectors/source/SftpFile.md @@ -273,7 +273,7 @@ Each extracted element is converted to a document-element row with the following - `page_number`: Page number (default: 1) - `position_index`: Position index within the document - `parent_id`: ID of the parent element -- `child_ids`: Comma-separated list of child element IDs +- `child_ids`: Direct child element IDs as an `ARRAY<STRING>`; an empty array for a leaf element When either `markdown_rag_metadata_enabled` or `pdf_rag_metadata_enabled` is set to `true`, SeaTunnel appends the following RAG metadata fields after `child_ids` for the corresponding file type: - `source_uri`: Source file path or URI diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index b24119253f..555f72484e 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -30,6 +30,10 @@ You need to check this document before you upgrade to related version. - **Impact**: Such tables are now read completely, but with one snapshot split instead of several. With `exactly_once = true`, a table whose only key is a nullable unique key, or a nullable column configured in `table-names-config.primaryKeys`, now fails at startup with `Exactly once is enabled, but not found non-nullable primary key or unique key for table ...` instead of running and losing rows. This also applies to an existing job whose configured `primaryKeys` column is nullable in [...] - **Migration Guide**: Declare a NOT NULL unique column, or configure `table-names-config.primaryKeys` / `snapshotSplitColumn` with a non-nullable column, to keep parallel snapshot reading. For `exactly_once = true`, add a primary key or a non-nullable unique key to the table, or make the configured `primaryKeys` column NOT NULL in the database. +### Markdown file source `child_ids` array + +The Markdown file source now declares `child_ids` as `ARRAY<STRING>` and emits a `String[]` for every row. Previously, it declared `STRING`, emitted comma-separated IDs for parents, and used `null` for leaves. Update downstream schemas and consumers to read an array; leaves now contain an empty array. The element IDs, types, page numbers, and position indexes retain their existing formats. + ### Runtime Requirements - **Breaking Change: Minimum Java runtime raised from Java 8 to Java 11** diff --git a/docs/zh/connectors/source/CosFile.md b/docs/zh/connectors/source/CosFile.md index fb219452d8..847f7fda81 100644 --- a/docs/zh/connectors/source/CosFile.md +++ b/docs/zh/connectors/source/CosFile.md @@ -197,7 +197,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码 - `page_number`：页码（默认：1） - `position_index`：文档中的位置索引 - `parent_id`：父元素的 ID -- `child_ids`：子元素 ID 的逗号分隔列表 +- `child_ids`：直接子元素的 ID，类型为 `ARRAY<STRING>`；叶子元素为空数组 当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段： - `source_uri`：源文件路径或 URI diff --git a/docs/zh/connectors/source/FtpFile.md b/docs/zh/connectors/source/FtpFile.md index bd06bf39b3..afa3d0ff89 100644 --- a/docs/zh/connectors/source/FtpFile.md +++ b/docs/zh/connectors/source/FtpFile.md @@ -282,7 +282,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码 - `page_number`：页码（默认：1） - `position_index`：文档中的位置索引 - `parent_id`：父元素的 ID -- `child_ids`：子元素 ID 的逗号分隔列表 +- `child_ids`：直接子元素的 ID，类型为 `ARRAY<STRING>`；叶子元素为空数组 当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段： - `source_uri`：源文件路径或 URI diff --git a/docs/zh/connectors/source/HdfsFile.md b/docs/zh/connectors/source/HdfsFile.md index 18d0504890..49b5d97493 100644 --- a/docs/zh/connectors/source/HdfsFile.md +++ b/docs/zh/connectors/source/HdfsFile.md @@ -132,7 +132,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码 - `page_number`：页码（默认：1） - `position_index`：文档中的位置索引 - `parent_id`：父元素的 ID -- `child_ids`：子元素 ID 的逗号分隔列表 +- `child_ids`：直接子元素的 ID，类型为 `ARRAY<STRING>`；叶子元素为空数组 当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段： - `source_uri`：源文件路径或 URI diff --git a/docs/zh/connectors/source/Hive.md b/docs/zh/connectors/source/Hive.md index a5bd5880c6..3fb36054da 100644 --- a/docs/zh/connectors/source/Hive.md +++ b/docs/zh/connectors/source/Hive.md @@ -16,7 +16,7 @@ import ChangeLog from '../changelog/connector-hive.md'; - `page_number`：页码（默认：1） - `position_index`：文档中的位置索引 - `parent_id`：父元素的 ID -- `child_ids`：子元素 ID 的逗号分隔列表 +- `child_ids`：直接子元素的 ID，类型为 `ARRAY<STRING>`；叶子元素为空数组 注意：Markdown 格式仅支持读取，不支持写入。 diff --git a/docs/zh/connectors/source/LocalFile.md b/docs/zh/connectors/source/LocalFile.md index 3519e80980..1855e3bf7e 100644 --- a/docs/zh/connectors/source/LocalFile.md +++ b/docs/zh/connectors/source/LocalFile.md @@ -213,7 +213,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码 - `page_number`：页码（默认：1） - `position_index`：文档中的位置索引 - `parent_id`：父元素的 ID -- `child_ids`：子元素 ID 的逗号分隔列表 +- `child_ids`（`ARRAY<STRING>`）：直接子元素的 ID；叶子元素为空数组。 当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段： - `source_uri`：源文件路径或 URI diff --git a/docs/zh/connectors/source/ObsFile.md b/docs/zh/connectors/source/ObsFile.md index 1c22092ac7..e09a3c9847 100644 --- a/docs/zh/connectors/source/ObsFile.md +++ b/docs/zh/connectors/source/ObsFile.md @@ -120,7 +120,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码 - `page_number`：页码（默认：1） - `position_index`：文档中的位置索引 - `parent_id`：父元素的 ID -- `child_ids`：子元素 ID 的逗号分隔列表 +- `child_ids`：直接子元素的 ID，类型为 `ARRAY<STRING>`；叶子元素为空数组 当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段： - `source_uri`：源文件路径或 URI diff --git a/docs/zh/connectors/source/OssFile.md b/docs/zh/connectors/source/OssFile.md index a3b5915e25..54ba631ea7 100644 --- a/docs/zh/connectors/source/OssFile.md +++ b/docs/zh/connectors/source/OssFile.md @@ -288,7 +288,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码 - `page_number`：页码（默认：1） - `position_index`：文档中的位置索引 - `parent_id`：父元素的 ID -- `child_ids`：子元素 ID 的逗号分隔列表 +- `child_ids`：直接子元素的 ID，类型为 `ARRAY<STRING>`；叶子元素为空数组 当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段： - `source_uri`：源文件路径或 URI diff --git a/docs/zh/connectors/source/OssJindoFile.md b/docs/zh/connectors/source/OssJindoFile.md index a872c98592..d6773acbf2 100644 --- a/docs/zh/connectors/source/OssJindoFile.md +++ b/docs/zh/connectors/source/OssJindoFile.md @@ -126,7 +126,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码 - `page_number`：页码（默认：1） - `position_index`：文档中的位置索引 - `parent_id`：父元素的 ID -- `child_ids`：子元素 ID 的逗号分隔列表 +- `child_ids`：直接子元素的 ID，类型为 `ARRAY<STRING>`；叶子元素为空数组 当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段： - `source_uri`：源文件路径或 URI diff --git a/docs/zh/connectors/source/S3File.md b/docs/zh/connectors/source/S3File.md index 8dc9fd2752..aae121c25e 100644 --- a/docs/zh/connectors/source/S3File.md +++ b/docs/zh/connectors/source/S3File.md @@ -454,7 +454,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码 - `page_number`：页码（默认：1） - `position_index`：文档中的位置索引 - `parent_id`：父元素的 ID -- `child_ids`：子元素 ID 的逗号分隔列表 +- `child_ids`：直接子元素的 ID，类型为 `ARRAY<STRING>`；叶子元素为空数组 当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段： - `source_uri`：源文件路径或 URI diff --git a/docs/zh/connectors/source/SftpFile.md b/docs/zh/connectors/source/SftpFile.md index 4f9cf54d9f..c94e04060f 100644 --- a/docs/zh/connectors/source/SftpFile.md +++ b/docs/zh/connectors/source/SftpFile.md @@ -274,7 +274,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码 - `page_number`：页码（默认：1） - `position_index`：文档中的位置索引 - `parent_id`：父元素的 ID -- `child_ids`：子元素 ID 的逗号分隔列表 +- `child_ids`：直接子元素的 ID，类型为 `ARRAY<STRING>`；叶子元素为空数组 当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段： - `source_uri`：源文件路径或 URI diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index f2c287a693..bdd479c86f 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -29,6 +29,10 @@ - **影响**：这类表现在会被完整读取，但快照只有一个分片而不是多个。开启 `exactly_once = true` 时，若表唯一的键是可空唯一键，或 `table-names-config.primaryKeys` 中配置的是可空列，作业会在启动时报错 `Exactly once is enabled, but not found non-nullable primary key or unique key for table ...`，而不是继续运行并丢失数据。对于已有作业，即使配置的 `primaryKeys` 列中实际没有 NULL，只要该列在数据库中可空，也会如此。 - **迁移指南**：如需保持快照并行读取，请声明一个 NOT NULL 的唯一列，或通过 `table-names-config.primaryKeys` / `snapshotSplitColumn` 配置非空列。对于 `exactly_once = true`，请为表添加主键或非空唯一键，或将配置的 `primaryKeys` 列在数据库中改为 NOT NULL。 +### Markdown File Source 的 `child_ids` 数组 + +Markdown File Source 现在将 `child_ids` 声明为 `ARRAY<STRING>`，每行输出 `String[]`。此前该字段声明为 `STRING`，父元素输出逗号分隔的 ID，叶子元素输出 `null`。请将下游 schema 和消费逻辑改为读取数组；叶子元素现在输出空数组。元素 ID、类型、页码和位置索引保持原有格式。 + ### 运行环境要求 - **破坏性变更：最低 Java 运行时从 Java 8 提升到 Java 11** diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/MarkdownReadStrategy.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/MarkdownReadStrategy.java index 5d0aeaf055..aa065b2466 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/MarkdownReadStrategy.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/MarkdownReadStrategy.java @@ -20,6 +20,7 @@ package org.apache.seatunnel.connectors.seatunnel.file.source.reader; import org.apache.seatunnel.shade.com.typesafe.config.Config; import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.table.type.ArrayType; import org.apache.seatunnel.api.table.type.BasicType; import org.apache.seatunnel.api.table.type.KnowledgeSyncMetadataField; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; @@ -56,7 +57,9 @@ import java.io.InputStream; import java.nio.charset.StandardCharsets; import java.security.DigestInputStream; import java.security.MessageDigest; +import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Deque; import java.util.HashMap; import java.util.IdentityHashMap; import java.util.List; @@ -85,7 +88,7 @@ public class MarkdownReadStrategy extends AbstractReadStrategy { BasicType.INT_TYPE, BasicType.INT_TYPE, BasicType.STRING_TYPE, - BasicType.STRING_TYPE + ArrayType.STRING_ARRAY_TYPE }; /** Stable metadata fields appended for downstream RAG/document indexing pipelines. */ private static final String[] RAG_METADATA_FIELD_NAMES = { @@ -218,6 +221,7 @@ public class MarkdownReadStrategy extends AbstractReadStrategy { List<SeaTunnelRow> rows = new ArrayList<>(); assignIdsAndCollectTree(document, null, nodeInfoMap, DEFAULT_POSITION, typeCounters); + assignHeadingParents(document, nodeInfoMap); generateRows( document, rows, @@ -261,6 +265,31 @@ public class MarkdownReadStrategy extends AbstractReadStrategy { } } + /** Assigns section parents to top-level blocks after preserving nested AST relationships. */ + private void assignHeadingParents(Node document, Map<Node, NodeInfo> nodeInfoMap) { + Deque<Heading> headings = new ArrayDeque<>(); + for (Node node = document.getFirstChild(); node != null; node = node.getNext()) { + if (!isEligibleForRow(node)) { + continue; + } + if (node instanceof Heading) { + Heading heading = (Heading) node; + while (!headings.isEmpty() && headings.peek().getLevel() >= heading.getLevel()) { + headings.pop(); + } + } + if (!headings.isEmpty()) { + NodeInfo parentInfo = nodeInfoMap.get(headings.peek()); + NodeInfo nodeInfo = nodeInfoMap.get(node); + nodeInfo.parentId = parentInfo.elementId; + parentInfo.childIds.add(nodeInfo.elementId); + } + if (node instanceof Heading) { + headings.push((Heading) node); + } + } + } + private void generateRows( Node node, List<SeaTunnelRow> rows, @@ -288,7 +317,7 @@ public class MarkdownReadStrategy extends AbstractReadStrategy { pageNumber, nodeInfo.positionIndex, nodeInfo.parentId, - nodeInfo.childIds.isEmpty() ? null : String.join(",", nodeInfo.childIds) + nodeInfo.childIds.toArray(new String[0]) }; String contentHash = null; if (markdownRagMetadataEnabled) { diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSourceTest.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSourceTest.java index 93e6c76dd0..eaa73d1a55 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSourceTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSourceTest.java @@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.seatunnel.file.source; import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.type.ArrayType; import org.apache.seatunnel.api.table.type.KnowledgeSyncMetadataField; import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions; import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf; @@ -89,6 +90,8 @@ public class BaseFileSourceTest { Assertions.assertArrayEquals( MARKDOWN_FIELD_NAMES, catalogTable.getSeaTunnelRowType().getFieldNames()); + Assertions.assertEquals( + ArrayType.STRING_ARRAY_TYPE, catalogTable.getSeaTunnelRowType().getFieldType(7)); Assertions.assertTrue(catalogTable.getMetadataSchema().getColumns().isEmpty()); } @@ -101,6 +104,8 @@ public class BaseFileSourceTest { Assertions.assertArrayEquals( concat(MARKDOWN_FIELD_NAMES, MARKDOWN_RAG_METADATA_FIELD_NAMES), catalogTable.getSeaTunnelRowType().getFieldNames()); + Assertions.assertEquals( + ArrayType.STRING_ARRAY_TYPE, catalogTable.getSeaTunnelRowType().getFieldType(7)); assertMarkdownKnowledgeSyncMetadata(catalogTable); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/MarkdownReadStrategyTest.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/MarkdownReadStrategyTest.java index 7b71cade1d..910bc1d5f4 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/MarkdownReadStrategyTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/MarkdownReadStrategyTest.java @@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.seatunnel.file.source.reader; import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; +import org.apache.seatunnel.api.table.type.ArrayType; import org.apache.seatunnel.api.table.type.KnowledgeSyncMetadataField; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; @@ -38,6 +39,7 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.security.MessageDigest; import java.util.Arrays; +import java.util.List; import java.util.Map; import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_DEFAULT; @@ -75,6 +77,7 @@ class MarkdownReadStrategyTest { markdownReadStrategy.read(path, "", tempCollector); Assertions.assertArrayEquals(DEFAULT_FIELD_NAMES, rowType.getFieldNames()); + Assertions.assertEquals(ArrayType.STRING_ARRAY_TYPE, rowType.getFieldType(7)); Assertions.assertEquals(75, tempCollector.getRows().size()); Assertions.assertEquals( DEFAULT_FIELD_NAMES.length, tempCollector.getRows().get(0).getArity()); @@ -89,7 +92,13 @@ class MarkdownReadStrategyTest { Assertions.assertEquals(1, tempCollector.getRows().get(0).getField(4)); Assertions.assertEquals(1, tempCollector.getRows().get(0).getField(5)); Assertions.assertNull(tempCollector.getRows().get(0).getField(6)); - Assertions.assertNull(tempCollector.getRows().get(0).getField(7)); + Assertions.assertArrayEquals( + new String[] { + "Paragraph_1", "Heading_2", "Heading_3", "Heading_4", "Heading_5", + "Heading_6", "Heading_7", "Heading_8", "Heading_9", "Heading_10", + "Heading_11" + }, + (String[]) tempCollector.getRows().get(0).getField(7)); Assertions.assertEquals("OrderedList_1", tempCollector.getRows().get(3).getField(0)); Assertions.assertEquals("OrderedList", tempCollector.getRows().get(3).getField(1)); @@ -107,10 +116,14 @@ class MarkdownReadStrategyTest { tempCollector.getRows().get(3).getField(3)); Assertions.assertEquals(1, tempCollector.getRows().get(3).getField(4)); Assertions.assertEquals(5, tempCollector.getRows().get(3).getField(5)); - Assertions.assertNull(tempCollector.getRows().get(3).getField(6)); - Assertions.assertEquals( - "OrderedListItem_1,OrderedListItem_2,OrderedListItem_3,OrderedListItem_4,OrderedListItem_5,OrderedListItem_6,OrderedListItem_7,OrderedListItem_8,OrderedListItem_9", - tempCollector.getRows().get(3).getField(7)); + Assertions.assertEquals("Heading_2", tempCollector.getRows().get(3).getField(6)); + Assertions.assertArrayEquals( + new String[] { + "OrderedListItem_1", "OrderedListItem_2", "OrderedListItem_3", + "OrderedListItem_4", "OrderedListItem_5", "OrderedListItem_6", + "OrderedListItem_7", "OrderedListItem_8", "OrderedListItem_9" + }, + (String[]) tempCollector.getRows().get(3).getField(7)); Assertions.assertEquals("OrderedListItem_1", tempCollector.getRows().get(4).getField(0)); Assertions.assertEquals("OrderedListItem", tempCollector.getRows().get(4).getField(1)); @@ -120,7 +133,77 @@ class MarkdownReadStrategyTest { Assertions.assertEquals(1, tempCollector.getRows().get(4).getField(4)); Assertions.assertEquals(1, tempCollector.getRows().get(4).getField(5)); Assertions.assertEquals("OrderedList_1", tempCollector.getRows().get(4).getField(6)); - Assertions.assertNull(tempCollector.getRows().get(4).getField(7)); + Assertions.assertArrayEquals( + new String[0], (String[]) tempCollector.getRows().get(4).getField(7)); + } + + @Test + public void testHeadingHierarchyKeepsOriginalRowFields() throws Exception { + Path markdownFile = tempDir.resolve("heading-hierarchy.md"); + Files.write( + markdownFile, + ("# Data Source Configuration\n\n" + + "## Kafka Configuration\n" + + "Users can configure Kafka.\n\n" + + "## MySQL Configuration\n" + + "MySQL requires a JDBC URL.\n\n" + + "### Notes\n" + + "Test the connection before deploying.\n") + .getBytes(StandardCharsets.UTF_8)); + + TempCollector collector = new TempCollector(); + createMarkdownReadStrategy().read(markdownFile.toString(), "", collector); + List<SeaTunnelRow> rows = collector.getRows(); + + Assertions.assertEquals(7, rows.size()); + Assertions.assertEquals("Heading_1", rows.get(0).getField(0)); + Assertions.assertEquals("Heading", rows.get(0).getField(1)); + Assertions.assertEquals(1, rows.get(0).getField(4)); + Assertions.assertEquals(1, rows.get(0).getField(5)); + Assertions.assertNull(rows.get(0).getField(6)); + Assertions.assertArrayEquals( + new String[] {"Heading_2", "Heading_3"}, (String[]) rows.get(0).getField(7)); + + Assertions.assertEquals("Heading_1", rows.get(1).getField(6)); + Assertions.assertArrayEquals( + new String[] {"Paragraph_1"}, (String[]) rows.get(1).getField(7)); + Assertions.assertEquals("Heading_2", rows.get(2).getField(6)); + Assertions.assertArrayEquals(new String[0], (String[]) rows.get(2).getField(7)); + Assertions.assertEquals("Heading_1", rows.get(3).getField(6)); + Assertions.assertArrayEquals( + new String[] {"Paragraph_2", "Heading_4"}, (String[]) rows.get(3).getField(7)); + Assertions.assertEquals("Heading_3", rows.get(4).getField(6)); + Assertions.assertEquals("Heading_3", rows.get(5).getField(6)); + Assertions.assertArrayEquals( + new String[] {"Paragraph_3"}, (String[]) rows.get(5).getField(7)); + Assertions.assertEquals("Heading_4", rows.get(6).getField(6)); + } + + @Test + public void testHeadingHierarchyPreservesNestedAstRelations() throws Exception { + Path markdownFile = tempDir.resolve("nested-heading.md"); + Files.write( + markdownFile, + ("# Main\n\n- First\n- Second\n\n> ## Quoted\n\nAfter quote\n") + .getBytes(StandardCharsets.UTF_8)); + + TempCollector collector = new TempCollector(); + createMarkdownReadStrategy().read(markdownFile.toString(), "", collector); + List<SeaTunnelRow> rows = collector.getRows(); + + Assertions.assertArrayEquals( + new String[] {"BulletList_1", "BlockQuote_1", "Paragraph_1"}, + (String[]) rows.get(0).getField(7)); + Assertions.assertEquals("Heading_1", rows.get(1).getField(6)); + Assertions.assertArrayEquals( + new String[] {"BulletListItem_1", "BulletListItem_2"}, + (String[]) rows.get(1).getField(7)); + Assertions.assertEquals("BulletList_1", rows.get(2).getField(6)); + Assertions.assertEquals("Heading_1", rows.get(4).getField(6)); + Assertions.assertArrayEquals( + new String[] {"Heading_2"}, (String[]) rows.get(4).getField(7)); + Assertions.assertEquals("BlockQuote_1", rows.get(5).getField(6)); + Assertions.assertEquals("Heading_1", rows.get(6).getField(6)); } @Test @@ -147,7 +230,9 @@ class MarkdownReadStrategyTest { Assertions.assertArrayEquals( concat(DEFAULT_FIELD_NAMES, RAG_FIELD_NAMES), rowType.getFieldNames()); + Assertions.assertEquals(ArrayType.STRING_ARRAY_TYPE, rowType.getFieldType(7)); Assertions.assertEquals(75, firstCollector.getRows().size()); + Assertions.assertInstanceOf(String[].class, firstCollector.getRows().get(0).getField(7)); Assertions.assertEquals(13, firstCollector.getRows().get(0).getArity()); Assertions.assertEquals(path, firstCollector.getRows().get(0).getField(8)); Assertions.assertTrue( diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/local/LocalFileIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/local/LocalFileIT.java index 3eb325e074..50d24fe30f 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/local/LocalFileIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/local/LocalFileIT.java @@ -337,7 +337,7 @@ public class LocalFileIT extends TestSuiteBase { container.execInContainer( "sh", "-c", - "mkdir -p /seatunnel/read/markdown && printf '# E2E Markdown RAG\\n' > /seatunnel/read/markdown/e2e.md"); + "mkdir -p /seatunnel/read/markdown && printf '# E2E Markdown RAG\\n## Child\\nBody\\n' > /seatunnel/read/markdown/e2e.md"); ContainerUtil.copyFileIntoContainers( "/text/e2e.txt", "/seatunnel/read/recursive/e2e.txt", container); diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/resources/markdown/local_file_markdown_rag_metadata_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/resources/markdown/local_file_markdown_rag_metadata_to_assert.conf index 33f5aedc1b..b23ce9507c 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/resources/markdown/local_file_markdown_rag_metadata_to_assert.conf +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/resources/markdown/local_file_markdown_rag_metadata_to_assert.conf @@ -45,11 +45,41 @@ transform { ChunkHash = "chunk_hash" } } + Sql { + plugin_input = "markdown_rows_with_logical_metadata" + plugin_output = "heading_rows" + query = "select * from markdown_rows_with_logical_metadata where element_id = 'Heading_1'" + } + Sql { + plugin_input = "markdown_rows_with_logical_metadata" + plugin_output = "child_heading_rows" + query = "select * from markdown_rows_with_logical_metadata where element_id = 'Heading_2'" + } + Sql { + plugin_input = "markdown_rows_with_logical_metadata" + plugin_output = "paragraph_rows" + query = "select * from markdown_rows_with_logical_metadata where element_id = 'Paragraph_1'" + } + TableRename { + plugin_input = "heading_rows" + plugin_output = "heading_assert_rows" + suffix = "_heading" + } + TableRename { + plugin_input = "child_heading_rows" + plugin_output = "child_heading_assert_rows" + suffix = "_child_heading" + } + TableRename { + plugin_input = "paragraph_rows" + plugin_output = "paragraph_assert_rows" + suffix = "_paragraph" + } } sink { Assert { - plugin_input = "markdown_rows_with_logical_metadata" + plugin_input = "heading_assert_rows" rules { row_rules = [ { @@ -107,6 +137,11 @@ sink { } ] }, + { + field_name = child_ids + field_type = array<string> + field_value = [{ equals_to = ["Heading_2"] }] + }, { field_name = source_uri field_type = string @@ -175,7 +210,7 @@ sink { field_type = string field_value = [ { - equals_to = "37c48a8c1958a60f1cbd66f00afaef821c1e9dc1e549e8df96a58dc004be3609" + equals_to = "35f4024ea492a3a66048f502cce55f396f8eb1b03bffc22dc9a36e83d29decc9" } ] }, @@ -191,4 +226,46 @@ sink { ] } } + Assert { + plugin_input = "child_heading_assert_rows" + rules { + row_rules = [ + { rule_type = MAX_ROW, rule_value = 1 } + { rule_type = MIN_ROW, rule_value = 1 } + ] + field_rules = [ + { + field_name = parent_id + field_type = string + field_value = [{ equals_to = "Heading_1" }] + } + { + field_name = child_ids + field_type = array<string> + field_value = [{ equals_to = ["Paragraph_1"] }] + } + ] + } + } + Assert { + plugin_input = "paragraph_assert_rows" + rules { + row_rules = [ + { rule_type = MAX_ROW, rule_value = 1 } + { rule_type = MIN_ROW, rule_value = 1 } + ] + field_rules = [ + { + field_name = parent_id + field_type = string + field_value = [{ equals_to = "Heading_2" }] + } + { + field_name = child_ids + field_type = array<string> + field_value = [{ equals_to = [] }] + } + ] + } + } }
