Teradata Merge语句迁移至MySQL的实现问题咨询

首先，你的Update语句其实是没问题的，它已经正确实现了Merge里WHEN MATCHED THEN UPDATE的逻辑。问题出在Insert语句的连接条件上——你用了!=来判断不匹配，这完全不符合左连接找未匹配记录的逻辑，会导致错误的结果。

分步修正方案： #

1. 保留正确的Update语句

这条语句可以准确匹配目标表和源表的主键（SUBJECT_ID, WORKFLOW_ID, WORKFLOW_RUN_ID, WORKLET_RUN_ID, INSTANCE_ID, TASK_ID, START_TIME），并更新指定字段：

UPDATE schema.INFA_TASK_RUN tgt 
INNER JOIN schema.INFA_TASK_RUN_STG src 
    ON tgt.SUBJECT_ID = src.SUBJECT_ID 
    AND tgt.WORKFLOW_ID = src.WORKFLOW_ID 
    AND tgt.WORKFLOW_RUN_ID = src.WORKFLOW_RUN_ID 
    AND tgt.WORKLET_RUN_ID = src.WORKLET_RUN_ID 
    AND tgt.INSTANCE_ID = src.INSTANCE_ID 
    AND tgt.TASK_ID = src.TASK_ID 
    AND tgt.START_TIME = src.START_TIME 
SET 
    tgt.END_TIME = src.END_TIME, 
    tgt.RUN_ERR_CODE = src.RUN_ERR_CODE, 
    tgt.RUN_ERR_MSG = src.RUN_ERR_MSG, 
    tgt.RUN_STATUS_CODE = src.RUN_STATUS_CODE;

2. 修正Insert语句实现WHEN NOT MATCHED THEN INSERT

要找到源表中不存在于目标表的记录，我们需要用左连接+筛选目标表主键为NULL的逻辑——左连接后，未匹配到目标表的源记录对应的目标表字段都会是NULL。修正后的语句如下：

INSERT INTO schema.INFA_TASK_RUN (
    SUBJECT_AREA, WORKFLOW_NAME, VERSION_NUMBER, SUBJECT_ID, 
    WORKFLOW_ID, WORKFLOW_RUN_ID, WORKLET_RUN_ID, CHILD_RUN_ID, 
    INSTANCE_ID, INSTANCE_NAME, TASK_ID, TASK_TYPE_NAME, 
    TASK_TYPE, START_TIME, END_TIME, RUN_ERR_CODE, 
    RUN_ERR_MSG, RUN_STATUS_CODE, TASK_NAME, TASK_VERSION_NUMBER, 
    SERVER_ID, SERVER_NAME
)
SELECT 
    src.SUBJECT_AREA, src.WORKFLOW_NAME, src.VERSION_NUMBER, src.SUBJECT_ID, 
    src.WORKFLOW_ID, src.WORKFLOW_RUN_ID, src.WORKLET_RUN_ID, src.CHILD_RUN_ID, 
    src.INSTANCE_ID, src.INSTANCE_NAME, src.TASK_ID, src.TASK_TYPE_NAME, 
    src.TASK_TYPE, src.START_TIME, src.END_TIME, src.RUN_ERR_CODE, 
    src.RUN_ERR_MSG, src.RUN_STATUS_CODE, src.TASK_NAME, src.TASK_VERSION_NUMBER, 
    src.SERVER_ID, src.SERVER_NAME
FROM schema.INFA_TASK_RUN_STG src 
LEFT JOIN schema.INFA_TASK_RUN tgt 
    ON tgt.SUBJECT_ID = src.SUBJECT_ID 
    AND tgt.WORKFLOW_ID = src.WORKFLOW_ID 
    AND tgt.WORKFLOW_RUN_ID = src.WORKFLOW_RUN_ID 
    AND tgt.WORKLET_RUN_ID = src.WORKLET_RUN_ID 
    AND tgt.INSTANCE_ID = src.INSTANCE_ID 
    AND tgt.TASK_ID = src.TASK_ID 
    AND tgt.START_TIME = src.START_TIME
WHERE tgt.SUBJECT_ID IS NULL; -- 筛选出目标表中无匹配的源记录

额外建议：保证操作原子性 #

为了避免Update执行成功但Insert失败导致的数据不一致，建议把这两条语句放在一个事务中执行：

START TRANSACTION;

-- 执行Update
UPDATE schema.INFA_TASK_RUN tgt 
INNER JOIN schema.INFA_TASK_RUN_STG src 
    ON tgt.SUBJECT_ID = src.SUBJECT_ID 
    AND tgt.WORKFLOW_ID = src.WORKFLOW_ID 
    AND tgt.WORKFLOW_RUN_ID = src.WORKFLOW_RUN_ID 
    AND tgt.WORKLET_RUN_ID = src.WORKLET_RUN_ID 
    AND tgt.INSTANCE_ID = src.INSTANCE_ID 
    AND tgt.TASK_ID = src.TASK_ID 
    AND tgt.START_TIME = src.START_TIME 
SET 
    tgt.END_TIME = src.END_TIME, 
    tgt.RUN_ERR_CODE = src.RUN_ERR_CODE, 
    tgt.RUN_ERR_MSG = src.RUN_ERR_MSG, 
    tgt.RUN_STATUS_CODE = src.RUN_STATUS_CODE;

-- 执行Insert
INSERT INTO schema.INFA_TASK_RUN (
    SUBJECT_AREA, WORKFLOW_NAME, VERSION_NUMBER, SUBJECT_ID, 
    WORKFLOW_ID, WORKFLOW_RUN_ID, WORKLET_RUN_ID, CHILD_RUN_ID, 
    INSTANCE_ID, INSTANCE_NAME, TASK_ID, TASK_TYPE_NAME, 
    TASK_TYPE, START_TIME, END_TIME, RUN_ERR_CODE, 
    RUN_ERR_MSG, RUN_STATUS_CODE, TASK_NAME, TASK_VERSION_NUMBER, 
    SERVER_ID, SERVER_NAME
)
SELECT 
    src.SUBJECT_AREA, src.WORKFLOW_NAME, src.VERSION_NUMBER, src.SUBJECT_ID, 
    src.WORKFLOW_ID, src.WORKFLOW_RUN_ID, src.WORKLET_RUN_ID, src.CHILD_RUN_ID, 
    src.INSTANCE_ID, src.INSTANCE_NAME, src.TASK_ID, src.TASK_TYPE_NAME, 
    src.TASK_TYPE, src.START_TIME, src.END_TIME, src.RUN_ERR_CODE, 
    src.RUN_ERR_MSG, src.RUN_STATUS_CODE, src.TASK_NAME, src.TASK_VERSION_NUMBER, 
    src.SERVER_ID, src.SERVER_NAME
FROM schema.INFA_TASK_RUN_STG src 
LEFT JOIN schema.INFA_TASK_RUN tgt 
    ON tgt.SUBJECT_ID = src.SUBJECT_ID 
    AND tgt.WORKFLOW_ID = src.WORKFLOW_ID 
    AND tgt.WORKFLOW_RUN_ID = src.WORKFLOW_RUN_ID 
    AND tgt.WORKLET_RUN_ID = src.WORKLET_RUN_ID 
    AND tgt.INSTANCE_ID = src.INSTANCE_ID 
    AND tgt.TASK_ID = src.TASK_ID 
    AND tgt.START_TIME = src.START_TIME
WHERE tgt.SUBJECT_ID IS NULL;

COMMIT;

这样就能完全等价于Teradata中Merge语句的逻辑了。

内容的提问来源于stack exchange，提问作者Ashu