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 cbfda997..d8534d69 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 @@ -4,15 +4,12 @@ import com.lingniu.ingest.api.consumer.EnvelopeIngestResult; import com.lingniu.ingest.api.consumer.EnvelopeIngestor; import com.lingniu.ingest.sink.kafka.proto.RawArchiveRef; import com.lingniu.ingest.sink.kafka.proto.RawFrameFactPayload; -import com.lingniu.ingest.sink.kafka.proto.TelemetryField; -import com.lingniu.ingest.sink.kafka.proto.TelemetrySnapshot; import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope; import com.lingniu.ingest.sink.kafka.proto.LocationPayload; import com.lingniu.ingest.sink.kafka.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.util.ArrayList; @@ -79,16 +76,6 @@ 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()); @@ -101,7 +88,6 @@ class EventHistoryEnvelopeIngestorTest { .extracting(TdengineLocationRow::factId) .containsExactly("jt808-location-1"); assertThat(tdengineWriter.locations.getFirst().longitude()).isEqualTo(113.12); - assertThat(tdengineWriter.telemetryFields).isEmpty(); } @Test @@ -117,23 +103,6 @@ class EventHistoryEnvelopeIngestorTest { .containsExactly("frame-jt808-1"); assertThat(tdengineWriter.locations).extracting(TdengineLocationRow::factId) .containsExactly("jt808-location-1"); - assertThat(tdengineWriter.telemetryFields).isEmpty(); - } - - @Test - void tryIngestDoesNotWriteTelemetryFieldFacts() { - CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter(); - EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(tdengineWriter); - - EnvelopeIngestResult result = ingestor.tryIngest( - jt808LocationEnvelope("jt808-location-1", "frame-jt808-1", "013800000001").toByteArray()); - - assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED); - assertThat(tdengineWriter.rawFrames).extracting(TdengineRawFrameRow::frameId) - .containsExactly("frame-jt808-1"); - assertThat(tdengineWriter.locations).extracting(TdengineLocationRow::factId) - .containsExactly("jt808-location-1"); - assertThat(tdengineWriter.telemetryFields).isEmpty(); } @Test @@ -158,7 +127,6 @@ class EventHistoryEnvelopeIngestorTest { .containsExactly("frame-jt808-1", "frame-jt808-2"); assertThat(tdengineWriter.locations).extracting(TdengineLocationRow::factId) .containsExactly("jt808-location-1", "jt808-location-2"); - assertThat(tdengineWriter.telemetryFields).isEmpty(); } private static VehicleEnvelope jt808LocationEnvelope(String eventId, String frameId, String phone) { @@ -187,23 +155,12 @@ class EventHistoryEnvelopeIngestorTest { .setLongitude(113.12) .setLatitude(23.45) .setSpeedKmh(42.5)) - .setTelemetrySnapshot(TelemetrySnapshot.newBuilder() - .setEventType("LOCATION") - .setRawArchiveUri("archive://jt808/2026/06/29/" + frameId + ".bin") - .addFields(TelemetryField.newBuilder() - .setKey("location.speedKmh") - .setValueType("DOUBLE") - .setValue("42.5") - .setUnit("km/h") - .setQuality("GOOD") - .setSourcePath("JT808.0x0200.speed"))) .build(); } 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) { @@ -215,9 +172,5 @@ class EventHistoryEnvelopeIngestorTest { locations.addAll(rows); } - @Override - public void appendTelemetryFields(List rows) { - telemetryFields.addAll(rows); - } } } diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java index 7973b4d8..99162807 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java @@ -18,8 +18,6 @@ import com.lingniu.ingest.tdenginehistory.TdenginePageCursor; import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder; import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery; import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow; -import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery; -import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -683,9 +681,5 @@ class Gb32960DecodedFrameServiceTest { throw new UnsupportedOperationException(); } - @Override - public TdenginePage queryTelemetryFields(TdengineTelemetryFieldQuery query) { - throw new UnsupportedOperationException(); - } } } diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808LocationHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808LocationHistoryControllerTest.java index bb1fa2f0..489afc97 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808LocationHistoryControllerTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808LocationHistoryControllerTest.java @@ -136,10 +136,5 @@ class Jt808LocationHistoryControllerTest { return response; } - @Override - public TdenginePage queryTelemetryFields( - com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery query) { - throw new UnsupportedOperationException(); - } } } diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryControllerTest.java index ea323ed6..6cee6ebe 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryControllerTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryControllerTest.java @@ -8,8 +8,6 @@ import com.lingniu.ingest.tdenginehistory.TdenginePageCursor; import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder; import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery; import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow; -import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery; -import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow; import org.junit.jupiter.api.Test; import org.springframework.http.HttpStatus; import org.springframework.web.server.ResponseStatusException; @@ -122,9 +120,5 @@ class Jt808RawFrameHistoryControllerTest { throw new UnsupportedOperationException(); } - @Override - public TdenginePage queryTelemetryFields(TdengineTelemetryFieldQuery query) { - throw new UnsupportedOperationException(); - } } } diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/LocationHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/LocationHistoryControllerTest.java index 0bce2b69..f419c274 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/LocationHistoryControllerTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/LocationHistoryControllerTest.java @@ -128,10 +128,5 @@ class LocationHistoryControllerTest { return page; } - @Override - public com.lingniu.ingest.tdenginehistory.TdenginePage queryTelemetryFields( - com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery query) { - throw new UnsupportedOperationException(); - } } } diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/RawFrameHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/RawFrameHistoryControllerTest.java index 777fbdea..75ad0b7f 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/RawFrameHistoryControllerTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/RawFrameHistoryControllerTest.java @@ -8,8 +8,6 @@ import com.lingniu.ingest.tdenginehistory.TdenginePageCursor; import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder; import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery; import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow; -import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery; -import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow; import org.junit.jupiter.api.Test; import org.springframework.http.HttpStatus; import org.springframework.web.server.ResponseStatusException; @@ -67,7 +65,6 @@ class RawFrameHistoryControllerTest { .containsEntry("raw.rawArchiveUri", item.rawUri()) .containsEntry("raw.plateNo", "粤BD12345"); assertThat(reader.locationRawUri).isNull(); - assertThat(reader.telemetryRawUri).isNull(); assertThat(response.nextCursor()).isEqualTo(new RawFrameHistoryController.CursorResponse( "2026-06-29T12:00:01Z", "frame-1")); } @@ -114,25 +111,6 @@ class RawFrameHistoryControllerTest { "VIN-JT808-1", "VIN-JT808-1", "g7gps")); - reader.telemetryByRawUri.add(new TdengineTelemetryFieldRow( - ts, - "tf-1", - "frame-1", - ts.plusMillis(500), - "VEHICLE.speedKmh", - "DOUBLE", - "25.0", - 25.0, - null, - "km/h", - "GOOD", - "VEHICLE.speedKmh", - raw.rawUri(), - "{}", - "GB32960", - "VIN-GB-1", - "VIN-GB-1", - "")); RawFrameHistoryController controller = new RawFrameHistoryController(reader); RawFrameHistoryController.RawFramePageResponse response = controller.rawFrames( @@ -151,7 +129,6 @@ class RawFrameHistoryControllerTest { RawFrameHistoryController.RawFrameResponse item = response.items().getFirst(); assertThat(reader.locationRawUri).isNull(); - assertThat(reader.telemetryRawUri).isNull(); assertThat(item.parsedFields()) .containsKey("location") .containsKey("extras"); @@ -273,10 +250,8 @@ class RawFrameHistoryControllerTest { private static final class CapturingReader implements TdengineHistoryReader { private final TdenginePage response; private final List locationsByRawUri = new ArrayList<>(); - private final List telemetryByRawUri = new ArrayList<>(); private TdengineRawFrameQuery query; private String locationRawUri; - private String telemetryRawUri; private CapturingReader(TdenginePage response) { this.response = response; @@ -293,21 +268,10 @@ class RawFrameHistoryControllerTest { throw new UnsupportedOperationException(); } - @Override - public TdenginePage queryTelemetryFields(TdengineTelemetryFieldQuery query) { - throw new UnsupportedOperationException(); - } - @Override public List queryLocationsByRawUri(String protocol, String rawUri, int limit) { locationRawUri = rawUri; return locationsByRawUri; } - - @Override - public List queryTelemetryFieldsByRawUri(String protocol, String rawUri, int limit) { - telemetryRawUri = rawUri; - return telemetryByRawUri; - } } } 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 9d654148..b0d11c03 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,13 +4,10 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.lingniu.ingest.sink.kafka.proto.LocationPayload; import com.lingniu.ingest.sink.kafka.proto.RawFrameFactPayload; -import com.lingniu.ingest.sink.kafka.proto.TelemetryField; import com.lingniu.ingest.sink.kafka.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; @@ -82,62 +79,6 @@ 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 vin = envelope.getVin(); - String phone = firstNonBlank( - envelope.getMetadataOrDefault("phone", ""), - envelope.hasRawFrameFact() ? envelope.getRawFrameFact().getPhone() : ""); - String vehicleKey = vehicleKey(envelope, - firstNonBlank( - envelope.getMetadataOrDefault("vehicle_key", ""), - envelope.hasRawFrameFact() ? envelope.getRawFrameFact().getVehicleKey() : ""), - phone); - 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); } @@ -195,41 +136,6 @@ public final class TdengineEnvelopeRows { return firstNonBlank(metadata.get("parsedJson"), metadata.get("parsed_json")); } - 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/TdengineHistoryQueries.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java index 3bea7022..6bc889c3 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java @@ -11,9 +11,6 @@ public final class TdengineHistoryQueries { private static final String LOCATION_COLUMNS = "ts, fact_id, frame_id, received_at, longitude, latitude, " + "altitude_m, speed_kmh, direction_deg, alarm_flag, status_flag, total_mileage_km, raw_uri, " + "protocol, vehicle_key, vin, phone"; - private static final String TELEMETRY_FIELD_COLUMNS = "ts, fact_id, frame_id, received_at, field_key, " - + "value_type, value_text, value_double, value_long, unit, quality, source_path, raw_uri, " - + "metadata_json, protocol, vehicle_key, vin, phone"; private final TdengineHistorySchema schema; @@ -39,20 +36,10 @@ public final class TdengineHistoryQueries { query.order(), query.limit(), query.cursor()); } - public TdengineQueryStatement telemetryFields(TdengineTelemetryFieldQuery query) { - String table = schema.telemetryFieldTable(query.protocol(), query.vehicleKey(), query.fieldKey()); - return query(table, TELEMETRY_FIELD_COLUMNS, "fact_id", query.from(), query.to(), - query.order(), query.limit(), query.cursor()); - } - public TdengineQueryStatement locationsByRawUri(String protocol, String rawUri, int limit) { return byRawUri(schema.locationsStableTable(), LOCATION_COLUMNS, protocol, rawUri, "fact_id", limit); } - public TdengineQueryStatement telemetryFieldsByRawUri(String protocol, String rawUri, int limit) { - return byRawUri(schema.telemetryFieldsStableTable(), TELEMETRY_FIELD_COLUMNS, protocol, rawUri, "fact_id", limit); - } - private static TdengineQueryStatement query(String table, String columns, String tieBreakerColumn, diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryReader.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryReader.java index d12c8824..5044a28c 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryReader.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryReader.java @@ -9,15 +9,8 @@ public interface TdengineHistoryReader { TdenginePage queryLocations(TdengineLocationQuery query) throws IOException; - TdenginePage queryTelemetryFields(TdengineTelemetryFieldQuery query) throws IOException; - default List queryLocationsByRawUri(String protocol, String rawUri, int limit) throws IOException { return List.of(); } - - default List queryTelemetryFieldsByRawUri(String protocol, String rawUri, int limit) - throws IOException { - return List.of(); - } } 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 f9ee1319..4e135832 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,8 +18,7 @@ public final class TdengineHistorySchema { "CREATE DATABASE IF NOT EXISTS " + database + " PRECISION 'ms'", "USE " + database, rawFramesStableSql(), - vehicleLocationsStableSql(), - telemetryFieldsStableSql() + vehicleLocationsStableSql() ); } @@ -35,20 +34,10 @@ public final class TdengineHistorySchema { return "vehicle_locations"; } - public String telemetryFieldsStableTable() { - return "telemetry_fields"; - } - public String locationTable(String protocol, String vehicleKey) { 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 ( @@ -98,28 +87,4 @@ public final class TdengineHistorySchema { )"""; } - 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 14eb1744..af5aaefe 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,10 +13,6 @@ public final class TdengineHistoryStatements { + "altitude_m, speed_kmh, direction_deg, alarm_flag, status_flag, total_mileage_km, raw_uri"; 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) { @@ -54,29 +50,10 @@ 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 5f27d910..e8d51a7c 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,10 +16,4 @@ 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/TdengineJdbcHistoryReader.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReader.java index 50efd057..cf3897e5 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 @@ -40,15 +40,6 @@ public final class TdengineJdbcHistoryReader implements TdengineHistoryReader { return readPage(statement, pageSize, this::location, row -> new TdenginePageCursor(row.ts(), row.factId())); } - @Override - public TdenginePage queryTelemetryFields(TdengineTelemetryFieldQuery query) - throws IOException { - int pageSize = Math.min(query.limit(), 1000); - TdengineQueryStatement statement = queries.telemetryFields(query.withLimit(pageSize + 1)); - return readPage(statement, pageSize, this::telemetryField, - row -> new TdenginePageCursor(row.ts(), row.factId())); - } - @Override public List queryLocationsByRawUri(String protocol, String rawUri, int limit) throws IOException { @@ -59,16 +50,6 @@ public final class TdengineJdbcHistoryReader implements TdengineHistoryReader { return readList(statement, this::location); } - @Override - public List queryTelemetryFieldsByRawUri(String protocol, String rawUri, int limit) - throws IOException { - if (rawUri == null || rawUri.isBlank()) { - return List.of(); - } - TdengineQueryStatement statement = queries.telemetryFieldsByRawUri(protocol, rawUri, limit); - return readList(statement, this::telemetryField); - } - private TdenginePage readPage(TdengineQueryStatement statement, int pageSize, SqlRowMapper mapper, @@ -187,29 +168,6 @@ public final class TdengineJdbcHistoryReader implements TdengineHistoryReader { ); } - private TdengineTelemetryFieldRow telemetryField(ResultSet rs) throws SQLException { - return new TdengineTelemetryFieldRow( - instant(rs, "ts"), - rs.getString("fact_id"), - rs.getString("frame_id"), - instant(rs, "received_at"), - rs.getString("field_key"), - rs.getString("value_type"), - rs.getString("value_text"), - nullableDouble(rs, "value_double"), - nullableLong(rs, "value_long"), - rs.getString("unit"), - rs.getString("quality"), - rs.getString("source_path"), - rs.getString("raw_uri"), - rs.getString("metadata_json"), - rs.getString("protocol"), - rs.getString("vehicle_key"), - rs.getString("vin"), - rs.getString("phone") - ); - } - private static Instant instant(ResultSet rs, String column) throws SQLException { Timestamp timestamp = rs.getTimestamp(column); return timestamp == null ? null : timestamp.toInstant(); @@ -220,11 +178,6 @@ public final class TdengineJdbcHistoryReader implements TdengineHistoryReader { return value instanceof Number number ? number.doubleValue() : null; } - private static Long nullableLong(ResultSet rs, String column) throws SQLException { - Object value = rs.getObject(column); - return value instanceof Number number ? number.longValue() : null; - } - @FunctionalInterface private interface SqlRowMapper { T map(ResultSet rs) throws SQLException; 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 a9166808..5d4ed92e 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 @@ -59,11 +59,6 @@ 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/TdengineTelemetryFieldQuery.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineTelemetryFieldQuery.java deleted file mode 100644 index dcdeb8ab..00000000 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineTelemetryFieldQuery.java +++ /dev/null @@ -1,35 +0,0 @@ -package com.lingniu.ingest.tdenginehistory; - -import java.time.Instant; - -public record TdengineTelemetryFieldQuery( - String protocol, - String vehicleKey, - String fieldKey, - Instant from, - Instant to, - TdengineQueryOrder order, - int limit, - TdenginePageCursor cursor -) { - public TdengineTelemetryFieldQuery { - 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 (fieldKey == null || fieldKey.isBlank()) { - throw new IllegalArgumentException("fieldKey must not be blank"); - } - if (from == null || to == null || !from.isBefore(to)) { - throw new IllegalArgumentException("query time range must be valid"); - } - order = order == null ? TdengineQueryOrder.DESC : order; - limit = Math.max(1, Math.min(limit, 1001)); - } - - TdengineTelemetryFieldQuery withLimit(int newLimit) { - return new TdengineTelemetryFieldQuery(protocol, vehicleKey, fieldKey, from, to, order, newLimit, cursor); - } -} 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 deleted file mode 100644 index 632154f9..00000000 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineTelemetryFieldRow.java +++ /dev/null @@ -1,25 +0,0 @@ -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 023b8dfd..dca1641f 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,8 +4,6 @@ import com.lingniu.ingest.sink.kafka.proto.LocationPayload; import com.lingniu.ingest.sink.kafka.proto.ParseStatusProto; import com.lingniu.ingest.sink.kafka.proto.RawArchiveRef; import com.lingniu.ingest.sink.kafka.proto.RawFrameFactPayload; -import com.lingniu.ingest.sink.kafka.proto.TelemetryField; -import com.lingniu.ingest.sink.kafka.proto.TelemetrySnapshot; import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope; import org.junit.jupiter.api.Test; @@ -123,52 +121,6 @@ class TdengineEnvelopeRowsTest { assertThat(row.rawUri()).isEqualTo("archive://jt808/metadata-only.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"); - } - @Test void mapsJt808UnknownVinRowsToPhoneVehicleKey() { VehicleEnvelope envelope = VehicleEnvelope.newBuilder() @@ -190,19 +142,12 @@ class TdengineEnvelopeRowsTest { .setLocation(LocationPayload.newBuilder() .setLongitude(119.1) .setLatitude(29.2)) - .setTelemetrySnapshot(TelemetrySnapshot.newBuilder() - .addFields(TelemetryField.newBuilder() - .setKey("speed_kmh") - .setValueType("DOUBLE") - .setValue("12.3"))) .build(); assertThat(TdengineEnvelopeRows.rawFrame(envelope).orElseThrow().vehicleKey()) .isEqualTo("jt808:13079963291"); assertThat(TdengineEnvelopeRows.location(envelope).orElseThrow().vehicleKey()) .isEqualTo("jt808:13079963291"); - assertThat(TdengineEnvelopeRows.telemetryFields(envelope).getFirst().vehicleKey()) - .isEqualTo("jt808:13079963291"); } private static VehicleEnvelope rawFrameEnvelope(String rawArchiveEventId) { diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java index ee592efd..84f2b279 100644 --- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java @@ -92,30 +92,4 @@ class TdengineHistoryQueriesTest { assertThat(statement.values().getLast()).isEqualTo(1001); } - @Test - void telemetryFieldQueryUsesFieldChildTableAndKeysetPagination() { - TdengineTelemetryFieldQuery query = new TdengineTelemetryFieldQuery( - "GB32960", - "VIN123", - "VEHICLE.totalMileageKm", - Instant.parse("2026-06-29T00:00:00Z"), - Instant.parse("2026-06-30T00:00:00Z"), - TdengineQueryOrder.ASC, - 100, - new TdenginePageCursor(Instant.parse("2026-06-29T01:00:00Z"), "evt-1#0")); - - TdengineQueryStatement statement = queries.telemetryFields(query); - - assertThat(statement.sql()) - .startsWith("SELECT ts, fact_id, frame_id, received_at, field_key") - .contains(" FROM tf_gb32960_") - .contains(" WHERE ts >= ? AND ts < ?") - .contains(" AND (ts > ? OR (ts = ? AND fact_id > ?))") - .contains(" ORDER BY ts ASC, fact_id ASC LIMIT ?"); - assertThat(statement.values()) - .containsExactly( - query.from(), query.to(), - query.cursor().ts(), query.cursor().ts(), query.cursor().tieBreaker(), - 100); - } } 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 e1c44f2f..35931d48 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 @@ -2,12 +2,15 @@ package com.lingniu.ingest.tdenginehistory; import org.junit.jupiter.api.Test; +import java.nio.file.Files; +import java.nio.file.Path; + import static org.assertj.core.api.Assertions.assertThat; class TdengineHistorySchemaTest { @Test - void bootstrapSqlCreatesRawLocationAndTelemetryStablesFromSpec() { + void bootstrapSqlCreatesOnlyRawAndLocationStablesFromSpec() { TdengineHistorySchema schema = new TdengineHistorySchema("vehicle_history"); assertThat(schema.bootstrapSql()) @@ -30,11 +33,10 @@ 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)"); + assertThat(schema.bootstrapSql()).hasSize(4); + assertThat(String.join("\n", schema.bootstrapSql())) + .doesNotContain("telemetry_fields") + .doesNotContain("field_key"); } @Test @@ -45,7 +47,20 @@ 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}"); + } + + @Test + void historyStorePublicApiDoesNotExposeTelemetryFields() throws Exception { + String writer = Files.readString(Path.of( + "src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryWriter.java")); + String reader = Files.readString(Path.of( + "src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryReader.java")); + + assertThat(writer) + .doesNotContain("TelemetryField") + .doesNotContain("appendTelemetry"); + assertThat(reader) + .doesNotContain("TelemetryField") + .doesNotContain("queryTelemetry"); } } 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 ccb22a66..af42941b 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 @@ -82,40 +82,4 @@ class TdengineHistoryStatementsTest { row.totalMileageKm(), row.rawUri()); } - @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/TdengineJdbcHistoryReaderTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReaderTest.java index 19dc929f..1ec60192 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 @@ -69,54 +69,6 @@ class TdengineJdbcHistoryReaderTest { assertThat(page.items().getFirst().longitude()).isEqualTo(113.12); } - @Test - void readsTelemetryFieldRowsWithNextCursor() throws Exception { - RecordingQueryJdbc jdbc = new RecordingQueryJdbc(); - jdbc.rows.add(telemetryFieldResult("evt-1#0", "2026-06-29T05:00:01Z")); - jdbc.rows.add(telemetryFieldResult("evt-2#0", "2026-06-29T05:00:02Z")); - TdengineJdbcHistoryReader reader = new TdengineJdbcHistoryReader( - jdbc.dataSource(), new TdengineHistorySchema("vehicle_history")); - - TdenginePage page = reader.queryTelemetryFields(new TdengineTelemetryFieldQuery( - "GB32960", - "VIN123", - "VEHICLE.totalMileageKm", - Instant.parse("2026-06-29T00:00:00Z"), - Instant.parse("2026-06-30T00:00:00Z"), - TdengineQueryOrder.ASC, - 1, - null)); - - assertThat(page.items()) - .extracting(TdengineTelemetryFieldRow::factId) - .containsExactly("evt-1#0"); - assertThat(page.items().getFirst().fieldKey()).isEqualTo("VEHICLE.totalMileageKm"); - assertThat(page.items().getFirst().valueDouble()).isEqualTo(123.4); - assertThat(page.nextCursor()) - .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(); @@ -183,29 +135,6 @@ class TdengineJdbcHistoryReaderTest { return row; } - private static Map telemetryFieldResult(String factId, String ts) { - Map row = new LinkedHashMap<>(); - row.put("ts", Timestamp.from(Instant.parse(ts))); - row.put("fact_id", factId); - row.put("frame_id", "frame-1"); - row.put("received_at", Timestamp.from(Instant.parse(ts).plusMillis(500))); - row.put("field_key", "VEHICLE.totalMileageKm"); - row.put("value_type", "DOUBLE"); - row.put("value_text", "123.4"); - row.put("value_double", 123.4); - row.put("value_long", null); - row.put("unit", "km"); - row.put("quality", "GOOD"); - row.put("source_path", "VEHICLE.totalMileageKm"); - row.put("raw_uri", "archive://gb32960/frame-1.bin"); - row.put("metadata_json", "{}"); - row.put("protocol", "GB32960"); - row.put("vehicle_key", "VIN123"); - row.put("vin", "VIN123"); - row.put("phone", ""); - return row; - } - private static final class RecordingQueryJdbc { private final List> rows = new ArrayList<>(); private final Map boundValues = new HashMap<>(); 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 1d20685f..bd91e249 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 @@ -92,28 +92,6 @@ class TdengineJdbcHistoryWriterTest { && sql.contains("archive://gb32960/frame-1.bin")); } - @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).startsWith(schema.bootstrapSql().toArray(String[]::new)); - assertThat(jdbc.executedSql) - .filteredOn(sql -> sql.contains("USING telemetry_fields TAGS")) - .hasSize(1); - assertThat(jdbc.preparedBatches).isEmpty(); - assertThat(jdbc.executedSql) - .anyMatch(sql -> sql.startsWith("INSERT INTO tf_gb32960_") - && sql.contains("evt-1#0") - && sql.contains("evt-2#0") - && sql.contains("123.4") - && sql.contains("'124.5'")); - } - @Test void schemaBootstrapRunsOnlyOnceAcrossWrites() throws Exception { RecordingJdbc jdbc = new RecordingJdbc(); @@ -223,31 +201,6 @@ 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<>();