[pull] dev from apache:dev

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

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 数据同步，提取所有元数据

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);
+    }
+}

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 static String[] getPartition(SeaTunnelRowAccessor row) {
         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);
     }