From 7abac30ea4fa2df684d4770decada2e599c00a77 Mon Sep 17 00:00:00 2001 From: lingniu Date: Mon, 29 Jun 2026 14:47:54 +0800 Subject: [PATCH] fix: drop history export and tolerate empty tdengine tables --- .../eventhistory/EventHistoryController.java | 111 ------------------ .../EventHistoryControllerTest.java | 78 ------------ .../TdengineJdbcHistoryReader.java | 18 +++ .../TdengineJdbcHistoryReaderTest.java | 51 +++++++- 4 files changed, 68 insertions(+), 190 deletions(-) diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java index 3d41105f..3ae8342a 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java @@ -1,7 +1,5 @@ package com.lingniu.ingest.eventhistory; -import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; import com.lingniu.ingest.api.ProtocolId; import com.lingniu.ingest.eventfilestore.EventFileQuery; import com.lingniu.ingest.eventfilestore.EventFileRecord; @@ -14,8 +12,6 @@ import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import java.io.IOException; -import java.util.ArrayList; -import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -31,8 +27,6 @@ import java.util.Map; @RequestMapping("/api/event-history") public class EventHistoryController { - private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); - private final EventFileStore store; public EventHistoryController(EventFileStore store) { @@ -52,51 +46,6 @@ public class EventHistoryController { .toList(); } - @GetMapping(value = "/export.csv", produces = "text/csv;charset=UTF-8") - public String exportCsv( - @RequestParam ProtocolId protocol, - @RequestParam String dateFrom, - @RequestParam String dateTo, - @RequestParam(defaultValue = "ASC") EventFileQuery.Order order, - @RequestParam(defaultValue = "100000") int limit, - @RequestParam(required = false) String vin, - @RequestParam(required = false) String fields) throws IOException { - List records = queryRecords(protocol, dateFrom, dateTo, order, limit, vin); - List selectedFields = parseFields(fields); - StringBuilder csv = new StringBuilder(256 + records.size() * 128); - csv.append("event_id,protocol,event_type,vin,event_time,ingest_time,raw_archive_uri"); - for (String field : selectedFields) { - csv.append(',').append(csv(field)); - } - csv.append(",payload_json\n"); - for (EventFileRecord record : records) { - Map telemetryFields = selectedFields.isEmpty() - ? Map.of() - : telemetryFields(record.payloadJson()); - csv.append(csv(record.eventId())).append(',') - .append(csv(record.protocol().name())).append(',') - .append(csv(record.eventType())).append(',') - .append(csv(record.vin())).append(',') - .append(csv(record.eventTime().toString())).append(',') - .append(csv(record.ingestTime().toString())).append(',') - .append(csv(record.rawArchiveUri())); - for (String field : selectedFields) { - csv.append(',').append(csv(fieldValue(record, telemetryFields, field))); - } - csv.append(',') - .append(csv(record.payloadJson())).append('\n'); - } - return csv.toString(); - } - - public String exportCsv(ProtocolId protocol, - String dateFrom, - String dateTo, - EventFileQuery.Order order, - int limit) throws IOException { - return exportCsv(protocol, dateFrom, dateTo, order, limit, null, null); - } - private List queryRecords(ProtocolId protocol, String dateFrom, String dateTo, @@ -140,64 +89,4 @@ public class EventHistoryController { record.payloadJson()); } } - - private static List parseFields(String fields) { - if (fields == null || fields.isBlank()) { - return List.of(); - } - List out = new ArrayList<>(); - for (String field : fields.split(",")) { - String normalized = field.trim(); - if (!normalized.isBlank()) { - out.add(normalized); - } - } - return List.copyOf(out); - } - - private static Map telemetryFields(String payloadJson) throws IOException { - if (payloadJson == null || payloadJson.isBlank()) { - return Map.of(); - } - JsonNode root = OBJECT_MAPPER.readTree(payloadJson); - JsonNode fields = root.path("fields"); - if (!fields.isArray()) { - return Map.of(); - } - // 通用导出只识别 telemetry snapshot 的扁平 fields；复杂 GB32960 block 字段用专用 CSV 接口。 - Map out = new LinkedHashMap<>(); - for (JsonNode field : fields) { - String key = field.path("key").asText(""); - if (!key.isBlank()) { - out.put(key, field.path("value").asText("")); - } - } - return out; - } - - private static String fieldValue(EventFileRecord record, - Map telemetryFields, - String field) { - if (field == null || field.isBlank()) { - return ""; - } - if (field.startsWith("metadata.")) { - return record.metadata().get(field.substring("metadata.".length())); - } - return telemetryFields.get(field); - } - - private static String csv(String value) { - if (value == null) { - return ""; - } - boolean quote = value.indexOf(',') >= 0 - || value.indexOf('"') >= 0 - || value.indexOf('\n') >= 0 - || value.indexOf('\r') >= 0; - if (!quote) { - return value; - } - return "\"" + value.replace("\"", "\"\"") + "\""; - } } diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java index 06041560..c46f990b 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java @@ -39,84 +39,6 @@ class EventHistoryControllerTest { assertThat(store.lastQuery.vin()).isEqualTo("VIN001"); } - @Test - void exportsRecordsAsCsv() throws Exception { - EventHistoryController controller = new EventHistoryController(new CapturingStore(List.of(record("event-1")))); - - String csv = controller.exportCsv( - ProtocolId.GB32960, - "2026-06-22", - "2026-06-22", - EventFileQuery.Order.ASC, - 100); - - assertThat(csv).startsWith("event_id,protocol,event_type,vin,event_time,ingest_time,raw_archive_uri,payload_json\n"); - assertThat(csv).contains("event-1,GB32960,REALTIME,VIN001"); - assertThat(csv).contains("\"{\"\"fields\"\":[]}\""); - } - - @Test - void exportsConfiguredTelemetryFieldsAsCsvColumns() throws Exception { - EventHistoryController controller = new EventHistoryController(new CapturingStore(List.of( - new EventFileRecord( - "media-1", - ProtocolId.JT1078, - "MEDIA_META", - "VIN001", - Instant.parse("2026-06-22T08:00:00Z"), - Instant.parse("2026-06-22T08:00:01Z"), - "file:///archive/jt1078/media-1.rtp", - Map.of(), - """ - {"fields":[ - {"key":"media_archive_ref","value":"file:///archive/jt1078/media-1.rtp"}, - {"key":"passthrough_size_bytes","value":"128"} - ]} - """)))); - - String csv = controller.exportCsv( - ProtocolId.JT1078, - "2026-06-22", - "2026-06-22", - EventFileQuery.Order.ASC, - 100, - null, - "media_archive_ref,passthrough_size_bytes"); - - assertThat(csv).startsWith("event_id,protocol,event_type,vin,event_time,ingest_time,raw_archive_uri,media_archive_ref,passthrough_size_bytes,payload_json\n"); - assertThat(csv).contains("media-1,JT1078,MEDIA_META,VIN001,2026-06-22T08:00:00Z,2026-06-22T08:00:01Z,file:///archive/jt1078/media-1.rtp,file:///archive/jt1078/media-1.rtp,128,"); - } - - @Test - void exportsConfiguredMetadataFieldsAsCsvColumns() throws Exception { - EventHistoryController controller = new EventHistoryController(new CapturingStore(List.of( - new EventFileRecord( - "media-2", - ProtocolId.JT1078, - "MEDIA_META", - "VIN001", - Instant.parse("2026-06-22T08:00:00Z"), - Instant.parse("2026-06-22T08:00:01Z"), - "file:///archive/jt1078/media-2.rtp", - Map.of( - "channelId", "3", - "segmentSizeBytes", "512", - "archiveKey", "jt1078/2026/06/22/VIN001/ch-3/dt-0/1782115200.media"), - "{\"fields\":[]}")))); - - String csv = controller.exportCsv( - ProtocolId.JT1078, - "2026-06-22", - "2026-06-22", - EventFileQuery.Order.ASC, - 100, - null, - "metadata.channelId,metadata.segmentSizeBytes,metadata.archiveKey"); - - assertThat(csv).startsWith("event_id,protocol,event_type,vin,event_time,ingest_time,raw_archive_uri,metadata.channelId,metadata.segmentSizeBytes,metadata.archiveKey,payload_json\n"); - assertThat(csv).contains("media-2,JT1078,MEDIA_META,VIN001,2026-06-22T08:00:00Z,2026-06-22T08:00:01Z,file:///archive/jt1078/media-2.rtp,3,512,jt1078/2026/06/22/VIN001/ch-3/dt-0/1782115200.media,"); - } - private static EventFileRecord record(String eventId) { return new EventFileRecord( eventId, diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReader.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReader.java index c6e16d14..09e09dd9 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReader.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReader.java @@ -63,6 +63,9 @@ public final class TdengineJdbcHistoryReader implements TdengineHistoryReader { } } } catch (SQLException e) { + if (isTableMissing(e)) { + return new TdenginePage<>(List.of(), Optional.empty()); + } throw new IOException("tdengine history query failed", e); } boolean hasMore = rows.size() > pageSize; @@ -73,6 +76,21 @@ public final class TdengineJdbcHistoryReader implements TdengineHistoryReader { return new TdenginePage<>(items, next); } + private static boolean isTableMissing(SQLException e) { + for (Throwable current = e; current != null; current = current.getCause()) { + String message = current.getMessage(); + if (message != null) { + String normalized = message.toLowerCase(); + if (normalized.contains("table does not exist") + || normalized.contains("0x2603") + || normalized.contains("table not exist")) { + return true; + } + } + } + return false; + } + private static void bind(PreparedStatement prepared, List values) throws SQLException { for (int i = 0; i < values.size(); i++) { Object value = values.get(i); diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReaderTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReaderTest.java index f2b4e91d..19dc929f 100644 --- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReaderTest.java +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReaderTest.java @@ -7,6 +7,7 @@ import java.lang.reflect.Proxy; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; +import java.sql.SQLException; import java.sql.Timestamp; import java.time.Instant; import java.util.ArrayList; @@ -95,6 +96,48 @@ class TdengineJdbcHistoryReaderTest { .contains(new TdenginePageCursor(Instant.parse("2026-06-29T05:00:01Z"), "evt-1#0")); } + @Test + void returnsEmptyPageWhenTdengineChildTableDoesNotExist() throws Exception { + RecordingQueryJdbc jdbc = new RecordingQueryJdbc(); + jdbc.executeQueryFailure = new SQLException("(0x2603):Fail to get table info, error: Table does not exist"); + TdengineJdbcHistoryReader reader = new TdengineJdbcHistoryReader( + jdbc.dataSource(), new TdengineHistorySchema("vehicle_history")); + + TdenginePage page = reader.queryTelemetryFields(new TdengineTelemetryFieldQuery( + "JT808", + "jt808:g7gps", + "location.speed_kmh", + Instant.parse("2026-06-29T00:00:00Z"), + Instant.parse("2026-06-30T00:00:00Z"), + TdengineQueryOrder.ASC, + 10, + null)); + + assertThat(page.items()).isEmpty(); + assertThat(page.nextCursor()).isEmpty(); + } + + @Test + void returnsEmptyPageWhenTdengineChildTableDoesNotExistInNestedCause() throws Exception { + RecordingQueryJdbc jdbc = new RecordingQueryJdbc(); + jdbc.executeQueryFailure = new SQLException("wrapper", + new RuntimeException("TDengine error: table not exist")); + TdengineJdbcHistoryReader reader = new TdengineJdbcHistoryReader( + jdbc.dataSource(), new TdengineHistorySchema("vehicle_history")); + + TdenginePage page = reader.queryLocations(new TdengineLocationQuery( + "JT808", + "jt808:g7gps", + Instant.parse("2026-06-29T00:00:00Z"), + Instant.parse("2026-06-30T00:00:00Z"), + TdengineQueryOrder.ASC, + 10, + null)); + + assertThat(page.items()).isEmpty(); + assertThat(page.nextCursor()).isEmpty(); + } + private static Map rawFrameResult(String frameId, String ts) { Map row = new LinkedHashMap<>(); row.put("ts", Timestamp.from(Instant.parse(ts))); @@ -166,6 +209,7 @@ class TdengineJdbcHistoryReaderTest { private static final class RecordingQueryJdbc { private final List> rows = new ArrayList<>(); private final Map boundValues = new HashMap<>(); + private SQLException executeQueryFailure; private String preparedSql; DataSource dataSource() { @@ -199,7 +243,12 @@ class TdengineJdbcHistoryReaderTest { boundValues.put((Integer) args[0], args[1]); yield null; } - case "executeQuery" -> resultSet(); + case "executeQuery" -> { + if (executeQueryFailure != null) { + throw executeQueryFailure; + } + yield resultSet(); + } default -> null; }); }