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

feat: add tdengine history keyset queries

Browse Source
This commit is contained in:
lingniu
2026-06-29 13:38:39 +08:00
parent e708091f2d
commit 8466d622f6
13 changed files with 573 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1,65 @@
package com.lingniu.ingest.tdenginehistory;
import java.util.ArrayList;
import java.util.List;
public final class TdengineHistoryQueries {
private static final String RAW_COLUMNS = "ts, frame_id, received_at, message_id, sub_type, event_time, "
+ "raw_uri, checksum, raw_size_bytes, parse_status, parse_error, peer, metadata_json, "
+ "protocol, vehicle_key, vin, phone";
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, "
+ "metadata_json, protocol, vehicle_key, vin, phone";
private final TdengineHistorySchema schema;
public TdengineHistoryQueries(TdengineHistorySchema schema) {
if (schema == null) {
throw new IllegalArgumentException("schema must not be null");
}
this.schema = schema;
}
public TdengineQueryStatement rawFrames(TdengineRawFrameQuery query) {
String table = schema.rawFrameTable(query.protocol(), query.vehicleKey());
return query(table, RAW_COLUMNS, "frame_id", query.from(), query.to(),
query.order(), query.limit(), query.cursor());
}
public TdengineQueryStatement locations(TdengineLocationQuery query) {
String table = schema.locationTable(query.protocol(), query.vehicleKey());
return query(table, LOCATION_COLUMNS, "fact_id", query.from(), query.to(),
query.order(), query.limit(), query.cursor());
}
private static TdengineQueryStatement query(String table,
String columns,
String tieBreakerColumn,
Object from,
Object to,
TdengineQueryOrder order,
int limit,
TdenginePageCursor cursor) {
List<Object> values = new ArrayList<>();
StringBuilder sql = new StringBuilder(256)
.append("SELECT ").append(columns)
.append(" FROM ").append(table)
.append(" WHERE ts >= ? AND ts < ?");
values.add(from);
values.add(to);
if (cursor != null) {
String op = order == TdengineQueryOrder.ASC ? ">" : "<";
sql.append(" AND (ts ").append(op).append(" ? OR (ts = ? AND ")
.append(tieBreakerColumn).append(' ').append(op).append(" ?))");
values.add(cursor.ts());
values.add(cursor.ts());
values.add(cursor.tieBreaker());
}
sql.append(" ORDER BY ts ").append(order.name())
.append(", ").append(tieBreakerColumn).append(' ').append(order.name())
.append(" LIMIT ?");
values.add(limit);
return new TdengineQueryStatement(sql.toString(), values);
}
}

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

@@ -0,0 +1,10 @@
package com.lingniu.ingest.tdenginehistory;
import java.io.IOException;
public interface TdengineHistoryReader {
TdenginePage<TdengineRawFrameRow> queryRawFrames(TdengineRawFrameQuery query) throws IOException;
TdenginePage<TdengineLocationRow> queryLocations(TdengineLocationQuery query) throws IOException;
}

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

