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

feat: add optional parsed fields to raw frame query

Browse Source
This commit is contained in:
lingniu
2026-06-30 10:51:03 +08:00
parent b6cb661869
commit e47e341034
6 changed files with 260 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -45,6 +45,14 @@ public final class TdengineHistoryQueries {
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,
@@ -113,4 +121,22 @@ public final class TdengineHistoryQueries {
values.add(value);
}
}
private static TdengineQueryStatement byRawUri(String table,
String columns,
String protocol,
String rawUri,
String tieBreakerColumn,
int limit) {
List<Object> values = new ArrayList<>();
StringBuilder sql = new StringBuilder(256)
.append("SELECT ").append(columns)
.append(" FROM ").append(table)
.append(" WHERE raw_uri = ?");
values.add(rawUri);
addFilter(sql, values, "protocol", protocol == null ? "" : protocol.trim().toUpperCase());
sql.append(" ORDER BY ts DESC, ").append(tieBreakerColumn).append(" DESC LIMIT ?");
values.add(Math.max(1, Math.min(limit, 1000)));
return new TdengineQueryStatement(sql.toString(), values);
}
}

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

@@ -1,6 +1,7 @@
package com.lingniu.ingest.tdenginehistory;
import java.io.IOException;
import java.util.List;
public interface TdengineHistoryReader {
@@ -9,4 +10,14 @@ 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();
}
}

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

@@ -31,6 +31,14 @@ public final class TdengineHistorySchema {
return "raw_frames";
}
public String locationsStableTable() {
return "vehicle_locations";
}
public String telemetryFieldsStableTable() {
return "telemetry_fields";
}
public String locationTable(String protocol, String vehicleKey) {
return "loc_" + TdengineIdentifier.fragment(protocol) + "_" + TdengineIdentifier.hash16(vehicleKey);
}

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

@@ -49,6 +49,26 @@ public final class TdengineJdbcHistoryReader implements TdengineHistoryReader {
row -> new TdenginePageCursor(row.ts(), row.factId()));
}
@Override
public List<TdengineLocationRow> queryLocationsByRawUri(String protocol, String rawUri, int limit)
throws IOException {
if (rawUri == null || rawUri.isBlank()) {
return List.of();
}
TdengineQueryStatement statement = queries.locationsByRawUri(protocol, rawUri, limit);
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,
@@ -76,6 +96,25 @@ public final class TdengineJdbcHistoryReader implements TdengineHistoryReader {
return new TdenginePage<>(items, next);
}
private <T> List<T> readList(TdengineQueryStatement statement, SqlRowMapper<T> mapper) 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) {
if (isTableMissing(e)) {
return List.of();
}
throw new IOException("tdengine history query failed", e);
}
return List.copyOf(rows);
}
private static boolean isTableMissing(SQLException e) {
for (Throwable current = e; current != null; current = current.getCause()) {
String message = current.getMessage();