njnu-seafish commented on code in PR #18549:
URL:
https://github.com/apache/dolphinscheduler/pull/18549#discussion_r3966482649
##########
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml:
##########
@@ -51,6 +51,42 @@
having count(*) = 0
</insert>
+ <!-- MySQL -->
+ <insert id="insertTaskResultAlertIfAbsent" databaseId="mysql">
+ INSERT INTO t_ds_alert(sign, title, content, alert_status,
warning_type, log, alertgroup_id,
+ create_time, update_time, project_code,
workflow_definition_code,
+ workflow_instance_id, alert_type)
+ VALUES (#{alert.sign}, #{alert.title}, #{alert.content},
#{alert.alertStatus.code},
+ #{alert.warningType.code}, #{alert.log},
#{alert.alertGroupId}, #{alert.createTime},
+ #{alert.updateTime}, #{alert.projectCode},
#{alert.workflowDefinitionCode},
+ #{alert.workflowInstanceId}, #{alert.alertType.code})
+ ON DUPLICATE KEY UPDATE id = id
+ </insert>
+
+ <!-- H2 -->
+ <insert id="insertTaskResultAlertIfAbsent" databaseId="h2">
+ MERGE INTO t_ds_alert(sign, title, content, alert_status,
warning_type, log, alertgroup_id,
+ create_time, update_time, project_code, workflow_definition_code,
+ workflow_instance_id, alert_type)
+ KEY(sign, workflow_instance_id, alert_type)
+ VALUES (#{alert.sign}, #{alert.title}, #{alert.content},
#{alert.alertStatus.code},
+ #{alert.warningType.code}, #{alert.log}, #{alert.alertGroupId},
#{alert.createTime},
+ #{alert.updateTime}, #{alert.projectCode},
#{alert.workflowDefinitionCode},
+ #{alert.workflowInstanceId}, #{alert.alertType.code})
+ </insert>
+
+ <!-- PostgreSQL -->
+ <insert id="insertTaskResultAlertIfAbsent" databaseId="postgresql">
+ INSERT INTO t_ds_alert(sign, title, content, alert_status,
warning_type, log, alertgroup_id,
+ create_time, update_time, project_code,
workflow_definition_code,
+ workflow_instance_id, alert_type)
+ VALUES (#{alert.sign}, #{alert.title}, #{alert.content},
#{alert.alertStatus.code},
+ #{alert.warningType.code}, #{alert.log},
#{alert.alertGroupId}, #{alert.createTime},
+ #{alert.updateTime}, #{alert.projectCode},
#{alert.workflowDefinitionCode},
+ #{alert.workflowInstanceId}, #{alert.alertType.code})
+ ON CONFLICT (sign, workflow_instance_id, alert_type) DO NOTHING
+ </insert>
Review Comment:
> Don't use this writing method, which will lead to poor database
performance.
1. Unique Index Added
A unique composite index uk_alert_dedup has been created on the columns
(sign, workflow_instance_id, alert_type) in the t_ds_alert table. This serves
as the ultimate safeguard at the database level, ensuring that no duplicate
alert records can ever be persisted, regardless of what happens at the
application or SQL layer.
> ALTER TABLE `t_ds_alert` ADD UNIQUE INDEX `uk_alert_dedup` (`sign`,
`workflow_instance_id`, `alert_type`);
2. Optimistic Check in SQL — Low Cost, No Exception Overhead
The INSERT statement uses a WHERE NOT EXISTS clause to perform an optimistic
duplicate check before inserting. In the vast majority of cases, if the alert
already exists, the subquery detects it via the unique index and simply returns
zero affected rows — no exception is thrown, no lock contention is incurred,
and no error-handling overhead is introduced. This makes the common path both
fast and lightweight.
> <insert id="insertTaskResultAlertIfAbsent">
INSERT INTO t_ds_alert(sign, title, content, alert_status,
warning_type, log, alertgroup_id,
create_time, update_time, project_code,
workflow_definition_code,
workflow_instance_id, alert_type)
SELECT #{alert.sign}, #{alert.title}, #{alert.content},
#{alert.alertStatus.code},
#{alert.warningType.code}, #{alert.log},
#{alert.alertGroupId}, #{alert.createTime},
#{alert.updateTime}, #{alert.projectCode},
#{alert.workflowDefinitionCode},
#{alert.workflowInstanceId}, #{alert.alertType.code}
WHERE NOT EXISTS (
SELECT 1 FROM t_ds_alert
WHERE sign = #{alert.sign}
AND workflow_instance_id = #{alert.workflowInstanceId}
AND alert_type = #{alert.alertType.code}
)
</insert>
3. Pessimistic Fallback via DuplicateKeyException — Ensuring Consistency
Under Extreme Concurrency
In rare race conditions where two or more threads simultaneously pass the
NOT EXISTS check and attempt to insert the same alert, the unique index will
reject one of them, causing a DuplicateKeyException. The application catches
this exception as a pessimistic fallback, treating it as a skip rather than a
failure. This guarantees data consistency even under extreme concurrent load.
> try {
int count = alertMapper.insertTaskResultAlertIfAbsent(alert);
if (count > 0) {
log.info("add task result alert to db , alert: {}", alert);
} else {
log.info("skip duplicate task result alert, sign: {},
workflowInstanceId: {}", sign,
alert.getWorkflowInstanceId());
}
return count;
} catch (DuplicateKeyException e) {
// Concurrent race: NOT EXISTS passed but another thread
inserted first.
// The uk_alert_dedup unique constraint caught it — treat as a
skip.
log.info("skip duplicate task result alert (concurrent race),
sign: {}, workflowInstanceId: {}", sign,
alert.getWorkflowInstanceId());
return 0;
}
I clean up redundant comments and code.
@SbloodyS When you have a moment, could you please review the code again?
Thanks so much!
--
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]
