我用python编写了一个airbyte自定义目的地。我已经使用此查询实现了增量同步重复数据消除操作，

MERGE INTO {self.schema_name}.{table_name} AS target
USING {self.schema_name}.{table_name}_temp AS source
ON target.data_{primary_keys[0][0]}=source.data_{primary_keys[0][0]}
WHEN MATCHED THEN
        {query_placeholder_refined}
WHEN NOT MATCHED THEN
        INSERT *

这里，query_placeholder_refined变量被UPDATE SET查询语句替换，其中目标表的所有列分别被更新，例如采用该查询的简化版本，

MERGE INTO integration.issues as target
USING integration.issues_temp as source
ON target.data_id=source.data_id
WHEN MATCHED THEN
        UPDATE SET target.data_issue_url=source.data_issue_url, target.data_user_id=source.data_user_id 
WHEN NOT MATCHED THEN 
        INSERT *

查询在一些流上运行得很好，但对于其他流，它给出了这个错误，pyspark.sql.utils.AnalysisException: Updates are in conflict for these columns: data_user_id