feat: add jt808 raw frame history query

modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java

@@ -25,9 +25,12 @@ public final class TdengineHistoryQueries {
     }
 
     public TdengineQueryStatement rawFrames(TdengineRawFrameQuery query) {
-        String table = schema.rawFrameTable(query.protocol(), query.vehicleKey());
-        return query(table, RAW_COLUMNS, "frame_id", query.from(), query.to(),
-                query.order(), query.limit(), query.cursor());
+        if (query.childTableQuery()) {
+            String table = schema.rawFrameTable(query.protocol(), query.vehicleKey());
+            return query(table, RAW_COLUMNS, "frame_id", query.from(), query.to(),
+                    query.order(), query.limit(), query.cursor());
+        }
+        return rawFramesStable(query);
     }
 
     public TdengineQueryStatement locations(TdengineLocationQuery query) {
@@ -71,4 +74,43 @@ public final class TdengineHistoryQueries {
         values.add(limit);
         return new TdengineQueryStatement(sql.toString(), values);
     }
+
+    private TdengineQueryStatement rawFramesStable(TdengineRawFrameQuery query) {
+        List<Object> values = new ArrayList<>();
+        StringBuilder sql = new StringBuilder(384)
+                .append("SELECT ").append(RAW_COLUMNS)
+                .append(" FROM ").append(schema.rawFramesStableTable())
+                .append(" WHERE ts >= ? AND ts < ?");
+        values.add(query.from());
+        values.add(query.to());
+        addFilter(sql, values, "protocol", query.protocol());
+        addFilter(sql, values, "vehicle_key", query.vehicleKey());
+        addFilter(sql, values, "vin", query.vin());
+        addFilter(sql, values, "phone", query.phone());
+        if (query.messageId() != null) {
+            sql.append(" AND message_id = ?");
+            values.add(query.messageId());
+        }
+        addFilter(sql, values, "parse_status", query.parseStatus());
+        if (query.cursor() != null) {
+            String op = query.order() == TdengineQueryOrder.ASC ? ">" : "<";
+            sql.append(" AND (ts ").append(op).append(" ? OR (ts = ? AND frame_id ")
+                    .append(op).append(" ?))");
+            values.add(query.cursor().ts());
+            values.add(query.cursor().ts());
+            values.add(query.cursor().tieBreaker());
+        }
+        sql.append(" ORDER BY ts ").append(query.order().name())
+                .append(", frame_id ").append(query.order().name())
+                .append(" LIMIT ?");
+        values.add(query.limit());
+        return new TdengineQueryStatement(sql.toString(), values);
+    }
+
+    private static void addFilter(StringBuilder sql, List<Object> values, String column, String value) {
+        if (value != null && !value.isBlank()) {
+            sql.append(" AND ").append(column).append(" = ?");
+            values.add(value);
+        }
+    }
 }

modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java

@@ -27,6 +27,10 @@ public final class TdengineHistorySchema {
         return "raw_" + TdengineIdentifier.fragment(protocol) + "_" + TdengineIdentifier.hash16(vehicleKey);
     }
 
+    public String rawFramesStableTable() {
+        return "raw_frames";
+    }
+
     public String locationTable(String protocol, String vehicleKey) {
         return "loc_" + TdengineIdentifier.fragment(protocol) + "_" + TdengineIdentifier.hash16(vehicleKey);
     }

modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameQuery.java

@@ -5,27 +5,56 @@ import java.time.Instant;
 public record TdengineRawFrameQuery(
         String protocol,
         String vehicleKey,
+        String vin,
+        String phone,
+        Integer messageId,
+        String parseStatus,
         Instant from,
         Instant to,
         TdengineQueryOrder order,
         int limit,
         TdenginePageCursor cursor
 ) {
+    public TdengineRawFrameQuery(String protocol,
+                                 String vehicleKey,
+                                 Instant from,
+                                 Instant to,
+                                 TdengineQueryOrder order,
+                                 int limit,
+                                 TdenginePageCursor cursor) {
+        this(protocol, vehicleKey, "", "", null, "", from, to, order, limit, cursor);
+    }
+
     public TdengineRawFrameQuery {
         if (protocol == null || protocol.isBlank()) {
             throw new IllegalArgumentException("protocol must not be blank");
         }
-        if (vehicleKey == null || vehicleKey.isBlank()) {
-            throw new IllegalArgumentException("vehicleKey must not be blank");
-        }
         if (from == null || to == null || !from.isBefore(to)) {
             throw new IllegalArgumentException("query time range must be valid");
         }
+        protocol = protocol.trim().toUpperCase();
+        vehicleKey = trim(vehicleKey);
+        vin = trim(vin);
+        phone = trim(phone);
+        parseStatus = trim(parseStatus);
         order = order == null ? TdengineQueryOrder.DESC : order;
         limit = Math.max(1, Math.min(limit, 1001));
     }
 
     TdengineRawFrameQuery withLimit(int newLimit) {
-        return new TdengineRawFrameQuery(protocol, vehicleKey, from, to, order, newLimit, cursor);
+        return new TdengineRawFrameQuery(protocol, vehicleKey, vin, phone, messageId, parseStatus,
+                from, to, order, newLimit, cursor);
+    }
+
+    boolean childTableQuery() {
+        return !vehicleKey.isBlank()
+                && vin.isBlank()
+                && phone.isBlank()
+                && messageId == null
+                && parseStatus.isBlank();
+    }
+
+    private static String trim(String value) {
+        return value == null ? "" : value.trim();
     }
 }