diff --git a/docs/en/transforms/metadata.md b/docs/en/transforms/metadata.md index 5a311465b04f..b0e3fa45f12e 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 d51c3264a7a9..b7a25dcece2a 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 000000000000..5ba464198b60 --- /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. + * + *

{@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. + * + *

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 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 9e80335a49ef..9a06e58586e2 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 static String[] getPartition(SeaTunnelRowAccessor row) { return (String[]) row.getOptions().get(PARTITION.getName()); } + /** + * Returns whether the logical metadata key is supported for explicit metadata projection. + * + *

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 000000000000..dee28bde170b --- /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 6db61c900d2e..bb7b8cb41126 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 @@ protected Object[] getOutputFieldValues(SeaTunnelRowAccessor inputRow) { for (Map.Entry 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 @@ protected Column[] getOutputColumns() { 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 63277273f15b..9641b14c9163 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.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 @@ void testMetadataTransform() { Assertions.assertEquals(eventTime, outputRow.getField(9)); Assertions.assertEquals(150L, outputRow.getField(10)); } + + @Test + void shouldProjectKnowledgeSyncMetadataFromRowOptions() { + Map metadataMapping = new LinkedHashMap<>(); + metadataMapping.put( + KnowledgeSyncMetadataField.DOCUMENT_ID.getName(), + KnowledgeSyncMetadataField.DOCUMENT_ID.getPhysicalName()); + metadataMapping.put( + KnowledgeSyncMetadataField.CHUNK_HASH.getName(), + KnowledgeSyncMetadataField.CHUNK_HASH.getPhysicalName()); + Map 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 metadataMapping = new LinkedHashMap<>(); + metadataMapping.put( + KnowledgeSyncMetadataField.DOCUMENT_ID.getName(), + KnowledgeSyncMetadataField.DOCUMENT_ID.getPhysicalName()); + Map 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 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()); + } }