bugfix:fix error while the "context" is key word in DM8 when delete u…

…ndolog (#7010)

changes/en-us/2.x.md

@@ -25,6 +25,7 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#6984](https://github.com/apache/incubator-seata/pull/6984)] support building docker image on openjdk23
 - [[#6994](https://github.com/apache/incubator-seata/pull/6994)] fix the problem of building undoLog exception when update join does not update data
 - [[#7005](https://github.com/apache/incubator-seata/pull/7005)] fix the Raft NPE issue caused by two-phase concurrency
+- [[#7010](https://github.com/apache/incubator-seata/pull/7010)] fix error while the "context" is key word in DM8 when delete undolog
 
 ### optimize:
 - [[#6826](https://github.com/apache/incubator-seata/pull/6826)] remove the branch registration operation of the XA read-only transaction
@@ -73,6 +74,7 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [xingfudeshi](https://github.com/xingfudeshi)
 - [o-jimin](https://github.com/o-jimin)
 - [lixingjia77](https://github.com/lixingjia77)
+- [whaon](https://github.com/whaon)
 
 
 

changes/zh-cn/2.x.md

@@ -25,6 +25,7 @@
 - [[#6984](https://github.com/apache/incubator-seata/pull/6984)] 修复 openjdk23 版本下无法构建 docker 镜像的问题
 - [[#6994](https://github.com/apache/incubator-seata/pull/6994)] 修复updateJoin语句未更新到数据时prepareUndoLog异常
 - [[#7005](https://github.com/apache/incubator-seata/pull/7005)] 修复Raft模式下两阶段并发可能导致NPE的问题
+- [[#7010](https://github.com/apache/incubator-seata/pull/7010)] 修复使用达梦数据库时删除undolog发生SQL语法错误
 
 
 ### optimize:
@@ -77,6 +78,7 @@
 - [xingfudeshi](https://github.com/xingfudeshi)
 - [o-jimin](https://github.com/o-jimin)
 - [lixingjia77](https://github.com/lixingjia77)
+- [whaon](https://github.com/whaon)
 
 
 同时，我们收到了社区反馈的很多有价值的issue和建议，非常感谢大家。

rm-datasource/src/main/java/org/apache/seata/rm/datasource/undo/dm/DmUndoLogManager.java

@@ -18,9 +18,11 @@
 
 
 import org.apache.seata.common.loader.LoadLevel;
+import org.apache.seata.common.util.CollectionUtils;
 import org.apache.seata.core.compressor.CompressorType;
 import org.apache.seata.core.constants.ClientTableColumnsName;
 import org.apache.seata.rm.datasource.undo.AbstractUndoLogManager;
+import org.apache.seata.rm.datasource.undo.UndoLogConstants;
 import org.apache.seata.rm.datasource.undo.UndoLogParser;
 import org.apache.seata.sqlparser.util.JdbcConstants;
 import org.slf4j.Logger;
@@ -30,13 +32,16 @@
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.util.Date;
+import java.util.Set;
 
 
 @LoadLevel(name = JdbcConstants.DM)
 public class DmUndoLogManager extends AbstractUndoLogManager {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DmUndoLogManager.class);
 
+    protected static final String DELETE_SUB_UNDO_LOG_SQL = "DELETE FROM " + UNDO_LOG_TABLE_NAME + " WHERE \""
+            + ClientTableColumnsName.UNDO_LOG_CONTEXT.toUpperCase() + "\" = ? AND " + ClientTableColumnsName.UNDO_LOG_XID + " = ?";
 
     private static final String INSERT_UNDO_LOG_SQL = "INSERT INTO " + UNDO_LOG_TABLE_NAME +
             " (" + ClientTableColumnsName.UNDO_LOG_BRANCH_XID + ", "
@@ -48,6 +53,88 @@ public class DmUndoLogManager extends AbstractUndoLogManager {
     private static final String DELETE_UNDO_LOG_BY_CREATE_SQL = "DELETE FROM " + UNDO_LOG_TABLE_NAME +
             " WHERE " + ClientTableColumnsName.UNDO_LOG_LOG_CREATED + " <= ? and ROWNUM <= ?";
 
+    /**
+     * Delete undo log.
+     *
+     * @param xid      the xid
+     * @param branchId the branch id
+     * @param conn     the conn
+     * @throws SQLException the sql exception
+     */
+    @Override
+    public void deleteUndoLog(String xid, long branchId, Connection conn) throws SQLException {
+        try (PreparedStatement deletePST = conn.prepareStatement(DELETE_UNDO_LOG_SQL);
+             PreparedStatement deleteSubPST = conn.prepareStatement(DELETE_SUB_UNDO_LOG_SQL)) {
+            deletePST.setLong(1, branchId);
+            deletePST.setString(2, xid);
+            deletePST.executeUpdate();
+
+            deleteSubPST.setString(1, UndoLogConstants.BRANCH_ID_KEY + CollectionUtils.KV_SPLIT + branchId);
+            deleteSubPST.setString(2, xid);
+            deleteSubPST.executeUpdate();
+        } catch (Exception e) {
+            if (!(e instanceof SQLException)) {
+                e = new SQLException(e);
+            }
+            throw (SQLException) e;
+        }
+    }
+
+    /**
+     * batch Delete undo log.
+     *
+     * @param xids      xid
+     * @param branchIds branch Id
+     * @param conn      connection
+     */
+    @Override
+    public void batchDeleteUndoLog(Set<String> xids, Set<Long> branchIds, Connection conn) throws SQLException {
+        if (CollectionUtils.isEmpty(xids) || CollectionUtils.isEmpty(branchIds)) {
+            return;
+        }
+        int xidSize = xids.size();
+        int branchIdSize = branchIds.size();
+        String batchDeleteSql = toBatchDeleteUndoLogSql(xidSize, branchIdSize);
+        String batchDeleteSubSql = toBatchDeleteSubUndoLogSql(xidSize, branchIdSize);
+        try (PreparedStatement deletePST = conn.prepareStatement(batchDeleteSql);
+             PreparedStatement deleteSubPST = conn.prepareStatement(batchDeleteSubSql)) {
+            int paramsIndex = 1;
+            for (Long branchId : branchIds) {
+                deletePST.setLong(paramsIndex, branchId);
+                deleteSubPST.setString(paramsIndex, UndoLogConstants.BRANCH_ID_KEY + CollectionUtils.KV_SPLIT + branchId);
+                paramsIndex++;
+            }
+            for (String xid : xids) {
+                deletePST.setString(paramsIndex, xid);
+                deleteSubPST.setString(paramsIndex, xid);
+                paramsIndex++;
+            }
+            int deleteRows = deletePST.executeUpdate();
+            if (LOGGER.isDebugEnabled()) {
+                LOGGER.debug("batch delete undo log size {}", deleteRows);
+            }
+            int deleteSubRows = deleteSubPST.executeUpdate();
+            if (LOGGER.isDebugEnabled()) {
+                LOGGER.debug("batch delete sub undo log size {}", deleteSubRows);
+            }
+        } catch (Exception e) {
+            if (!(e instanceof SQLException)) {
+                e = new SQLException(e);
+            }
+            throw (SQLException) e;
+        }
+    }
+
+    protected static String toBatchDeleteSubUndoLogSql(int xidSize, int branchIdSize) {
+        StringBuilder sqlBuilder = new StringBuilder(64);
+        sqlBuilder.append("DELETE FROM ").append(UNDO_LOG_TABLE_NAME).append(" WHERE \"").append(
+                ClientTableColumnsName.UNDO_LOG_CONTEXT.toUpperCase()).append("\" IN ");
+        appendInParam(branchIdSize, sqlBuilder);
+        sqlBuilder.append(" AND ").append(ClientTableColumnsName.UNDO_LOG_XID).append(" IN ");
+        appendInParam(xidSize, sqlBuilder);
+        return sqlBuilder.toString();
+    }
+
     @Override
     public int deleteUndoLogByLogCreated(Date logCreated, int limitRows, Connection conn) throws SQLException {
         try (PreparedStatement deletePST = conn.prepareStatement(DELETE_UNDO_LOG_BY_CREATE_SQL)) {