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

refactor: remove telemetry fields from tdengine history

Browse Source
This commit is contained in:
lingniu
2026-07-01 14:03:32 +08:00
parent 68f45ee79c
commit c3d67e4339
22 changed files with 24 additions and 639 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<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 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<String, String> metadata) {
try {
return OBJECT_MAPPER.writeValueAsString(metadata == null ? Map.of() : metadata);

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

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

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

@@ -9,15 +9,8 @@ public interface TdengineHistoryReader {
TdenginePage<TdengineLocationRow> queryLocations(TdengineLocationQuery query) throws IOException;
TdenginePage<TdengineTelemetryFieldRow> queryTelemetryFields(TdengineTelemetryFieldQuery query) throws IOException;
default List<TdengineLocationRow> queryLocationsByRawUri(String protocol, String rawUri, int limit)
throws IOException {
return List.of();
}
default List<TdengineTelemetryFieldRow> queryTelemetryFieldsByRawUri(String protocol, String rawUri, int limit)
throws IOException {
return List.of();
}
}

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

@@ -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)
)""";
}
}

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

@@ -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("'", "''")) + "'";
}

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

@@ -16,10 +16,4 @@ 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;
}

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

@@ -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<TdengineTelemetryFieldRow> 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<TdengineLocationRow> 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<TdengineTelemetryFieldRow> 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 <T> TdenginePage<T> readPage(TdengineQueryStatement statement,
int pageSize,
SqlRowMapper<T> 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> {
T map(ResultSet rs) throws SQLException;

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

@@ -59,11 +59,6 @@ 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;

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

@@ -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);
}
}

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

@@ -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
) {
}

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

@@ -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) {

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

@@ -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);
}
}

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

@@ -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");
}
}

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

@@ -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());
}
}

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

@@ -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<TdengineTelemetryFieldRow> 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<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();
@@ -183,29 +135,6 @@ class TdengineJdbcHistoryReaderTest {
return row;
}
private static Map<String, Object> telemetryFieldResult(String factId, String ts) {
Map<String, Object> 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<Map<String, Object>> rows = new ArrayList<>();
private final Map<Integer, Object> boundValues = new HashMap<>();

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

@@ -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<String> executedSql = new ArrayList<>();
private final Map<String, PreparedBatch> preparedBatches = new HashMap<>();