Conflicts:

	iot-platform/src/main/java/com/iot/platform/task/VehicleSyncTask.java

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

@@ -9,13 +9,9 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 
-import java.io.ByteArrayOutputStream;
 import java.sql.*;
 import java.util.*;
 import java.util.concurrent.*;
-import java.util.zip.*;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
 
 @Service("tdengineService")
 public class TDengineService {
@@ -34,15 +30,15 @@ public class TDengineService {
     private HikariDataSource dataSource;
     private volatile boolean dataSourceInitialized = false;
 
-    // === 新增：缓存超级表结构 (key = dbName.stableName) ===
+    // === 缓存超级表结构 (key = dbName.stableName) ===
     private final Map<String, Set<String>> stableColumnCache = new ConcurrentHashMap<>();
     private static final int MAX_CACHE_SIZE = 1000;
 
-    // JSON 列名，用于存储所有动态字段
-    private static final String JSON_COLUMN_NAME = "ext_data";
+    // 允许的列名字符
+    private static final String ALLOWED_COLUMNS = "^[a-zA-Z_][a-zA-Z0-9_]*$";
 
-    // ObjectMapper 线程安全，可复用
-    private static final ObjectMapper objectMapper = new ObjectMapper();
+    // TDengine 超级表最大列数限制
+    private static final int MAX_COLUMNS_PER_STABLE = 4096;
 
     private synchronized void initDataSource() {
         if (dataSourceInitialized) {
@@ -72,7 +68,7 @@ public class TDengineService {
     }
 
     /**
-     * 手动获取连接
+     * 获取连接
      */
     public Connection getConnection() throws SQLException {
         if (!dataSourceInitialized) {
@@ -86,7 +82,7 @@ public class TDengineService {
     }
 
     /**
-     * 手动关闭连接
+     * 关闭连接
      */
     public void closeConnection(Connection conn) {
         if (conn != null) {
@@ -111,7 +107,7 @@ public class TDengineService {
     }
 
     private boolean isValidFieldName(String name) {
-        return name != null && name.matches("^[a-zA-Z0-9_]+$");
+        return name != null && name.matches(ALLOWED_COLUMNS);
     }
 
     // === 缓存工具方法 ===
@@ -129,15 +125,12 @@ public class TDengineService {
             return cached;
         }
 
-        // 缓存未命中，查 DB
         Set<String> columns = loadStableColumnsFromDB(dbName, stableName);
-        if (!columns.isEmpty()) {
-            if (stableColumnCache.size() >= MAX_CACHE_SIZE) {
-                stableColumnCache.clear();
-                log.warn("TDengine 超级表缓存已达上限({})，已清空", MAX_CACHE_SIZE);
-            }
-            stableColumnCache.put(key, columns);
+        if (stableColumnCache.size() >= MAX_CACHE_SIZE) {
+            stableColumnCache.clear();
+            log.warn("TDengine 超级表缓存已达上限({})，已清空", MAX_CACHE_SIZE);
         }
+        stableColumnCache.put(key, columns);
         return columns;
     }
 
@@ -151,7 +144,7 @@ public class TDengineService {
         try {
             conn = getConnection();
             stmt = conn.createStatement();
-            stmt.setQueryTimeout(5); // ⭐ 防止无限等待
+            stmt.setQueryTimeout(5);
             rs = stmt.executeQuery(sql);
             while (rs.next()) {
                 columns.add(rs.getString(1));
@@ -169,7 +162,7 @@ public class TDengineService {
     }
 
     // ==========================================
-    // 初始化表结构
+    // 初始化表结构（按列存储，无 ext_data）
     // ==========================================
     private void initTableStructure(String dbName, String supertablename, String table, Set<String> fieldNames) throws SQLException {
         Connection conn = null;
@@ -181,10 +174,9 @@ public class TDengineService {
 
             stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS " + wrapName(dbName));
 
-            // 创建超级表：固定 ts + surfacename + ext_data(VARCHAR)
-            // 注：TDengine 2.x JSON 类型只能用于 TAGS，不能用于普通列
+            // 创建超级表：固定 ts + surfacename，无 ext_data 列
             String stableSql = String.format(
-                    "CREATE STABLE IF NOT EXISTS %s.%s (ts TIMESTAMP, surfacename VARCHAR(64), ext_data VARCHAR(8192)) TAGS (location BINARY(64))",
+                    "CREATE STABLE IF NOT EXISTS %s.%s (ts TIMESTAMP, surfacename VARCHAR(64)) TAGS (location BINARY(64))",
                     wrapName(dbName),
                     wrapName(supertablename)
             );
@@ -192,16 +184,12 @@ public class TDengineService {
 
             // 更新缓存：固定列
             String key = getStableKey(dbName, supertablename);
-            if (stableColumnCache.size() >= MAX_CACHE_SIZE) {
-                stableColumnCache.clear();
-                log.warn("TDengine 超级表缓存已达上限({})，已清空", MAX_CACHE_SIZE);
-            }
             Set<String> fixedCols = new HashSet<>();
             fixedCols.add("ts");
             fixedCols.add("surfacename");
-            fixedCols.add("ext_data");
             stableColumnCache.put(key, fixedCols);
 
+            // 创建子表
             String tableSql = String.format(
                     "CREATE TABLE IF NOT EXISTS %s.%s USING %s.%s TAGS ('%s')",
                     wrapName(dbName),
@@ -219,17 +207,17 @@ public class TDengineService {
     }
 
     // ==========================================
-    // 批量插入
+    // 批量插入（按列存储）
     // ==========================================
     public void insertBatch(String dbName, String table, List<Map<String, Object>> dataList) throws SQLException {
         if (dataList == null || dataList.isEmpty()) return;
 
         String supertablename = table.contains("_") ? table.substring(0, table.lastIndexOf('_')) : table;
 
-        // 初始化表结构（固定为 ts + surfacename + ext_data）
+        // 初始化表结构
         initTableStructure(dbName, supertablename, table, Collections.emptySet());
 
-        // 分批处理，每批最多 50 条记录（避免 SQL 过长）
+        // 分批处理，每批最多 50 条记录
         int batchSize = 50;
         int totalBatches = (dataList.size() + batchSize - 1) / batchSize;
 
@@ -248,23 +236,65 @@ public class TDengineService {
      * 内部方法：插入一批数据
      */
     private void insertBatchInternal(String dbName, String supertablename, String table, List<Map<String, Object>> dataList) throws SQLException {
-        // 确保表存在（可能有竞态条件）
         ensureTableExists(dbName, supertablename, table);
 
+        // 收集所有动态列名及其数据类型
+        Map<String, String> columnTypes = new LinkedHashMap<>();
+        for (Map<String, Object> data : dataList) {
+            if (data == null) continue;
+            for (Map.Entry<String, Object> entry : data.entrySet()) {
+                String key = entry.getKey();
+                if (isValidFieldName(key) && !isReservedColumn(key) && !columnTypes.containsKey(key)) {
+                    columnTypes.put(key, getValueType(entry.getValue()));
+                }
+            }
+        }
+
+        // 获取当前列数
+        Set<String> existingColumns = getStableColumns(dbName, supertablename);
+        int totalColumns = existingColumns.size() + columnTypes.size();
+
+        // 检查列数是否超限
+        if (totalColumns > MAX_COLUMNS_PER_STABLE) {
+            log.error("超级表总列数({})超过限制({})，跳过本次插入 | 表: {}.{}",
+                    totalColumns, MAX_COLUMNS_PER_STABLE, dbName, table);
+            return;
+        }
+
+        // 确保所有列都存在（自动 ALTER 添加新列）
+        ensureColumnsExist(dbName, supertablename, columnTypes);
+
+        // 重新获取列数以确保一致性
+        existingColumns = getStableColumns(dbName, supertablename);
+        if (existingColumns.size() >= MAX_COLUMNS_PER_STABLE) {
+            log.error("超级表列数已达上限({})，停止插入数据 | 表: {}.{}",
+                    MAX_COLUMNS_PER_STABLE, dbName, table);
+            return;
+        }
+
+        // 构建 SQL
         StringBuilder sqlBuilder = new StringBuilder();
         sqlBuilder.append("INSERT INTO ").append(wrapName(dbName)).append(".").append(wrapName(table))
-                .append(" (ts, surfacename, ext_data) VALUES ");
+                .append(" (ts, surfacename");
+
+        for (String col : columnTypes.keySet()) {
+            sqlBuilder.append(", ").append(wrapName(col));
+        }
+        sqlBuilder.append(") VALUES ");
 
         boolean hasData = false;
         for (Map<String, Object> data : dataList) {
             if (data == null) continue;
 
-            // 将动态字段转为 JSON 并 GZIP 压缩
-            String extJson = buildDynamicJson(data);
-            String compressed = compressToBase64(extJson);
+            sqlBuilder.append("(NOW(), '").append(escapeValue(supertablename)).append("'");
 
-            sqlBuilder.append("(NOW(), '").append(escapeValue(supertablename)).append("', '")
-                    .append(escapeValue(compressed)).append("'),");
+            for (Map.Entry<String, String> entry : columnTypes.entrySet()) {
+                String col = entry.getKey();
+                String colType = entry.getValue();
+                Object value = data.get(col);
+                sqlBuilder.append(", ").append(formatValue(value, col, colType));
+            }
+            sqlBuilder.append("),");
             hasData = true;
         }
 
@@ -281,7 +311,6 @@ public class TDengineService {
             stmt.setQueryTimeout(30);
             stmt.executeUpdate(finalSql);
         } catch (SQLException e) {
-            // 表不存在时尝试重建表后重试
             if (e.getMessage().contains("Table does not exist")) {
                 log.warn("表不存在，重建表: {}", table);
                 initTableStructure(dbName, supertablename, table, Collections.emptySet());
@@ -296,20 +325,44 @@ public class TDengineService {
     }
 
     /**
-     * 重试插入（表重建后）
+     * 重试插入
      */
     private void insertBatchRetry(String dbName, String supertablename, String table, List<Map<String, Object>> dataList) throws SQLException {
+        Map<String, String> columnTypes = new LinkedHashMap<>();
+        for (Map<String, Object> data : dataList) {
+            if (data == null) continue;
+            for (Map.Entry<String, Object> entry : data.entrySet()) {
+                String key = entry.getKey();
+                if (isValidFieldName(key) && !isReservedColumn(key) && !columnTypes.containsKey(key)) {
+                    columnTypes.put(key, getValueType(entry.getValue()));
+                }
+            }
+        }
+
+        ensureColumnsExist(dbName, supertablename, columnTypes);
+
         StringBuilder sqlBuilder = new StringBuilder();
         sqlBuilder.append("INSERT INTO ").append(wrapName(dbName)).append(".").append(wrapName(table))
-                .append(" (ts, surfacename, ext_data) VALUES ");
+                .append(" (ts, surfacename");
+
+        for (String col : columnTypes.keySet()) {
+            sqlBuilder.append(", ").append(wrapName(col));
+        }
+        sqlBuilder.append(") VALUES ");
 
         boolean hasData = false;
         for (Map<String, Object> data : dataList) {
             if (data == null) continue;
-            String extJson = buildDynamicJson(data);
-            String compressed = compressToBase64(extJson);
-            sqlBuilder.append("(NOW(), '").append(escapeValue(supertablename)).append("', '")
-                    .append(escapeValue(compressed)).append("'),");
+
+            sqlBuilder.append("(NOW(), '").append(escapeValue(supertablename)).append("'");
+
+            for (Map.Entry<String, String> entry : columnTypes.entrySet()) {
+                String col = entry.getKey();
+                String colType = entry.getValue();
+                Object value = data.get(col);
+                sqlBuilder.append(", ").append(formatValue(value, col, colType));
+            }
+            sqlBuilder.append("),");
             hasData = true;
         }
 
@@ -334,7 +387,166 @@ public class TDengineService {
     }
 
     /**
-     * 确保表存在（检查+创建）
+     * 判断是否为保留列（ts, surfacename）
+     */
+    private boolean isReservedColumn(String columnName) {
+        return "ts".equalsIgnoreCase(columnName) || "surfacename".equalsIgnoreCase(columnName);
+    }
+
+    /**
+     * 判断值类型，返回 TDengine 对应的数据类型
+     */
+    private String getValueType(Object value) {
+        if (value == null) {
+            return "VARCHAR";
+        }
+        if (value instanceof Boolean) {
+            return "BOOL";
+        }
+        if (value instanceof Integer || value instanceof Long) {
+            return "BIGINT";
+        }
+        if (value instanceof Float || value instanceof Double) {
+            return "DOUBLE";
+        }
+        // 时间类型（Date, Timestamp, LocalDateTime 等）
+        if (value instanceof java.util.Date || value instanceof java.sql.Timestamp
+                || value instanceof java.time.LocalDateTime || value instanceof java.time.Instant) {
+            return "TIMESTAMP";
+        }
+        return "VARCHAR";
+    }
+
+    /**
+     * 根据 TDengine 列类型获取创建列的 SQL 类型
+     */
+    private String getColumnTypeForDDL(String tdType, String columnName) {
+        // 特殊列名使用较长长度
+        if ("device_id".equalsIgnoreCase(columnName)) {
+            return "VARCHAR(100)";
+        }
+        switch (tdType) {
+            case "BOOL":
+            case "BIGINT":
+            case "DOUBLE":
+            case "TIMESTAMP":
+                return tdType;
+            case "VARCHAR":
+            default:
+                return "VARCHAR(38)";
+        }
+    }
+
+    /**
+     * 格式化值
+     */
+    private String formatValue(Object value, String columnName, String columnType) {
+        if (value == null) {
+            return "NULL";
+        }
+        String strValue = value.toString();
+
+        // 数字类型不做长度校验
+        if ("DOUBLE".equals(columnType) || "BIGINT".equals(columnType) || "BOOL".equals(columnType)) {
+            if (value instanceof Boolean) {
+                return (Boolean) value ? "1" : "0";
+            }
+            return strValue;
+        }
+
+        // TIMESTAMP 类型处理
+        if ("TIMESTAMP".equals(columnType)) {
+            if (value instanceof java.util.Date) {
+                return String.valueOf(((java.util.Date) value).getTime());
+            }
+            if (value instanceof java.sql.Timestamp) {
+                return String.valueOf(((java.sql.Timestamp) value).getTime());
+            }
+            if (value instanceof java.time.LocalDateTime) {
+                return String.valueOf(((java.time.LocalDateTime) value).toInstant(java.time.ZoneOffset.of("+8")).toEpochMilli());
+            }
+            if (value instanceof java.time.Instant) {
+                return String.valueOf(((java.time.Instant) value).toEpochMilli());
+            }
+        }
+
+        // 字符串类型校验长度
+        int maxLen = "device_id".equalsIgnoreCase(columnName) ? 100 : 38;
+        if (strValue.length() > maxLen) {
+            log.warn("字段值超长，跳过存储 | 列: {} | 值长度: {} | 最大: {} | 前{}字符: {}",
+                    columnName, strValue.length(), maxLen, maxLen, strValue.substring(0, maxLen));
+            return "NULL";
+        }
+        return "'" + escapeValue(strValue) + "'";
+    }
+
+    /**
+     * 确保列存在，如有新列则 ALTER 添加（根据类型添加对应列）
+     */
+    private void ensureColumnsExist(String dbName, String supertablename, Map<String, String> columnTypes) throws SQLException {
+        Set<String> existingColumns = getStableColumns(dbName, supertablename);
+
+        // 检查是否已达到列数上限
+        if (existingColumns.size() >= MAX_COLUMNS_PER_STABLE) {
+            log.error("超级表列数已达上限({})，停止添加新列 | 表: {}.{}", MAX_COLUMNS_PER_STABLE, dbName, supertablename);
+            return;
+        }
+
+        Connection conn = null;
+        Statement stmt = null;
+        try {
+            conn = getConnection();
+            stmt = conn.createStatement();
+            stmt.setQueryTimeout(10);
+
+            for (Map.Entry<String, String> entry : columnTypes.entrySet()) {
+                String col = entry.getKey();
+                String colType = entry.getValue();
+
+                // 再次检查列数上限
+                if (existingColumns.size() >= MAX_COLUMNS_PER_STABLE) {
+                    log.error("超级表列数已达上限({})，跳过添加列 | 列: {} | 表: {}.{}",
+                            MAX_COLUMNS_PER_STABLE, col, dbName, supertablename);
+                    continue;
+                }
+
+                if (!existingColumns.contains(col)) {
+                    String ddlType = getColumnTypeForDDL(colType, col);
+                    String alterSql = String.format(
+                            "ALTER STABLE %s.%s ADD COLUMN %s %s",
+                            wrapName(dbName),
+                            wrapName(supertablename),
+                            wrapName(col),
+                            ddlType
+                    );
+                    try {
+                        stmt.executeUpdate(alterSql);
+                        log.info("超级表添加列: {}.{} -> {} ({})", dbName, supertablename, col, ddlType);
+                        existingColumns.add(col);
+                    } catch (SQLException e) {
+                        if (e.getMessage().contains("already exists") || e.getMessage().contains("duplicate")) {
+                            existingColumns.add(col);
+                        } else if (e.getMessage().contains("too many columns")) {
+                            log.error("超级表列数超限，无法添加列 | 列: {} | 表: {}.{} | 错误: {}",
+                                    col, dbName, supertablename, e.getMessage());
+                        } else {
+                            throw e;
+                        }
+                    }
+
+                    // 更新缓存
+                    String key = getStableKey(dbName, supertablename);
+                    stableColumnCache.put(key, existingColumns);
+                }
+            }
+        } finally {
+            if (stmt != null) try { stmt.close(); } catch (SQLException ignored) {}
+            closeConnection(conn);
+        }
+    }
+
+    /**
+     * 确保表存在
      */
     private void ensureTableExists(String dbName, String supertablename, String table) throws SQLException {
         Connection conn = null;
@@ -345,7 +557,7 @@ public class TDengineService {
             stmt = conn.createStatement();
             stmt.setQueryTimeout(5);
 
-            // 检查超级表是否存在（使用 information_schema）
+            // 检查超级表是否存在
             String checkStableSql = String.format(
                     "SELECT * FROM information_schema.ins_tables WHERE table_name = '%s' AND db_name = '%s'",
                     escapeValue(supertablename), escapeValue(dbName));
@@ -390,47 +602,6 @@ public class TDengineService {
         }
     }
 
-    /**
-     * 将动态字段构建为 JSON 字符串（排除 topic、ts、surfacename）
-     */
-    private String buildDynamicJson(Map<String, Object> data) {
-        if (data == null) return "{}";
-        Map<String, Object> dynamic = new LinkedHashMap<>();
-        for (Map.Entry<String, Object> entry : data.entrySet()) {
-            String key = entry.getKey();
-            if ("topic".equalsIgnoreCase(key) || "ts".equalsIgnoreCase(key)
-                    || "surfacename".equalsIgnoreCase(key)) {
-                continue;
-            }
-            Object value = entry.getValue();
-            if (value != null && !value.toString().trim().isEmpty()) {
-                dynamic.put(key, value);
-            }
-        }
-        try {
-            return objectMapper.writeValueAsString(dynamic);
-        } catch (Exception e) {
-            throw new IllegalStateException("JSON 序列化失败，数据可能丢失: " + dynamic, e);
-        }
-    }
-
-    /**
-     * GZIP 压缩字符串
-     */
-    private String compressToBase64(String data) {
-        if (data == null || data.isEmpty()) return "";
-        try {
-            ByteArrayOutputStream bos = new ByteArrayOutputStream();
-            try (GZIPOutputStream gzip = new GZIPOutputStream(bos)) {
-                gzip.write(data.getBytes("UTF-8"));
-            }
-            byte[] compressed = Base64.getEncoder().encode(bos.toByteArray());
-            return new String(compressed);
-        } catch (Exception e) {
-            throw new IllegalStateException("GZIP 压缩失败: " + e.getMessage(), e);
-        }
-    }
-
     @Deprecated
     public void addToBatch(String dbName, String table, Map<String, Object> dataMap) throws SQLException {
         List<Map<String, Object>> list = new ArrayList<>();
@@ -438,9 +609,9 @@ public class TDengineService {
         insertBatch(dbName, table, list);
     }
 
-    // ==========================================
-    // 可选：清除缓存（用于运维或测试）
-    // ==========================================
+    /**
+     * 清除缓存
+     */
     public void clearStableColumnCache() {
         stableColumnCache.clear();
         log.info("清除了 TDengine 超级表结构缓存");
@@ -448,7 +619,6 @@ public class TDengineService {
 
     public void close() {
         log.info("关闭 TDengine 服务...");
-        // batchExecutor 由 Spring 管理生命周期，不在这里 shutdown
         if (dataSource != null) {
             dataSource.close();
         }

iot-platform/src/test/java/com/iot/platform/service/TDengineServiceTest.java

@@ -83,69 +83,225 @@ class TDengineServiceTest {
     }
 
     @Test
-    @DisplayName("buildDynamicJson: 应排除 topic/ts/surfacename")
-    void buildDynamicJson_excludesReservedKeys() throws Exception {
-        Method method = TDengineService.class.getDeclaredMethod("buildDynamicJson", Map.class);
+    @DisplayName("getValueType: Boolean 返回 BOOL")
+    void getValueType_boolean_returnsBool() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
         method.setAccessible(true);
 
-        Map<String, Object> data = new LinkedHashMap<>();
-        data.put("topic", "test/topic");
-        data.put("ts", "1234567890");
-        data.put("surfacename", "s1");
-        data.put("temperature", "25.5");
-        data.put("humidity", "60");
+        assertThat(method.invoke(tdengineService, Boolean.TRUE)).isEqualTo("BOOL");
+        assertThat(method.invoke(tdengineService, Boolean.FALSE)).isEqualTo("BOOL");
+    }
+
+    @Test
+    @DisplayName("getValueType: Integer 返回 BIGINT")
+    void getValueType_integer_returnsBigint() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
+        method.setAccessible(true);
 
-        String result = (String) method.invoke(tdengineService, data);
-        assertThat(result).contains("temperature", "humidity");
-        assertThat(result).doesNotContain("topic", "ts", "surfacename");
+        assertThat(method.invoke(tdengineService, 123)).isEqualTo("BIGINT");
+        assertThat(method.invoke(tdengineService, Integer.valueOf(456))).isEqualTo("BIGINT");
     }
 
     @Test
-    @DisplayName("buildDynamicJson: 空值和空字符串应被过滤")
-    void buildDynamicJson_filtersEmptyValues() throws Exception {
-        Method method = TDengineService.class.getDeclaredMethod("buildDynamicJson", Map.class);
+    @DisplayName("getValueType: Long 返回 BIGINT")
+    void getValueType_long_returnsBigint() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
         method.setAccessible(true);
 
-        Map<String, Object> data = new LinkedHashMap<>();
-        data.put("valid", "value");
-        data.put("empty", "");
-        data.put("nullVal", null);
+        assertThat(method.invoke(tdengineService, 123L)).isEqualTo("BIGINT");
+    }
+
+    @Test
+    @DisplayName("getValueType: Float 返回 DOUBLE")
+    void getValueType_float_returnsDouble() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, 12.34f)).isEqualTo("DOUBLE");
+    }
+
+    @Test
+    @DisplayName("getValueType: Double 返回 DOUBLE")
+    void getValueType_double_returnsDouble() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
+        method.setAccessible(true);
 
-        String result = (String) method.invoke(tdengineService, data);
-        assertThat(result).contains("valid");
-        assertThat(result).doesNotContain("empty", "nullVal");
+        assertThat(method.invoke(tdengineService, 56.78)).isEqualTo("DOUBLE");
+        assertThat(method.invoke(tdengineService, Double.valueOf(99.99))).isEqualTo("DOUBLE");
     }
 
     @Test
-    @DisplayName("buildDynamicJson: null 输入应返回空 JSON")
-    void buildDynamicJson_nullInput_returnsEmptyJson() throws Exception {
-        Method method = TDengineService.class.getDeclaredMethod("buildDynamicJson", Map.class);
+    @DisplayName("getValueType: java.util.Date 返回 TIMESTAMP")
+    void getValueType_date_returnsTimestamp() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
         method.setAccessible(true);
 
-        assertThat(method.invoke(tdengineService, (Map<String, Object>) null)).isEqualTo("{}");
+        assertThat(method.invoke(tdengineService, new java.util.Date())).isEqualTo("TIMESTAMP");
     }
 
     @Test
-    @DisplayName("compressToBase64: 应压缩并返回非空字符串")
-    void compressToBase64_compressesData() throws Exception {
-        Method method = TDengineService.class.getDeclaredMethod("compressToBase64", String.class);
+    @DisplayName("getValueType: java.sql.Timestamp 返回 TIMESTAMP")
+    void getValueType_timestamp_returnsTimestamp() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
         method.setAccessible(true);
 
-        String input = "{\"temperature\":25.5,\"humidity\":60}";
-        String result = (String) method.invoke(tdengineService, input);
+        assertThat(method.invoke(tdengineService, new java.sql.Timestamp(System.currentTimeMillis()))).isEqualTo("TIMESTAMP");
+    }
+
+    @Test
+    @DisplayName("getValueType: java.time.LocalDateTime 返回 TIMESTAMP")
+    void getValueType_localDateTime_returnsTimestamp() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
+        method.setAccessible(true);
 
-        assertThat(result).isNotEmpty();
-        assertThat(result).isNotEqualTo(input); // should be compressed
+        assertThat(method.invoke(tdengineService, java.time.LocalDateTime.now())).isEqualTo("TIMESTAMP");
     }
 
     @Test
-    @DisplayName("compressToBase64: null 或空字符串应返回空")
-    void compressToBase64_nullOrEmpty_returnsEmpty() throws Exception {
-        Method method = TDengineService.class.getDeclaredMethod("compressToBase64", String.class);
+    @DisplayName("getValueType: java.time.Instant 返回 TIMESTAMP")
+    void getValueType_instant_returnsTimestamp() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
         method.setAccessible(true);
 
-        assertThat(method.invoke(tdengineService, (String) null)).isEqualTo("");
-        assertThat(method.invoke(tdengineService, "")).isEqualTo("");
+        assertThat(method.invoke(tdengineService, java.time.Instant.now())).isEqualTo("TIMESTAMP");
+    }
+
+    @Test
+    @DisplayName("getValueType: String 返回 VARCHAR")
+    void getValueType_string_returnsVarchar() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, "test")).isEqualTo("VARCHAR");
+        assertThat(method.invoke(tdengineService, "")).isEqualTo("VARCHAR");
+    }
+
+    @Test
+    @DisplayName("getValueType: null 返回 VARCHAR")
+    void getValueType_null_returnsVarchar() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getValueType", Object.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, null)).isEqualTo("VARCHAR");
+    }
+
+    @Test
+    @DisplayName("getColumnTypeForDDL: BOOL/BIGINT/DOUBLE/TIMESTAMP 直接返回")
+    void getColumnTypeForDDL_directTypes() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getColumnTypeForDDL", String.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, "BOOL")).isEqualTo("BOOL");
+        assertThat(method.invoke(tdengineService, "BIGINT")).isEqualTo("BIGINT");
+        assertThat(method.invoke(tdengineService, "DOUBLE")).isEqualTo("DOUBLE");
+        assertThat(method.invoke(tdengineService, "TIMESTAMP")).isEqualTo("TIMESTAMP");
+    }
+
+    @Test
+    @DisplayName("getColumnTypeForDDL: VARCHAR 返回 VARCHAR(32)")
+    void getColumnTypeForDDL_varchar_returnsWithLength() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("getColumnTypeForDDL", String.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, "VARCHAR")).isEqualTo("VARCHAR(32)");
+    }
+
+    @Test
+    @DisplayName("formatValue: null 返回 NULL")
+    void formatValue_null_returnsNull() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("formatValue", Object.class, String.class, String.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, null, "col", "VARCHAR")).isEqualTo("NULL");
+    }
+
+    @Test
+    @DisplayName("formatValue: DOUBLE 类型直接返回数字字符串")
+    void formatValue_double_returnsNumberString() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("formatValue", Object.class, String.class, String.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, 12.34, "col", "DOUBLE")).isEqualTo("12.34");
+    }
+
+    @Test
+    @DisplayName("formatValue: BIGINT 类型直接返回数字字符串")
+    void formatValue_bigint_returnsNumberString() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("formatValue", Object.class, String.class, String.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, 123L, "col", "BIGINT")).isEqualTo("123");
+    }
+
+    @Test
+    @DisplayName("formatValue: BOOL 类型 true 返回 1")
+    void formatValue_bool_true_returns1() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("formatValue", Object.class, String.class, String.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, true, "col", "BOOL")).isEqualTo("1");
+        assertThat(method.invoke(tdengineService, false, "col", "BOOL")).isEqualTo("0");
+    }
+
+    @Test
+    @DisplayName("formatValue: VARCHAR 类型超过32字符返回 NULL")
+    void formatValue_varchar_exceedsLength_returnsNull() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("formatValue", Object.class, String.class, String.class);
+        method.setAccessible(true);
+
+        String longStr = "12345678901234567890123456789012345"; // 35 chars
+        assertThat(method.invoke(tdengineService, longStr, "col", "VARCHAR")).isEqualTo("NULL");
+    }
+
+    @Test
+    @DisplayName("formatValue: VARCHAR 类型32字符内正常返回")
+    void formatValue_varchar_withinLength_returnsQuoted() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("formatValue", Object.class, String.class, String.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, "test", "col", "VARCHAR")).isEqualTo("'test'");
+    }
+
+    @Test
+    @DisplayName("formatValue: VARCHAR 类型包含单引号应转义")
+    void formatValue_varchar_withQuote_escapes() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("formatValue", Object.class, String.class, String.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, "it's", "col", "VARCHAR")).isEqualTo("'it''s'");
+    }
+
+    @Test
+    @DisplayName("formatValue: TIMESTAMP 类型 java.util.Date 转为毫秒")
+    void formatValue_timestamp_date_returnsMillis() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("formatValue", Object.class, String.class, String.class);
+        method.setAccessible(true);
+
+        java.util.Date date = new java.util.Date(1234567890000L);
+        String result = (String) method.invoke(tdengineService, date, "col", "TIMESTAMP");
+        assertThat(result).isEqualTo("1234567890000");
+    }
+
+    @Test
+    @DisplayName("isReservedColumn: ts 和 surfacename 返回 true")
+    void isReservedColumn_reservedNames() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("isReservedColumn", String.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, "ts")).isEqualTo(true);
+        assertThat(method.invoke(tdengineService, "TS")).isEqualTo(true);
+        assertThat(method.invoke(tdengineService, "surfacename")).isEqualTo(true);
+        assertThat(method.invoke(tdengineService, "Surfacename")).isEqualTo(true);
+    }
+
+    @Test
+    @DisplayName("isReservedColumn: 其他列名返回 false")
+    void isReservedColumn_otherNames() throws Exception {
+        Method method = TDengineService.class.getDeclaredMethod("isReservedColumn", String.class);
+        method.setAccessible(true);
+
+        assertThat(method.invoke(tdengineService, "temperature")).isEqualTo(false);
+        assertThat(method.invoke(tdengineService, "voltage")).isEqualTo(false);
     }
 
     @Test
