fix: 适配 IoT 外部数据库时间函数并整理 SQL 工具

YunaiV
1 parent 8e4d1e70
Showing 6 changed files with 235 additions and 39 deletions
yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/injector/SelectForUpdateMethod.java
yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/mapper/BaseMapperX.java
yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/util/SqlDialectUtils.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleAction.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleActionTest.java
yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/dal/mysql/cal/plan/MesCalPlanMapper.java
@@ -47,14 +47,17 @@ public class SelectForUpdateMethod extends AbstractMethod {
     @Override
     public MappedStatement injectMappedStatement(Class<?> mapperClass, Class<?> modelClass, TableInfo tableInfo) {
-        // 没有主键的表，无法查询首行主键
+        // 1. 没有主键的表，无法查询首行主键
         if (kind == Kind.FIRST_IDS && !tableInfo.havePK()) {
             return null;
         }
+        // 2. 拼接 SQL：查询列 + 条件 + 排序与锁定；首行主键查询只查询主键列
         String columns = kind == Kind.FIRST_IDS ? tableInfo.getKeyColumn() : sqlSelectColumns(tableInfo, true);
         String sql = "<script>" + sqlFirst() + " SELECT " + columns + " FROM " + tableInfo.getTableName()
                 + NEWLINE + sqlWhere(tableInfo) + NEWLINE + sqlTail(tableInfo) + sqlComment() + NEWLINE + "</script>";
+        // 3. 使用 LockSqlSource 标记为锁定读的内置查询，供 ForUpdateInnerInterceptor 识别
         SqlSource sqlSource = new LockSqlSource(kind, super.createSqlSource(configuration, sql, modelClass));
+        // 4. 注册 MappedStatement：首行主键查询的结果为主键类型，其它为实体（沿用表的 resultMap）
         if (kind == Kind.FIRST_IDS) {
             return addSelectMappedStatementForOther(mapperClass, methodName, sqlSource, tableInfo.getKeyType());
         }
@@ -68,16 +71,21 @@ public class SelectForUpdateMethod extends AbstractMethod {
      * @return WHERE 脚本
      */
     private String sqlWhere(TableInfo tableInfo) {
+        // 实体条件：Wrapper 设置了 entity 时生效
         String entityWhere = SqlScriptUtils.convertIf(tableInfo.getAllSqlWhere(true, true, true, WRAPPER_ENTITY_DOT),
                 WRAPPER_ENTITY + " != null", true);
+        // 普通条件：整体加括号，避免 OR 条件绕过后面的主键、逻辑删除条件
         String normalWhere = SqlScriptUtils.convertIf(" AND (${" + WRAPPER_DOT + "expression.normal.sqlSegment})",
                 WRAPPER_DOT + "nonEmptyOfNormal", true);
+        // 主键：仅加锁查询、且表有主键时，按传入的主键限制，用于 Oracle 的首行锁定
         String idWhere = kind == Kind.LOCK && tableInfo.havePK() ? NEWLINE + SqlScriptUtils.convertIf(
                 " AND " + tableInfo.getKeyColumn() + " = " + SqlScriptUtils.safeParam(PARAM_ID), PARAM_ID + " != null", true) : "";
+        // 拼接逻辑删除条件，由 <where> 去除开头多余的 AND
         return SqlScriptUtils.convertWhere(entityWhere + NEWLINE + normalWhere + idWhere
                 + NEWLINE + tableInfo.getLogicDeleteSql(true, true));
     }
+    // DONE @AI：方法内 注释；（处理：已按主键查询、加锁查询两种情况补充注释）
     /**
      * 排序与锁定：
      * 1. 加锁查询：Wrapper 的排序（未设置时不排序，避免改变数据库的加锁范围） + FOR UPDATE；首行再加 LIMIT 1
@@ -88,19 +96,23 @@ public class SelectForUpdateMethod extends AbstractMethod {
      */
     private String sqlTail(TableInfo tableInfo) {
         String orderBy = " ${" + WRAPPER_DOT + "expression.orderBy.sqlSegment}";
+        // 情况一：首行主键查询，不加锁；未设置排序时按主键排序
         if (kind == Kind.FIRST_IDS) {
             return SqlScriptUtils.convertChoose(WRAPPER_DOT + "expression.orderBy.isEmpty()",
                     " ORDER BY " + tableInfo.getKeyColumn(), orderBy);
         }
+        // 情况二：加锁查询，只使用 Wrapper 的排序，末尾为 FOR UPDATE；首行为 LIMIT 1 FOR UPDATE
         return SqlScriptUtils.convertIf(orderBy, "!" + WRAPPER_DOT + "expression.orderBy.isEmpty()", true)
                 + NEWLINE + (kind == Kind.LOCK_FIRST ? "LIMIT 1 FOR UPDATE" : "FOR UPDATE");
     }
+    // DONE @AI：方法内 注释；（处理：已补充同名 MappedStatement 跳过注入的注释，保留 flushCache、useCache 的说明）
     @Override
     protected MappedStatement addMappedStatement(Class<?> mapperClass, String id, SqlSource sqlSource,
                                                  SqlCommandType sqlCommandType, Class<?> parameterType,
                                                  String resultMap, Class<?> resultType, KeyGenerator keyGenerator,
                                                  String keyProperty, String keyColumn) {
+        // 已通过 XML、注解等定义同名的 MappedStatement 时，不重复注入
         String statementName = mapperClass.getName() + DOT + id;
         if (configuration.hasStatement(statementName, false)) {
             logger.warn(LEFT_SQ_BRACKET + statementName + "] Has been loaded by XML or SqlProvider or Mybatis's Annotation, so ignoring this injection for [" + getClass() + RIGHT_SQ_BRACKET);
@@ -145,6 +157,7 @@ public class SelectForUpdateMethod extends AbstractMethod {
      *
      * MyBatis Plus Join 等插件复制 MappedStatement 时保留同一个 SqlSource，标记仍然有效
      */
+    @SuppressWarnings("ClassCanBeRecord")
     @Getter
     @AllArgsConstructor
     public static class LockSqlSource implements SqlSource {
 package cn.iocoder.yudao.framework.mybatis.core.mapper;
 import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.util.ObjectUtil;
 import cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants;
 import cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil;
 import cn.iocoder.yudao.framework.common.pojo.PageParam;
@@ -151,10 +152,6 @@ public interface BaseMapperX&lt;T&gt; extends MPJBaseMapper&lt;T&gt; {
         return selectOne(new LambdaQueryWrapper<T>().eq(field1, value1).eq(field2, value2).eq(field3, value3));
     }
-    // TODO DONE @AI：能不能把现有的 selectOneForUpdate、selectFirstOneForUpdate、selectListForFirstLock、 这些，做一个整合。目前感觉有点散。。。不然有点乱；包括 cms 项目里，其实也有一些 lock；/Users/yunai/Java/yudao-cms-all/ruoyi-vue-pro-jdk25/yudao-module-cms；最好有个比较万金油的方案。。。。
-    // 已整合为 selectOneForUpdate、selectListForUpdate、selectFirstOneForUpdate 三个锁定读入口（含字段重载），boot、cms 的手写 FOR UPDATE 均改为调用它们；
-    // 锁定子句由内置查询 SelectForUpdateMethod 生成（不再暴露 selectListForFirstLock 等接口方法），SQL Server 由 ForUpdateInnerInterceptor 改写，Oracle 的首行锁定在本类按数据库类型走多次查询。
-
     // ========== 锁定读：需要在事务中调用；Wrapper 不能使用 last、groupBy、having、参数别名 ==========
     /**
@@ -227,31 +224,40 @@ public interface BaseMapperX&lt;T&gt; extends MPJBaseMapper&lt;T&gt; {
      * @param queryWrapper 查询条件
      * @return 实体
      */
+    // DONE @AI：方法内注释噢；（处理：已按“情况一 非 Oracle 单条锁定、情况二 Oracle 2.1～2.3 候选、加锁、重读”补充步骤注释）
     default T selectFirstOneForUpdate(LambdaQueryWrapper<T> queryWrapper) {
+        // 情况一：非 Oracle，单条 LIMIT 1 FOR UPDATE 锁定首行（SQL Server 由 ForUpdateInnerInterceptor 改写）
         DbType dbType = JdbcUtils.getDbType();
-        if (!ObjectUtils.equalsAny(dbType, DbType.ORACLE, DbType.ORACLE_12C)) {
+        // DONE @AI：ObjectUtils notequals any；（处理：已改为 ObjectUtils.notEqualsAny，dbType 为 null 时与原取反写法一致，均走单条锁定）
+        if (ObjectUtils.notEqualsAny(dbType, DbType.ORACLE, DbType.ORACLE_12C)) {
             return CollUtil.getFirst(selectListByForUpdateMethod(SelectForUpdateMethod.Kind.LOCK_FIRST, queryWrapper, null));
         }
+
+        // 情况二：Oracle，按“查询候选主键、按主键加锁、重新读取首行主键”重试，最多 5 次
         int maxAttempts = 5;
         for (int i = 0; i < maxAttempts; i++) {
-            List<Object> ids = selectListByForUpdateMethod(SelectForUpdateMethod.Kind.FIRST_IDS, queryWrapper, null);
-            if (ids.isEmpty()) {
+            // DONE @AI：getFirst 是不是可以简化下；看看其它地方，是不是也有地方；（处理：候选、重读主键与返回的锁定行均改为 CollUtil.getFirst；是否锁定到候选仍按列表是否为空判断）
+            // 2.1 查询首行主键作为候选，不加锁；主键不为 null，取不到时即没有满足条件的记录
+            Object id = CollUtil.getFirst(selectListByForUpdateMethod(SelectForUpdateMethod.Kind.FIRST_IDS, queryWrapper, null));
+            if (id == null) {
                 return null;
             }
-            Object id = ids.get(0);
-            // 是否锁定到候选，以列表是否为空判断：查询列均为 NULL 时，MyBatis 返回的元素为 null
+            // 2.2 按原条件、且主键等于候选加锁；列表为空说明候选已不满足条件，重试
+            // 注意：以列表是否为空判断是否锁定到候选，不能以元素判断：查询列均为 NULL 时，MyBatis 返回的元素为 null
             List<T> lockRows = selectListByForUpdateMethod(SelectForUpdateMethod.Kind.LOCK, queryWrapper, id);
-            if (lockRows.isEmpty()) {
+            if (CollUtil.isEmpty(lockRows)) {
                 continue;
             }
-            List<Object> latestIds = selectListByForUpdateMethod(SelectForUpdateMethod.Kind.FIRST_IDS, queryWrapper, null);
-            if (!latestIds.isEmpty() && Objects.equals(id, latestIds.get(0))) {
-                return lockRows.get(0);
+            // 2.3 锁定后重新读取首行主键，与候选相同才返回锁定的记录（查询列均为 NULL 时为 null）；不同时重试
+            Object latestId = CollUtil.getFirst(selectListByForUpdateMethod(SelectForUpdateMethod.Kind.FIRST_IDS, queryWrapper, null));
+            if (Objects.equals(id, latestId)) {
+                return CollUtil.getFirst(lockRows);
             }
         }
         throw ServiceExceptionUtil.exception(GlobalErrorCodeConstants.LOCKED);
     }
+    // DONE @AI：方法内注释噢；（处理：已补充校验、构建参数、执行内置查询三步注释）
     /**
      * 执行锁定读的内置查询，见 {@link SelectForUpdateMethod}
      *
@@ -262,18 +268,23 @@ public interface BaseMapperX&lt;T&gt; extends MPJBaseMapper&lt;T&gt; {
      */
     private <E> List<E> selectListByForUpdateMethod(SelectForUpdateMethod.Kind kind, LambdaQueryWrapper<T> queryWrapper,
                                                     Object id) {
+        // 1. 校验查询条件
         checkForUpdateWrapper(queryWrapper);
+        // 2. 构建参数：查询条件、主键；首行主键查询由分页插件限制为 1 条，不查询总数
         MapperMethod.ParamMap<Object> param = new MapperMethod.ParamMap<>();
         param.put(Constants.WRAPPER, queryWrapper);
         param.put(SelectForUpdateMethod.PARAM_ID, id);
         if (kind == SelectForUpdateMethod.Kind.FIRST_IDS) {
             param.put("page", MyBatisUtils.buildPage(0, 1, false));
         }
+        // 3. 通过 SqlSession 执行内置查询：未在 Mapper 接口声明，MappedStatement 编号为 Mapper 接口名 + 方法名
         MapperProxyMetadata mapperProxyMetadata = MybatisUtils.getMapperProxy(this);
         return mapperProxyMetadata.getSqlSession().selectList(
                 mapperProxyMetadata.getMapperInterface().getName() + Constants.DOT + kind.getMethod(), param);
     }
+    // DONE @AI：方法内注释噢；（处理：已补充 last、groupBy/having、参数别名的校验注释）
+    // DONE @AI：notequals；isNotEmpty 等等，减少取反；（处理：已改为 ObjectUtil.notEqual、CollUtil.isNotEmpty）
     /**
      * 校验锁定读的查询条件：内置查询只使用实体条件、普通条件与排序
      *
@@ -283,14 +294,17 @@ public interface BaseMapperX&lt;T&gt; extends MPJBaseMapper&lt;T&gt; {
         if (queryWrapper == null) {
             throw new IllegalArgumentException("锁定读的查询条件不能为空");
         }
+        // 校验 last：设置 last 时，Wrapper 的 SQL 片段比条件表达式多出 last 部分
         MergeSegments expression = queryWrapper.getExpression();
-        if (!Objects.equals(queryWrapper.getSqlSegment(), expression.getSqlSegment())) {
+        if (ObjectUtil.notEqual(queryWrapper.getSqlSegment(), expression.getSqlSegment())) {
             throw new IllegalArgumentException("锁定读的查询条件不能使用 last");
         }
-        if (!expression.getGroupBy().isEmpty() || !expression.getHaving().isEmpty()) {
+        // 校验 groupBy、having：内置查询不生成这两部分
+        if (CollUtil.isNotEmpty(expression.getGroupBy()) || CollUtil.isNotEmpty(expression.getHaving())) {
             throw new IllegalArgumentException("锁定读的查询条件不能使用 groupBy、having");
         }
-        if (!Constants.WRAPPER.equals(queryWrapper.getParamAlias())) {
+        // 校验参数别名：内置查询的模板固定使用 ew 作为参数名
+        if (ObjectUtil.notEqual(Constants.WRAPPER, queryWrapper.getParamAlias())) {
             throw new IllegalArgumentException("锁定读的查询条件不能设置参数别名");
         }
     }
@@ -305,7 +319,6 @@ public interface BaseMapperX&lt;T&gt; extends MPJBaseMapper&lt;T&gt; {
      * @return 实体
      */
     default T selectFirstOne(SFunction<T, ?> field, Object value) {
-        // 如果明确使用 MySQL 等场景，可以考虑使用 LIMIT 1 进行优化
         List<T> list = selectList(new LambdaQueryWrapper<T>().eq(field, value));
         return CollUtil.getFirst(list);
     }
@@ -15,8 +15,6 @@ import java.util.Map;
 import java.util.Set;
 import java.util.regex.Pattern;
-// TODO DONE @AI：方法注释；已补齐方言重载与内部辅助方法的参数、返回值及边界。
-// TODO DONE @AI：多余的方法，可以考虑去掉；当然，保留也可以。。。已核对生产 Java/XML 调用，现有方法均有用途，保留。
 /**
  * 日期统计、逗号分隔列表关联、JSON 数组匹配、行数限制、分批删除 SQL 片段的方言工具类
  *
@@ -165,7 +163,6 @@ public class SqlDialectUtils {
      * @param column 列名
      * @return SQL 片段
      */
-    // TODO DONE @AI：用不到的，可以暂时先去掉噢； CRM 业绩统计使用 yyyyMM，商城交易统计使用 yyyy-MM，均需保留。
     public static String formatYearMonth(String column) {
         return formatYearMonth(getDialect(), column);
     }
@@ -196,7 +193,6 @@ public class SqlDialectUtils {
      * @param column 列名
      * @return SQL 片段
      */
-    // TODO DONE @AI：用不到的，可以暂时先去掉噢； CRM 业绩统计使用 yyyyMM，商城交易统计使用 yyyy-MM，均需保留。
     public static String formatYearDashMonth(String column) {
         return formatYearDashMonth(getDialect(), column);
     }
@@ -248,7 +244,6 @@ public class SqlDialectUtils {
         return formatDate(dialect, column);
     }
-    // TODO DONE @AI：用不到的，可以先清理掉噢； AI1 首页与交易统计的 String 日期消费者仍在使用，保留。
     /**
      * 按天分组；原有片段为 CAST(column AS DATE)，新方言输出 yyyy-MM-dd 字符串
      *
@@ -286,12 +281,10 @@ public class SqlDialectUtils {
      * @param column 列名
      * @return SQL 片段
      */
-    // TODO DONE @AI：多余的是不是去掉？ 会员统计的 LocalDate 消费者仍在使用，保留 DATE 类型转换。
     public static String castDate(String column) {
         return castDate(getDialect(), column);
     }
-    // TODO DONE @AI：方法注释；已补齐方言重载与内部辅助方法的参数、返回值及边界。
     /**
      * 生成 LocalDate 消费者所需的日期值；Oracle 用 TRUNC 去掉时分秒。
      *
@@ -337,7 +330,6 @@ public class SqlDialectUtils {
      * @param column 列名
      * @return SQL 片段
      */
-    // TODO DONE @AI：方法注释；已补齐方言重载与内部辅助方法的参数、返回值及边界。
     public static String year(String column) {
         return year(getDialect(), column);
     }
@@ -363,7 +355,6 @@ public class SqlDialectUtils {
         return dayOfMonth(getDialect(), column);
     }
-    // TODO DONE @AI：方法注释；已补齐方言重载与内部辅助方法的参数、返回值及边界。
     /**
      * 按指定方言提取月内日，原路径保留 DAYOFMONTH。
      *
@@ -9,7 +9,10 @@ import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.ArgumentPreparedStatementSetter;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.PreparedStatementCallback;
+import org.springframework.jdbc.core.PreparedStatementCreator;
 import org.springframework.stereotype.Component;
 /**
@@ -26,6 +29,11 @@ import org.springframework.stereotype.Component;
 public class IotDatabaseDataRuleAction extends
         IotDataRuleCacheableAction<IotDataSinkDatabaseConfig, JdbcTemplate> {
+    /**
+     * 不支持 NOW()、改用 CURRENT_TIMESTAMP 的数据库产品名，与 JDBC 驱动的 DatabaseMetaData#getDatabaseProductName 一致
+     */
+    private static final String[] CURRENT_TIMESTAMP_PRODUCT_NAMES = {"Oracle", "Microsoft SQL Server"};
+
     @Override
     public Integer getType() {
         return IotDataSinkTypeEnum.DATABASE.getType();
@@ -42,17 +50,27 @@ public class IotDatabaseDataRuleAction extends
             JdbcTemplate jdbcTemplate = getProducer(config);
             // 2. 构建并执行 INSERT SQL
-            String sql = StrUtil.format(
-                    "INSERT INTO {} (id, device_id, tenant_id, method, report_time, data, create_time) VALUES (?, ?, ?, ?, ?, ?, NOW())",
-                    config.getTableName());
             String messageJson = JsonUtils.toJsonString(message);
-            jdbcTemplate.update(sql,
+            ArgumentPreparedStatementSetter parameterSetter = new ArgumentPreparedStatementSetter(new Object[]{
                     message.getId(),
                     message.getDeviceId(),
                     message.getTenantId(),
                     message.getMethod(),
                     message.getReportTime(),
-                    messageJson);
+                    messageJson});
+            // 2.1 在执行 INSERT 的同一外部连接上识别数据库产品，选择 create_time 使用的数据库时间函数
+            PreparedStatementCreator statementCreator = connection -> connection.prepareStatement(
+                    buildInsertSql(config.getTableName(), connection.getMetaData().getDatabaseProductName()));
+            // 2.2 Statement 由 JdbcTemplate 创建后再绑定参数：绑定或执行失败时，由 JdbcTemplate 关闭 Statement、释放连接
+            PreparedStatementCallback<Integer> statementCallback = statement -> {
+                try {
+                    parameterSetter.setValues(statement);
+                    return statement.executeUpdate();
+                } finally {
+                    parameterSetter.cleanupParameters();
+                }
+            };
+            jdbcTemplate.execute(statementCreator, statementCallback);
             log.info("[execute][message({}) config({}) 写入数据库成功，table: {}]",
                     message.getId(), config, config.getTableName());
         } catch (Exception e) {
@@ -61,6 +79,22 @@ public class IotDatabaseDataRuleAction extends
         }
     }
+    /**
+     * 构建 INSERT SQL
+     *
+     * create_time 使用外部数据库的时间：Oracle、SQL Server 不支持 NOW()，使用 CURRENT_TIMESTAMP；其它数据库保持 NOW()
+     *
+     * @param tableName 已校验的表名
+     * @param databaseProductName 外部数据库的产品名，即 DatabaseMetaData#getDatabaseProductName
+     * @return INSERT SQL
+     */
+    private static String buildInsertSql(String tableName, String databaseProductName) {
+        String now = StrUtil.equalsAny(databaseProductName, CURRENT_TIMESTAMP_PRODUCT_NAMES) ? "CURRENT_TIMESTAMP" : "NOW()";
+        return StrUtil.format(
+                "INSERT INTO {} (id, device_id, tenant_id, method, report_time, data, create_time) VALUES (?, ?, ?, ?, ?, ?, {})",
+                tableName, now);
+    }
+
     @Override
     protected JdbcTemplate initProducer(IotDataSinkDatabaseConfig config) throws Exception {
         // 使用 HikariCP 连接池，HikariCP 会根据 JDBC URL 自动检测并加载对应的数据库驱动
 package cn.iocoder.yudao.module.iot.service.rule.data.action;
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkDatabaseConfig;
 import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
 import com.zaxxer.hikari.HikariDataSource;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
+import org.springframework.dao.DataAccessException;
 import org.springframework.jdbc.core.JdbcTemplate;
 import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
 import java.time.LocalDateTime;
 import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.*;
@@ -26,11 +36,21 @@ import static org.mockito.Mockito.*;
  */
 class IotDatabaseDataRuleActionTest {
+    private static final String TABLE_NAME = "iot_device_message";
+
     private IotDatabaseDataRuleAction databaseDataRuleAction;
     @Mock
     private JdbcTemplate jdbcTemplate;
+    /**
+     * 外部数据库：由真实的 {@link JdbcTemplate} 管理连接、Statement 的生命周期
+     */
+    private DataSource externalDataSource;
+    private Connection externalConnection;
+    private DatabaseMetaData externalMetaData;
+    private PreparedStatement externalStatement;
+
     @BeforeEach
     public void setUp() {
         MockitoAnnotations.openMocks(this);
@@ -47,17 +67,15 @@ class IotDatabaseDataRuleActionTest {
     public void testExecute_tableNameValid() throws Exception {
         // 准备参数：普通表名，以及 schema.table 格式
         for (String tableName : Arrays.asList("iot_device_message", "_message01", "iot_data.device_message")) {
+            mockExternalDatabase("MySQL");
             IotDataSinkDatabaseConfig config = buildConfig(tableName);
-            IotDatabaseDataRuleAction action = spy(databaseDataRuleAction);
-            doReturn(jdbcTemplate).when(action).getProducer(same(config));
+            IotDatabaseDataRuleAction action = spyAction(config);
             // 调用
             action.execute(buildMessage(), config);
             // 断言：表名原样拼接，值使用参数绑定
-            verify(jdbcTemplate).update(eq("INSERT INTO " + tableName
-                            + " (id, device_id, tenant_id, method, report_time, data, create_time) VALUES (?, ?, ?, ?, ?, ?, NOW())"),
-                    any(), any(), any(), any(), any(), any());
+            verify(externalConnection).prepareStatement(buildInsertSql(tableName, "NOW()"));
         }
     }
@@ -79,6 +97,105 @@ class IotDatabaseDataRuleActionTest {
     }
     @Test
+    public void testExecute_createTimeByDatabaseProduct() throws Exception {
+        // 准备参数：数据库产品名 -> create_time 使用的时间函数
+        Map<String, String> cases = new LinkedHashMap<>();
+        cases.put("Oracle", "CURRENT_TIMESTAMP");
+        cases.put("Microsoft SQL Server", "CURRENT_TIMESTAMP");
+        cases.put("MySQL", "NOW()");
+        cases.put("PostgreSQL", "NOW()");
+        cases.put("DM DBMS", "NOW()"); // 未登记的产品，保持原 NOW()
+        cases.put(null, "NOW()");
+        for (Map.Entry<String, String> entry : cases.entrySet()) {
+            mockExternalDatabase(entry.getKey());
+            IotDeviceMessage message = buildMessage();
+            IotDataSinkDatabaseConfig config = buildConfig(TABLE_NAME);
+            IotDatabaseDataRuleAction action = spyAction(config);
+
+            // 调用
+            action.execute(message, config);
+
+            // 断言：只借用一次外部连接，在同一连接上读取产品名、创建 Statement
+            verify(externalDataSource, times(1)).getConnection();
+            InOrder inOrder = inOrder(externalConnection, externalMetaData, externalStatement);
+            inOrder.verify(externalConnection).getMetaData();
+            inOrder.verify(externalMetaData).getDatabaseProductName();
+            inOrder.verify(externalConnection).prepareStatement(buildInsertSql(TABLE_NAME, entry.getValue()));
+            // 断言：原 6 个参数的值、绑定方式、顺序不变
+            inOrder.verify(externalStatement).setString(1, "message_01");
+            inOrder.verify(externalStatement).setObject(2, 1L);
+            inOrder.verify(externalStatement).setObject(3, 2L);
+            inOrder.verify(externalStatement).setString(4, "thing.property.post");
+            inOrder.verify(externalStatement).setObject(5, message.getReportTime());
+            inOrder.verify(externalStatement).setString(6, JsonUtils.toJsonString(message));
+            inOrder.verify(externalStatement).executeUpdate();
+            // 断言：Statement、连接由 JdbcTemplate 关闭、归还
+            inOrder.verify(externalStatement).close();
+            inOrder.verify(externalConnection).close();
+            verify(externalConnection, times(1)).prepareStatement(anyString());
+        }
+    }
+
+    @Test
+    public void testExecute_getMetaDataFailure() throws Exception {
+        // 准备参数：读取外部数据库元数据失败
+        mockExternalDatabase("Oracle");
+        SQLException metaDataException = new SQLException("metadata failure");
+        when(externalConnection.getMetaData()).thenThrow(metaDataException);
+        IotDataSinkDatabaseConfig config = buildConfig(TABLE_NAME);
+        IotDatabaseDataRuleAction action = spyAction(config);
+
+        // 调用，并断言异常：不吞掉原始异常
+        DataAccessException exception = assertThrows(DataAccessException.class,
+                () -> action.execute(buildMessage(), config));
+        assertSame(metaDataException, exception.getCause());
+        // 断言：未创建 Statement，连接被归还
+        verify(externalConnection, never()).prepareStatement(anyString());
+        verify(externalConnection, times(1)).close();
+        verify(externalDataSource, times(1)).getConnection();
+    }
+
+    @Test
+    public void testExecute_bindFailure() throws Exception {
+        // 准备参数：绑定参数失败
+        mockExternalDatabase("Microsoft SQL Server");
+        SQLException bindException = new SQLException("bind failure");
+        doThrow(bindException).when(externalStatement).setString(1, "message_01");
+        IotDataSinkDatabaseConfig config = buildConfig(TABLE_NAME);
+        IotDatabaseDataRuleAction action = spyAction(config);
+
+        // 调用，并断言异常：不吞掉原始异常
+        DataAccessException exception = assertThrows(DataAccessException.class,
+                () -> action.execute(buildMessage(), config));
+        assertSame(bindException, exception.getCause());
+        // 断言：未执行 SQL，已创建的 Statement 被关闭，连接被归还
+        verify(externalStatement, never()).executeUpdate();
+        verify(externalStatement, times(1)).close();
+        verify(externalConnection, times(1)).close();
+        verify(externalDataSource, times(1)).getConnection();
+    }
+
+    @Test
+    public void testExecute_executeUpdateFailure() throws Exception {
+        // 准备参数：执行 SQL 失败
+        mockExternalDatabase("PostgreSQL");
+        SQLException executeException = new SQLException("execute failure");
+        when(externalStatement.executeUpdate()).thenThrow(executeException);
+        IotDataSinkDatabaseConfig config = buildConfig(TABLE_NAME);
+        IotDatabaseDataRuleAction action = spyAction(config);
+
+        // 调用，并断言异常：不吞掉原始异常
+        DataAccessException exception = assertThrows(DataAccessException.class,
+                () -> action.execute(buildMessage(), config));
+        assertSame(executeException, exception.getCause());
+        // 断言：Statement 被关闭，连接被归还
+        verify(externalConnection).prepareStatement(buildInsertSql(TABLE_NAME, "NOW()"));
+        verify(externalStatement, times(1)).close();
+        verify(externalConnection, times(1)).close();
+        verify(externalDataSource, times(1)).getConnection();
+    }
+
+    @Test
     public void testCloseProducer_whenHikari() throws Exception {
         // 准备：底层是 HikariDataSource
         HikariDataSource hikari = mock(HikariDataSource.class);
@@ -102,6 +219,35 @@ class IotDatabaseDataRuleActionTest {
         verifyNoInteractions(other);
     }
+    /**
+     * 模拟外部数据库：DataSource 每次返回同一个 Connection，产品名来自该 Connection 的元数据
+     */
+    private void mockExternalDatabase(String databaseProductName) throws SQLException {
+        externalDataSource = mock(DataSource.class);
+        externalConnection = mock(Connection.class);
+        externalMetaData = mock(DatabaseMetaData.class);
+        externalStatement = mock(PreparedStatement.class);
+        when(externalDataSource.getConnection()).thenReturn(externalConnection);
+        when(externalConnection.getMetaData()).thenReturn(externalMetaData);
+        when(externalMetaData.getDatabaseProductName()).thenReturn(databaseProductName);
+        when(externalConnection.prepareStatement(anyString())).thenReturn(externalStatement);
+        when(externalStatement.executeUpdate()).thenReturn(1);
+    }
+
+    /**
+     * 创建 Action，其 producer 为基于外部数据库的真实 {@link JdbcTemplate}
+     */
+    private IotDatabaseDataRuleAction spyAction(IotDataSinkDatabaseConfig config) throws Exception {
+        IotDatabaseDataRuleAction action = spy(databaseDataRuleAction);
+        doReturn(new JdbcTemplate(externalDataSource)).when(action).getProducer(same(config));
+        return action;
+    }
+
+    private static String buildInsertSql(String tableName, String now) {
+        return "INSERT INTO " + tableName
+                + " (id, device_id, tenant_id, method, report_time, data, create_time) VALUES (?, ?, ?, ?, ?, ?, " + now + ")";
+    }
+
     private static IotDataSinkDatabaseConfig buildConfig(String tableName) {
         IotDataSinkDatabaseConfig config = new IotDataSinkDatabaseConfig();
         config.setJdbcUrl("jdbc:mysql://127.0.0.1:3306/iot_data");
@@ -112,7 +258,7 @@ class IotDatabaseDataRuleActionTest {
     }
     private static IotDeviceMessage buildMessage() {
-        return new IotDeviceMessage().setId("message_01").setDeviceId(1L).setTenantId(1L)
+        return new IotDeviceMessage().setId("message_01").setDeviceId(1L).setTenantId(2L)
                 .setMethod("thing.property.post").setReportTime(LocalDateTime.now());
     }
@@ -38,7 +38,6 @@ public interface MesCalPlanMapper extends BaseMapperX&lt;MesCalPlanDO&gt; {
     default List<MesCalPlanDO> selectListByStatusAndDay(Integer status, LocalDate day) {
         return selectList(new LambdaQueryWrapperX<MesCalPlanDO>()
                 .eq(MesCalPlanDO::getStatus, status)
-                // TODO @AI：dayend 获取下？应该有方法，这样可以 le，更统一点；
                 .lt(MesCalPlanDO::getStartDate, day.plusDays(1).atStartOfDay())
                 .ge(MesCalPlanDO::getEndDate, day.atStartOfDay())
                 .orderByAsc(MesCalPlanDO::getId));