DanielLeens commented on PR #11401: URL: https://github.com/apache/seatunnel/pull/11401#issuecomment-6368905693
I prepared a conflict-resolution patch for current contributor head `1361f9dd0e452e701a16fc7218af84a118d6e871` merged with upstream `dev` commit `14a27ac26a0a71b56722f86cdf8ec2779094ceaf`. This is a merge proposal requiring fresh CI. It resolves the five conflicted files while keeping both the CDC identifier work and newer upstream functionality: - Both `Collections` and `HashMap` imports in Oracle, and both `CommonOptions` and `SchemaChangeType` in PostgreSQL. - The CDC source identifier contract in both languages, alongside the complete current Knowledge Sync / Markdown source documentation. - The PR's `testSourceIdentifierMetadataTransform` and its metadata fixture, alongside every upstream metadata/Knowledge Sync/connector-declared field test. No upstream assertion is removed or relaxed. The patch below applies to Git's conflicted working files **after** merging the two commits above, not directly to a clean checkout. Save the diff as `pr11401-conflict-resolution.patch` and, from the source branch at the stated head: ```bash git fetch https://github.com/apache/seatunnel.git dev git merge --no-commit --no-ff 14a27ac26a0a71b56722f86cdf8ec2779094ceaf # The five expected content conflicts are resolved by the patch below. git apply --check pr11401-conflict-resolution.patch git apply pr11401-conflict-resolution.patch ``` Then stage the five resolved files, review the full merge, and create a normal merge commit on your branch. No history rewrite is required. If either pinned commit changes, regenerate the resolution rather than applying it blindly. Validation completed: scoped Spotless for Oracle CDC, PostgreSQL CDC and transforms; forward patch-application check against the exact saved conflict files; no unresolved index entries; whitespace check against the pinned upstream baseline; preservation of all original and upstream test method names and all upstream assertion lines. This does not establish runtime correctness. Local compilation/tests were not run; the old unit/transform job logs now return HTTP 410, so I am not attributing their failures from old review text. Push the resolved branch from your fork and obtain fresh functional CI, including the metadata and CDC regression suites. No remote code ref has been changed by this work. Only this small resolution patch is included, not the full upstream sync diff. ```diff diff --git a/docs/en/transforms/metadata.md b/docs/en/transforms/metadata.md --- a/docs/en/transforms/metadata.md +++ b/docs/en/transforms/metadata.md @@ -82,33 +82,6 @@ ### Important Notes -<<<<<<< HEAD -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. **CDC source identifiers**: `Database`, `Schema`, `Table`, `Collection`, and `Namespace` are SeaTunnel-owned identifier metadata fields. Unavailable identifiers remain absent and are not filled with placeholders or duplicated values. -3. **Time fields**: `Delay` and `SourceTimestamp` are only available for CDC connectors. `EventTime` is also provided by the Kafka source via `ConsumerRecord.timestamp` when available. -4. **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. -5. **Binlog/GTID fields**: `BinlogFile`, `BinlogPos`, `BinlogRow`, and `Gtid` are MySQL-CDC specific. For `startup.mode = initial`, snapshot rows return `null` for all four fields. - -### CDC Source Identifier Contract - -CDC identifier metadata uses the following stable SeaTunnel field semantics: - -| Metadata Key | Semantics | -|:-------------|:----------| -| Database | Logical database name only. It must not contain schema, table, collection, or namespace composition. | -| Schema | Relational schema or owner name only. It is absent when the backend has no schema concept. | -| Table | Physical table name only. It must not be database-qualified or schema-qualified. | -| Collection | Document collection name only. | -| Namespace | Source-native combined namespace string. For MongoDB-CDC, this is `database.collection`. | - -Connector coverage for the first delivery: - -| Connector | Identifier Metadata | -|:----------|:--------------------| -| MySQL-CDC / TiDB-CDC | `Database`, `Table` | -| PostgreSQL-CDC / Opengauss-CDC / Oracle-CDC / SQLServer-CDC | `Database` where available, `Schema`, `Table` | -| MongoDB-CDC | `Database`, `Collection`, `Namespace` | -======= 1. **Metadata field names are case-sensitive**: Configuration must strictly follow the Key names in the tables above (e.g., `Database`, `Table`, `RowKind`) or the exact names declared in the input `MetadataSchema`. 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. @@ -119,6 +92,26 @@ 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`. 9. **Projection collisions**: An output name cannot duplicate an existing physical field. Because enabled Markdown already contains physical `source_uri` and `document_id`, project the logical values to aliases such as `ks_source_uri` and `ks_document_id`. +### CDC Source Identifier Contract + +CDC identifier metadata uses the following stable SeaTunnel field semantics: + +| Metadata Key | Semantics | +|:-------------|:----------| +| Database | Logical database name only. It must not contain schema, table, collection, or namespace composition. | +| Schema | Relational schema or owner name only. It is absent when the backend has no schema concept. | +| Table | Physical table name only. It must not be database-qualified or schema-qualified. | +| Collection | Document collection name only. | +| Namespace | Source-native combined namespace string. For MongoDB-CDC, this is `database.collection`. | + +Connector coverage for the first delivery: + +| Connector | Identifier Metadata | +|:----------|:--------------------| +| MySQL-CDC / TiDB-CDC | `Database`, `Table` | +| PostgreSQL-CDC / Opengauss-CDC / Oracle-CDC / SQLServer-CDC | `Database` where available, `Schema`, `Table` | +| MongoDB-CDC | `Database`, `Collection`, `Namespace` | + ### Markdown Source Bridge The Markdown source is a Knowledge Sync metadata producer when `file_format_type=markdown` and `markdown_rag_metadata_enabled=true`. @@ -159,7 +152,6 @@ ``` `ChunkHash` is valid for the immediate Markdown row only. If a downstream transform changes `text` or expands one row into multiple chunks, recompute the final `ChunkHash`, `ChunkId`, and `ChunkIndex` before sending rows to a lifecycle sink. This bridge is producer integration only; it does not implement incremental comparison, writer affinity, stale-chunk deletion, or tombstones. ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf ## Options diff --git a/docs/zh/transforms/metadata.md b/docs/zh/transforms/metadata.md --- a/docs/zh/transforms/metadata.md +++ b/docs/zh/transforms/metadata.md @@ -82,33 +82,6 @@ ### 重要说明 -<<<<<<< HEAD -1. **元数据字段区分大小写**：配置时必须严格按照上表中的 Key 名称（如 `Database`、`Table`、`RowKind` 等）。 -2. **CDC 源端标识字段**：`Database`、`Schema`、`Table`、`Collection`、`Namespace` 是 SeaTunnel 维护的稳定标识元数据。连接器无法提供的标识会保持缺失，不会填充占位值或重复值。 -3. **时间相关字段**：`Delay` 和 `SourceTimestamp` 仅在 CDC 连接器有效。`EventTime` 也会在 Kafka 源中使用 `ConsumerRecord.timestamp`（毫秒，非负时）写入。 -4. **Kafka 事件时间**：Kafka 源会在 `ConsumerRecord.timestamp` 非负时写入 `EventTime`，可通过 Metadata 转换将其暴露为普通字段。 -5. **Binlog/GTID 字段**：`BinlogFile`、`BinlogPos`、`BinlogRow`、`Gtid` 仅适用于 MySQL-CDC。使用 `startup.mode = initial` 时，快照行的这四个字段均为 `null`。 - -### CDC 源端标识契约 - -CDC 标识元数据使用以下稳定的 SeaTunnel 字段语义： - -| 元数据Key | 语义 | -|:----------|:-----| -| Database | 仅表示逻辑数据库名，不包含 schema、表、集合或命名空间组合。 | -| Schema | 仅表示关系型数据库的 schema 或 owner。后端没有 schema 概念时该字段缺失。 | -| Table | 仅表示物理表名，不包含 database 或 schema 前缀。 | -| Collection | 仅表示文档集合名称。 | -| Namespace | 源端原生组合命名空间。MongoDB-CDC 中固定为 `database.collection`。 | - -首批交付的连接器覆盖范围： - -| 连接器 | 标识元数据 | -|:-------|:-----------| -| MySQL-CDC / TiDB-CDC | `Database`、`Table` | -| PostgreSQL-CDC / Opengauss-CDC / Oracle-CDC / SQLServer-CDC | 可用时提供 `Database`，并提供 `Schema`、`Table` | -| MongoDB-CDC | `Database`、`Collection`、`Namespace` | -======= 1. **元数据字段区分大小写**：配置时必须严格按照上表中的 Key 名称（如 `Database`、`Table`、`RowKind`），或输入表 `MetadataSchema` 中声明的精确名称。 2. **时间相关字段**：`Delay` 和 `SourceTimestamp` 仅在 CDC 连接器有效。`EventTime` 也会在 Kafka 源中使用 `ConsumerRecord.timestamp`（毫秒，非负时）写入。 3. **Kafka 事件时间**：Kafka 源会在 `ConsumerRecord.timestamp` 非负时写入 `EventTime`，可通过 Metadata 转换将其暴露为普通字段。 @@ -119,6 +92,26 @@ 8. **Knowledge Sync 可空语义**：`DocumentId` 用于标识每个文档生命周期事件。声明 `Deleted` 后，producer 必须为普通行写入 `false`，为文档 tombstone 写入 `true`，不能使用 `null`。普通 chunk 行必须包含 `ChunkId`、`ChunkHash` 和 `ChunkIndex`；紧凑的文档 tombstone 可以将这些 chunk 字段留为 `null`。 9. **投影冲突**：输出字段名不能与已有物理字段重复。启用 Markdown RAG metadata 后已经存在物理 `source_uri` 和 `document_id`，因此应把逻辑值投影为 `ks_source_uri`、`ks_document_id` 等别名。 +### CDC 源端标识契约 + +CDC 标识元数据使用以下稳定的 SeaTunnel 字段语义： + +| 元数据Key | 语义 | +|:----------|:-----| +| Database | 仅表示逻辑数据库名，不包含 schema、表、集合或命名空间组合。 | +| Schema | 仅表示关系型数据库的 schema 或 owner。后端没有 schema 概念时该字段缺失。 | +| Table | 仅表示物理表名，不包含 database 或 schema 前缀。 | +| Collection | 仅表示文档集合名称。 | +| Namespace | 源端原生组合命名空间。MongoDB-CDC 中固定为 `database.collection`。 | + +首批交付的连接器覆盖范围： + +| 连接器 | 标识元数据 | +|:-------|:-----------| +| MySQL-CDC / TiDB-CDC | `Database`、`Table` | +| PostgreSQL-CDC / Opengauss-CDC / Oracle-CDC / SQLServer-CDC | 可用时提供 `Database`，并提供 `Schema`、`Table` | +| MongoDB-CDC | `Database`、`Collection`、`Namespace` | + ### Markdown Source Bridge 当 `file_format_type=markdown` 且 `markdown_rag_metadata_enabled=true` 时，Markdown source 会成为 Knowledge Sync 元数据 producer。 @@ -159,7 +152,6 @@ ``` `ChunkHash` 只对 Markdown 直接输出的当前行有效。如果下游 transform 修改 `text` 或把一行展开为多个 chunk，则必须在把行发送到 lifecycle sink 前重新计算最终 `ChunkHash`、`ChunkId` 和 `ChunkIndex`。该 bridge 仅完成 producer 集成，不实现增量比较、writer affinity、过期 chunk 删除或 tombstone。 ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf ## 配置选项 diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java @@ -54,11 +54,8 @@ import java.time.ZoneId; import java.util.Arrays; -<<<<<<< HEAD import java.util.Collections; -======= import java.util.HashMap; ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf import java.util.List; import java.util.Map; import java.util.Optional; diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java @@ -21,11 +21,8 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.source.SupportSchemaEvolution; import org.apache.seatunnel.api.table.catalog.CatalogTable; -<<<<<<< HEAD +import org.apache.seatunnel.api.table.schema.SchemaChangeType; import org.apache.seatunnel.api.table.type.CommonOptions; -======= -import org.apache.seatunnel.api.table.schema.SchemaChangeType; ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf import org.apache.seatunnel.common.utils.JdbcUrlUtil; import org.apache.seatunnel.common.utils.SeaTunnelException; import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig; 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 --- 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 @@ -224,16 +224,6 @@ } @Test -<<<<<<< HEAD - void testSourceIdentifierMetadataTransform() { - Map<String, String> metadataMapping = new LinkedHashMap<>(); - metadataMapping.put("Schema", "source_schema"); - metadataMapping.put("Collection", "source_collection"); - metadataMapping.put("Namespace", "source_namespace"); - Map<String, Object> config = new HashMap<>(); - config.put("metadata_fields", metadataMapping); - MetadataTransform transform = -======= void shouldProjectKnowledgeSyncMetadataFromRowOptions() { Map<String, String> metadataMapping = new LinkedHashMap<>(); metadataMapping.put( @@ -547,12 +537,55 @@ config.put("metadata_fields", metadataMapping); MetadataTransform transform = ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf new MetadataTransform(ReadonlyConfig.fromMap(config), catalogTable); transform.initRowContainerGenerator(); Column[] columns = transform.getOutputColumns(); -<<<<<<< HEAD + Assertions.assertEquals(BasicType.STRING_TYPE, columns[0].getDataType()); + Assertions.assertEquals(BasicType.STRING_TYPE, columns[1].getDataType()); + Assertions.assertEquals(BasicType.STRING_TYPE, columns[2].getDataType()); + Assertions.assertEquals(BasicType.LONG_TYPE, columns[3].getDataType()); + + SeaTunnelRow outputRow = transform.map(inputRow); + Assertions.assertEquals("default", outputRow.getField(5)); + Assertions.assertEquals("default", outputRow.getField(6)); + Assertions.assertEquals("+I", outputRow.getField(7)); + Assertions.assertEquals(eventTime, outputRow.getField(8)); + } + + @Test + void shouldRejectMarkdownCanonicalPhysicalNameCollision() { + 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); + + TransformException exception = + Assertions.assertThrows( + TransformException.class, + () -> + new MetadataTransform( + ReadonlyConfig.fromMap(config), markdownCatalogTable())); + + Assertions.assertTrue( + exception.getMessage().contains(KnowledgeSyncMetadataField.DOCUMENT_ID.getName())); + } + + @Test + void testSourceIdentifierMetadataTransform() { + Map<String, String> metadataMapping = new LinkedHashMap<>(); + metadataMapping.put("Schema", "source_schema"); + metadataMapping.put("Collection", "source_collection"); + metadataMapping.put("Namespace", "source_namespace"); + Map<String, Object> config = new HashMap<>(); + config.put("metadata_fields", metadataMapping); + MetadataTransform transform = + new MetadataTransform(ReadonlyConfig.fromMap(config), catalogTable); + transform.initRowContainerGenerator(); + + Column[] columns = transform.getOutputColumns(); Assertions.assertEquals("source_schema", columns[0].getName()); Assertions.assertEquals("source_collection", columns[1].getName()); Assertions.assertEquals("source_namespace", columns[2].getName()); @@ -565,37 +598,6 @@ Assertions.assertEquals("public", outputRow.getField(5)); Assertions.assertEquals("orders", outputRow.getField(6)); Assertions.assertEquals("inventory.orders", outputRow.getField(7)); -======= - Assertions.assertEquals(BasicType.STRING_TYPE, columns[0].getDataType()); - Assertions.assertEquals(BasicType.STRING_TYPE, columns[1].getDataType()); - Assertions.assertEquals(BasicType.STRING_TYPE, columns[2].getDataType()); - Assertions.assertEquals(BasicType.LONG_TYPE, columns[3].getDataType()); - - SeaTunnelRow outputRow = transform.map(inputRow); - Assertions.assertEquals("default", outputRow.getField(5)); - Assertions.assertEquals("default", outputRow.getField(6)); - Assertions.assertEquals("+I", outputRow.getField(7)); - Assertions.assertEquals(eventTime, outputRow.getField(8)); - } - - @Test - void shouldRejectMarkdownCanonicalPhysicalNameCollision() { - 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); - - TransformException exception = - Assertions.assertThrows( - TransformException.class, - () -> - new MetadataTransform( - ReadonlyConfig.fromMap(config), markdownCatalogTable())); - - Assertions.assertTrue( - exception.getMessage().contains(KnowledgeSyncMetadataField.DOCUMENT_ID.getName())); } private static Map<String, Object> metadataFieldsConfig(String metadataKey, String outputName) { @@ -700,6 +702,5 @@ "comment", "test", MetadataSchema.builder().columns(metadata).build()); ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf } } ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
