shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: drop history export and tolerate empty tdengine tables

Browse Source
This commit is contained in:
lingniu
2026-06-29 14:47:54 +08:00
parent b631d5d998
commit 7abac30ea4
4 changed files with 68 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java
View File

@@ -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<EventFileRecord> records = queryRecords(protocol, dateFrom, dateTo, order, limit, vin);
List<String> 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<String, String> 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<EventFileRecord> queryRecords(ProtocolId protocol,
String dateFrom,
String dateTo,
@@ -140,64 +89,4 @@ public class EventHistoryController {
record.payloadJson());
}
}
private static List<String> parseFields(String fields) {
if (fields == null || fields.isBlank()) {
return List.of();
}
List<String> 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<String, String> 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<String, String> 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<String, String> 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("\"", "\"\"") + "\"";
}
}

78
modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java
View File

@@ -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,

18
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReader.java
View File

@@ -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<Object> values) throws SQLException {
for (int i = 0; i < values.size(); i++) {
Object value = values.get(i);

51
modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReaderTest.java
View File

@@ -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<TdengineTelemetryFieldRow> 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<TdengineLocationRow> 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<String, Object> rawFrameResult(String frameId, String ts) {
Map<String, Object> 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<Map<String, Object>> rows = new ArrayList<>();
private final Map<Integer, Object> 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;
});
}