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

feat: persist telemetry fields to tdengine

Browse Source
This commit is contained in:
lingniu
2026-06-29 13:48:04 +08:00
parent ab9c669e76
commit 3904dcf869
12 changed files with 350 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -96,5 +96,9 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeIngestor {
if (location.isPresent()) {
tdengineWriter.appendLocations(List.of(location.get()));
}
var telemetryFields = TdengineEnvelopeRows.telemetryFields(envelope);
if (!telemetryFields.isEmpty()) {
tdengineWriter.appendTelemetryFields(telemetryFields);
}
}
}

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

@@ -19,6 +19,7 @@ import com.lingniu.ingest.sink.mq.proto.ParseStatusProto;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
import com.lingniu.ingest.tdenginehistory.TdengineLocationRow;
import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow;
import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow;
import org.junit.jupiter.api.Test;
import java.io.IOException;
@@ -125,7 +126,7 @@ class EventHistoryEnvelopeIngestorTest {
}
@Test
void tryIngestWritesRawAndLocationFactsToTdengineWhenWriterExists() throws Exception {
void tryIngestWritesRawLocationAndTelemetryFactsToTdengineWhenWriterExists() throws Exception {
CapturingStore store = new CapturingStore();
CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter();
EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(
@@ -156,6 +157,16 @@ class EventHistoryEnvelopeIngestorTest {
.setLongitude(113.12)
.setLatitude(23.45)
.setSpeedKmh(42.5))
.setTelemetrySnapshot(TelemetrySnapshot.newBuilder()
.setEventType("LOCATION")
.setRawArchiveUri("archive://jt808/2026/06/29/frame-jt808-1.bin")
.addFields(TelemetryField.newBuilder()
.setKey("location.speedKmh")
.setValueType("DOUBLE")
.setValue("42.5")
.setUnit("km/h")
.setQuality("GOOD")
.setSourcePath("JT808.0x0200.speed")))
.build();
EnvelopeIngestResult result = ingestor.tryIngest(envelope.toByteArray());
@@ -169,6 +180,10 @@ class EventHistoryEnvelopeIngestorTest {
.extracting(TdengineLocationRow::factId)
.containsExactly("jt808-location-1");
assertThat(tdengineWriter.locations.getFirst().longitude()).isEqualTo(113.12);
assertThat(tdengineWriter.telemetryFields)
.extracting(TdengineTelemetryFieldRow::fieldKey)
.containsExactly("location.speedKmh");
assertThat(tdengineWriter.telemetryFields.getFirst().valueDouble()).isEqualTo(42.5);
}
private static VehicleEnvelope envelope(String eventId) {
@@ -206,6 +221,7 @@ class EventHistoryEnvelopeIngestorTest {
private static final class CapturingTdengineWriter implements TdengineHistoryWriter {
private final List<TdengineRawFrameRow> rawFrames = new ArrayList<>();
private final List<TdengineLocationRow> locations = new ArrayList<>();
private final List<TdengineTelemetryFieldRow> telemetryFields = new ArrayList<>();
@Override
public void appendRawFrames(List<TdengineRawFrameRow> rows) {
@@ -216,5 +232,10 @@ class EventHistoryEnvelopeIngestorTest {
public void appendLocations(List<TdengineLocationRow> rows) {
locations.addAll(rows);
}
@Override
public void appendTelemetryFields(List<TdengineTelemetryFieldRow> rows) {
telemetryFields.addAll(rows);
}
}
}

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

@@ -4,10 +4,13 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lingniu.ingest.sink.mq.proto.LocationPayload;
import com.lingniu.ingest.sink.mq.proto.RawFrameFactPayload;
import com.lingniu.ingest.sink.mq.proto.TelemetryField;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -76,6 +79,62 @@ public final class TdengineEnvelopeRows {
));
}
public static List<TdengineTelemetryFieldRow> telemetryFields(VehicleEnvelope envelope) {
if (envelope == null || !envelope.hasTelemetrySnapshot()) {
return List.of();
}
var snapshot = envelope.getTelemetrySnapshot();
if (snapshot.getFieldsCount() == 0) {
return List.of();
}
List<TdengineTelemetryFieldRow> rows = new ArrayList<>(snapshot.getFieldsCount());
String protocol = protocol(envelope);
String vehicleKey = firstNonBlank(
envelope.getMetadataOrDefault("vehicle_key", ""),
envelope.hasRawFrameFact() ? envelope.getRawFrameFact().getVehicleKey() : "",
envelope.getVin(),
envelope.getMetadataOrDefault("phone", ""));
String vin = envelope.getVin();
String phone = firstNonBlank(
envelope.getMetadataOrDefault("phone", ""),
envelope.hasRawFrameFact() ? envelope.getRawFrameFact().getPhone() : "");
String frameId = firstNonBlank(
envelope.getMetadataOrDefault("frame_id", ""),
envelope.hasRawFrameFact() ? envelope.getRawFrameFact().getFrameId() : "",
envelope.getEventId());
String rawUri = firstNonBlank(
snapshot.getRawArchiveUri(),
envelope.hasRawArchive() ? envelope.getRawArchive().getUri() : "");
String metadataJson = json(envelope.getMetadataMap());
for (int i = 0; i < snapshot.getFieldsCount(); i++) {
TelemetryField field = snapshot.getFields(i);
if (field.getKey().isBlank()) {
continue;
}
rows.add(new TdengineTelemetryFieldRow(
instant(envelope.getEventTimeMs()),
envelope.getEventId() + "#" + i,
frameId,
instant(envelope.getIngestTimeMs()),
field.getKey(),
field.getValueType(),
field.getValue(),
valueDouble(field),
valueLong(field),
field.getUnit(),
field.getQuality(),
field.getSourcePath(),
rawUri,
metadataJson,
protocol,
vehicleKey,
vin,
phone
));
}
return List.copyOf(rows);
}
private static Instant instant(long epochMillis) {
return Instant.ofEpochMilli(epochMillis);
}
@@ -92,6 +151,41 @@ public final class TdengineEnvelopeRows {
return Double.parseDouble(value);
}
private static Double valueDouble(TelemetryField field) {
String valueType = field.getValueType();
if (!"DOUBLE".equals(valueType) && !"FLOAT".equals(valueType)) {
return null;
}
return parseDouble(field.getValue());
}
private static Long valueLong(TelemetryField field) {
String valueType = field.getValueType();
if (!"LONG".equals(valueType) && !"INT".equals(valueType) && !"INTEGER".equals(valueType)) {
return null;
}
String value = field.getValue();
if (value == null || value.isBlank()) {
return null;
}
try {
return Long.parseLong(value);
} catch (NumberFormatException ignored) {
return null;
}
}
private static Double parseDouble(String value) {
if (value == null || value.isBlank()) {
return null;
}
try {
return Double.parseDouble(value);
} catch (NumberFormatException ignored) {
return null;
}
}
private static String json(Map<String, String> metadata) {
try {
return OBJECT_MAPPER.writeValueAsString(metadata == null ? Map.of() : metadata);

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

@@ -18,7 +18,8 @@ public final class TdengineHistorySchema {
"CREATE DATABASE IF NOT EXISTS " + database + " PRECISION 'ms'",
"USE " + database,
rawFramesStableSql(),
vehicleLocationsStableSql()
vehicleLocationsStableSql(),
telemetryFieldsStableSql()
);
}
@@ -30,6 +31,12 @@ public final class TdengineHistorySchema {
return "loc_" + TdengineIdentifier.fragment(protocol) + "_" + TdengineIdentifier.hash16(vehicleKey);
}
public String telemetryFieldTable(String protocol, String vehicleKey, String fieldKey) {
return "tf_" + TdengineIdentifier.fragment(protocol)
+ "_" + TdengineIdentifier.hash16(vehicleKey)
+ "_" + TdengineIdentifier.hash16(fieldKey);
}
private static String rawFramesStableSql() {
return """
CREATE STABLE IF NOT EXISTS raw_frames (
@@ -78,4 +85,29 @@ public final class TdengineHistorySchema {
phone NCHAR(32)
)""";
}
private static String telemetryFieldsStableSql() {
return """
CREATE STABLE IF NOT EXISTS telemetry_fields (
ts TIMESTAMP,
fact_id NCHAR(96),
frame_id NCHAR(64),
received_at TIMESTAMP,
value_type NCHAR(16),
value_text NCHAR(1024),
value_double DOUBLE,
value_long BIGINT,
unit NCHAR(32),
quality NCHAR(16),
source_path NCHAR(256),
raw_uri NCHAR(512),
metadata_json NCHAR(4096)
) TAGS (
protocol NCHAR(16),
vehicle_key NCHAR(128),
vin NCHAR(64),
phone NCHAR(32),
field_key NCHAR(256)
)""";
}
}

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

@@ -13,6 +13,10 @@ public final class TdengineHistoryStatements {
+ "altitude_m, speed_kmh, direction_deg, alarm_flag, status_flag, total_mileage_km, raw_uri, metadata_json";
private static final String LOCATION_PLACEHOLDERS = "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?";
private static final String TELEMETRY_FIELD_COLUMNS = "ts, fact_id, frame_id, received_at, value_type, "
+ "value_text, value_double, value_long, unit, quality, source_path, raw_uri, metadata_json";
private static final String TELEMETRY_FIELD_PLACEHOLDERS = "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?";
private final TdengineHistorySchema schema;
public TdengineHistoryStatements(TdengineHistorySchema schema) {
@@ -50,10 +54,29 @@ public final class TdengineHistoryStatements {
);
}
public TdengineBatchStatement telemetryField(TdengineTelemetryFieldRow row) {
String table = schema.telemetryFieldTable(row.protocol(), row.vehicleKey(), row.fieldKey());
return new TdengineBatchStatement(
"CREATE TABLE IF NOT EXISTS " + table + " USING telemetry_fields TAGS ("
+ tags(row.protocol(), row.vehicleKey(), row.vin(), row.phone(), row.fieldKey()) + ")",
"INSERT INTO " + table + " (" + TELEMETRY_FIELD_COLUMNS + ") VALUES ("
+ TELEMETRY_FIELD_PLACEHOLDERS + ")",
values(
row.ts(), row.factId(), row.frameId(), row.receivedAt(), row.valueType(), row.valueText(),
row.valueDouble(), row.valueLong(), row.unit(), row.quality(), row.sourcePath(),
row.rawUri(), row.metadataJson()
)
);
}
private static String tags(String protocol, String vehicleKey, String vin, String phone) {
return quote(protocol) + ", " + quote(vehicleKey) + ", " + quote(vin) + ", " + quote(phone);
}
private static String tags(String protocol, String vehicleKey, String vin, String phone, String fieldKey) {
return tags(protocol, vehicleKey, vin, phone) + ", " + quote(fieldKey);
}
private static String quote(String value) {
return "'" + (value == null ? "" : value.replace("'", "''")) + "'";
}

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

@@ -16,4 +16,10 @@ public interface TdengineHistoryWriter {
}
void appendLocations(List<TdengineLocationRow> rows) throws IOException;
default void appendTelemetryField(TdengineTelemetryFieldRow row) throws IOException {
appendTelemetryFields(List.of(row));
}
void appendTelemetryFields(List<TdengineTelemetryFieldRow> rows) throws IOException;
}

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

@@ -53,6 +53,11 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter {
append(rows, statements::location);
}
@Override
public void appendTelemetryFields(List<TdengineTelemetryFieldRow> rows) throws IOException {
append(rows, statements::telemetryField);
}
private <T> void append(List<T> rows, Function<T, TdengineBatchStatement> mapper) throws IOException {
if (rows == null || rows.isEmpty()) {
return;

25
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineTelemetryFieldRow.java Normal file
View File

@@ -0,0 +1,25 @@
package com.lingniu.ingest.tdenginehistory;
import java.time.Instant;
public record TdengineTelemetryFieldRow(
Instant ts,
String factId,
String frameId,
Instant receivedAt,
String fieldKey,
String valueType,
String valueText,
Double valueDouble,
Long valueLong,
String unit,
String quality,
String sourcePath,
String rawUri,
String metadataJson,
String protocol,
String vehicleKey,
String vin,
String phone
) {
}

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

@@ -4,6 +4,8 @@ import com.lingniu.ingest.sink.mq.proto.LocationPayload;
import com.lingniu.ingest.sink.mq.proto.ParseStatusProto;
import com.lingniu.ingest.sink.mq.proto.RawArchiveRef;
import com.lingniu.ingest.sink.mq.proto.RawFrameFactPayload;
import com.lingniu.ingest.sink.mq.proto.TelemetryField;
import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
import org.junit.jupiter.api.Test;
@@ -86,4 +88,50 @@ class TdengineEnvelopeRowsTest {
assertThat(row.totalMileageKm()).isNull();
assertThat(row.rawUri()).isEqualTo("archive://gb32960/2026/06/29/frame-location-1.bin");
}
@Test
void mapsTelemetrySnapshotEnvelopeToFieldRows() {
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setEventId("evt-telemetry-1")
.setVin("VIN32960")
.setSource("GB32960")
.setEventTimeMs(1_772_000_003_000L)
.setIngestTimeMs(1_772_000_003_456L)
.putMetadata("vehicle_key", "VIN32960")
.putMetadata("frame_id", "frame-telemetry-1")
.setTelemetrySnapshot(TelemetrySnapshot.newBuilder()
.setEventType("REALTIME")
.setRawArchiveUri("archive://gb32960/2026/06/29/frame-telemetry-1.bin")
.addFields(TelemetryField.newBuilder()
.setKey("VEHICLE.totalMileageKm")
.setValueType("DOUBLE")
.setValue("123.4")
.setUnit("km")
.setQuality("GOOD")
.setSourcePath("VEHICLE.totalMileageKm"))
.addFields(TelemetryField.newBuilder()
.setKey("VEHICLE.runningMode")
.setValueType("STRING")
.setValue("FUEL_CELL")
.setQuality("GOOD")
.setSourcePath("VEHICLE.runningMode")))
.build();
var rows = TdengineEnvelopeRows.telemetryFields(envelope);
assertThat(rows).hasSize(2);
TdengineTelemetryFieldRow totalMileage = rows.getFirst();
assertThat(totalMileage.ts()).isEqualTo(Instant.ofEpochMilli(1_772_000_003_000L));
assertThat(totalMileage.factId()).isEqualTo("evt-telemetry-1#0");
assertThat(totalMileage.frameId()).isEqualTo("frame-telemetry-1");
assertThat(totalMileage.protocol()).isEqualTo("GB32960");
assertThat(totalMileage.vehicleKey()).isEqualTo("VIN32960");
assertThat(totalMileage.fieldKey()).isEqualTo("VEHICLE.totalMileageKm");
assertThat(totalMileage.valueText()).isEqualTo("123.4");
assertThat(totalMileage.valueDouble()).isEqualTo(123.4);
assertThat(totalMileage.valueLong()).isNull();
assertThat(totalMileage.rawUri()).isEqualTo("archive://gb32960/2026/06/29/frame-telemetry-1.bin");
assertThat(rows.get(1).valueDouble()).isNull();
assertThat(rows.get(1).valueText()).isEqualTo("FUEL_CELL");
}
}

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

@@ -7,7 +7,7 @@ import static org.assertj.core.api.Assertions.assertThat;
class TdengineHistorySchemaTest {
@Test
void bootstrapSqlCreatesRawAndLocationStablesFromSpec() {
void bootstrapSqlCreatesRawLocationAndTelemetryStablesFromSpec() {
TdengineHistorySchema schema = new TdengineHistorySchema("vehicle_history");
assertThat(schema.bootstrapSql())
@@ -29,6 +29,11 @@ class TdengineHistorySchemaTest {
.contains("longitude DOUBLE")
.contains("total_mileage_km DOUBLE")
.contains("raw_uri NCHAR(512)");
assertThat(schema.bootstrapSql().get(4))
.contains("CREATE STABLE IF NOT EXISTS telemetry_fields")
.contains("value_double DOUBLE")
.contains("value_long BIGINT")
.contains("field_key NCHAR(256)");
}
@Test
@@ -39,5 +44,7 @@ class TdengineHistorySchemaTest {
.matches("raw_jt808_[0-9a-f]{16}");
assertThat(schema.locationTable("GB32960", "VIN WITH SPACE"))
.matches("loc_gb32960_[0-9a-f]{16}");
assertThat(schema.telemetryFieldTable("GB32960", "VIN WITH SPACE", "VEHICLE.totalMileageKm"))
.matches("tf_gb32960_[0-9a-f]{16}_[0-9a-f]{16}");
}
}

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

@@ -81,4 +81,41 @@ class TdengineHistoryStatementsTest {
row.altitudeM(), row.speedKmh(), row.directionDeg(), row.alarmFlag(), row.statusFlag(),
row.totalMileageKm(), row.rawUri(), row.metadataJson());
}
@Test
void telemetryFieldBatchStatementUsesFieldKeyTagAndTypedValues() {
TdengineTelemetryFieldRow row = new TdengineTelemetryFieldRow(
Instant.parse("2026-06-29T05:00:01Z"),
"evt-1#0",
"frame-1",
Instant.parse("2026-06-29T05:00:02Z"),
"VEHICLE.totalMileageKm",
"DOUBLE",
"123.4",
123.4,
null,
"km",
"GOOD",
"VEHICLE.totalMileageKm",
"archive://gb32960/frame-1.bin",
"{}",
"GB32960",
"VIN123",
"VIN123",
"");
TdengineBatchStatement batch = statements.telemetryField(row);
assertThat(batch.createChildTableSql())
.startsWith("CREATE TABLE IF NOT EXISTS tf_gb32960_")
.contains(" USING telemetry_fields TAGS ('GB32960', 'VIN123', 'VIN123', '', 'VEHICLE.totalMileageKm')");
assertThat(batch.insertSql())
.contains("(ts, fact_id, frame_id, received_at, value_type, value_text, value_double, value_long, unit, quality, source_path, raw_uri, metadata_json)")
.endsWith("VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)");
assertThat(batch.values())
.containsExactly(
row.ts(), row.factId(), row.frameId(), row.receivedAt(), row.valueType(), row.valueText(),
row.valueDouble(), row.valueLong(), row.unit(), row.quality(), row.sourcePath(),
row.rawUri(), row.metadataJson());
}
}

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

@@ -87,6 +87,26 @@ class TdengineJdbcHistoryWriterTest {
assertThat(batch.rows().getFirst().get(12)).isNull();
}
@Test
void writesTelemetryFieldsPerVehicleAndFieldChildTable() throws Exception {
RecordingJdbc jdbc = new RecordingJdbc();
TdengineJdbcHistoryWriter writer = new TdengineJdbcHistoryWriter(jdbc.dataSource(), schema);
TdengineTelemetryFieldRow first = telemetryField("evt-1#0", "VEHICLE.totalMileageKm", "123.4", 123.4);
TdengineTelemetryFieldRow second = telemetryField("evt-2#0", "VEHICLE.totalMileageKm", "124.5", 124.5);
writer.appendTelemetryFields(List.of(first, second));
assertThat(jdbc.executedSql)
.filteredOn(sql -> sql.contains("USING telemetry_fields TAGS"))
.hasSize(1);
RecordingJdbc.PreparedBatch batch = jdbc.preparedBatches.values().iterator().next();
assertThat(batch.sql()).contains("INSERT INTO tf_gb32960_");
assertThat(batch.rows()).hasSize(2);
assertThat(batch.rows().getFirst().get(2)).isEqualTo("evt-1#0");
assertThat(batch.rows().getFirst().get(7)).isEqualTo(123.4);
assertThat(batch.rows().get(1).get(6)).isEqualTo("124.5");
}
private static TdengineRawFrameRow rawFrame(String frameId, Instant ts) {
return new TdengineRawFrameRow(
ts,
@@ -108,6 +128,31 @@ class TdengineJdbcHistoryWriterTest {
"013800000000");
}
private static TdengineTelemetryFieldRow telemetryField(String factId,
String fieldKey,
String value,
Double valueDouble) {
return new TdengineTelemetryFieldRow(
Instant.parse("2026-06-29T05:00:01Z"),
factId,
"frame-1",
Instant.parse("2026-06-29T05:00:02Z"),
fieldKey,
"DOUBLE",
value,
valueDouble,
null,
"km",
"GOOD",
fieldKey,
"archive://gb32960/frame-1.bin",
"{}",
"GB32960",
"VIN123",
"VIN123",
"");
}
private static final class RecordingJdbc {
private final List<String> executedSql = new ArrayList<>();
private final Map<String, PreparedBatch> preparedBatches = new HashMap<>();