@@ -161,8 +317,6 @@ class TDengineServiceTest {
         Map<String, Object> data = new HashMap<>();
         data.put("key", "value");
 
-        // addToBatch 现在声明 throws SQLException，尝试连接 TDengine 会失败
-        // 本地环境缺少 TDengine 驱动时会抛出各种异常（UnsatisfiedLinkError / NoClassDefFoundError / IllegalStateException）
         assertThatThrownBy(() -> tdengineService.addToBatch("db", "table", data))
                 .isInstanceOf(Throwable.class);
     }
@@ -171,21 +325,18 @@ class TDengineServiceTest {
     @DisplayName("clearStableColumnCache: 应清空缓存不抛异常")
     void clearStableColumnCache_clearsWithoutError() {
         tdengineService.clearStableColumnCache();
-        // Should not throw
     }
 
     @Test
     @DisplayName("close: 应关闭数据源不抛异常")
     void close_closesDataSource() {
         tdengineService.close();
-        // Should not throw even if dataSource is null
     }
 
     @Test
     @DisplayName("closeConnection: null 连接应安全处理")
     void closeConnection_nullConnection_safe() {
         tdengineService.closeConnection(null);
-        // Should not throw
     }
 
     @Test
@@ -193,8 +344,7 @@ class TDengineServiceTest {
     void getConnection_initFails_throwsException() {
         when(tdengineConfig.getUrl()).thenReturn("jdbc:TAOS://invalid:6030/test");
 
-        // 本地环境缺少 TDengine 驱动时会抛出 NoClassDefFoundError / UnsatisfiedLinkError
         assertThatThrownBy(() -> tdengineService.getConnection())
                 .isInstanceOf(Throwable.class);
     }
-}
+}