Merge remote-tracking branch 'origin/mqy20260511'

iot-platform/src/main/java/com/iot/platform/service/TdEngineService.java

@@ -195,6 +195,44 @@ public class TdEngineService {
         return columns;
     }
 
+    // 缓存超级表列类型 (key = dbName.stableName, value = Map<列名, 类型>)
+    private final Map<String, Map<String, String>> stableColumnTypeCache = new ConcurrentHashMap<>();
+
+    /**
+     * 获取超级表列类型缓存
+     */
+    private Map<String, String> getStableColumnTypes(String dbName, String stableName) {
+        String key = getStableKey(dbName, stableName);
+        Map<String, String> cached = stableColumnTypeCache.get(key);
+        if (cached != null) {
+            return cached;
+        }
+        return stableColumnTypeCache.computeIfAbsent(key, k -> loadStableColumnTypesFromDB(dbName, stableName));
+    }
+
+    /**
+     * 从数据库加载列类型
+     */
+    private Map<String, String> loadStableColumnTypesFromDB(String dbName, String stableName) {
+        Map<String, String> columnTypes = new HashMap<>();
+        String sql = String.format("DESCRIBE %s.%s", wrapName(dbName), wrapName(stableName));
+        try (Connection conn = getConnection();
+             Statement stmt = conn.createStatement();
+             ResultSet rs = stmt.executeQuery(sql)) {
+            stmt.setQueryTimeout(5);
+            while (rs.next()) {
+                String colName = rs.getString(1);
+                String colType = rs.getString(2);
+                columnTypes.put(colName, colType);
+            }
+        } catch (SQLException e) {
+            if (!e.getMessage().contains("Table does not exist") && !e.getMessage().contains("invalid name")) {
+                log.debug("查询超级表列类型失败: {}.{} | {}", dbName, stableName, e.getMessage());
+            }
+        }
+        return columnTypes;
+    }
+
     // ==========================================
     // 初始化表结构（按列存储，无 ext_data）
     // ==========================================
@@ -273,8 +311,7 @@ public class TdEngineService {
             for (Map.Entry<String, Object> entry : data.entrySet()) {
                 String key = entry.getKey();
                 if (isValidFieldName(key) && !isReservedColumn(key) && !columnTypes.containsKey(key)) {
-                    // 统一使用 VARCHAR，避免类型推断导致的问题
-                    columnTypes.put(key, "VARCHAR");
+                    columnTypes.put(key, getValueType(entry.getValue()));
                 }
             }
         }
@@ -292,6 +329,9 @@ public class TdEngineService {
             return null;
         }
 
+        // 获取数据库中实际的列类型，用于格式化值
+        Map<String, String> dbColumnTypes = getStableColumnTypes(dbName, superTableName);
+
         StringBuilder sql = new StringBuilder();
         sql.append("INSERT INTO ").append(wrapName(dbName)).append(".").append(wrapName(table))
            .append(" (ts, surfacename");
@@ -311,7 +351,9 @@ public class TdEngineService {
             for (Map.Entry<String, String> entry : columnTypes.entrySet()) {
                 String col = entry.getKey();
                 Object value = data.get(col);
-                sql.append(", ").append(formatValue(value, col, entry.getValue()));
+                // 优先使用数据库中的列类型，其次使用数据类型推断
+                String colType = dbColumnTypes.getOrDefault(col, entry.getValue());
+                sql.append(", ").append(formatValue(value, col, colType));
             }
             sql.append("),");
             hasData = true;
@@ -609,6 +651,7 @@ public class TdEngineService {
      */
     public void clearStableColumnCache() {
         stableColumnCache.clear();
+        stableColumnTypeCache.clear();
         log.info("清除了 TdEngine 超级表结构缓存");
     }