@@ -0,0 +1,138 @@
package com.lingniu.ingest.tdenginehistory;
import javax.sql.DataSource;
import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.function.Function;
public final class TdengineJdbcHistoryReader implements TdengineHistoryReader {
private final DataSource dataSource;
private final TdengineHistoryQueries queries;
public TdengineJdbcHistoryReader(DataSource dataSource, TdengineHistorySchema schema) {
if (dataSource == null) {
throw new IllegalArgumentException("dataSource must not be null");
}
this.dataSource = dataSource;
this.queries = new TdengineHistoryQueries(schema);
}
@Override
public TdenginePage<TdengineRawFrameRow> queryRawFrames(TdengineRawFrameQuery query) throws IOException {
int pageSize = Math.min(query.limit(), 1000);
TdengineQueryStatement statement = queries.rawFrames(query.withLimit(pageSize + 1));
return readPage(statement, pageSize, this::rawFrame, row -> new TdenginePageCursor(row.ts(), row.frameId()));
}
@Override
public TdenginePage<TdengineLocationRow> queryLocations(TdengineLocationQuery query) throws IOException {
int pageSize = Math.min(query.limit(), 1000);
TdengineQueryStatement statement = queries.locations(query.withLimit(pageSize + 1));
return readPage(statement, pageSize, this::location, row -> new TdenginePageCursor(row.ts(), row.factId()));
}
private <T> TdenginePage<T> readPage(TdengineQueryStatement statement,
int pageSize,
SqlRowMapper<T> mapper,
Function<T, TdenginePageCursor> cursor) throws IOException {
List<T> rows = new ArrayList<>();
try (Connection connection = dataSource.getConnection();
PreparedStatement prepared = connection.prepareStatement(statement.sql())) {
bind(prepared, statement.values());
try (ResultSet resultSet = prepared.executeQuery()) {
while (resultSet.next()) {
rows.add(mapper.map(resultSet));
}
}
} catch (SQLException e) {
throw new IOException("tdengine history query failed", e);
}
boolean hasMore = rows.size() > pageSize;
List<T> items = hasMore ? List.copyOf(rows.subList(0, pageSize)) : List.copyOf(rows);
Optional<TdenginePageCursor> next = hasMore && !items.isEmpty()
? Optional.of(cursor.apply(items.getLast()))
: Optional.empty();
return new TdenginePage<>(items, next);
}
private static void bind(PreparedStatement prepared, List<Object> values) throws SQLException {
for (int i = 0; i < values.size(); i++) {
Object value = values.get(i);
int parameterIndex = i + 1;
if (value instanceof Instant instant) {
prepared.setTimestamp(parameterIndex, Timestamp.from(instant));
} else {
prepared.setObject(parameterIndex, value);
}
}
}
private TdengineRawFrameRow rawFrame(ResultSet rs) throws SQLException {
return new TdengineRawFrameRow(
instant(rs, "ts"),
rs.getString("frame_id"),
instant(rs, "received_at"),
rs.getInt("message_id"),
rs.getInt("sub_type"),
instant(rs, "event_time"),
rs.getString("raw_uri"),
rs.getString("checksum"),
rs.getLong("raw_size_bytes"),
rs.getString("parse_status"),
rs.getString("parse_error"),
rs.getString("peer"),
rs.getString("metadata_json"),
rs.getString("protocol"),
rs.getString("vehicle_key"),
rs.getString("vin"),
rs.getString("phone")
);
}
private TdengineLocationRow location(ResultSet rs) throws SQLException {
return new TdengineLocationRow(
instant(rs, "ts"),
rs.getString("fact_id"),
rs.getString("frame_id"),
instant(rs, "received_at"),
rs.getDouble("longitude"),
rs.getDouble("latitude"),
rs.getDouble("altitude_m"),
rs.getDouble("speed_kmh"),
rs.getDouble("direction_deg"),
rs.getLong("alarm_flag"),
rs.getLong("status_flag"),
nullableDouble(rs, "total_mileage_km"),
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();
}
private static Double nullableDouble(ResultSet rs, String column) throws SQLException {
Object value = rs.getObject(column);
return value instanceof Number number ? number.doubleValue() : null;
}
@FunctionalInterface
private interface SqlRowMapper<T> {
T map(ResultSet rs) throws SQLException;
}
}

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

@@ -0,0 +1,31 @@
package com.lingniu.ingest.tdenginehistory;
import java.time.Instant;
public record TdengineLocationQuery(
String protocol,
String vehicleKey,
Instant from,
Instant to,
TdengineQueryOrder order,
int limit,
TdenginePageCursor cursor
) {
public TdengineLocationQuery {
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 (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));
}
TdengineLocationQuery withLimit(int newLimit) {
return new TdengineLocationQuery(protocol, vehicleKey, from, to, order, newLimit, cursor);
}
}

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

