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 2b060343dc [Feature][Connector-File-Base] Add optional PDF RAG
metadata for file source (#11571)
2b060343dc is described below
commit 2b060343dcc99c58fe53ebede9e8b6886e19bfeb
Author: JeremyXin <[email protected]>
AuthorDate: Tue Aug 4 16:07:38 2026 +0800
[Feature][Connector-File-Base] Add optional PDF RAG metadata for file
source (#11571)
---
docs/en/connectors/source/CosFile.md | 3 +-
docs/en/connectors/source/FtpFile.md | 3 +-
docs/en/connectors/source/HdfsFile.md | 3 +-
docs/en/connectors/source/LocalFile.md | 3 +-
docs/en/connectors/source/ObsFile.md | 3 +-
docs/en/connectors/source/OssFile.md | 3 +-
docs/en/connectors/source/OssJindoFile.md | 3 +-
docs/en/connectors/source/S3File.md | 3 +-
docs/en/connectors/source/SftpFile.md | 3 +-
docs/zh/connectors/source/CosFile.md | 3 +-
docs/zh/connectors/source/FtpFile.md | 3 +-
docs/zh/connectors/source/HdfsFile.md | 3 +-
docs/zh/connectors/source/LocalFile.md | 3 +-
docs/zh/connectors/source/ObsFile.md | 3 +-
docs/zh/connectors/source/OssFile.md | 3 +-
docs/zh/connectors/source/OssJindoFile.md | 3 +-
docs/zh/connectors/source/S3File.md | 3 +-
docs/zh/connectors/source/SftpFile.md | 3 +-
.../file/config/BaseFileSourceConfig.java | 5 +-
.../file/config/FileBaseSourceOptions.java | 8 +
.../seatunnel/file/source/BaseFileSource.java | 4 +-
.../file/source/reader/PdfReadStrategy.java | 164 ++++++++++++++++++---
.../seatunnel/file/source/BaseFileSourceTest.java | 42 ++++++
.../file/source/reader/PdfReadStrategyTest.java | 109 ++++++++++++++
.../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 +
.../file/obs/source/ObsFileSourceFactory.java | 4 +
.../file/oss/source/OssFileSourceFactory.java | 4 +
.../file/s3/source/S3FileSourceFactory.java | 4 +
.../file/sftp/source/SftpFileSourceFactory.java | 4 +
33 files changed, 380 insertions(+), 42 deletions(-)
diff --git a/docs/en/connectors/source/CosFile.md
b/docs/en/connectors/source/CosFile.md
index e1759f2e06..685e9ccf84 100644
--- a/docs/en/connectors/source/CosFile.md
+++ b/docs/en/connectors/source/CosFile.md
@@ -198,7 +198,7 @@ Each extracted element is converted to a document-element
row with the following
- `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`:
+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
- `document_id`: Stable document identifier derived from `source_uri`
- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
@@ -213,6 +213,7 @@ Note: Markdown format only supports reading, not writing.
If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
PDF uses the same document-element row schema described above.
+For PDF input, enable `pdf_rag_metadata_enabled` to append the RAG metadata
fields described above.
The main PDF-specific behaviors are:
diff --git a/docs/en/connectors/source/FtpFile.md
b/docs/en/connectors/source/FtpFile.md
index 59cbe2d19d..e84c2149be 100644
--- a/docs/en/connectors/source/FtpFile.md
+++ b/docs/en/connectors/source/FtpFile.md
@@ -300,7 +300,7 @@ Each extracted element is converted to a document-element
row with the following
- `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`:
+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
- `document_id`: Stable document identifier derived from `source_uri`
- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
@@ -315,6 +315,7 @@ Note: Markdown format only supports reading, not writing.
If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
PDF uses the same document-element row schema described above.
+For PDF input, enable `pdf_rag_metadata_enabled` to append the RAG metadata
fields described above.
The main PDF-specific behaviors are:
diff --git a/docs/en/connectors/source/HdfsFile.md
b/docs/en/connectors/source/HdfsFile.md
index d25f4b0ccf..04ed4bc55e 100644
--- a/docs/en/connectors/source/HdfsFile.md
+++ b/docs/en/connectors/source/HdfsFile.md
@@ -125,7 +125,7 @@ Each extracted element is converted to a document-element
row with the following
- `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`:
+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
- `document_id`: Stable document identifier derived from `source_uri`
- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
@@ -140,6 +140,7 @@ Note: Markdown format only supports reading, not writing.
If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
PDF uses the same document-element row schema described above.
+For PDF input, enable `pdf_rag_metadata_enabled` to append the RAG metadata
fields described above.
The main PDF-specific behaviors are:
diff --git a/docs/en/connectors/source/LocalFile.md
b/docs/en/connectors/source/LocalFile.md
index 63d7fec912..7959a1b184 100644
--- a/docs/en/connectors/source/LocalFile.md
+++ b/docs/en/connectors/source/LocalFile.md
@@ -212,7 +212,7 @@ Each extracted element is converted to a document-element
row with the following
- `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`:
+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
- `document_id`: Stable document identifier derived from `source_uri`
- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
@@ -227,6 +227,7 @@ Note: Markdown format only supports reading, not writing.
If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
PDF uses the same document-element row schema described above.
+For PDF input, enable `pdf_rag_metadata_enabled` to append the RAG metadata
fields described above.
The main PDF-specific behaviors are:
diff --git a/docs/en/connectors/source/ObsFile.md
b/docs/en/connectors/source/ObsFile.md
index bf9b51882a..1226fc55e8 100644
--- a/docs/en/connectors/source/ObsFile.md
+++ b/docs/en/connectors/source/ObsFile.md
@@ -236,7 +236,7 @@ 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`:
+> 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
> - `document_id`: Stable document identifier derived from `source_uri`
> - `chunk_id`: Stable chunk identifier derived from document identity, chunk
> order, and content hash
@@ -251,6 +251,7 @@ schema {
>
> If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
> structured document elements.
> PDF uses the same document-element row schema described above.
+> For PDF input, enable `pdf_rag_metadata_enabled` to append the RAG metadata
fields described above.
>
> The main PDF-specific behaviors are:
>
diff --git a/docs/en/connectors/source/OssFile.md
b/docs/en/connectors/source/OssFile.md
index c65574c611..5c4e6fcc8d 100644
--- a/docs/en/connectors/source/OssFile.md
+++ b/docs/en/connectors/source/OssFile.md
@@ -239,7 +239,7 @@ Each extracted element is converted to a document-element
row with the following
- `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`:
+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
- `document_id`: Stable document identifier derived from `source_uri`
- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
@@ -254,6 +254,7 @@ Note: Markdown format only supports reading, not writing.
If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
PDF uses the same document-element row schema described above.
+For PDF input, enable `pdf_rag_metadata_enabled` to append the RAG metadata
fields described above.
The main PDF-specific behaviors are:
diff --git a/docs/en/connectors/source/OssJindoFile.md
b/docs/en/connectors/source/OssJindoFile.md
index f9179ea0bf..55858e8c94 100644
--- a/docs/en/connectors/source/OssJindoFile.md
+++ b/docs/en/connectors/source/OssJindoFile.md
@@ -200,7 +200,7 @@ Each extracted element is converted to a document-element
row with the following
- `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`:
+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
- `document_id`: Stable document identifier derived from `source_uri`
- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
@@ -215,6 +215,7 @@ Note: Markdown format only supports reading, not writing.
If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
PDF uses the same document-element row schema described above.
+For PDF input, enable `pdf_rag_metadata_enabled` to append the RAG metadata
fields described above.
The main PDF-specific behaviors are:
diff --git a/docs/en/connectors/source/S3File.md
b/docs/en/connectors/source/S3File.md
index cab46613ba..36f96c2448 100644
--- a/docs/en/connectors/source/S3File.md
+++ b/docs/en/connectors/source/S3File.md
@@ -249,7 +249,7 @@ Each extracted element is converted to a document-element
row with the following
- `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`:
+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
- `document_id`: Stable document identifier derived from `source_uri`
- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
@@ -264,6 +264,7 @@ Note: Markdown format only supports reading, not writing.
If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
PDF uses the same document-element row schema described above.
+For PDF input, enable `pdf_rag_metadata_enabled` to append the RAG metadata
fields described above.
The main PDF-specific behaviors are:
diff --git a/docs/en/connectors/source/SftpFile.md
b/docs/en/connectors/source/SftpFile.md
index 6b6c0ce739..b6c71f8ff0 100644
--- a/docs/en/connectors/source/SftpFile.md
+++ b/docs/en/connectors/source/SftpFile.md
@@ -268,7 +268,7 @@ Each extracted element is converted to a document-element
row with the following
- `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`:
+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
- `document_id`: Stable document identifier derived from `source_uri`
- `chunk_id`: Stable chunk identifier derived from document identity, chunk
order, and content hash
@@ -283,6 +283,7 @@ Note: Markdown format only supports reading, not writing.
If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
PDF uses the same document-element row schema described above.
+For PDF input, enable `pdf_rag_metadata_enabled` to append the RAG metadata
fields described above.
The main PDF-specific behaviors are:
diff --git a/docs/zh/connectors/source/CosFile.md
b/docs/zh/connectors/source/CosFile.md
index 2108778f3a..1c5e95afaa 100644
--- a/docs/zh/connectors/source/CosFile.md
+++ b/docs/zh/connectors/source/CosFile.md
@@ -196,7 +196,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
-当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true`
时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段：
- `source_uri`：源文件路径或 URI
- `document_id`：由 `source_uri` 派生的稳定文档标识符
- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
@@ -211,6 +211,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
PDF 使用与上文相同的文档元素 schema。
+对于 PDF 输入，启用 `pdf_rag_metadata_enabled` 即可追加上文所述的 RAG 元数据字段。
PDF 特有的解析行为如下：
diff --git a/docs/zh/connectors/source/FtpFile.md
b/docs/zh/connectors/source/FtpFile.md
index 623386927b..7fbfe85c67 100644
--- a/docs/zh/connectors/source/FtpFile.md
+++ b/docs/zh/connectors/source/FtpFile.md
@@ -281,7 +281,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
-当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true`
时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段：
- `source_uri`：源文件路径或 URI
- `document_id`：由 `source_uri` 派生的稳定文档标识符
- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
@@ -296,6 +296,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
PDF 使用与上文相同的文档元素 schema。
+对于 PDF 输入，启用 `pdf_rag_metadata_enabled` 即可追加上文所述的 RAG 元数据字段。
PDF 特有的解析行为如下：
diff --git a/docs/zh/connectors/source/HdfsFile.md
b/docs/zh/connectors/source/HdfsFile.md
index ec3d42b873..4b3fa99498 100644
--- a/docs/zh/connectors/source/HdfsFile.md
+++ b/docs/zh/connectors/source/HdfsFile.md
@@ -125,7 +125,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
-当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true`
时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段：
- `source_uri`：源文件路径或 URI
- `document_id`：由 `source_uri` 派生的稳定文档标识符
- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
@@ -140,6 +140,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
PDF 使用与上文相同的文档元素 schema。
+对于 PDF 输入，启用 `pdf_rag_metadata_enabled` 即可追加上文所述的 RAG 元数据字段。
PDF 特有的解析行为如下：
diff --git a/docs/zh/connectors/source/LocalFile.md
b/docs/zh/connectors/source/LocalFile.md
index ded499c9b8..f0c9ee9ddd 100644
--- a/docs/zh/connectors/source/LocalFile.md
+++ b/docs/zh/connectors/source/LocalFile.md
@@ -212,7 +212,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
-当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true`
时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段：
- `source_uri`：源文件路径或 URI
- `document_id`：由 `source_uri` 派生的稳定文档标识符
- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
@@ -227,6 +227,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
PDF 使用与上文相同的文档元素 schema。
+对于 PDF 输入，启用 `pdf_rag_metadata_enabled` 即可追加上文所述的 RAG 元数据字段。
PDF 特有的解析行为如下：
diff --git a/docs/zh/connectors/source/ObsFile.md
b/docs/zh/connectors/source/ObsFile.md
index bcaa432aa0..b3af3fe993 100644
--- a/docs/zh/connectors/source/ObsFile.md
+++ b/docs/zh/connectors/source/ObsFile.md
@@ -108,7 +108,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
-当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true`
时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段：
- `source_uri`：源文件路径或 URI
- `document_id`：由 `source_uri` 派生的稳定文档标识符
- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
@@ -123,6 +123,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
PDF 使用与上文相同的文档元素 schema。
+对于 PDF 输入，启用 `pdf_rag_metadata_enabled` 即可追加上文所述的 RAG 元数据字段。
PDF 特有的解析行为如下：
diff --git a/docs/zh/connectors/source/OssFile.md
b/docs/zh/connectors/source/OssFile.md
index edb354530a..510458b0d7 100644
--- a/docs/zh/connectors/source/OssFile.md
+++ b/docs/zh/connectors/source/OssFile.md
@@ -266,7 +266,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
-当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true`
时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段：
- `source_uri`：源文件路径或 URI
- `document_id`：由 `source_uri` 派生的稳定文档标识符
- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
@@ -281,6 +281,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
PDF 使用与上文相同的文档元素 schema。
+对于 PDF 输入，启用 `pdf_rag_metadata_enabled` 即可追加上文所述的 RAG 元数据字段。
PDF 特有的解析行为如下：
diff --git a/docs/zh/connectors/source/OssJindoFile.md
b/docs/zh/connectors/source/OssJindoFile.md
index 648eb531a4..81fbcf24ac 100644
--- a/docs/zh/connectors/source/OssJindoFile.md
+++ b/docs/zh/connectors/source/OssJindoFile.md
@@ -117,7 +117,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
-当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true`
时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段：
- `source_uri`：源文件路径或 URI
- `document_id`：由 `source_uri` 派生的稳定文档标识符
- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
@@ -132,6 +132,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
PDF 使用与上文相同的文档元素 schema。
+对于 PDF 输入，启用 `pdf_rag_metadata_enabled` 即可追加上文所述的 RAG 元数据字段。
PDF 特有的解析行为如下：
diff --git a/docs/zh/connectors/source/S3File.md
b/docs/zh/connectors/source/S3File.md
index 10bc105ec8..a3f392dbeb 100644
--- a/docs/zh/connectors/source/S3File.md
+++ b/docs/zh/connectors/source/S3File.md
@@ -399,7 +399,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
-当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true`
时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段：
- `source_uri`：源文件路径或 URI
- `document_id`：由 `source_uri` 派生的稳定文档标识符
- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
@@ -414,6 +414,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
PDF 使用与上文相同的文档元素 schema。
+对于 PDF 输入，启用 `pdf_rag_metadata_enabled` 即可追加上文所述的 RAG 元数据字段。
PDF 特有的解析行为如下：
diff --git a/docs/zh/connectors/source/SftpFile.md
b/docs/zh/connectors/source/SftpFile.md
index 05edf11e0e..2e8c123a58 100644
--- a/docs/zh/connectors/source/SftpFile.md
+++ b/docs/zh/connectors/source/SftpFile.md
@@ -266,7 +266,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
- `parent_id`：父元素的 ID
- `child_ids`：子元素 ID 的逗号分隔列表
-当 `markdown_rag_metadata_enabled` 设置为 `true` 时，SeaTunnel 会在 `child_ids` 之后追加以下
RAG 元数据字段：
+当 `markdown_rag_metadata_enabled` 或 `pdf_rag_metadata_enabled` 设置为 `true`
时，SeaTunnel 会针对对应文件类型在 `child_ids` 之后追加以下 RAG 元数据字段：
- `source_uri`：源文件路径或 URI
- `document_id`：由 `source_uri` 派生的稳定文档标识符
- `chunk_id`：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
@@ -281,6 +281,7 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
PDF 使用与上文相同的文档元素 schema。
+对于 PDF 输入，启用 `pdf_rag_metadata_enabled` 即可追加上文所述的 RAG 元数据字段。
PDF 特有的解析行为如下：
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 08a0902f44..9a405df916 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
@@ -137,7 +137,9 @@ 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 || fileFormat ==
FileFormat.MARKDOWN) {
+ if (fileFormat == FileFormat.BINARY
+ || fileFormat == FileFormat.MARKDOWN
+ || fileFormat == FileFormat.PDF) {
return newCatalogTable(catalogTable,
getSchemaForEmptyFilePath(readonlyConfig));
}
return catalogTable;
@@ -159,6 +161,7 @@ public abstract class BaseFileSourceConfig implements
Serializable {
filePaths.get(0),
configSchema ?
catalogTable.getSeaTunnelRowType() : null));
case MARKDOWN:
+ case PDF:
return newCatalogTable(
catalogTable,
readStrategy.getSeaTunnelRowTypeInfo(filePaths.get(0)));
default:
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 64082dd489..ec8f2f75c7 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
@@ -116,6 +116,14 @@ public class FileBaseSourceOptions extends FileBaseOptions
{
"Whether to append RAG-oriented metadata columns
when reading markdown files. "
+ "Only valid when file_format_type is
markdown.");
+ public static final Option<Boolean> PDF_RAG_METADATA_ENABLED =
+ Options.key("pdf_rag_metadata_enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to append RAG-oriented metadata columns
when reading PDF files. "
+ + "Only valid when file_format_type is
pdf.");
+
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 63310aba43..e13d00cc18 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
@@ -130,7 +130,9 @@ public abstract class BaseFileSource
}
private CatalogTable buildCatalogTableForEmptyPath(String path, FileFormat
fileFormat) {
- if (fileFormat != FileFormat.BINARY && fileFormat !=
FileFormat.MARKDOWN) {
+ if (fileFormat != FileFormat.BINARY
+ && fileFormat != FileFormat.MARKDOWN
+ && fileFormat != FileFormat.PDF) {
// Preserve the legacy simple-text fallback for formats that still
infer schema from
// the first concrete file.
return CatalogTableUtil.buildSimpleTextTable();
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategy.java
index 479eaf4af8..8f081647cb 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategy.java
@@ -17,6 +17,7 @@
package org.apache.seatunnel.connectors.seatunnel.file.source.reader;
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
import org.apache.seatunnel.api.source.Collector;
@@ -26,6 +27,7 @@ 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.DocumentElement;
+import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
@@ -63,9 +65,14 @@ import lombok.extern.slf4j.Slf4j;
import java.awt.geom.Rectangle2D;
import java.io.IOException;
import java.io.InputStream;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
@@ -81,6 +88,41 @@ import java.util.stream.IntStream;
@Slf4j
public class PdfReadStrategy extends AbstractReadStrategy {
+ 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,
+ ArrayType.STRING_ARRAY_TYPE
+ };
+ 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 pdfRagMetadataEnabled =
+ FileBaseSourceOptions.PDF_RAG_METADATA_ENABLED.defaultValue();
+
@Override
public void read(String path, String tableId, Collector<SeaTunnelRow>
output)
throws IOException, FileConnectorException {
@@ -94,14 +136,23 @@ public class PdfReadStrategy extends AbstractReadStrategy {
try (PDDocument document = Loader.loadPDF(tempPdfPath.toFile())) {
List<DocumentElement> elements =
extractPdfDocumentElements(document);
+ String sourceUri = normalizeSourceUri(path);
+ String documentId = buildDocumentId(sourceUri);
log.info(
"PDF file '{}' processed successfully, generated {}
elements",
path,
elements.size());
+ int chunkIndex = 1;
for (DocumentElement element : elements) {
- output.collect(element.toSeaTunnelRow());
+ output.collect(
+ toSeaTunnelRow(
+ element,
+ sourceUri,
+ documentId,
+ chunkIndex++,
+ pdfRagMetadataEnabled));
}
}
} catch (Exception e) {
@@ -126,27 +177,21 @@ public class PdfReadStrategy 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 SeaTunnelDataType[] {
- BasicType.STRING_TYPE,
- BasicType.STRING_TYPE,
- BasicType.INT_TYPE,
- BasicType.STRING_TYPE,
- BasicType.INT_TYPE,
- BasicType.INT_TYPE,
- BasicType.STRING_TYPE,
- ArrayType.STRING_ARRAY_TYPE
- });
+ if (pdfRagMetadataEnabled) {
+ 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.PDF_RAG_METADATA_ENABLED.key())) {
+ pdfRagMetadataEnabled =
+
pluginConfig.getBoolean(FileBaseSourceOptions.PDF_RAG_METADATA_ENABLED.key());
+ }
}
/** Extract all document elements including headings, paragraphs, and
images */
@@ -924,6 +969,81 @@ public class PdfReadStrategy extends AbstractReadStrategy {
}
}
+ private SeaTunnelRow toSeaTunnelRow(
+ DocumentElement element,
+ String sourceUri,
+ String documentId,
+ int chunkIndex,
+ boolean appendRagMetadata) {
+ SeaTunnelRow row = element.toSeaTunnelRow();
+ if (!appendRagMetadata) {
+ return row;
+ }
+ return new SeaTunnelRow(
+ appendRagMetadata(row, sourceUri, documentId, chunkIndex,
element.getText()));
+ }
+
+ private Object[] appendRagMetadata(
+ SeaTunnelRow row, String sourceUri, String documentId, int
chunkIndex, String text) {
+ Object[] enriched = new Object[row.getArity() +
RAG_METADATA_FIELD_NAMES.length];
+ for (int i = 0; i < row.getArity(); i++) {
+ enriched[i] = row.getField(i);
+ }
+ String contentHash = sha256Hex(text == null ? "" : text);
+ String chunkId = "chunk_" + sha256Hex(documentId + ":" + chunkIndex +
":" + contentHash);
+ enriched[row.getArity()] = sourceUri;
+ enriched[row.getArity() + 1] = documentId;
+ enriched[row.getArity() + 2] = chunkId;
+ enriched[row.getArity() + 3] = chunkIndex;
+ enriched[row.getArity() + 4] = contentHash;
+ return enriched;
+ }
+
+ private static String buildDocumentId(String sourceUri) {
+ return "doc_" + sha256Hex(sourceUri);
+ }
+
+ private static String normalizeSourceUri(String sourceUri) {
+ 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;
+ }
+
/** Finds the paragraph boundary below an outline heading based on text
position changes. */
private static class ParagraphCoordinateExtractor {
private final PDDocument document;
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 e7ede52914..3b22784f0d 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
@@ -56,6 +56,19 @@ class BaseFileSourceTest {
private static final String[] MARKDOWN_RAG_METADATA_FIELD_NAMES = {
"source_uri", "document_id", "chunk_id", "chunk_index", "content_hash"
};
+ private static final String[] PDF_FIELD_NAMES = {
+ "element_id",
+ "element_type",
+ "heading_level",
+ "text",
+ "page_number",
+ "position_index",
+ "parent_id",
+ "child_ids"
+ };
+ private static final String[] PDF_RAG_METADATA_FIELD_NAMES = {
+ "source_uri", "document_id", "chunk_id", "chunk_index", "content_hash"
+ };
@Test
void testMarkdownSourceDiscoversSchemaFromEmptyDirectory() {
@@ -78,6 +91,27 @@ class BaseFileSourceTest {
catalogTable.getSeaTunnelRowType().getFieldNames());
}
+ @Test
+ void testPdfSourceDiscoversSchemaFromEmptyDirectory() {
+ BaseFileSource source = new TestFileSource(createPdfConfig(false));
+
+ CatalogTable catalogTable = source.getProducedCatalogTables().get(0);
+
+ Assertions.assertArrayEquals(
+ PDF_FIELD_NAMES,
catalogTable.getSeaTunnelRowType().getFieldNames());
+ }
+
+ @Test
+ void testPdfSourceDiscoversRagMetadataSchemaFromEmptyDirectory() {
+ BaseFileSource source = new TestFileSource(createPdfConfig(true));
+
+ CatalogTable catalogTable = source.getProducedCatalogTables().get(0);
+
+ Assertions.assertArrayEquals(
+ concat(PDF_FIELD_NAMES, PDF_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);
@@ -86,6 +120,14 @@ class BaseFileSourceTest {
return ReadonlyConfig.fromMap(map);
}
+ private ReadonlyConfig createPdfConfig(boolean ragMetadataEnabled) {
+ Map<String, Object> map = new HashMap<>();
+ map.put(FileBaseSourceOptions.FILE_PATH.key(), EMPTY_PATH);
+ map.put(FileBaseSourceOptions.FILE_FORMAT_TYPE.key(), "pdf");
+ map.put(FileBaseSourceOptions.PDF_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);
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategyTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategyTest.java
index bb7b7442be..ea0382f4f7 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategyTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategyTest.java
@@ -17,7 +17,10 @@
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.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.connectors.seatunnel.file.config.DocumentElement;
import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
@@ -57,6 +60,20 @@ import static
org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME
@Slf4j
public class PdfReadStrategyTest {
+ 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"
+ };
+
@Test
public void testReadPdfStrategy()
throws URISyntaxException, IOException, FileConnectorException {
@@ -120,6 +137,64 @@ public class PdfReadStrategyTest {
() -> strategy.read("/nonexistent/path/file.pdf", "",
collector));
}
+ @Test
+ public void testReadPdfWithRagMetadata()
+ throws URISyntaxException, IOException, FileConnectorException {
+ URL resource =
this.getClass().getResource("/pdf_read_strategy_test.pdf");
+ String path = Paths.get(resource.toURI()).toString();
+ PdfReadStrategy pdfReadStrategy = createPdfReadStrategy(true);
+ SeaTunnelRowType rowType =
pdfReadStrategy.getSeaTunnelRowTypeInfo(path);
+ TempCollector firstCollector = new TempCollector();
+ pdfReadStrategy.read(path, "", firstCollector);
+
+ Assertions.assertArrayEquals(
+ concat(DEFAULT_FIELD_NAMES, RAG_FIELD_NAMES),
rowType.getFieldNames());
+ Assertions.assertFalse(firstCollector.getRows().isEmpty());
+ 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());
+
+ PdfReadStrategy secondReadStrategy = createPdfReadStrategy(true);
+ 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 testReadPdfWithRagMetadataNormalizesFileUri()
+ throws IOException, FileConnectorException {
+ Path tempPdf = createSimplePdf("Title in file uri pdf");
+ try {
+ PdfReadStrategy pdfReadStrategy = createPdfReadStrategy(true);
+ TempCollector tempCollector = new TempCollector();
+ pdfReadStrategy.read(tempPdf.toUri().toString(), "",
tempCollector);
+
+ PdfReadStrategy expectedReadStrategy = createPdfReadStrategy(true);
+ TempCollector expectedCollector = new TempCollector();
+ expectedReadStrategy.read(tempPdf.toString(), "",
expectedCollector);
+
+ Assertions.assertEquals(tempPdf.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));
+ }
+ } finally {
+ Files.deleteIfExists(tempPdf);
+ }
+ }
+
@Test
public void testChildIdsIsArrayType() throws URISyntaxException,
IOException {
URL resource =
this.getClass().getResource("/pdf_read_strategy_test.pdf");
@@ -298,10 +373,44 @@ public class PdfReadStrategyTest {
return tempFile;
}
+ private Path createSimplePdf(String text) throws IOException {
+ Path tempFile = Files.createTempFile("pdf_rag_metadata_test_", ".pdf");
+ try (PDDocument document = new PDDocument()) {
+ PDPage page = new PDPage(PDRectangle.A4);
+ document.addPage(page);
+ try (PDPageContentStream contentStream = new
PDPageContentStream(document, page)) {
+ contentStream.beginText();
+ contentStream.setFont(new
PDType1Font(Standard14Fonts.FontName.HELVETICA), 12);
+ contentStream.newLineAtOffset(50, 700);
+ contentStream.showText(text);
+ contentStream.endText();
+ }
+ document.save(tempFile.toFile());
+ }
+ return tempFile;
+ }
+
private List<SeaTunnelRow> getHeadingElements(List<SeaTunnelRow> rows) {
return rows.stream().filter(row -> row.getField(2) !=
null).collect(Collectors.toList());
}
+ private PdfReadStrategy createPdfReadStrategy(boolean ragMetadataEnabled) {
+ PdfReadStrategy pdfReadStrategy = new PdfReadStrategy();
+ pdfReadStrategy.init(new LocalConf(FS_DEFAULT_NAME_DEFAULT));
+ if (ragMetadataEnabled) {
+ pdfReadStrategy.setPluginConfig(
+ ConfigFactory.parseString("pdf_rag_metadata_enabled =
true"));
+ }
+ return pdfReadStrategy;
+ }
+
+ 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";
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 50dfd7e91e..e3efd70652 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
@@ -99,6 +99,10 @@ public class CosFileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.FILE_FORMAT_TYPE,
FileFormat.MARKDOWN,
FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.PDF,
+ FileBaseSourceOptions.PDF_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
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 aa17762e12..209e9b3eb4 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
@@ -105,6 +105,10 @@ public class FtpFileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.FILE_FORMAT_TYPE,
FileFormat.MARKDOWN,
FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.PDF,
+ FileBaseSourceOptions.PDF_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 b99417be7a..cad52a6931 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
@@ -114,6 +114,10 @@ public class HdfsFileSourceFactory implements
TableSourceFactory {
HdfsFileSourceOptions.FILE_FORMAT_TYPE,
FileFormat.MARKDOWN,
HdfsFileSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
+ .conditional(
+ HdfsFileSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.PDF,
+ HdfsFileSourceOptions.PDF_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 75939c1ffb..32bb57f6db 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
@@ -94,6 +94,10 @@ public class OssFileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.FILE_FORMAT_TYPE,
FileFormat.MARKDOWN,
FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.PDF,
+ FileBaseSourceOptions.PDF_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
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 0d340ec5c7..6c50e98b87 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
@@ -114,6 +114,10 @@ public class LocalFileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.FILE_FORMAT_TYPE,
FileFormat.MARKDOWN,
FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.PDF,
+ FileBaseSourceOptions.PDF_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-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 44e079c6c1..0fdf4b4eaf 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
@@ -83,6 +83,10 @@ public class ObsFileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.FILE_FORMAT_TYPE,
FileFormat.MARKDOWN,
FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.PDF,
+ FileBaseSourceOptions.PDF_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
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 8fc59b4db2..bfaa6a46da 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
@@ -103,6 +103,10 @@ public class OssFileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.FILE_FORMAT_TYPE,
FileFormat.MARKDOWN,
FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.PDF,
+ FileBaseSourceOptions.PDF_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 7a8bc1a679..e0d5e1cf47 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
@@ -119,6 +119,10 @@ public class S3FileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.FILE_FORMAT_TYPE,
FileFormat.MARKDOWN,
FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.PDF,
+ FileBaseSourceOptions.PDF_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 b1fd263c4a..f860d3bed3 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
@@ -95,6 +95,10 @@ public class SftpFileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.FILE_FORMAT_TYPE,
FileFormat.MARKDOWN,
FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED)
+ .conditional(
+ FileBaseSourceOptions.FILE_FORMAT_TYPE,
+ FileFormat.PDF,
+ FileBaseSourceOptions.PDF_RAG_METADATA_ENABLED)
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(ConnectorCommonOptions.METALAKE_TYPE)
