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 befd31c453 [Feature][API] Define Knowledge Sync metadata field
contract (#10990)
befd31c453 is described below
commit befd31c453e407f5194af76be55d2bef18903a45
Author: yzeng1618 <[email protected]>
AuthorDate: Wed Aug 5 14:34:30 2026 +0800
[Feature][API] Define Knowledge Sync metadata field contract (#10990)
Co-authored-by: zengyi <[email protected]>
---
docs/en/transforms/metadata.md | 45 ++++++++
docs/zh/transforms/metadata.md | 45 ++++++++
.../api/table/type/KnowledgeSyncMetadataField.java | 100 ++++++++++++++++++
.../seatunnel/api/table/type/MetadataUtil.java | 12 +++
.../table/type/KnowledgeSyncMetadataFieldTest.java | 117 +++++++++++++++++++++
.../transform/metadata/MetadataTransform.java | 81 +++++++-------
.../transform/metadata/MetadataTransformTest.java | 86 +++++++++++++++
7 files changed, 445 insertions(+), 41 deletions(-)
diff --git a/docs/en/transforms/metadata.md b/docs/en/transforms/metadata.md
index 5a311465b0..b0e3fa45f1 100644
--- a/docs/en/transforms/metadata.md
+++ b/docs/en/transforms/metadata.md
@@ -33,12 +33,35 @@ The Metadata transform plugin is used to extract metadata
information from data
| Gtid | string | Global Transaction ID (`server_uuid:transaction_id`).
`null` when GTID is disabled or for snapshot rows. | MySQL-CDC only |
| Partition | string | Partition information of the data, multiple
partition fields separated by commas | Connectors supporting partitions |
+## Knowledge Sync Metadata Fields
+
+Knowledge Sync pipelines can use the following logical metadata keys to carry
document and chunk identity. These keys become physical fields only after they
are explicitly projected by the `Metadata` transform.
+
+The `Metadata` transform does not generate Knowledge Sync metadata by itself.
The upstream source or transform must declare these fields in
`CatalogTable.metadataSchema` and write the corresponding values into
`SeaTunnelRow.options`.
+
+| Metadata Key | Canonical Physical Field | Output Type | Description |
+|:---:|:---:|:---:|:---|
+| DocumentId | `document_id` | string | Non-null stable document identity for
every document lifecycle event. |
+| DocumentHash | `document_hash` | string | Stable document version or content
hash. |
+| SourceUri | `source_uri` | string | Credential-free stable source URI or
path. |
+| SourceVersion | `source_version` | string | Source-side version, etag,
revision, or similar marker. |
+| SourceModifiedAt | `source_modified_at` | long | Source modified time in
epoch milliseconds. |
+| MimeType | `mime_type` | string | Source MIME type. |
+| Deleted | `deleted` | boolean | Non-null lifecycle marker: `false` for
normal rows and `true` for document tombstones. |
+| ChunkId | `chunk_id` | string | Stable chunk identity. Required for normal
chunk rows and nullable for document tombstones. |
+| ChunkHash | `chunk_hash` | string | Stable chunk content hash. Required for
normal chunk rows and nullable for document tombstones. |
+| ChunkIndex | `chunk_index` | int | Zero-based chunk index. Required for
normal chunk rows and nullable for document tombstones. |
+
### Important Notes
1. **Metadata field names are case-sensitive**: Configuration must strictly
follow the Key names in the table above (e.g., `Database`, `Table`, `RowKind`,
etc.)
2. **Time fields**: `Delay` and `SourceTimestamp` are only available for CDC
connectors. `EventTime` is also provided by the Kafka source via
`ConsumerRecord.timestamp` when available.
3. **Kafka event time**: The Kafka source writes `ConsumerRecord.timestamp`
(milliseconds) into `EventTime` when it is non-negative, so you can surface it
with the `Metadata` transform.
4. **Binlog/GTID fields**: `BinlogFile`, `BinlogPos`, `BinlogRow`, and `Gtid`
are MySQL-CDC specific. For `startup.mode = initial`, snapshot rows return
`null` for all four fields.
+5. **Knowledge Sync projection is explicit**: Knowledge Sync metadata fields
are projected only when they are configured in `metadata_fields`, declared in
the input table metadata schema, and present in row options. This transform
reads logical row metadata; it does not read existing physical columns with the
same names.
+6. **Markdown RAG compatibility**: Existing Markdown RAG output currently
exposes physical fields such as `source_uri`, `document_id`, `chunk_id`,
`chunk_index`, and `content_hash`. This transform does not migrate those
physical fields into logical Knowledge Sync metadata, and this change does not
rename them.
+7. **Source URI security**: Producers must remove URI user info, access
tokens, signatures, and other transient authentication material before writing
`SourceUri` into row options. Non-sensitive query parameters that are part of
the stable resource identity may be retained.
+8. **Knowledge Sync nullability**: `DocumentId` identifies every document
lifecycle event. When `Deleted` is declared, producers must write `false` for
normal rows and `true` for document tombstones rather than `null`. Normal chunk
rows require `ChunkId`, `ChunkHash`, and `ChunkIndex`; compact document
tombstones may leave those chunk fields `null`.
## Options
@@ -76,6 +99,28 @@ metadata_fields {
- The right side is a custom output field name, which cannot duplicate
existing field names
- You can select only the metadata fields you need, not all of them must be
configured
+### Knowledge Sync Projection Example
+
+Project Knowledge Sync logical metadata keys into canonical physical columns.
The upstream producer must already provide the metadata values through row
options and declare them in the table metadata schema.
+
+```hocon
+transform {
+ Metadata {
+ plugin_input = "knowledge_chunks"
+ plugin_output = "knowledge_chunks_with_meta"
+ metadata_fields = {
+ DocumentId = "document_id"
+ DocumentHash = "document_hash"
+ ChunkId = "chunk_id"
+ ChunkHash = "chunk_hash"
+ ChunkIndex = "chunk_index"
+ }
+ }
+}
+```
+
+After this transform, downstream components can read `document_id`,
`chunk_id`, and `chunk_hash` as regular physical fields in the input schema.
+
## Complete Examples
### Example 1: MySQL CDC Data Synchronization, Extracting All Metadata
diff --git a/docs/zh/transforms/metadata.md b/docs/zh/transforms/metadata.md
index d51c3264a7..b7a25dcece 100644
--- a/docs/zh/transforms/metadata.md
+++ b/docs/zh/transforms/metadata.md
@@ -33,12 +33,35 @@ Metadata 转换插件用于将数据行中的元数据信息提取为普通字
| Gtid | string | 全局事务 ID（格式：`server_uuid:transaction_id`）。GTID
未启用或快照行时返回 `null`。 | 仅 MySQL-CDC |
| Partition | string | 数据所属的分区信息，多个分区字段使用逗号分隔 | 支持分区的连接器 |
+## Knowledge Sync 元数据字段
+
+Knowledge Sync 流程可以使用下面的逻辑元数据 Key 携带文档和 chunk 身份信息。这些 Key 只有通过 `Metadata`
转换显式投影后，才会成为真实的物理列。
+
+`Metadata` 转换本身不会生成 Knowledge Sync 元数据。上游 Source 或 Transform 必须先在
`CatalogTable.metadataSchema` 中声明这些元数据字段，并将对应值写入 `SeaTunnelRow.options`。
+
+| 元数据 Key | 标准物理字段名 | 输出类型 | 说明 |
+|:---:|:---:|:---:|:---|
+| DocumentId | `document_id` | string | 每个文档生命周期事件都必须包含的非空稳定文档标识。 |
+| DocumentHash | `document_hash` | string | 稳定的文档版本或内容哈希。 |
+| SourceUri | `source_uri` | string | 不包含凭证的稳定来源 URI 或路径。 |
+| SourceVersion | `source_version` | string | 来源侧版本、etag、revision 等版本标记。 |
+| SourceModifiedAt | `source_modified_at` | long | 来源修改时间，epoch 毫秒。 |
+| MimeType | `mime_type` | string | 来源 MIME 类型。 |
+| Deleted | `deleted` | boolean | 非空生命周期标记：普通行为 `false`，文档 tombstone 为 `true`。
|
+| ChunkId | `chunk_id` | string | 稳定的 chunk 标识。普通 chunk 行必填，文档 tombstone 可为
`null`。 |
+| ChunkHash | `chunk_hash` | string | 稳定的 chunk 内容哈希。普通 chunk 行必填，文档 tombstone
可为 `null`。 |
+| ChunkIndex | `chunk_index` | int | 文档内从 0 开始的 chunk 序号。普通 chunk 行必填，文档
tombstone 可为 `null`。 |
+
### 重要说明
1. **元数据字段区分大小写**：配置时必须严格按照上表中的 Key 名称（如 `Database`、`Table`、`RowKind` 等）。
2. **时间相关字段**：`Delay` 和 `SourceTimestamp` 仅在 CDC 连接器有效。`EventTime` 也会在 Kafka
源中使用 `ConsumerRecord.timestamp`（毫秒，非负时）写入。
3. **Kafka 事件时间**：Kafka 源会在 `ConsumerRecord.timestamp` 非负时写入 `EventTime`，可通过
Metadata 转换将其暴露为普通字段。
4. **Binlog/GTID 字段**：`BinlogFile`、`BinlogPos`、`BinlogRow`、`Gtid` 仅适用于
MySQL-CDC。使用 `startup.mode = initial` 时，快照行的这四个字段均为 `null`。
+5. **Knowledge Sync 投影需要显式配置**：Knowledge Sync 元数据字段只有在 `metadata_fields`
中配置、输入表 metadata schema 中声明了对应 Key，并且行 options
中存在对应值时才会被投影。该转换读取的是逻辑行元数据，不会读取同名的已有物理列。
+6. **兼容 Markdown RAG 物理字段**：现有 Markdown RAG 输出当前以物理字段暴露
`source_uri`、`document_id`、`chunk_id`、`chunk_index` 和
`content_hash`。该转换不会把这些物理字段迁移为逻辑 Knowledge Sync 元数据，本次变更也不会重命名这些字段。
+7. **来源 URI 安全**：producer 在将 `SourceUri` 写入行 options 前，必须移除 URI
userinfo、访问令牌、签名以及其他临时鉴权信息。属于稳定资源标识且不敏感的查询参数可以保留。
+8. **Knowledge Sync 可空语义**：`DocumentId` 用于标识每个文档生命周期事件。声明 `Deleted` 后，producer
必须为普通行写入 `false`，为文档 tombstone 写入 `true`，不能使用 `null`。普通 chunk 行必须包含
`ChunkId`、`ChunkHash` 和 `ChunkIndex`；紧凑的文档 tombstone 可以将这些 chunk 字段留为 `null`。
## 配置选项
@@ -76,6 +99,28 @@ metadata_fields {
- 右侧是自定义的输出字段名，不能与原有字段重名
- 可以只选择需要的元数据字段，不必全部配置
+### Knowledge Sync 投影示例
+
+将 Knowledge Sync 逻辑元数据 Key 投影为标准物理列。上游 producer 必须已经通过行 options 提供这些元数据值，并在表
metadata schema 中声明这些字段。
+
+```hocon
+transform {
+ Metadata {
+ plugin_input = "knowledge_chunks"
+ plugin_output = "knowledge_chunks_with_meta"
+ metadata_fields = {
+ DocumentId = "document_id"
+ DocumentHash = "document_hash"
+ ChunkId = "chunk_id"
+ ChunkHash = "chunk_hash"
+ ChunkIndex = "chunk_index"
+ }
+ }
+}
+```
+
+完成该转换后，下游组件可以把 `document_id`、`chunk_id`、`chunk_hash` 当作输入 schema 中的普通物理字段读取。
+
## 完整示例
### 示例 1：MySQL CDC 数据同步，提取所有元数据
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/KnowledgeSyncMetadataField.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/KnowledgeSyncMetadataField.java
new file mode 100644
index 0000000000..5ba464198b
--- /dev/null
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/KnowledgeSyncMetadataField.java
@@ -0,0 +1,100 @@
+/*
+ * 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.api.table.type;
+
+import org.apache.seatunnel.api.table.catalog.MetadataColumn;
+
+import java.util.Collections;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Standard metadata contract for Knowledge Sync document and chunk identity
fields.
+ *
+ * <p>{@link #DOCUMENT_ID} identifies every document lifecycle event. {@link
#DELETED} must be
+ * explicitly {@code false} for normal rows or {@code true} for document
tombstones. Chunk fields
+ * remain nullable because a document tombstone does not represent an
individual chunk.
+ */
+public enum KnowledgeSyncMetadataField {
+ DOCUMENT_ID("DocumentId", "document_id", BasicType.STRING_TYPE, false),
+ DOCUMENT_HASH("DocumentHash", "document_hash", BasicType.STRING_TYPE,
true),
+ /**
+ * Credential-free stable source URI or path.
+ *
+ * <p>Producers must remove URI user info, access tokens, signatures, and
other transient
+ * authentication material before storing this value in row options.
+ */
+ SOURCE_URI("SourceUri", "source_uri", BasicType.STRING_TYPE, true),
+ SOURCE_VERSION("SourceVersion", "source_version", BasicType.STRING_TYPE,
true),
+ SOURCE_MODIFIED_AT("SourceModifiedAt", "source_modified_at",
BasicType.LONG_TYPE, true),
+ MIME_TYPE("MimeType", "mime_type", BasicType.STRING_TYPE, true),
+ DELETED("Deleted", "deleted", BasicType.BOOLEAN_TYPE, false),
+ CHUNK_ID("ChunkId", "chunk_id", BasicType.STRING_TYPE, true),
+ CHUNK_HASH("ChunkHash", "chunk_hash", BasicType.STRING_TYPE, true),
+ CHUNK_INDEX("ChunkIndex", "chunk_index", BasicType.INT_TYPE, true);
+
+ private static final Set<String> FIELD_NAMES =
+ Collections.unmodifiableSet(
+ Stream.of(values())
+ .map(KnowledgeSyncMetadataField::getName)
+ .collect(Collectors.toSet()));
+
+ private final String name;
+ private final String physicalName;
+ private final SeaTunnelDataType<?> dataType;
+ private final boolean nullable;
+
+ KnowledgeSyncMetadataField(
+ String name, String physicalName, SeaTunnelDataType<?> dataType,
boolean nullable) {
+ this.name = name;
+ this.physicalName = physicalName;
+ this.dataType = dataType;
+ this.nullable = nullable;
+ }
+
+ /** Returns the logical metadata key stored in row options and metadata
schemas. */
+ public String getName() {
+ return name;
+ }
+
+ /** Returns the canonical physical field name used when projecting this
metadata. */
+ public String getPhysicalName() {
+ return physicalName;
+ }
+
+ /** Returns the field data type defined by the Knowledge Sync contract. */
+ public SeaTunnelDataType<?> getDataType() {
+ return dataType;
+ }
+
+ /** Returns whether the metadata column may contain {@code null}. */
+ public boolean isNullable() {
+ return nullable;
+ }
+
+ /** Creates the metadata column declaration for a Knowledge Sync producer
schema. */
+ public MetadataColumn toMetadataColumn() {
+ return MetadataColumn.of(name, dataType, (Long) null, nullable, null,
null);
+ }
+
+ /** Returns whether the name is a logical Knowledge Sync metadata key. */
+ public static boolean isKnowledgeSyncMetadataField(String name) {
+ return FIELD_NAMES.contains(name);
+ }
+}
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/MetadataUtil.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/MetadataUtil.java
index 9e80335a49..9a06e58586 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/MetadataUtil.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/MetadataUtil.java
@@ -44,6 +44,9 @@ public class MetadataUtil {
.filter(CommonOptions::isSupportMetadataTrans)
.map(CommonOptions::getName)
.forEach(METADATA_FIELDS::add);
+ Stream.of(KnowledgeSyncMetadataField.values())
+ .map(KnowledgeSyncMetadataField::getName)
+ .forEach(METADATA_FIELDS::add);
}
public static void setDelay(SeaTunnelRow row, Long delay) {
@@ -116,6 +119,15 @@ public class MetadataUtil {
return (String[]) row.getOptions().get(PARTITION.getName());
}
+ /**
+ * Returns whether the logical metadata key is supported for explicit
metadata projection.
+ *
+ * <p>This method does not inspect or classify physical table columns that
happen to use the
+ * same name.
+ *
+ * @param fieldName logical metadata key
+ * @return whether the key is supported for metadata projection
+ */
public static boolean isMetadataField(String fieldName) {
return METADATA_FIELDS.contains(fieldName);
}
diff --git
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/type/KnowledgeSyncMetadataFieldTest.java
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/type/KnowledgeSyncMetadataFieldTest.java
new file mode 100644
index 0000000000..dee28bde17
--- /dev/null
+++
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/type/KnowledgeSyncMetadataFieldTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.api.table.type;
+
+import org.apache.seatunnel.api.table.catalog.MetadataColumn;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class KnowledgeSyncMetadataFieldTest {
+
+ @Test
+ void shouldRegisterKnowledgeSyncMetadataFields() {
+ Assertions.assertTrue(MetadataUtil.isMetadataField("DocumentId"));
+ Assertions.assertTrue(MetadataUtil.isMetadataField("DocumentHash"));
+ Assertions.assertTrue(MetadataUtil.isMetadataField("SourceUri"));
+ Assertions.assertTrue(MetadataUtil.isMetadataField("SourceVersion"));
+
Assertions.assertTrue(MetadataUtil.isMetadataField("SourceModifiedAt"));
+ Assertions.assertTrue(MetadataUtil.isMetadataField("MimeType"));
+ Assertions.assertTrue(MetadataUtil.isMetadataField("Deleted"));
+ Assertions.assertTrue(MetadataUtil.isMetadataField("ChunkId"));
+ Assertions.assertTrue(MetadataUtil.isMetadataField("ChunkHash"));
+ Assertions.assertTrue(MetadataUtil.isMetadataField("ChunkIndex"));
+ }
+
+ @Test
+ void shouldRecognizeKnowledgeSyncMetadataFieldNames() {
+ for (KnowledgeSyncMetadataField field :
KnowledgeSyncMetadataField.values()) {
+ Assertions.assertTrue(
+
KnowledgeSyncMetadataField.isKnowledgeSyncMetadataField(field.getName()));
+ }
+ Assertions.assertFalse(
+
KnowledgeSyncMetadataField.isKnowledgeSyncMetadataField("UnknownMetadata"));
+
Assertions.assertFalse(KnowledgeSyncMetadataField.isKnowledgeSyncMetadataField(null));
+ }
+
+ @Test
+ void shouldExposeCanonicalPhysicalNamesAndTypes() {
+ Assertions.assertEquals(
+ "document_id",
KnowledgeSyncMetadataField.DOCUMENT_ID.getPhysicalName());
+ Assertions.assertEquals(
+ "document_hash",
KnowledgeSyncMetadataField.DOCUMENT_HASH.getPhysicalName());
+ Assertions.assertEquals(
+ "source_uri",
KnowledgeSyncMetadataField.SOURCE_URI.getPhysicalName());
+ Assertions.assertEquals(
+ "source_version",
KnowledgeSyncMetadataField.SOURCE_VERSION.getPhysicalName());
+ Assertions.assertEquals(
+ "source_modified_at",
+
KnowledgeSyncMetadataField.SOURCE_MODIFIED_AT.getPhysicalName());
+ Assertions.assertEquals(
+ "mime_type",
KnowledgeSyncMetadataField.MIME_TYPE.getPhysicalName());
+ Assertions.assertEquals("deleted",
KnowledgeSyncMetadataField.DELETED.getPhysicalName());
+ Assertions.assertEquals("chunk_id",
KnowledgeSyncMetadataField.CHUNK_ID.getPhysicalName());
+ Assertions.assertEquals(
+ "chunk_hash",
KnowledgeSyncMetadataField.CHUNK_HASH.getPhysicalName());
+ Assertions.assertEquals(
+ "chunk_index",
KnowledgeSyncMetadataField.CHUNK_INDEX.getPhysicalName());
+
+ Assertions.assertEquals(
+ BasicType.STRING_TYPE,
KnowledgeSyncMetadataField.DOCUMENT_ID.getDataType());
+ Assertions.assertEquals(
+ BasicType.STRING_TYPE,
KnowledgeSyncMetadataField.DOCUMENT_HASH.getDataType());
+ Assertions.assertEquals(
+ BasicType.STRING_TYPE,
KnowledgeSyncMetadataField.SOURCE_URI.getDataType());
+ Assertions.assertEquals(
+ BasicType.STRING_TYPE,
KnowledgeSyncMetadataField.SOURCE_VERSION.getDataType());
+ Assertions.assertEquals(
+ BasicType.LONG_TYPE,
KnowledgeSyncMetadataField.SOURCE_MODIFIED_AT.getDataType());
+ Assertions.assertEquals(
+ BasicType.STRING_TYPE,
KnowledgeSyncMetadataField.MIME_TYPE.getDataType());
+ Assertions.assertEquals(
+ BasicType.BOOLEAN_TYPE,
KnowledgeSyncMetadataField.DELETED.getDataType());
+ Assertions.assertEquals(
+ BasicType.STRING_TYPE,
KnowledgeSyncMetadataField.CHUNK_ID.getDataType());
+ Assertions.assertEquals(
+ BasicType.STRING_TYPE,
KnowledgeSyncMetadataField.CHUNK_HASH.getDataType());
+ Assertions.assertEquals(
+ BasicType.INT_TYPE,
KnowledgeSyncMetadataField.CHUNK_INDEX.getDataType());
+ }
+
+ @Test
+ void shouldCreateMetadataColumnsForProducerSchemas() {
+ MetadataColumn column =
KnowledgeSyncMetadataField.DOCUMENT_ID.toMetadataColumn();
+
+ Assertions.assertEquals("DocumentId", column.getName());
+ Assertions.assertEquals(BasicType.STRING_TYPE, column.getDataType());
+ Assertions.assertFalse(column.isNullable());
+ Assertions.assertFalse(column.isPhysical());
+ }
+
+ @Test
+ void shouldDefineLifecycleNullability() {
+ Assertions.assertFalse(
+
KnowledgeSyncMetadataField.DOCUMENT_ID.toMetadataColumn().isNullable());
+
Assertions.assertFalse(KnowledgeSyncMetadataField.DELETED.toMetadataColumn().isNullable());
+
Assertions.assertTrue(KnowledgeSyncMetadataField.CHUNK_ID.toMetadataColumn().isNullable());
+ Assertions.assertTrue(
+
KnowledgeSyncMetadataField.CHUNK_HASH.toMetadataColumn().isNullable());
+ Assertions.assertTrue(
+
KnowledgeSyncMetadataField.CHUNK_INDEX.toMetadataColumn().isNullable());
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/metadata/MetadataTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/metadata/MetadataTransform.java
index 6db61c900d..bb7b8cb411 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/metadata/MetadataTransform.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/metadata/MetadataTransform.java
@@ -83,21 +83,7 @@ public class MetadataTransform extends
MultipleFieldOutputTransform {
for (Map.Entry<String, String> mapping :
metadataFieldMapping.entrySet()) {
String metadataFieldName = mapping.getKey();
int i = fieldNames.indexOf(metadataFieldName);
- Object fieldValue;
- switch (CommonOptions.fromName(metadataFieldName)) {
- case DATABASE:
- fieldValue = MetadataUtil.getDatabase(inputRow);
- break;
- case TABLE:
- fieldValue = MetadataUtil.getTable(inputRow);
- break;
- case ROW_KIND:
- fieldValue = MetadataUtil.getRowKind(inputRow);
- break;
- default:
- fieldValue = inputRow.getOptions().get(metadataFieldName);
- }
- value[i] = fieldValue;
+ value[i] = getMetadataFieldValue(metadataFieldName, inputRow);
}
return value;
}
@@ -111,38 +97,51 @@ public class MetadataTransform extends
MultipleFieldOutputTransform {
int i = fieldNames.indexOf(metadataFieldName);
Column column;
- switch (CommonOptions.fromName(metadataFieldName)) {
- case DATABASE:
- case TABLE:
- case ROW_KIND:
- column =
- PhysicalColumn.of(
- mappingFieldName,
- BasicType.STRING_TYPE,
- (Long) null,
- null,
- true,
- null,
- null);
- break;
- default:
- if (metadataSchema.contains(metadataFieldName)) {
- column =
- ((MetadataColumn)
- metadataSchema
-
.getColumn(metadataFieldName)
-
.rename(mappingFieldName))
- .toPhysicalColumn();
- } else {
- throw
TransformCommonError.cannotFindMetadataFieldError(
- getPluginName(), mappingFieldName);
- }
+ if (isComputedCommonMetadataField(metadataFieldName)) {
+ column =
+ PhysicalColumn.of(
+ mappingFieldName,
+ BasicType.STRING_TYPE,
+ (Long) null,
+ null,
+ true,
+ null,
+ null);
+ } else if (metadataSchema != null &&
metadataSchema.contains(metadataFieldName)) {
+ column =
+ ((MetadataColumn)
+ metadataSchema
+ .getColumn(metadataFieldName)
+ .rename(mappingFieldName))
+ .toPhysicalColumn();
+ } else {
+ throw TransformCommonError.cannotFindMetadataFieldError(
+ getPluginName(), metadataFieldName);
}
columns[i] = column;
}
return columns;
}
+ private Object getMetadataFieldValue(String metadataFieldName,
SeaTunnelRowAccessor inputRow) {
+ if (CommonOptions.DATABASE.getName().equals(metadataFieldName)) {
+ return MetadataUtil.getDatabase(inputRow);
+ }
+ if (CommonOptions.TABLE.getName().equals(metadataFieldName)) {
+ return MetadataUtil.getTable(inputRow);
+ }
+ if (CommonOptions.ROW_KIND.getName().equals(metadataFieldName)) {
+ return MetadataUtil.getRowKind(inputRow);
+ }
+ return inputRow.getOptions().get(metadataFieldName);
+ }
+
+ private boolean isComputedCommonMetadataField(String metadataFieldName) {
+ return CommonOptions.DATABASE.getName().equals(metadataFieldName)
+ || CommonOptions.TABLE.getName().equals(metadataFieldName)
+ || CommonOptions.ROW_KIND.getName().equals(metadataFieldName);
+ }
+
@VisibleForTesting
public void initRowContainerGenerator() {
transformTableSchema();
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/metadata/MetadataTransformTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/metadata/MetadataTransformTest.java
index 63277273f1..9641b14c91 100644
---
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/metadata/MetadataTransformTest.java
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/metadata/MetadataTransformTest.java
@@ -29,9 +29,11 @@ import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.type.ArrayType;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.CommonOptions;
+import org.apache.seatunnel.api.table.type.KnowledgeSyncMetadataField;
import org.apache.seatunnel.api.table.type.MetadataUtil;
import org.apache.seatunnel.api.table.type.RowKind;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.transform.exception.TransformException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
@@ -192,4 +194,88 @@ public class MetadataTransformTest {
Assertions.assertEquals(eventTime, outputRow.getField(9));
Assertions.assertEquals(150L, outputRow.getField(10));
}
+
+ @Test
+ void shouldProjectKnowledgeSyncMetadataFromRowOptions() {
+ Map<String, String> metadataMapping = new LinkedHashMap<>();
+ metadataMapping.put(
+ KnowledgeSyncMetadataField.DOCUMENT_ID.getName(),
+ KnowledgeSyncMetadataField.DOCUMENT_ID.getPhysicalName());
+ metadataMapping.put(
+ KnowledgeSyncMetadataField.CHUNK_HASH.getName(),
+ KnowledgeSyncMetadataField.CHUNK_HASH.getPhysicalName());
+ Map<String, Object> config = new HashMap<>();
+ config.put("metadata_fields", metadataMapping);
+ MetadataTransform transform =
+ new MetadataTransform(
+ ReadonlyConfig.fromMap(config),
knowledgeSyncCatalogTable(true));
+ transform.initRowContainerGenerator();
+
+ Column[] columns = transform.getOutputColumns();
+ Assertions.assertEquals("document_id", columns[0].getName());
+ Assertions.assertEquals("chunk_hash", columns[1].getName());
+ Assertions.assertEquals(BasicType.STRING_TYPE,
columns[0].getDataType());
+ Assertions.assertEquals(BasicType.STRING_TYPE,
columns[1].getDataType());
+ Assertions.assertInstanceOf(PhysicalColumn.class, columns[0]);
+ Assertions.assertInstanceOf(PhysicalColumn.class, columns[1]);
+ Assertions.assertFalse(columns[0].isNullable());
+ Assertions.assertTrue(columns[1].isNullable());
+
+ SeaTunnelRow input = new SeaTunnelRow(new Object[] {"chunk text"});
+
input.getOptions().put(KnowledgeSyncMetadataField.DOCUMENT_ID.getName(),
"doc_faq");
+
input.getOptions().put(KnowledgeSyncMetadataField.CHUNK_HASH.getName(),
"hash_chunk_0");
+
+ SeaTunnelRow output = transform.map(input);
+ Assertions.assertEquals(3, output.getArity());
+ Assertions.assertEquals("chunk text", output.getField(0));
+ Assertions.assertEquals("doc_faq", output.getField(1));
+ Assertions.assertEquals("hash_chunk_0", output.getField(2));
+ Assertions.assertArrayEquals(
+ new String[] {"text", "document_id", "chunk_hash"},
+
transform.getProducedCatalogTable().getTableSchema().getFieldNames());
+ }
+
+ @Test
+ void shouldRejectKnowledgeSyncMetadataWhenSchemaDoesNotDeclareIt() {
+ Map<String, String> metadataMapping = new LinkedHashMap<>();
+ metadataMapping.put(
+ KnowledgeSyncMetadataField.DOCUMENT_ID.getName(),
+ KnowledgeSyncMetadataField.DOCUMENT_ID.getPhysicalName());
+ Map<String, Object> config = new HashMap<>();
+ config.put("metadata_fields", metadataMapping);
+ MetadataTransform transform =
+ new MetadataTransform(
+ ReadonlyConfig.fromMap(config),
knowledgeSyncCatalogTable(false));
+
+ TransformException exception =
+ Assertions.assertThrows(
+ TransformException.class,
transform::initRowContainerGenerator);
+ Assertions.assertTrue(
+
exception.getMessage().contains(KnowledgeSyncMetadataField.DOCUMENT_ID.getName()));
+ }
+
+ private static CatalogTable knowledgeSyncCatalogTable(boolean
includeKnowledgeSyncMetadata) {
+ List<Column> metadata = new ArrayList<>();
+ if (includeKnowledgeSyncMetadata) {
+
metadata.add(KnowledgeSyncMetadataField.DOCUMENT_ID.toMetadataColumn());
+
metadata.add(KnowledgeSyncMetadataField.CHUNK_HASH.toMetadataColumn());
+ }
+ return CatalogTable.of(
+ TableIdentifier.of("catalog", TablePath.DEFAULT),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "text",
+ BasicType.STRING_TYPE,
+ (Long) null,
+ true,
+ null,
+ null))
+ .build(),
+ new HashMap<>(),
+ new ArrayList<>(),
+ "comment",
+ "test",
+ MetadataSchema.builder().columns(metadata).build());
+ }
}