@@ -0,0 +1,12 @@
package com.lingniu.ingest.tdenginehistory;
import java.util.List;
import java.util.Optional;
public record TdenginePage<T>(List<T> items, Optional<TdenginePageCursor> nextCursor) {
public TdenginePage {
items = items == null ? List.of() : List.copyOf(items);
nextCursor = nextCursor == null ? Optional.empty() : nextCursor;
}
}

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

@@ -0,0 +1,13 @@
package com.lingniu.ingest.tdenginehistory;
import java.time.Instant;
public record TdenginePageCursor(Instant ts, String tieBreaker) {
public TdenginePageCursor {
if (ts == null) {
throw new IllegalArgumentException("cursor ts must not be null");
}
tieBreaker = tieBreaker == null ? "" : tieBreaker;
}
}

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

@@ -0,0 +1,6 @@
package com.lingniu.ingest.tdenginehistory;
public enum TdengineQueryOrder {
ASC,
DESC
}

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

@@ -0,0 +1,15 @@
package com.lingniu.ingest.tdenginehistory;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
public record TdengineQueryStatement(String sql, List<Object> values) {
public TdengineQueryStatement {
if (sql == null || sql.isBlank()) {
throw new IllegalArgumentException("sql must not be blank");
}
values = values == null ? List.of() : Collections.unmodifiableList(new ArrayList<>(values));
}
}

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

@@ -0,0 +1,31 @@
package com.lingniu.ingest.tdenginehistory;
import java.time.Instant;
public record TdengineRawFrameQuery(
String protocol,
String vehicleKey,
Instant from,
Instant to,
TdengineQueryOrder order,
int limit,
TdenginePageCursor cursor
) {
public TdengineRawFrameQuery {
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 (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));
}
TdengineRawFrameQuery withLimit(int newLimit) {
return new TdengineRawFrameQuery(protocol, vehicleKey, from, to, order, newLimit, cursor);
}
}

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

@@ -2,7 +2,9 @@ package com.lingniu.ingest.tdenginehistory.config;
import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryStatements;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
import com.lingniu.ingest.tdenginehistory.TdengineJdbcHistoryReader;
import com.lingniu.ingest.tdenginehistory.TdengineJdbcHistoryWriter;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
@@ -47,4 +49,15 @@ public class TdengineHistoryAutoConfiguration {
public TdengineHistoryWriter tdengineHistoryWriter(DataSource dataSource, TdengineHistorySchema schema) {
return new TdengineJdbcHistoryWriter(dataSource, schema);
}
@Bean
@ConditionalOnMissingBean
@ConditionalOnBean(DataSource.class)
@ConditionalOnProperty(
prefix = "lingniu.ingest.tdengine-history",
name = "enabled",
havingValue = "true")
public TdengineHistoryReader tdengineHistoryReader(DataSource dataSource, TdengineHistorySchema schema) {
return new TdengineJdbcHistoryReader(dataSource, schema);
}
}

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

