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 970cadb1a4 [Feature][Connector-V2] Add optional Markdown RAG metadata
for file source (#10844)
970cadb1a4 is described below
commit 970cadb1a4b27f0347a3b1e65427bea8a68281c3
Author: yzeng1618 <[email protected]>
AuthorDate: Wed May 13 11:33:49 2026 +0800
[Feature][Connector-V2] Add optional Markdown RAG metadata for file source
(#10844)
Co-authored-by: zengyi <[email protected]>
---
docs/en/connectors/source/CosFile.md | 9 +
docs/en/connectors/source/FtpFile.md | 9 +
docs/en/connectors/source/HdfsFile.md | 9 +
docs/en/connectors/source/LocalFile.md | 9 +
docs/en/connectors/source/ObsFile.md | 9 +
docs/en/connectors/source/OssFile.md | 9 +
docs/en/connectors/source/OssJindoFile.md | 9 +
docs/en/connectors/source/S3File.md | 9 +
docs/en/connectors/source/SftpFile.md | 9 +
docs/zh/connectors/source/CosFile.md | 9 +
docs/zh/connectors/source/FtpFile.md | 9 +
docs/zh/connectors/source/HdfsFile.md | 9 +
docs/zh/connectors/source/LocalFile.md | 9 +
docs/zh/connectors/source/ObsFile.md | 28 +++
docs/zh/connectors/source/OssFile.md | 9 +
docs/zh/connectors/source/OssJindoFile.md | 29 +++
docs/zh/connectors/source/S3File.md | 9 +
docs/zh/connectors/source/SftpFile.md | 9 +
.../file/config/BaseFileSourceConfig.java | 14 +-
.../file/config/FileBaseSourceOptions.java | 8 +
.../seatunnel/file/source/BaseFileSource.java | 15 +-
.../file/source/reader/MarkdownReadStrategy.java | 203 ++++++++++++++++----
.../seatunnel/file/source/BaseFileSourceTest.java | 206 +++++++++++++++++++++
.../source/reader/MarkdownReadStrategyTest.java | 164 +++++++++++++++-
.../file/cos/source/CosFileSourceFactory.java | 4 +
.../file/ftp/source/FtpFileSourceFactory.java | 4 +
.../file/hdfs/source/HdfsFileSourceFactory.java | 4 +
.../oss/jindo/source/OssFileSourceFactory.java | 4 +
.../file/local/source/LocalFileSourceFactory.java | 4 +
.../seatunnel/file/local/LocalFileFactoryTest.java | 10 +
.../seatunnel/file/local/LocalFileSourceTest.java | 89 +++++++++
.../file/obs/source/ObsFileSourceFactory.java | 4 +
.../file/oss/source/OssFileSourceFactory.java | 4 +
.../file/s3/source/S3FileSourceFactory.java | 4 +
.../file/sftp/source/SftpFileSourceFactory.java | 4 +
.../e2e/connector/file/local/LocalFileIT.java | 12 ++
...local_file_markdown_rag_metadata_to_assert.conf | 143 ++++++++++++++
37 files changed, 1053 insertions(+), 48 deletions(-)
diff --git a/docs/en/connectors/source/CosFile.md
b/docs/en/connectors/source/CosFile.md
index b81254e02d..14097b4390 100644
--- a/docs/en/connectors/source/CosFile.md
+++ b/docs/en/connectors/source/CosFile.md
@@ -195,6 +195,15 @@ Each element is converted to a row with the following
schema:
- `parent_id`: ID of the parent element
- `child_ids`: Comma-separated list of child element IDs
+When `markdown_rag_metadata_enabled` is set to `true`, SeaTunnel appends the
following RAG metadata fields after `child_ids`:
+- `source_uri`: Source file path or URI
+- `document_id`: Stable document identifier derived from `source_uri`
+- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
+- `chunk_index`: One-based chunk order in the parsed document
+- `content_hash`: SHA-256 hash of the emitted `text` value
+
+The option defaults to `false`, so the original Markdown schema is unchanged
unless you enable it.
+
Note: Markdown format only supports reading, not writing.
### bucket [string]
diff --git a/docs/en/connectors/source/FtpFile.md
b/docs/en/connectors/source/FtpFile.md
index a59aa86cc8..d38c18b666 100644
--- a/docs/en/connectors/source/FtpFile.md
+++ b/docs/en/connectors/source/FtpFile.md
@@ -281,6 +281,15 @@ Each element is converted to a row with the following
schema:
- `parent_id`: ID of the parent element
- `child_ids`: Comma-separated list of child element IDs
+When `markdown_rag_metadata_enabled` is set to `true`, SeaTunnel appends the
following RAG metadata fields after `child_ids`:
+- `source_uri`: Source file path or URI
+- `document_id`: Stable document identifier derived from `source_uri`
+- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
+- `chunk_index`: One-based chunk order in the parsed document
+- `content_hash`: SHA-256 hash of the emitted `text` value
+
+The option defaults to `false`, so the original Markdown schema is unchanged
unless you enable it.
+
Note: Markdown format only supports reading, not writing.
### connection_mode [string]
diff --git a/docs/en/connectors/source/HdfsFile.md
b/docs/en/connectors/source/HdfsFile.md
index 5d0fe84de9..9a4835af34 100644
--- a/docs/en/connectors/source/HdfsFile.md
+++ b/docs/en/connectors/source/HdfsFile.md
@@ -114,6 +114,15 @@ Each element is converted to a row with the following
schema:
- `parent_id`: ID of the parent element
- `child_ids`: Comma-separated list of child element IDs
+When `markdown_rag_metadata_enabled` is set to `true`, SeaTunnel appends the
following RAG metadata fields after `child_ids`:
+- `source_uri`: Source file path or URI
+- `document_id`: Stable document identifier derived from `source_uri`
+- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
+- `chunk_index`: One-based chunk order in the parsed document
+- `content_hash`: SHA-256 hash of the emitted `text` value
+
+The option defaults to `false`, so the original Markdown schema is unchanged
unless you enable it.
+
Note: Markdown format only supports reading, not writing.
### delimiter/field_delimiter [string]
diff --git a/docs/en/connectors/source/LocalFile.md
b/docs/en/connectors/source/LocalFile.md
index 46aba942d6..33e4085509 100644
--- a/docs/en/connectors/source/LocalFile.md
+++ b/docs/en/connectors/source/LocalFile.md
@@ -201,6 +201,15 @@ Each element is converted to a row with the following
schema:
- `parent_id`: ID of the parent element
- `child_ids`: Comma-separated list of child element IDs
+When `markdown_rag_metadata_enabled` is set to `true`, SeaTunnel appends the
following RAG metadata fields after `child_ids`:
+- `source_uri`: Source file path or URI
+- `document_id`: Stable document identifier derived from `source_uri`
+- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
+- `chunk_index`: One-based chunk order in the parsed document
+- `content_hash`: SHA-256 hash of the emitted `text` value
+
+The option defaults to `false`, so the original Markdown schema is unchanged
unless you enable it.
+
Note: Markdown format only supports reading, not writing.
### read_columns [list]
diff --git a/docs/en/connectors/source/ObsFile.md
b/docs/en/connectors/source/ObsFile.md
index 573d618843..824c63f07c 100644
--- a/docs/en/connectors/source/ObsFile.md
+++ b/docs/en/connectors/source/ObsFile.md
@@ -237,6 +237,15 @@ schema {
> - `parent_id`: ID of the parent element
> - `child_ids`: Comma-separated list of child element IDs
>
+> When `markdown_rag_metadata_enabled` is set to `true`, SeaTunnel appends the
following RAG metadata fields after `child_ids`:
+> - `source_uri`: Source file path or URI
+> - `document_id`: Stable document identifier derived from `source_uri`
+> - `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
+> - `chunk_index`: One-based chunk order in the parsed document
+> - `content_hash`: SHA-256 hash of the emitted `text` value
+>
+> The option defaults to `false`, so the original Markdown schema is unchanged
unless you enable it.
+>
> Note: Markdown format only supports reading, not writing.
#### <span id="schema"> schema </span>
diff --git a/docs/en/connectors/source/OssFile.md
b/docs/en/connectors/source/OssFile.md
index 90ed790adc..2ed32e9ca9 100644
--- a/docs/en/connectors/source/OssFile.md
+++ b/docs/en/connectors/source/OssFile.md
@@ -236,6 +236,15 @@ Each element is converted to a row with the following
schema:
- `parent_id`: ID of the parent element
- `child_ids`: Comma-separated list of child element IDs
+When `markdown_rag_metadata_enabled` is set to `true`, SeaTunnel appends the
following RAG metadata fields after `child_ids`:
+- `source_uri`: Source file path or URI
+- `document_id`: Stable document identifier derived from `source_uri`
+- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
+- `chunk_index`: One-based chunk order in the parsed document
+- `content_hash`: SHA-256 hash of the emitted `text` value
+
+The option defaults to `false`, so the original Markdown schema is unchanged
unless you enable it.
+
Note: Markdown format only supports reading, not writing.
### compress_codec [string]
diff --git a/docs/en/connectors/source/OssJindoFile.md
b/docs/en/connectors/source/OssJindoFile.md
index f9e3f8c86b..e3b9867f23 100644
--- a/docs/en/connectors/source/OssJindoFile.md
+++ b/docs/en/connectors/source/OssJindoFile.md
@@ -197,6 +197,15 @@ Each element is converted to a row with the following
schema:
- `parent_id`: ID of the parent element
- `child_ids`: Comma-separated list of child element IDs
+When `markdown_rag_metadata_enabled` is set to `true`, SeaTunnel appends the
following RAG metadata fields after `child_ids`:
+- `source_uri`: Source file path or URI
+- `document_id`: Stable document identifier derived from `source_uri`
+- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
+- `chunk_index`: One-based chunk order in the parsed document
+- `content_hash`: SHA-256 hash of the emitted `text` value
+
+The option defaults to `false`, so the original Markdown schema is unchanged
unless you enable it.
+
Note: Markdown format only supports reading, not writing.
### bucket [string]
diff --git a/docs/en/connectors/source/S3File.md
b/docs/en/connectors/source/S3File.md
index 239886a80d..61275f37c1 100644
--- a/docs/en/connectors/source/S3File.md
+++ b/docs/en/connectors/source/S3File.md
@@ -245,6 +245,15 @@ Each element is converted to a row with the following
schema:
- `parent_id`: ID of the parent element
- `child_ids`: Comma-separated list of child element IDs
+When `markdown_rag_metadata_enabled` is set to `true`, SeaTunnel appends the
following RAG metadata fields after `child_ids`:
+- `source_uri`: Source file path or URI
+- `document_id`: Stable document identifier derived from `source_uri`
+- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
+- `chunk_index`: One-based chunk order in the parsed document
+- `content_hash`: SHA-256 hash of the emitted `text` value
+
+The option defaults to `false`, so the original Markdown schema is unchanged
unless you enable it.
+
Note: Markdown format only supports reading, not writing.
### delimiter/field_delimiter [string]
diff --git a/docs/en/connectors/source/SftpFile.md
b/docs/en/connectors/source/SftpFile.md
index 272ea7cbe5..b819cb3ce5 100644
--- a/docs/en/connectors/source/SftpFile.md
+++ b/docs/en/connectors/source/SftpFile.md
@@ -257,6 +257,15 @@ Each element is converted to a row with the following
schema:
- `parent_id`: ID of the parent element
- `child_ids`: Comma-separated list of child element IDs
+When `markdown_rag_metadata_enabled` is set to `true`, SeaTunnel appends the
following RAG metadata fields after `child_ids`:
+- `source_uri`: Source file path or URI
+- `document_id`: Stable document identifier derived from `source_uri`
+- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
+- `chunk_index`: One-based chunk order in the parsed document
+- `content_hash`: SHA-256 hash of the emitted `text` value
+
+The option defaults to `false`, so the original Markdown schema is unchanged
unless you enable it.
+
Note: Markdown format only supports reading, not writing.
### compress_codec [string]
diff --git a/docs/zh/connectors/source/CosFile.md
b/docs/zh/connectors/source/CosFile.md
index 1066a0ad5d..5a2e8943bc 100644
--- a/docs/zh/connectors/source/CosFile.md
+++ b/docs/zh/connectors/source/CosFile.md
@@ -192,6 +192,15 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
+当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+- `source_uri`：源文件路径或 URI
+- `document_id`：由 `source_uri` 派生的稳定文档标识符
+- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
+- `chunk_index`：解析后文档中的一基 chunk 顺序
+- `content_hash`：已输出 `text` 值的 SHA-256 哈希
+
+该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
+
注意：Markdown 格式仅支持读取，不支持写入。
根据此要求，您需要确保源端和目标端使用“二进制”格式进行文件同步同时。您可以在下面的示例中找到具体用法。
diff --git a/docs/zh/connectors/source/FtpFile.md
b/docs/zh/connectors/source/FtpFile.md
index a015ec91ad..64666a8c85 100644
--- a/docs/zh/connectors/source/FtpFile.md
+++ b/docs/zh/connectors/source/FtpFile.md
@@ -258,6 +258,15 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
+当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+- `source_uri`：源文件路径或 URI
+- `document_id`：由 `source_uri` 派生的稳定文档标识符
+- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
+- `chunk_index`：解析后文档中的一基 chunk 顺序
+- `content_hash`：已输出 `text` 值的 SHA-256 哈希
+
+该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
+
注意：Markdown 格式仅支持读取，不支持写入。
### connection_mode [string]
diff --git a/docs/zh/connectors/source/HdfsFile.md
b/docs/zh/connectors/source/HdfsFile.md
index 0a376cd92d..c2c649ffc9 100644
--- a/docs/zh/connectors/source/HdfsFile.md
+++ b/docs/zh/connectors/source/HdfsFile.md
@@ -114,6 +114,15 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
+当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+- `source_uri`：源文件路径或 URI
+- `document_id`：由 `source_uri` 派生的稳定文档标识符
+- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
+- `chunk_index`：解析后文档中的一基 chunk 顺序
+- `content_hash`：已输出 `text` 值的 SHA-256 哈希
+
+该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
+
注意：Markdown 格式仅支持读取，不支持写入。
### delimiter/field_delimiter [string]
diff --git a/docs/zh/connectors/source/LocalFile.md
b/docs/zh/connectors/source/LocalFile.md
index e39ffea331..14f1b074be 100644
--- a/docs/zh/connectors/source/LocalFile.md
+++ b/docs/zh/connectors/source/LocalFile.md
@@ -201,6 +201,15 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
+当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+- `source_uri`：源文件路径或 URI
+- `document_id`：由 `source_uri` 派生的稳定文档标识符
+- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
+- `chunk_index`：解析后文档中的一基 chunk 顺序
+- `content_hash`：已输出 `text` 值的 SHA-256 哈希
+
+该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
+
注意：Markdown 格式仅支持读取，不支持写入。
### read_columns [list]
diff --git a/docs/zh/connectors/source/ObsFile.md
b/docs/zh/connectors/source/ObsFile.md
index ab8d6becfd..6349284912 100644
--- a/docs/zh/connectors/source/ObsFile.md
+++ b/docs/zh/connectors/source/ObsFile.md
@@ -81,6 +81,34 @@ import ChangeLog from '../changelog/connector-file-obs.md';
| quote_char | string | 否 | " | 用于包裹 CSV
字段的单字符，可保证包含逗号、换行符或引号的字段被正确解析。 |
| escape_char | string | 否 | - | 用于在 CSV
字段内转义引号或其他特殊字符，使其不会结束字段。 |
+### file_format_type [string]
+
+文件类型，支持以下文件类型：
+
+`text` `csv` `parquet` `orc` `json` `excel` `markdown`
+
+如果您将文件类型指定为 `markdown`，SeaTunnel 可以解析 markdown 文件并提取结构化数据。
+markdown 解析器提取各种元素，包括标题、段落、列表、代码块、表格等。
+每个元素都转换为具有以下架构的行：
+- `element_id`：元素的唯一标识符
+- `element_type`：元素类型（Heading、Paragraph、ListItem 等）
+- `heading_level`：标题级别（1-6，非标题元素为 null）
+- `text`：元素的文本内容
+- `page_number`：页码（默认：1）
+- `position_index`：文档中的位置索引
+- `parent_id`：父元素的 ID
+- `child_ids`：子元素 ID 的逗号分隔列表
+
+当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+- `source_uri`：源文件路径或 URI
+- `document_id`：由 `source_uri` 派生的稳定文档标识符
+- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
+- `chunk_index`：解析后文档中的一基 chunk 顺序
+- `content_hash`：已输出 `text` 值的 SHA-256 哈希
+
+该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
+
+注意：Markdown 格式仅支持读取，不支持写入。
## 变更日志
diff --git a/docs/zh/connectors/source/OssFile.md
b/docs/zh/connectors/source/OssFile.md
index 45990aaf9f..aa583cb0e5 100644
--- a/docs/zh/connectors/source/OssFile.md
+++ b/docs/zh/connectors/source/OssFile.md
@@ -262,6 +262,15 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
+当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+- `source_uri`：源文件路径或 URI
+- `document_id`：由 `source_uri` 派生的稳定文档标识符
+- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
+- `chunk_index`：解析后文档中的一基 chunk 顺序
+- `content_hash`：已输出 `text` 值的 SHA-256 哈希
+
+该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
+
注意：Markdown 格式仅支持读取，不支持写入。
### quote_char [string]
diff --git a/docs/zh/connectors/source/OssJindoFile.md
b/docs/zh/connectors/source/OssJindoFile.md
index cbd5e25d18..096488d268 100644
--- a/docs/zh/connectors/source/OssJindoFile.md
+++ b/docs/zh/connectors/source/OssJindoFile.md
@@ -81,6 +81,35 @@ import ChangeLog from
'../changelog/connector-file-oss-jindo.md';
| quote_char | string | 否 | " |
用于包裹 CSV 字段的单字符，可保证包含逗号、换行符或引号的字段被正确解析。 |
| escape_char | string | 否 | - | 用于在
CSV 字段内转义引号或其他特殊字符，使其不会结束字段。 |
+### file_format_type [string]
+
+文件类型，支持以下文件类型：
+
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+
+如果您将文件类型指定为 `markdown`，SeaTunnel 可以解析 markdown 文件并提取结构化数据。
+markdown 解析器提取各种元素，包括标题、段落、列表、代码块、表格等。
+每个元素都转换为具有以下架构的行：
+- `element_id`：元素的唯一标识符
+- `element_type`：元素类型（Heading、Paragraph、ListItem 等）
+- `heading_level`：标题级别（1-6，非标题元素为 null）
+- `text`：元素的文本内容
+- `page_number`：页码（默认：1）
+- `position_index`：文档中的位置索引
+- `parent_id`：父元素的 ID
+- `child_ids`：子元素 ID 的逗号分隔列表
+
+当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+- `source_uri`：源文件路径或 URI
+- `document_id`：由 `source_uri` 派生的稳定文档标识符
+- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
+- `chunk_index`：解析后文档中的一基 chunk 顺序
+- `content_hash`：已输出 `text` 值的 SHA-256 哈希
+
+该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
+
+注意：Markdown 格式仅支持读取，不支持写入。
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/source/S3File.md
b/docs/zh/connectors/source/S3File.md
index 5b1dd33eb5..6fae586557 100644
--- a/docs/zh/connectors/source/S3File.md
+++ b/docs/zh/connectors/source/S3File.md
@@ -383,6 +383,15 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
+当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+- `source_uri`：源文件路径或 URI
+- `document_id`：由 `source_uri` 派生的稳定文档标识符
+- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
+- `chunk_index`：解析后文档中的一基 chunk 顺序
+- `content_hash`：已输出 `text` 值的 SHA-256 哈希
+
+该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
+
注意：Markdown 格式仅支持读取，不支持写入。
### schema [config]
diff --git a/docs/zh/connectors/source/SftpFile.md
b/docs/zh/connectors/source/SftpFile.md
index 7b3b13c1d9..9112f06d34 100644
--- a/docs/zh/connectors/source/SftpFile.md
+++ b/docs/zh/connectors/source/SftpFile.md
@@ -255,6 +255,15 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
+当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+- `source_uri`：源文件路径或 URI
+- `document_id`：由 `source_uri` 派生的稳定文档标识符
+- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
+- `chunk_index`：解析后文档中的一基 chunk 顺序
+- `content_hash`：已输出 `text` 值的 SHA-256 哈希
+
+该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
+
注意：Markdown 格式仅支持读取，不支持写入。
在此要求下，您需要确保源和接收器同时使用`binary`格式进行文件同步。
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseFileSourceConfig.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseFileSourceConfig.java
index 5b2223d605..604cf90db3 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseFileSourceConfig.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseFileSourceConfig.java
@@ -91,10 +91,8 @@ public abstract class BaseFileSourceConfig implements
Serializable {
if (CollectionUtils.isEmpty(filePaths)) {
// When there are no files (including sync_mode=update filtered
all files), choose a
// compatible schema so that downstream can initialize correctly.
- if (fileFormat == FileFormat.BINARY) {
- String rootPath =
readonlyConfig.get(FileBaseSourceOptions.FILE_PATH);
- return newCatalogTable(
- catalogTable,
readStrategy.getSeaTunnelRowTypeInfo(rootPath));
+ if (fileFormat == FileFormat.BINARY || fileFormat ==
FileFormat.MARKDOWN) {
+ return newCatalogTable(catalogTable,
getSchemaForEmptyFilePath(readonlyConfig));
}
return catalogTable;
}
@@ -114,6 +112,9 @@ public abstract class BaseFileSourceConfig implements
Serializable {
readStrategy.getSeaTunnelRowTypeInfoWithUserConfigRowType(
filePaths.get(0),
configSchema ?
catalogTable.getSeaTunnelRowType() : null));
+ case MARKDOWN:
+ return newCatalogTable(
+ catalogTable,
readStrategy.getSeaTunnelRowTypeInfo(filePaths.get(0)));
default:
throw new FileConnectorException(
FileConnectorErrorCode.FORMAT_NOT_SUPPORT,
@@ -121,6 +122,11 @@ public abstract class BaseFileSourceConfig implements
Serializable {
}
}
+ private SeaTunnelRowType getSchemaForEmptyFilePath(ReadonlyConfig
readonlyConfig) {
+ String rootPath = readonlyConfig.get(FileBaseSourceOptions.FILE_PATH);
+ return readStrategy.getSeaTunnelRowTypeInfo(rootPath);
+ }
+
private CatalogTable newCatalogTable(
CatalogTable catalogTable, SeaTunnelRowType seaTunnelRowType) {
TableSchema tableSchema = catalogTable.getTableSchema();
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileBaseSourceOptions.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileBaseSourceOptions.java
index c98d82b989..62590e5aa5 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileBaseSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileBaseSourceOptions.java
@@ -108,6 +108,14 @@ public class FileBaseSourceOptions extends FileBaseOptions
{
.noDefaultValue()
.withDescription("The columns list that the user want to
read");
+ public static final Option<Boolean> MARKDOWN_RAG_METADATA_ENABLED =
+ Options.key("markdown_rag_metadata_enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to append RAG-oriented metadata columns
when reading markdown files. "
+ + "Only valid when file_format_type is
markdown.");
+
public static final Option<ExcelEngine> EXCEL_ENGINE =
Options.key("excel_engine")
.enumType(ExcelEngine.class)
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSource.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSource.java
index f482b36310..ecbacb8e98 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSource.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSource.java
@@ -106,9 +106,8 @@ public abstract class BaseFileSource
"SeaTunnel does not supported this file format");
}
} else {
- if (filePaths.isEmpty() && fileFormat != FileFormat.BINARY) {
- // When the directory is empty, distribute default behavior
schema
- userDefinedCatalogTable =
CatalogTableUtil.buildSimpleTextTable();
+ if (filePaths.isEmpty()) {
+ userDefinedCatalogTable = buildCatalogTableForEmptyPath(path,
fileFormat);
} else {
try {
SeaTunnelRowType rowType =
@@ -126,6 +125,16 @@ public abstract class BaseFileSource
this.catalogTable = userDefinedCatalogTable;
}
+ private CatalogTable buildCatalogTableForEmptyPath(String path, FileFormat
fileFormat) {
+ if (fileFormat != FileFormat.BINARY && fileFormat !=
FileFormat.MARKDOWN) {
+ // Preserve the legacy simple-text fallback for formats that still
infer schema from
+ // the first concrete file.
+ return CatalogTableUtil.buildSimpleTextTable();
+ }
+ SeaTunnelRowType rowType = readStrategy.getSeaTunnelRowTypeInfo(path);
+ return CatalogTableUtil.getCatalogTable("default", rowType);
+ }
+
protected abstract HadoopConf initHadoopConf();
@Override
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 675fe5c69c..fcb1f5e95b 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
@@ -17,12 +17,18 @@
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.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.connectors.seatunnel.file.config.FileBaseSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
+import org.apache.commons.io.IOUtils;
+
import com.vladsch.flexmark.ast.BlockQuote;
import com.vladsch.flexmark.ast.BulletList;
import com.vladsch.flexmark.ast.Code;
@@ -42,8 +48,12 @@ import com.vladsch.flexmark.util.ast.Node;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
-import java.nio.file.Files;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Paths;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.IdentityHashMap;
@@ -55,6 +65,42 @@ public class MarkdownReadStrategy extends
AbstractReadStrategy {
private static final int DEFAULT_PAGE_NUMBER = 1;
private static final int DEFAULT_POSITION = 1;
+ private static final char[] HEX_CHARS = "0123456789abcdef".toCharArray();
+ private static final String[] DEFAULT_FIELD_NAMES = {
+ "element_id",
+ "element_type",
+ "heading_level",
+ "text",
+ "page_number",
+ "position_index",
+ "parent_id",
+ "child_ids"
+ };
+ private static final SeaTunnelDataType[] DEFAULT_FIELD_TYPES = {
+ BasicType.STRING_TYPE,
+ BasicType.STRING_TYPE,
+ BasicType.INT_TYPE,
+ BasicType.STRING_TYPE,
+ BasicType.INT_TYPE,
+ BasicType.INT_TYPE,
+ BasicType.STRING_TYPE,
+ BasicType.STRING_TYPE
+ };
+ /** Stable metadata fields appended for downstream RAG/document indexing
pipelines. */
+ private static final String[] RAG_METADATA_FIELD_NAMES = {
+ "source_uri", "document_id", "chunk_id", "chunk_index", "content_hash"
+ };
+
+ private static final SeaTunnelDataType[] RAG_METADATA_FIELD_TYPES = {
+ BasicType.STRING_TYPE,
+ BasicType.STRING_TYPE,
+ BasicType.STRING_TYPE,
+ BasicType.INT_TYPE,
+ BasicType.STRING_TYPE
+ };
+
+ private boolean markdownRagMetadataEnabled =
+ FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED.defaultValue();
private static class NodeInfo {
String elementId;
@@ -72,16 +118,26 @@ public class MarkdownReadStrategy extends
AbstractReadStrategy {
@Override
public void read(String path, String tableId, Collector<SeaTunnelRow>
output)
throws IOException, FileConnectorException {
- String markdown = new String(Files.readAllBytes(Paths.get(path)));
+ String markdown;
+ try (InputStream inputStream =
hadoopFileSystemProxy.getInputStream(path)) {
+ markdown = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
+ }
Parser parser = Parser.builder().build();
Node document = parser.parse(markdown);
+ String sourceUri = normalizeSourceUri(path);
Map<Node, NodeInfo> nodeInfoMap = new IdentityHashMap<>();
Map<String, Integer> typeCounters = new HashMap<>();
List<SeaTunnelRow> rows = new ArrayList<>();
assignIdsAndCollectTree(document, null, nodeInfoMap, DEFAULT_POSITION,
typeCounters);
- generateRows(document, rows, nodeInfoMap, DEFAULT_PAGE_NUMBER);
+ generateRows(
+ document,
+ rows,
+ nodeInfoMap,
+ DEFAULT_PAGE_NUMBER,
+ sourceUri,
+ buildDocumentId(sourceUri));
for (SeaTunnelRow row : rows) {
output.collect(row);
@@ -118,7 +174,12 @@ public class MarkdownReadStrategy extends
AbstractReadStrategy {
}
private void generateRows(
- Node node, List<SeaTunnelRow> rows, Map<Node, NodeInfo>
nodeInfoMap, int pageNumber) {
+ Node node,
+ List<SeaTunnelRow> rows,
+ Map<Node, NodeInfo> nodeInfoMap,
+ int pageNumber,
+ String sourceUri,
+ String documentId) {
if (isEligibleForRow(node)) {
NodeInfo nodeInfo = nodeInfoMap.get(node);
String elementType = node.getClass().getSimpleName();
@@ -129,20 +190,22 @@ public class MarkdownReadStrategy extends
AbstractReadStrategy {
headingLevel = ((Heading) node).getLevel();
}
- rows.add(
- new SeaTunnelRow(
- new Object[] {
- nodeInfo.elementId,
- elementType,
- headingLevel,
- text,
- pageNumber,
- nodeInfo.positionIndex,
- nodeInfo.parentId,
- nodeInfo.childIds.isEmpty()
- ? null
- : String.join(",", nodeInfo.childIds)
- }));
+ Object[] fields =
+ new Object[] {
+ nodeInfo.elementId,
+ elementType,
+ headingLevel,
+ text,
+ pageNumber,
+ nodeInfo.positionIndex,
+ nodeInfo.parentId,
+ nodeInfo.childIds.isEmpty() ? null : String.join(",",
nodeInfo.childIds)
+ };
+ if (markdownRagMetadataEnabled) {
+ fields = appendRagMetadata(fields, sourceUri, documentId,
rows.size() + 1, text);
+ }
+
+ rows.add(new SeaTunnelRow(fields));
log.debug(
"Added row: element_id={} type={} heading_level={} text={}
parent_id={} child_ids={}",
nodeInfo.elementId,
@@ -154,7 +217,7 @@ public class MarkdownReadStrategy extends
AbstractReadStrategy {
}
for (Node child = node.getFirstChild(); child != null; child =
child.getNext()) {
- generateRows(child, rows, nodeInfoMap, pageNumber);
+ generateRows(child, rows, nodeInfoMap, pageNumber, sourceUri,
documentId);
}
}
@@ -254,26 +317,86 @@ public class MarkdownReadStrategy extends
AbstractReadStrategy {
@Override
public SeaTunnelRowType getSeaTunnelRowTypeInfo(String path) throws
FileConnectorException {
- return new SeaTunnelRowType(
- new String[] {
- "element_id",
- "element_type",
- "heading_level",
- "text",
- "page_number",
- "position_index",
- "parent_id",
- "child_ids"
- },
- new org.apache.seatunnel.api.table.type.SeaTunnelDataType[] {
- BasicType.STRING_TYPE,
- BasicType.STRING_TYPE,
- BasicType.INT_TYPE,
- BasicType.STRING_TYPE,
- BasicType.INT_TYPE,
- BasicType.INT_TYPE,
- BasicType.STRING_TYPE,
- BasicType.STRING_TYPE
- });
+ if (markdownRagMetadataEnabled) {
+ return new SeaTunnelRowType(
+ concat(DEFAULT_FIELD_NAMES, RAG_METADATA_FIELD_NAMES),
+ concat(DEFAULT_FIELD_TYPES, RAG_METADATA_FIELD_TYPES));
+ }
+ return new SeaTunnelRowType(DEFAULT_FIELD_NAMES, DEFAULT_FIELD_TYPES);
+ }
+
+ @Override
+ public void setPluginConfig(Config pluginConfig) {
+ super.setPluginConfig(pluginConfig);
+ if
(pluginConfig.hasPath(FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED.key()))
{
+ markdownRagMetadataEnabled =
+ pluginConfig.getBoolean(
+
FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED.key());
+ }
+ }
+
+ private Object[] appendRagMetadata(
+ Object[] fields, String sourceUri, String documentId, int
chunkIndex, String text) {
+ String contentHash = sha256Hex(text == null ? "" : text);
+ // Keep chunk ids stable across re-reads of the same logical document
while still changing
+ // when the chunk content changes.
+ String chunkId = "chunk_" + sha256Hex(documentId + ":" + chunkIndex +
":" + contentHash);
+ Object[] enriched = new Object[fields.length +
RAG_METADATA_FIELD_NAMES.length];
+ System.arraycopy(fields, 0, enriched, 0, fields.length);
+ enriched[fields.length] = sourceUri;
+ enriched[fields.length + 1] = documentId;
+ enriched[fields.length + 2] = chunkId;
+ enriched[fields.length + 3] = chunkIndex;
+ enriched[fields.length + 4] = contentHash;
+ return enriched;
+ }
+
+ private static String buildDocumentId(String sourceUri) {
+ // Document ids stay anchored to the normalized source location so
every chunk from the same
+ // file shares one stable parent id.
+ return "doc_" + sha256Hex(sourceUri);
+ }
+
+ private static String normalizeSourceUri(String sourceUri) {
+ // Normalize local file URIs to the path form emitted by existing
local-file reads so the
+ // metadata contract stays stable between "file:/..." and plain local
paths.
+ if (!sourceUri.startsWith("file:")) {
+ return sourceUri;
+ }
+ try {
+ return Paths.get(URI.create(sourceUri)).toString();
+ } catch (IllegalArgumentException e) {
+ return sourceUri;
+ }
+ }
+
+ private static String sha256Hex(String value) {
+ try {
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+ byte[] bytes =
digest.digest(value.getBytes(StandardCharsets.UTF_8));
+ char[] chars = new char[bytes.length * 2];
+ for (int i = 0; i < bytes.length; i++) {
+ int unsigned = bytes[i] & 0xFF;
+ chars[i * 2] = HEX_CHARS[unsigned >>> 4];
+ chars[i * 2 + 1] = HEX_CHARS[unsigned & 0x0F];
+ }
+ return new String(chars);
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException("SHA-256 is not available", e);
+ }
+ }
+
+ private static String[] concat(String[] left, String[] right) {
+ String[] result = new String[left.length + right.length];
+ System.arraycopy(left, 0, result, 0, left.length);
+ System.arraycopy(right, 0, result, left.length, right.length);
+ return result;
+ }
+
+ private static SeaTunnelDataType[] concat(SeaTunnelDataType[] left,
SeaTunnelDataType[] right) {
+ SeaTunnelDataType[] result = new SeaTunnelDataType[left.length +
right.length];
+ System.arraycopy(left, 0, result, 0, left.length);
+ System.arraycopy(right, 0, result, left.length, right.length);
+ return result;
}
}
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
new file mode 100644
index 0000000000..e7ede52914
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSourceTest.java
@@ -0,0 +1,206 @@
+/*
+ * 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.connectors.seatunnel.file.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
+import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.util.Progressable;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+class BaseFileSourceTest {
+
+ private static final String EMPTY_FILE_SYSTEM_URI = "empty:///";
+ private static final String EMPTY_PATH = "empty:///empty-dir";
+ private static final String[] MARKDOWN_FIELD_NAMES = {
+ "element_id",
+ "element_type",
+ "heading_level",
+ "text",
+ "page_number",
+ "position_index",
+ "parent_id",
+ "child_ids"
+ };
+ private static final String[] MARKDOWN_RAG_METADATA_FIELD_NAMES = {
+ "source_uri", "document_id", "chunk_id", "chunk_index", "content_hash"
+ };
+
+ @Test
+ void testMarkdownSourceDiscoversSchemaFromEmptyDirectory() {
+ BaseFileSource source = new
TestFileSource(createMarkdownConfig(false));
+
+ CatalogTable catalogTable = source.getProducedCatalogTables().get(0);
+
+ Assertions.assertArrayEquals(
+ MARKDOWN_FIELD_NAMES,
catalogTable.getSeaTunnelRowType().getFieldNames());
+ }
+
+ @Test
+ void testMarkdownSourceDiscoversRagMetadataSchemaFromEmptyDirectory() {
+ BaseFileSource source = new TestFileSource(createMarkdownConfig(true));
+
+ CatalogTable catalogTable = source.getProducedCatalogTables().get(0);
+
+ Assertions.assertArrayEquals(
+ concat(MARKDOWN_FIELD_NAMES,
MARKDOWN_RAG_METADATA_FIELD_NAMES),
+ catalogTable.getSeaTunnelRowType().getFieldNames());
+ }
+
+ private ReadonlyConfig createMarkdownConfig(boolean ragMetadataEnabled) {
+ Map<String, Object> map = new HashMap<>();
+ map.put(FileBaseSourceOptions.FILE_PATH.key(), EMPTY_PATH);
+ map.put(FileBaseSourceOptions.FILE_FORMAT_TYPE.key(), "markdown");
+ map.put(FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED.key(),
ragMetadataEnabled);
+ return ReadonlyConfig.fromMap(map);
+ }
+
+ private static String[] concat(String[] left, String[] right) {
+ String[] result = new String[left.length + right.length];
+ System.arraycopy(left, 0, result, 0, left.length);
+ System.arraycopy(right, 0, result, left.length, right.length);
+ return result;
+ }
+
+ private static class TestFileSource extends BaseFileSource {
+
+ private TestFileSource(ReadonlyConfig pluginConfig) {
+ super(pluginConfig);
+ }
+
+ @Override
+ protected HadoopConf initHadoopConf() {
+ return new EmptyConf(EMPTY_FILE_SYSTEM_URI);
+ }
+
+ @Override
+ public String getPluginName() {
+ return "TestFileSource";
+ }
+ }
+
+ private static class EmptyConf extends HadoopConf {
+
+ private static final String SCHEMA = "empty";
+
+ private EmptyConf(String hdfsNameKey) {
+ super(hdfsNameKey);
+ }
+
+ @Override
+ public String getFsHdfsImpl() {
+ return EmptyFileSystem.class.getName();
+ }
+
+ @Override
+ public String getSchema() {
+ return SCHEMA;
+ }
+ }
+
+ public static class EmptyFileSystem extends FileSystem {
+
+ private URI uri;
+ private Path workingDirectory;
+
+ @Override
+ public void initialize(URI name, Configuration conf) throws
IOException {
+ super.initialize(name, conf);
+ this.uri = name;
+ this.workingDirectory = new Path("/");
+ }
+
+ @Override
+ public URI getUri() {
+ return uri;
+ }
+
+ @Override
+ public FSDataInputStream open(Path path, int bufferSize) {
+ throw new UnsupportedOperationException("open is not needed for
this test");
+ }
+
+ @Override
+ public FSDataOutputStream create(
+ Path path,
+ FsPermission permission,
+ boolean overwrite,
+ int bufferSize,
+ short replication,
+ long blockSize,
+ Progressable progress) {
+ throw new UnsupportedOperationException("create is not needed for
this test");
+ }
+
+ @Override
+ public FSDataOutputStream append(Path path, int bufferSize,
Progressable progress) {
+ throw new UnsupportedOperationException("append is not needed for
this test");
+ }
+
+ @Override
+ public boolean rename(Path source, Path target) {
+ throw new UnsupportedOperationException("rename is not needed for
this test");
+ }
+
+ @Override
+ public boolean delete(Path path, boolean recursive) {
+ throw new UnsupportedOperationException("delete is not needed for
this test");
+ }
+
+ @Override
+ public FileStatus[] listStatus(Path path) {
+ return new FileStatus[0];
+ }
+
+ @Override
+ public void setWorkingDirectory(Path newDirectory) {
+ this.workingDirectory = newDirectory;
+ }
+
+ @Override
+ public Path getWorkingDirectory() {
+ return workingDirectory;
+ }
+
+ @Override
+ public boolean mkdirs(Path path, FsPermission permission) {
+ return true;
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) {
+ return new FileStatus(0, true, 1, 0, 0, path.makeQualified(uri,
workingDirectory));
+ }
+ }
+}
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 bdd7189cb7..9c37687be5 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
@@ -17,23 +17,55 @@
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.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.nio.file.Paths;
+import java.util.Arrays;
+
+import static
org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_DEFAULT;
class MarkdownReadStrategyTest {
+ private static final String[] DEFAULT_FIELD_NAMES = {
+ "element_id",
+ "element_type",
+ "heading_level",
+ "text",
+ "page_number",
+ "position_index",
+ "parent_id",
+ "child_ids"
+ };
+ private static final String[] RAG_FIELD_NAMES = {
+ "source_uri", "document_id", "chunk_id", "chunk_index", "content_hash"
+ };
+
+ @TempDir private Path tempDir;
+
@Test
public void testReadMarkdown() throws Exception {
URL resource = this.getClass().getResource("/test.md");
String path = Paths.get(resource.toURI()).toString();
- AbstractReadStrategy markdownReadStrategy = new MarkdownReadStrategy();
+ AbstractReadStrategy markdownReadStrategy =
createMarkdownReadStrategy();
+ SeaTunnelRowType rowType =
markdownReadStrategy.getSeaTunnelRowTypeInfo(path);
TempCollector tempCollector = new TempCollector();
markdownReadStrategy.read(path, "", tempCollector);
+ Assertions.assertArrayEquals(DEFAULT_FIELD_NAMES,
rowType.getFieldNames());
Assertions.assertEquals(75, tempCollector.getRows().size());
+ Assertions.assertEquals(
+ DEFAULT_FIELD_NAMES.length,
tempCollector.getRows().get(0).getArity());
Assertions.assertEquals("Heading_1",
tempCollector.getRows().get(0).getField(0));
Assertions.assertEquals("Heading",
tempCollector.getRows().get(0).getField(1));
@@ -77,4 +109,134 @@ class MarkdownReadStrategyTest {
Assertions.assertEquals("OrderedList_1",
tempCollector.getRows().get(4).getField(6));
Assertions.assertNull(tempCollector.getRows().get(4).getField(7));
}
+
+ @Test
+ public void testReadMarkdownWithFileUri() throws Exception {
+ Path markdownFile = tempDir.resolve("doc.md");
+ Files.write(markdownFile, Arrays.asList("# Title"),
StandardCharsets.UTF_8);
+
+ AbstractReadStrategy markdownReadStrategy =
createMarkdownReadStrategy();
+ TempCollector tempCollector = new TempCollector();
+ markdownReadStrategy.read(markdownFile.toUri().toString(), "",
tempCollector);
+
+ Assertions.assertEquals(1, tempCollector.getRows().size());
+ Assertions.assertEquals("Title",
tempCollector.getRows().get(0).getField(3));
+ }
+
+ @Test
+ public void testReadMarkdownWithRagMetadata() throws Exception {
+ URL resource = this.getClass().getResource("/test.md");
+ String path = Paths.get(resource.toURI()).toString();
+ AbstractReadStrategy markdownReadStrategy =
createRagMetadataMarkdownReadStrategy();
+ SeaTunnelRowType rowType =
markdownReadStrategy.getSeaTunnelRowTypeInfo(path);
+ TempCollector firstCollector = new TempCollector();
+ markdownReadStrategy.read(path, "", firstCollector);
+
+ Assertions.assertArrayEquals(
+ concat(DEFAULT_FIELD_NAMES, RAG_FIELD_NAMES),
rowType.getFieldNames());
+ Assertions.assertEquals(75, firstCollector.getRows().size());
+ Assertions.assertEquals(13,
firstCollector.getRows().get(0).getArity());
+ Assertions.assertEquals(path,
firstCollector.getRows().get(0).getField(8));
+ Assertions.assertTrue(
+
String.valueOf(firstCollector.getRows().get(0).getField(9)).startsWith("doc_"));
+ Assertions.assertTrue(
+
String.valueOf(firstCollector.getRows().get(0).getField(10)).startsWith("chunk_"));
+ Assertions.assertEquals(1,
firstCollector.getRows().get(0).getField(11));
+ Assertions.assertEquals(
+ 64,
String.valueOf(firstCollector.getRows().get(0).getField(12)).length());
+
+ AbstractReadStrategy secondReadStrategy =
createRagMetadataMarkdownReadStrategy();
+ TempCollector secondCollector = new TempCollector();
+ secondReadStrategy.read(path, "", secondCollector);
+
+ for (int fieldIndex = 8; fieldIndex < 13; fieldIndex++) {
+ Assertions.assertEquals(
+ firstCollector.getRows().get(0).getField(fieldIndex),
+ secondCollector.getRows().get(0).getField(fieldIndex));
+ }
+ }
+
+ @Test
+ public void testReadMarkdownWithRagMetadataNormalizesFileUri() throws
Exception {
+ Path markdownFile = tempDir.resolve("doc.md");
+ Files.write(markdownFile, Arrays.asList("# Title"),
StandardCharsets.UTF_8);
+
+ AbstractReadStrategy markdownReadStrategy =
createRagMetadataMarkdownReadStrategy();
+ TempCollector tempCollector = new TempCollector();
+ markdownReadStrategy.read(markdownFile.toUri().toString(), "",
tempCollector);
+
+ AbstractReadStrategy expectedReadStrategy =
createRagMetadataMarkdownReadStrategy();
+ TempCollector expectedCollector = new TempCollector();
+ expectedReadStrategy.read(markdownFile.toString(), "",
expectedCollector);
+
+ Assertions.assertEquals(
+ markdownFile.toString(),
tempCollector.getRows().get(0).getField(8));
+ for (int fieldIndex = 8; fieldIndex < 11; fieldIndex++) {
+ Assertions.assertEquals(
+ expectedCollector.getRows().get(0).getField(fieldIndex),
+ tempCollector.getRows().get(0).getField(fieldIndex));
+ }
+ }
+
+ @Test
+ public void testRagMetadataContentHashChangesWithText() throws Exception {
+ Path markdownFile = tempDir.resolve("doc.md");
+ Files.write(markdownFile, Arrays.asList("# First Title"),
StandardCharsets.UTF_8);
+
+ AbstractReadStrategy firstReadStrategy =
createRagMetadataMarkdownReadStrategy();
+ TempCollector firstCollector = new TempCollector();
+ firstReadStrategy.read(markdownFile.toString(), "", firstCollector);
+ Object firstDocumentId = firstCollector.getRows().get(0).getField(9);
+ Object firstChunkIndex = firstCollector.getRows().get(0).getField(11);
+ Object firstContentHash = firstCollector.getRows().get(0).getField(12);
+
+ Files.write(markdownFile, Arrays.asList("# Second Title"),
StandardCharsets.UTF_8);
+
+ AbstractReadStrategy secondReadStrategy =
createRagMetadataMarkdownReadStrategy();
+ TempCollector secondCollector = new TempCollector();
+ secondReadStrategy.read(markdownFile.toString(), "", secondCollector);
+
+ Assertions.assertEquals(firstDocumentId,
secondCollector.getRows().get(0).getField(9));
+ Assertions.assertEquals(firstChunkIndex,
secondCollector.getRows().get(0).getField(11));
+ Assertions.assertNotEquals(firstContentHash,
secondCollector.getRows().get(0).getField(12));
+ }
+
+ private static AbstractReadStrategy createMarkdownReadStrategy() {
+ AbstractReadStrategy markdownReadStrategy = new MarkdownReadStrategy();
+ markdownReadStrategy.init(new LocalConf(FS_DEFAULT_NAME_DEFAULT));
+ return markdownReadStrategy;
+ }
+
+ private static AbstractReadStrategy
createRagMetadataMarkdownReadStrategy() {
+ AbstractReadStrategy markdownReadStrategy =
createMarkdownReadStrategy();
+ markdownReadStrategy.setPluginConfig(
+ ConfigFactory.parseString("markdown_rag_metadata_enabled =
true"));
+ return markdownReadStrategy;
+ }
+
+ private static String[] concat(String[] left, String[] right) {
+ String[] result = new String[left.length + right.length];
+ System.arraycopy(left, 0, result, 0, left.length);
+ System.arraycopy(right, 0, result, left.length, right.length);
+ return result;
+ }
+
+ public static class LocalConf extends HadoopConf {
+ private static final String HDFS_IMPL =
"org.apache.hadoop.fs.LocalFileSystem";
+ private static final String SCHEMA = "file";
+
+ public LocalConf(String hdfsNameKey) {
+ super(hdfsNameKey);
+ }
+
+ @Override
+ public String getFsHdfsImpl() {
+ return HDFS_IMPL;
+ }
+
+ @Override
+ public String getSchema() {
+ return SCHEMA;
+ }
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
index 3d069ae954..68704b920e 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
@@ -95,6 +95,10 @@ public class CosFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.MARKDOWN,
+ FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.build();
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java
index 2ad2088e95..79499cbf47 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java
@@ -100,6 +100,10 @@ public class FtpFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.MARKDOWN,
+ FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
.optional(FtpFileSourceOptions.FTP_REMOTE_VERIFICATION_ENABLED)
.optional(FtpFileSourceOptions.FTP_CONTROL_ENCODING)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java
index a08fa2c790..de9a088b7e 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java
@@ -108,6 +108,10 @@ public class HdfsFileSourceFactory implements
TableSourceFactory {
.optional(HdfsFileSourceOptions.NULL_FORMAT)
.optional(HdfsFileSourceOptions.FILENAME_EXTENSION)
.optional(HdfsFileSourceOptions.READ_COLUMNS)
+ .conditional(
+ HdfsFileSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.MARKDOWN,
+ HdfsFileSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
.optional(
HdfsFileSourceOptions.DISCOVERY_MODE,
HdfsFileSourceOptions.SCAN_INTERVAL,
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java
index ecbcd908c6..d07ab8fffe 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java
@@ -90,6 +90,10 @@ public class OssFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.MARKDOWN,
+ FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.build();
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java
index 972b569f95..5cd72b6513 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java
@@ -109,6 +109,10 @@ public class LocalFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.MARKDOWN,
+ FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(ConnectorCommonOptions.METALAKE_TYPE)
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java
index 5c3a97968c..1fbbe1a12b 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java
@@ -49,6 +49,16 @@ class LocalFileFactoryTest {
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.SCAN_INTERVAL));
Assertions.assertTrue(
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.START_MODE));
+ Assertions.assertTrue(
+ optionRule.getRequiredOptions().stream()
+
.filter(RequiredOption.ConditionalRequiredOptions.class::isInstance)
+
.map(RequiredOption.ConditionalRequiredOptions.class::cast)
+ .anyMatch(
+ required ->
+ required.getOptions()
+ .contains(
+ FileBaseSourceOptions
+
.MARKDOWN_RAG_METADATA_ENABLED)));
Expression expectExpression =
Expression.of(FileBaseSourceOptions.SYNC_MODE,
FileSyncMode.UPDATE);
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileSourceTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileSourceTest.java
index a64af3b81a..2a2629638e 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileSourceTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileSourceTest.java
@@ -17,11 +17,14 @@
package org.apache.seatunnel.connectors.seatunnel.file.local;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
import
org.apache.seatunnel.connectors.seatunnel.file.config.ArchiveCompressFormat;
import org.apache.seatunnel.connectors.seatunnel.file.config.CompressFormat;
import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
import
org.apache.seatunnel.connectors.seatunnel.file.local.config.LocalFileHadoopConf;
+import
org.apache.seatunnel.connectors.seatunnel.file.local.source.config.LocalFileSourceConfig;
import
org.apache.seatunnel.connectors.seatunnel.file.source.split.AccordingToSplitSizeSplitStrategy;
import
org.apache.seatunnel.connectors.seatunnel.file.source.split.DefaultFileSplitStrategy;
import
org.apache.seatunnel.connectors.seatunnel.file.source.split.FileSplitStrategy;
@@ -30,13 +33,35 @@ import
org.apache.seatunnel.connectors.seatunnel.file.source.split.ParquetFileSp
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import java.io.Closeable;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
public class LocalFileSourceTest {
+ private static final String[] MARKDOWN_FIELD_NAMES = {
+ "element_id",
+ "element_type",
+ "heading_level",
+ "text",
+ "page_number",
+ "position_index",
+ "parent_id",
+ "child_ids"
+ };
+ private static final String[] MARKDOWN_RAG_METADATA_FIELD_NAMES = {
+ "source_uri", "document_id", "chunk_id", "chunk_index", "content_hash"
+ };
+
+ @TempDir private Path tempDir;
+
@Test
void testInitFileSplitStrategy() {
// test orc
@@ -116,6 +141,63 @@ public class LocalFileSourceTest {
closeQuietly(fileSplitStrategy);
}
+ @Test
+ void testMarkdownSourceDiscoversRagMetadataSchema() throws IOException {
+ Path markdownFile = tempDir.resolve("doc.md");
+ Files.write(markdownFile, Collections.singletonList("# Title"),
StandardCharsets.UTF_8);
+
+ Map<String, Object> map = new HashMap<>();
+ map.put(FileBaseSourceOptions.FILE_PATH.key(),
markdownFile.toUri().toString());
+ map.put(FileBaseSourceOptions.FILE_FORMAT_TYPE.key(), "markdown");
+ map.put(FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED.key(),
true);
+
+ CatalogTable catalogTable =
+ new LocalFileSourceConfig(
+ ReadonlyConfig.fromMap(map),
+ CatalogTableUtil.buildSimpleTextTable())
+ .getCatalogTable();
+
+ Assertions.assertArrayEquals(
+ concat(MARKDOWN_FIELD_NAMES,
MARKDOWN_RAG_METADATA_FIELD_NAMES),
+ catalogTable.getSeaTunnelRowType().getFieldNames());
+ }
+
+ @Test
+ void testContinuousMarkdownSourceDiscoversSchemaWithoutFiles() {
+ Map<String, Object> map = new HashMap<>();
+ map.put(FileBaseSourceOptions.FILE_PATH.key(), tempDir.toString());
+ map.put(FileBaseSourceOptions.FILE_FORMAT_TYPE.key(), "markdown");
+ map.put(FileBaseSourceOptions.DISCOVERY_MODE.key(), "continuous");
+
+ CatalogTable catalogTable =
+ new LocalFileSourceConfig(
+ ReadonlyConfig.fromMap(map),
+ CatalogTableUtil.buildSimpleTextTable())
+ .getCatalogTable();
+
+ Assertions.assertArrayEquals(
+ MARKDOWN_FIELD_NAMES,
catalogTable.getSeaTunnelRowType().getFieldNames());
+ }
+
+ @Test
+ void testContinuousMarkdownSourceDiscoversRagMetadataSchemaWithoutFiles() {
+ Map<String, Object> map = new HashMap<>();
+ map.put(FileBaseSourceOptions.FILE_PATH.key(), tempDir.toString());
+ map.put(FileBaseSourceOptions.FILE_FORMAT_TYPE.key(), "markdown");
+ map.put(FileBaseSourceOptions.DISCOVERY_MODE.key(), "continuous");
+ map.put(FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED.key(),
true);
+
+ CatalogTable catalogTable =
+ new LocalFileSourceConfig(
+ ReadonlyConfig.fromMap(map),
+ CatalogTableUtil.buildSimpleTextTable())
+ .getCatalogTable();
+
+ Assertions.assertArrayEquals(
+ concat(MARKDOWN_FIELD_NAMES,
MARKDOWN_RAG_METADATA_FIELD_NAMES),
+ catalogTable.getSeaTunnelRowType().getFieldNames());
+ }
+
private void closeQuietly(FileSplitStrategy strategy) {
try {
if (strategy instanceof Closeable) {
@@ -129,4 +211,11 @@ public class LocalFileSourceTest {
// ignore
}
}
+
+ private static String[] concat(String[] left, String[] right) {
+ String[] result = new String[left.length + right.length];
+ System.arraycopy(left, 0, result, 0, left.length);
+ System.arraycopy(right, 0, result, left.length, right.length);
+ return result;
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java
index 98897d2577..1f97737952 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java
@@ -79,6 +79,10 @@ public class ObsFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.MARKDOWN,
+ FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.build();
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java
index bf6e6701a5..b620f06289 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java
@@ -99,6 +99,10 @@ public class OssFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.MARKDOWN,
+ FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(ConnectorCommonOptions.METALAKE_TYPE)
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java
index 303ec74d39..f982fcce4a 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java
@@ -115,6 +115,10 @@ public class S3FileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.MARKDOWN,
+ FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(ConnectorCommonOptions.METALAKE_TYPE)
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java
index 44d7782d99..16e22ae664 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java
@@ -89,6 +89,10 @@ public class SftpFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.MARKDOWN,
+ FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(ConnectorCommonOptions.METALAKE_TYPE)
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 dc08fa7782..c0cc2d77c2 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
@@ -319,6 +319,11 @@ public class LocalFileIT extends TestSuiteBase {
"/seatunnel/read/e2e_null_format/e2e_null_format.txt",
container);
+ container.execInContainer(
+ "sh",
+ "-c",
+ "mkdir -p /seatunnel/read/markdown && printf '# E2E
Markdown RAG\\n' > /seatunnel/read/markdown/e2e.md");
+
container.execInContainer("mkdir", "-p", "/tmp/fake_empty");
};
@@ -427,6 +432,13 @@ public class LocalFileIT extends TestSuiteBase {
helper.execute("/excel/local_excel_xlsx_gz_to_assert.conf");
}
+ @TestTemplate
+ public void testLocalFileReadMarkdownRagMetadata(TestContainer container)
+ throws IOException, InterruptedException {
+ TestHelper helper = new TestHelper(container);
+
helper.execute("/markdown/local_file_markdown_rag_metadata_to_assert.conf");
+ }
+
@TestTemplate
@DisabledOnContainer(
value = {},
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
new file mode 100644
index 0000000000..ef0a6d543c
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/resources/markdown/local_file_markdown_rag_metadata_to_assert.conf
@@ -0,0 +1,143 @@
+#
+# 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"
+ spark.app.name = "SeaTunnel"
+ spark.executor.instances = 2
+ spark.executor.cores = 1
+ spark.executor.memory = "1g"
+ spark.master = local
+}
+
+source {
+ LocalFile {
+ path = "/seatunnel/read/markdown/e2e.md"
+ file_format_type = "markdown"
+ markdown_rag_metadata_enabled = true
+ }
+}
+
+sink {
+ Assert {
+ rules {
+ row_rules = [
+ {
+ rule_type = MAX_ROW
+ rule_value = 1
+ },
+ {
+ rule_type = MIN_ROW
+ rule_value = 1
+ }
+ ],
+ field_rules = [
+ {
+ field_name = element_id
+ field_type = string
+ field_value = [
+ {
+ equals_to = "Heading_1"
+ }
+ ]
+ },
+ {
+ field_name = element_type
+ field_type = string
+ field_value = [
+ {
+ equals_to = "Heading"
+ }
+ ]
+ },
+ {
+ field_name = heading_level
+ field_type = int
+ field_value = [
+ {
+ equals_to = 1
+ }
+ ]
+ },
+ {
+ field_name = text
+ field_type = string
+ field_value = [
+ {
+ equals_to = "E2E Markdown RAG"
+ }
+ ]
+ },
+ {
+ field_name = parent_id
+ field_type = string
+ field_value = [
+ {
+ rule_type = NULL
+ }
+ ]
+ },
+ {
+ field_name = source_uri
+ field_type = string
+ field_value = [
+ {
+ equals_to = "/seatunnel/read/markdown/e2e.md"
+ }
+ ]
+ },
+ {
+ field_name = document_id
+ field_type = string
+ field_value = [
+ {
+ equals_to =
"doc_f9e9a2324852c0fdff7ecea9013e95f3a2361f570b0d5226f2f671ae38d1d1c8"
+ }
+ ]
+ },
+ {
+ field_name = chunk_id
+ field_type = string
+ field_value = [
+ {
+ equals_to =
"chunk_b48c799b9daccdb6a04d665be5892bc9efc84795f266f7ec27c5b4fd2f0a5af0"
+ }
+ ]
+ },
+ {
+ field_name = chunk_index
+ field_type = int
+ field_value = [
+ {
+ equals_to = 1
+ }
+ ]
+ },
+ {
+ field_name = content_hash
+ field_type = string
+ field_value = [
+ {
+ equals_to =
"6fc6a22e9e5515086f32ea5400fcd490c92cbe0f3707febffb8f97b83abe2f46"
+ }
+ ]
+ }
+ ]
+ }
+ }
+}
