Merge remote-tracking branch 'origin/mqy20260511'

iot-platform/pom.xml

@@ -198,7 +198,7 @@
                         </configuration>
                     </execution>
                 </executions>
-            </plugin>
+            </plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration><source>8</source><target>8</target></configuration></plugin>
         </plugins>
         <finalName>${project.artifactId}</finalName>
     </build>

iot-platform/src/main/java/com/iot/platform/mqtt/MqttChargeStationConsumer.java

@@ -73,6 +73,6 @@ public class MqttChargeStationConsumer extends AbstractDynamicMqttConsumer {
         String superTable = topicParts[3];
         LocalDate date = LocalDate.now();
         String tableName = superTable + "_" + date.getYear() + String.format("%02d", date.getMonthValue());
-        tdengineService.insertBatch(dbName, tableName, batchToInsert);
+        tdengineService.insertBatch(dbName, superTable,tableName, batchToInsert);
     }
 }

iot-platform/src/main/java/com/iot/platform/mqtt/MqttDynamicConsumer.java

@@ -110,7 +110,9 @@ public class MqttDynamicConsumer extends AbstractDynamicMqttConsumer {
         String tableName = ctx.superTable + "_" + date.getYear() + String.format("%02d", date.getMonthValue());
 
         List<Map<String, Object>> batch = Collections.singletonList(data);
-        tdengineService.insertBatch(ctx.dbName, tableName, batch);
+
+//        log.info("数据：({})",batch);
+        tdengineService.insertBatch(ctx.dbName, ctx.superTable,tableName, batch);
     }
 
     private void writeToRedis(MessageContext ctx) {

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

@@ -32,8 +32,8 @@ public class TdEngineService {
     private final Map<String, Set<String>> stableColumnCache = new ConcurrentHashMap<>();
     private static final int MAX_CACHE_SIZE = 1000;
 
-    // 允许的列名字符（仅 ASCII 字母、数字、下划线）
-    private static final String ALLOWED_COLUMNS = "^[a-zA-Z_][a-zA-Z0-9_]*$";
+    // 允许的列名字符（支持中文、ASCII字母、数字、下划线）
+    private static final String ALLOWED_COLUMNS = "^[a-zA-Z_一-龥][a-zA-Z0-9_一-龥]*$";
 
     // 允许的表名/数据库名字符（允许数字开头和连字符，如 UUID 格式的表名）
     private static final String ALLOWED_TABLE_NAME = "^[a-zA-Z0-9][a-zA-Z0-9_-]*$";
@@ -122,7 +122,7 @@ public class TdEngineService {
         if (name == null || name.isEmpty()) {
             return "`unknown`";
         }
-        return "`" + name.replace("`", "") + "`";
+        return "`" + name.replaceAll("`", "") + "`";
     }
 
     private boolean isValidFieldName(String name) {
@@ -133,6 +133,18 @@ public class TdEngineService {
         return name != null && name.matches(ALLOWED_TABLE_NAME);
     }
 
+    private boolean isNumeric(String str) {
+        if (str == null || str.isEmpty()) {
+            return false;
+        }
+        try {
+            Double.parseDouble(str);
+            return true;
+        } catch (NumberFormatException e) {
+            return false;
+        }
+    }
+
     // === 缓存工具方法 ===
     private String getStableKey(String dbName, String stableName) {
         return dbName + "." + stableName;
@@ -196,7 +208,7 @@ public class TdEngineService {
 
             // 创建超级表：固定 ts + surfacename，无 ext_data 列
             String stableSql = String.format(
-                    "CREATE STABLE IF NOT EXISTS %s.%s (ts TIMESTAMP, surfacename VARCHAR(64)) TAGS (location BINARY(64))",
+                    "CREATE STABLE IF NOT EXISTS %s.%s (ts TIMESTAMP, surfacename VARCHAR(64)) TAGS (location VARCHAR(255))",
                     wrapName(dbName),
                     wrapName(superTableName)
             );
@@ -225,7 +237,7 @@ public class TdEngineService {
     // ==========================================
     // 批量插入（按列存储）
     // ==========================================
-    public void insertBatch(String dbName, String table, List<Map<String, Object>> dataList)
+    public void insertBatch(String dbName, String superTable,String table, List<Map<String, Object>> dataList)
             throws SQLException {
 
         if (dataList == null || dataList.isEmpty()) {
@@ -233,12 +245,12 @@ public class TdEngineService {
             return;
         }
 
-        String superTableName = extractSuperTableName(table);
+//        String superTableName = extractSuperTableName(table);
 
         int batchSize = DEFAULT_BATCH_SIZE;
         for (int i = 0; i < dataList.size(); i += batchSize) {
             List<Map<String, Object>> batch = dataList.subList(i, Math.min(i + batchSize, dataList.size()));
-            insertBatchInternal(dbName, superTableName, table, batch);
+            insertBatchInternal(dbName, superTable, table, batch);
         }
 
         log.info("批量写入成功: {} | 条数: {}", table, dataList.size());
@@ -261,10 +273,13 @@ public class TdEngineService {
             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()));
+                    // 统一使用 VARCHAR，避免类型推断导致的问题
+                    columnTypes.put(key, "VARCHAR");
                 }
             }
         }
+
+        log.debug("收集到的列: {} | 共 {} 列", columnTypes.keySet(), columnTypes.size());
         return columnTypes;
     }
 
@@ -305,7 +320,7 @@ public class TdEngineService {
         if (!hasData) {
             return null;
         }
-
+        log.info("生成的 INSERT SQL | 列类型: {} | SQL 前100字符: {}", columnTypes, sql.toString().substring(0, Math.min(100, sql.toString().length())));
         sql.setLength(sql.length() - 1);
         return sql.toString();
     }
@@ -329,7 +344,9 @@ public class TdEngineService {
             return;
         }
 
+        log.info("收集到的列类型: {}", columnTypes);
         Set<String> existingColumns = getStableColumns(dbName, superTableName);
+        log.info("超级表已有列: {}", existingColumns);
         // 计算实际新增列数（扣除已有列的交集）
         int newColumns = 0;
         for (String col : columnTypes.keySet()) {
@@ -542,8 +559,8 @@ public class TdEngineService {
         }
 
         try (Connection conn = getConnection()) {
-            // 检查超级表是否存在 — 使用 PreparedStatement 防止 SQL 注入
-            String checkStableSql = "SELECT * FROM information_schema.ins_tables WHERE table_name = ? AND db_name = ?";
+            // 检查超级表是否存在（stable_name 字段标识超级表）
+            String checkStableSql = "SELECT * FROM information_schema.ins_tables WHERE stable_name = ? AND db_name = ?";
             try (PreparedStatement pStmt = conn.prepareStatement(checkStableSql)) {
                 pStmt.setString(1, superTableName);
                 pStmt.setString(2, dbName);
@@ -557,9 +574,9 @@ public class TdEngineService {
                 }
             }
 
-            // 检查子表是否存在 — 使用 PreparedStatement 防止 SQL 注入
+            // 检查子表是否存在
             String checkTableSql = "SELECT * FROM information_schema.ins_tables " +
-                    "WHERE table_name = ? AND db_name = ? AND table_type = 'CHILD_TABLE'";
+                    "WHERE table_name = ? AND db_name = ? AND type = 'CHILD_TABLE'";
             try (PreparedStatement pStmt = conn.prepareStatement(checkTableSql)) {
                 pStmt.setString(1, table);
                 pStmt.setString(2, dbName);

iot-platform/src/test/java/com/iot/platform/mqtt/MqttChargeStationConsumerTest.java

@@ -101,7 +101,7 @@ class MqttChargeStationConsumerTest {
 
         method.invoke(consumer, dataList, topic);
 
-        verify(tdengineService).insertBatch(eq("myDb"), eq(expectedTable), anyList());
+        verify(tdengineService).insertBatch(eq("myDb"), "mySuperTable_",eq(expectedTable), anyList());
     }
 
     @Test
@@ -113,7 +113,7 @@ class MqttChargeStationConsumerTest {
 
         method.invoke(consumer, Collections.emptyList(), "prefix/db/suffix/table/extra");
 
-        verify(tdengineService, never()).insertBatch(anyString(), anyString(), anyList());
+        verify(tdengineService, never()).insertBatch(anyString(), "mySuperTable_",anyString(), anyList());
     }
 
     @Test
@@ -151,6 +151,6 @@ class MqttChargeStationConsumerTest {
 
         method.invoke(consumer, dataList, topic);
 
-        verify(tdengineService).insertBatch(eq("myDb"), eq(expectedTable), argThat(list -> list.size() == 1));
+        verify(tdengineService).insertBatch(eq("myDb"), "mySuperTable_",eq(expectedTable), argThat(list -> list.size() == 1));
     }
 }

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

@@ -305,8 +305,8 @@ class TDengineServiceTest {
     @Test
     @DisplayName("insertBatch: 空列表应直接返回不抛异常")
     void insertBatch_emptyList_returnsWithoutError() throws Exception {
-        tdengineService.insertBatch("db", "table", Collections.emptyList());
-        tdengineService.insertBatch("db", "table", null);
+        tdengineService.insertBatch("db", "supertable","table", Collections.emptyList());
+        tdengineService.insertBatch("db", "supertable","table", null);
     }
 
     @Test