@@ -0,0 +1,59 @@
package com.lingniu.ingest.tdenginehistory;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import static org.assertj.core.api.Assertions.assertThat;
class TdengineHistoryQueriesTest {
private final TdengineHistorySchema schema = new TdengineHistorySchema("vehicle_history");
private final TdengineHistoryQueries queries = new TdengineHistoryQueries(schema);
@Test
void rawFrameQueryUsesVehicleChildTableAndKeysetPagination() {
TdengineRawFrameQuery query = new TdengineRawFrameQuery(
"JT808",
"jt808:g7gps",
Instant.parse("2026-06-29T00:00:00Z"),
Instant.parse("2026-06-30T00:00:00Z"),
TdengineQueryOrder.ASC,
101,
new TdenginePageCursor(Instant.parse("2026-06-29T01:00:00Z"), "frame-100"));
TdengineQueryStatement statement = queries.rawFrames(query);
assertThat(statement.sql())
.startsWith("SELECT ts, frame_id, received_at, message_id")
.contains(" FROM raw_jt808_")
.contains(" WHERE ts >= ? AND ts < ?")
.contains(" AND (ts > ? OR (ts = ? AND frame_id > ?))")
.contains(" ORDER BY ts ASC, frame_id ASC LIMIT ?");
assertThat(statement.values())
.containsExactly(
query.from(), query.to(),
query.cursor().ts(), query.cursor().ts(), query.cursor().tieBreaker(),
101);
}
@Test
void locationQueryCapsLimitAndUsesDescendingCursor() {
TdengineLocationQuery query = new TdengineLocationQuery(
"GB32960",
"vin:VIN123",
Instant.parse("2026-06-29T00:00:00Z"),
Instant.parse("2026-06-30T00:00:00Z"),
TdengineQueryOrder.DESC,
5000,
new TdenginePageCursor(Instant.parse("2026-06-29T20:00:00Z"), "fact-200"));
TdengineQueryStatement statement = queries.locations(query);
assertThat(statement.sql())
.contains(" FROM loc_gb32960_")
.contains(" AND (ts < ? OR (ts = ? AND fact_id < ?))")
.contains(" ORDER BY ts DESC, fact_id DESC LIMIT ?");
assertThat(statement.values().getLast()).isEqualTo(1001);
}
}

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

