From 3904dcf86920d83c9a8889c94310e2c29cd33216 Mon Sep 17 00:00:00 2001 From: lingniu Date: Mon, 29 Jun 2026 13:48:04 +0800 Subject: [PATCH] feat: persist telemetry fields to tdengine --- .../EventHistoryEnvelopeIngestor.java | 4 + .../EventHistoryEnvelopeIngestorTest.java | 23 ++++- .../tdenginehistory/TdengineEnvelopeRows.java | 94 +++++++++++++++++++ .../TdengineHistorySchema.java | 34 ++++++- .../TdengineHistoryStatements.java | 23 +++++ .../TdengineHistoryWriter.java | 6 ++ .../TdengineJdbcHistoryWriter.java | 5 + .../TdengineTelemetryFieldRow.java | 25 +++++ .../TdengineEnvelopeRowsTest.java | 48 ++++++++++ .../TdengineHistorySchemaTest.java | 9 +- .../TdengineHistoryStatementsTest.java | 37 ++++++++ .../TdengineJdbcHistoryWriterTest.java | 45 +++++++++ 12 files changed, 350 insertions(+), 3 deletions(-) create mode 100644 modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineTelemetryFieldRow.java diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java index 2e8ecd97..5a3acc2b 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java @@ -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); + } } } diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java index cbcf2d93..cc7d8526 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java @@ -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 rawFrames = new ArrayList<>(); private final List locations = new ArrayList<>(); + private final List telemetryFields = new ArrayList<>(); @Override public void appendRawFrames(List rows) { @@ -216,5 +232,10 @@ class EventHistoryEnvelopeIngestorTest { public void appendLocations(List rows) { locations.addAll(rows); } + + @Override + public void appendTelemetryFields(List rows) { + telemetryFields.addAll(rows); + } } } diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java index 8d38b9d9..af12fa8e 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java @@ -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 telemetryFields(VehicleEnvelope envelope) { + if (envelope == null || !envelope.hasTelemetrySnapshot()) { + return List.of(); + } + var snapshot = envelope.getTelemetrySnapshot(); + if (snapshot.getFieldsCount() == 0) { + return List.of(); + } + List 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 metadata) { try { return OBJECT_MAPPER.writeValueAsString(metadata == null ? Map.of() : metadata); diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java index 9887b0fe..cb786aae 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java @@ -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) + )"""; + } } diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatements.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatements.java index e0220666..cc4eea41 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatements.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatements.java @@ -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("'", "''")) + "'"; } diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryWriter.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryWriter.java index e8d51a7c..5f27d910 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryWriter.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryWriter.java @@ -16,4 +16,10 @@ public interface TdengineHistoryWriter { } void appendLocations(List rows) throws IOException; + + default void appendTelemetryField(TdengineTelemetryFieldRow row) throws IOException { + appendTelemetryFields(List.of(row)); + } + + void appendTelemetryFields(List rows) throws IOException; } diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriter.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriter.java index fa0d6e9c..4dd47fb9 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriter.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriter.java @@ -53,6 +53,11 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter { append(rows, statements::location); } + @Override + public void appendTelemetryFields(List rows) throws IOException { + append(rows, statements::telemetryField); + } + private void append(List rows, Function mapper) throws IOException { if (rows == null || rows.isEmpty()) { return; diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineTelemetryFieldRow.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineTelemetryFieldRow.java new file mode 100644 index 00000000..632154f9 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineTelemetryFieldRow.java @@ -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 +) { +} diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java index 84a39e0f..c97745ec 100644 --- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java @@ -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"); + } } diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchemaTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchemaTest.java index 31945341..f97ebaad 100644 --- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchemaTest.java +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchemaTest.java @@ -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}"); } } diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatementsTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatementsTest.java index a86fa5be..0381ec07 100644 --- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatementsTest.java +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatementsTest.java @@ -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()); + } } diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriterTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriterTest.java index 1294353e..3d1a1dcf 100644 --- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriterTest.java +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriterTest.java @@ -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 executedSql = new ArrayList<>(); private final Map preparedBatches = new HashMap<>();