@@ -0,0 +1,171 @@
package com.lingniu.ingest.tdenginehistory;
import org.junit.jupiter.api.Test;
import javax.sql.DataSource;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Timestamp;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
class TdengineJdbcHistoryReaderTest {
@Test
void readsRawFramesAndReturnsNextCursorWhenLimitHasMoreRow() throws Exception {
RecordingQueryJdbc jdbc = new RecordingQueryJdbc();
jdbc.rows.add(rawFrameResult("frame-1", "2026-06-29T05:00:01Z"));
jdbc.rows.add(rawFrameResult("frame-2", "2026-06-29T05:00:02Z"));
TdengineJdbcHistoryReader reader = new TdengineJdbcHistoryReader(
jdbc.dataSource(), new TdengineHistorySchema("vehicle_history"));
TdenginePage<TdengineRawFrameRow> page = reader.queryRawFrames(new TdengineRawFrameQuery(
"JT808",
"jt808:g7gps",
Instant.parse("2026-06-29T00:00:00Z"),
Instant.parse("2026-06-30T00:00:00Z"),
TdengineQueryOrder.ASC,
1,
null));
assertThat(page.items())
.extracting(TdengineRawFrameRow::frameId)
.containsExactly("frame-1");
assertThat(page.nextCursor())
.contains(new TdenginePageCursor(Instant.parse("2026-06-29T05:00:01Z"), "frame-1"));
assertThat(jdbc.preparedSql).contains("LIMIT ?");
assertThat(jdbc.boundValues.get(3)).isEqualTo(2);
}
@Test
void readsLocationRowsWithoutNextCursorWhenPageIsComplete() throws Exception {
RecordingQueryJdbc jdbc = new RecordingQueryJdbc();
jdbc.rows.add(locationResult("fact-1", "2026-06-29T05:00:01Z"));
TdengineJdbcHistoryReader reader = new TdengineJdbcHistoryReader(
jdbc.dataSource(), new TdengineHistorySchema("vehicle_history"));
TdenginePage<TdengineLocationRow> page = reader.queryLocations(new TdengineLocationQuery(
"GB32960",
"vin:VIN123",
Instant.parse("2026-06-29T00:00:00Z"),
Instant.parse("2026-06-30T00:00:00Z"),
TdengineQueryOrder.DESC,
10,
null));
assertThat(page.items())
.extracting(TdengineLocationRow::factId)
.containsExactly("fact-1");
assertThat(page.nextCursor()).isEmpty();
assertThat(page.items().getFirst().longitude()).isEqualTo(113.12);
}
private static Map<String, Object> rawFrameResult(String frameId, String ts) {
Map<String, Object> row = new LinkedHashMap<>();
row.put("ts", Timestamp.from(Instant.parse(ts)));
row.put("frame_id", frameId);
row.put("received_at", Timestamp.from(Instant.parse(ts).plusMillis(500)));
row.put("message_id", 0x0200);
row.put("sub_type", 0);
row.put("event_time", Timestamp.from(Instant.parse(ts)));
row.put("raw_uri", "archive://jt808/" + frameId + ".bin");
row.put("checksum", "sha256:" + frameId);
row.put("raw_size_bytes", 67L);
row.put("parse_status", "SUCCEEDED");
row.put("parse_error", "");
row.put("peer", "10.0.0.1:808");
row.put("metadata_json", "{}");
row.put("protocol", "JT808");
row.put("vehicle_key", "jt808:g7gps");
row.put("vin", "VIN123");
row.put("phone", "013800000000");
return row;
}
private static Map<String, Object> locationResult(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("longitude", 113.12);
row.put("latitude", 23.45);
row.put("altitude_m", 8.0);
row.put("speed_kmh", 42.5);
row.put("direction_deg", 90.0);
row.put("alarm_flag", 1L);
row.put("status_flag", 3L);
row.put("total_mileage_km", null);
row.put("raw_uri", "archive://gb32960/frame-1.bin");
row.put("metadata_json", "{}");
row.put("protocol", "GB32960");
row.put("vehicle_key", "vin: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<>();
private String preparedSql;
DataSource dataSource() {
return (DataSource) Proxy.newProxyInstance(
getClass().getClassLoader(),
new Class<?>[]{DataSource.class},
(proxy, method, args) -> switch (method.getName()) {
case "getConnection" -> connection();
default -> null;
});
}
private Connection connection() {
return (Connection) Proxy.newProxyInstance(
getClass().getClassLoader(),
new Class<?>[]{Connection.class},
(proxy, method, args) -> switch (method.getName()) {
case "prepareStatement" -> preparedStatement((String) args[0]);
case "getAutoCommit" -> true;
default -> null;
});
}
private PreparedStatement preparedStatement(String sql) {
preparedSql = sql;
return (PreparedStatement) Proxy.newProxyInstance(
getClass().getClassLoader(),
new Class<?>[]{PreparedStatement.class},
(proxy, method, args) -> switch (method.getName()) {
case "setTimestamp", "setObject" -> {
boundValues.put((Integer) args[0], args[1]);
yield null;
}
case "executeQuery" -> resultSet();
default -> null;
});
}
private ResultSet resultSet() {
final int[] index = {-1};
return (ResultSet) Proxy.newProxyInstance(
getClass().getClassLoader(),
new Class<?>[]{ResultSet.class},
(proxy, method, args) -> switch (method.getName()) {
case "next" -> ++index[0] < rows.size();
case "getTimestamp", "getString", "getInt", "getLong", "getDouble", "getObject" ->
rows.get(index[0]).get((String) args[0]);
case "wasNull" -> false;
default -> null;
});
}
}
}

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

@@ -1,6 +1,7 @@
package com.lingniu.ingest.tdenginehistory.config;
import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
@@ -44,4 +45,12 @@ class TdengineHistoryAutoConfigurationTest {
.withPropertyValues("lingniu.ingest.tdengine-history.enabled=true")
.run(context -> assertThat(context).hasSingleBean(TdengineHistoryWriter.class));
}
@Test
void createsReaderWhenEnabledAndDataSourceExists() {
contextRunner
.withBean(DataSource.class, () -> mock(DataSource.class))
.withPropertyValues("lingniu.ingest.tdengine-history.enabled=true")
.run(context -> assertThat(context).hasSingleBean(TdengineHistoryReader.class));
}
}