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 af12fa8e..52a9bbfe 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 @@ -44,7 +44,7 @@ public final class TdengineEnvelopeRows { raw.getPeer(), json(metadata), protocol(envelope), - firstNonBlank(raw.getVehicleKey(), envelope.getMetadataOrDefault("vehicle_key", ""), envelope.getVin()), + vehicleKey(envelope, raw.getVehicleKey(), raw.getPhone()), firstNonBlank(raw.getVin(), envelope.getVin()), raw.getPhone() )); @@ -73,7 +73,8 @@ public final class TdengineEnvelopeRows { rawUri, json(envelope.getMetadataMap()), protocol(envelope), - firstNonBlank(envelope.getMetadataOrDefault("vehicle_key", ""), envelope.getVin()), + vehicleKey(envelope, envelope.getMetadataOrDefault("vehicle_key", ""), + envelope.getMetadataOrDefault("phone", "")), envelope.getVin(), envelope.getMetadataOrDefault("phone", "") )); @@ -89,15 +90,15 @@ public final class TdengineEnvelopeRows { } List rows = new ArrayList<>(snapshot.getFieldsCount()); String protocol = protocol(envelope); - String vehicleKey = firstNonBlank( - envelope.getMetadataOrDefault("vehicle_key", ""), - envelope.hasRawFrameFact() ? envelope.getRawFrameFact().getVehicleKey() : "", - envelope.getVin(), - envelope.getMetadataOrDefault("phone", "")); String vin = envelope.getVin(); String phone = firstNonBlank( envelope.getMetadataOrDefault("phone", ""), envelope.hasRawFrameFact() ? envelope.getRawFrameFact().getPhone() : ""); + String 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() : "", @@ -143,6 +144,22 @@ public final class TdengineEnvelopeRows { return firstNonBlank(envelope.getSource(), "UNKNOWN"); } + private static String vehicleKey(VehicleEnvelope envelope, String explicitVehicleKey, String phone) { + String key = firstKnownNonBlank(explicitVehicleKey, envelope.getMetadataOrDefault("vehicle_key", ""), + envelope.getVin()); + if (!key.isBlank()) { + return key; + } + if ("JT808".equalsIgnoreCase(protocol(envelope))) { + String phoneValue = firstKnownNonBlank(phone, envelope.getMetadataOrDefault("phone", "")); + if (!phoneValue.isBlank()) { + return phoneValue.startsWith("jt808:") ? phoneValue : "jt808:" + phoneValue; + } + } + return firstNonBlank(explicitVehicleKey, envelope.getMetadataOrDefault("vehicle_key", ""), + envelope.getVin(), phone); + } + private static Double totalMileage(VehicleEnvelope envelope) { String value = envelope.getMetadataOrDefault("total_mileage_km", ""); if (value.isBlank()) { @@ -202,4 +219,13 @@ public final class TdengineEnvelopeRows { } return ""; } + + private static String firstKnownNonBlank(String... values) { + for (String value : values) { + if (value != null && !value.isBlank() && !"unknown".equalsIgnoreCase(value.trim())) { + return value; + } + } + return ""; + } } 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 63cea842..01acad61 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 @@ -14,14 +14,19 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.function.Function; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter { + private static final Logger log = LoggerFactory.getLogger(TdengineJdbcHistoryWriter.class); + private final DataSource dataSource; private final TdengineHistorySchema schema; private final TdengineHistoryStatements statements; private final Object schemaInitializationMonitor = new Object(); private volatile boolean schemaInitialized; + private volatile boolean literalFallbackLogged; public TdengineJdbcHistoryWriter(DataSource dataSource, TdengineHistorySchema schema) { if (dataSource == null) { @@ -90,6 +95,8 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter { } Set createdChildTables = new LinkedHashSet<>(); Map preparedStatements = new LinkedHashMap<>(); + Map literalBatches = new LinkedHashMap<>(); + Map literalOnlyBatches = new LinkedHashMap<>(); try { for (T row : rows) { if (row == null) { @@ -101,6 +108,10 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter { statement.execute(batch.createChildTableSql()); } } + if (hasEmptyString(batch.values())) { + literalOnlyBatches.computeIfAbsent(batch.insertSql(), LiteralBatch::new).add(batch.values()); + continue; + } PreparedStatement prepared = preparedStatements.computeIfAbsent(batch.insertSql(), sql -> { try { return connection.prepareStatement(sql); @@ -110,9 +121,17 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter { }); bind(prepared, batch.values()); prepared.addBatch(); + literalBatches.computeIfAbsent(batch.insertSql(), LiteralBatch::new).add(batch.values()); } - for (PreparedStatement prepared : preparedStatements.values()) { - prepared.executeBatch(); + for (Map.Entry entry : preparedStatements.entrySet()) { + try { + entry.getValue().executeBatch(); + } catch (SQLException e) { + executeLiteralFallback(connection, literalBatches.get(entry.getKey()), e); + } + } + for (LiteralBatch batch : literalOnlyBatches.values()) { + executeLiteralBatch(connection, batch); } } finally { for (PreparedStatement prepared : preparedStatements.values()) { @@ -134,6 +153,76 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter { } } + private static boolean hasEmptyString(List values) { + for (Object value : values) { + if (value instanceof String stringValue && stringValue.isEmpty()) { + return true; + } + } + return false; + } + + private void executeLiteralFallback(Connection connection, + LiteralBatch batch, + SQLException preparedFailure) throws SQLException { + if (batch == null || batch.rows().isEmpty()) { + throw preparedFailure; + } + if (!literalFallbackLogged) { + literalFallbackLogged = true; + log.warn("TDengine prepared batch failed; falling back to literal inserts: {}", + preparedFailure.getMessage()); + log.debug("TDengine prepared batch failure stacktrace", preparedFailure); + } + executeLiteralBatch(connection, batch); + } + + private static void executeLiteralBatch(Connection connection, LiteralBatch batch) throws SQLException { + try (Statement statement = connection.createStatement()) { + for (List row : batch.rows()) { + statement.execute(toLiteralInsertSql(batch.insertSql(), row)); + } + } + } + + private static String toLiteralInsertSql(String insertSql, List values) { + int valuesIndex = insertSql.lastIndexOf("VALUES"); + if (valuesIndex < 0) { + throw new IllegalArgumentException("insert SQL must contain VALUES: " + insertSql); + } + StringBuilder sql = new StringBuilder(insertSql.substring(0, valuesIndex)) + .append("VALUES ("); + for (int i = 0; i < values.size(); i++) { + if (i > 0) { + sql.append(", "); + } + sql.append(literal(values.get(i))); + } + return sql.append(')').toString(); + } + + private static String literal(Object value) { + if (value == null) { + return "NULL"; + } + if (value instanceof Instant instant) { + return Long.toString(instant.toEpochMilli()); + } + if (value instanceof Timestamp timestamp) { + return Long.toString(timestamp.toInstant().toEpochMilli()); + } + if (value instanceof Double doubleValue) { + return Double.isFinite(doubleValue) ? doubleValue.toString() : "NULL"; + } + if (value instanceof Float floatValue) { + return Float.isFinite(floatValue) ? floatValue.toString() : "NULL"; + } + if (value instanceof Number number) { + return number.toString(); + } + return "'" + value.toString().replace("'", "''") + "'"; + } + private void inTransaction(SqlWork work) throws IOException { try (Connection connection = dataSource.getConnection()) { boolean autoCommit = connection.getAutoCommit(); @@ -174,4 +263,25 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter { return (SQLException) super.getCause(); } } + + private static final class LiteralBatch { + private final String insertSql; + private final List> rows = new java.util.ArrayList<>(); + + private LiteralBatch(String insertSql) { + this.insertSql = insertSql; + } + + private void add(List row) { + rows.add(new java.util.ArrayList<>(row)); + } + + private String insertSql() { + return insertSql; + } + + private List> rows() { + return rows; + } + } } 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 c97745ec..2d720e00 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 @@ -134,4 +134,40 @@ class TdengineEnvelopeRowsTest { assertThat(rows.get(1).valueDouble()).isNull(); assertThat(rows.get(1).valueText()).isEqualTo("FUEL_CELL"); } + + @Test + void mapsJt808UnknownVinRowsToPhoneVehicleKey() { + VehicleEnvelope envelope = VehicleEnvelope.newBuilder() + .setEventId("evt-jt808-phone-1") + .setVin("unknown") + .setSource("JT808") + .setEventTimeMs(1_772_000_004_000L) + .setIngestTimeMs(1_772_000_004_500L) + .putMetadata("phone", "13079963291") + .putMetadata("frame_id", "frame-jt808-phone-1") + .setRawFrameFact(RawFrameFactPayload.newBuilder() + .setFrameId("frame-jt808-phone-1") + .setPhone("13079963291") + .setMessageId(0x0200) + .setRawUri("archive://jt808/frame-jt808-phone-1.bin") + .setChecksum("sha256:phone") + .setRawSizeBytes(88) + .setParseStatus(ParseStatusProto.PARSE_STATUS_SUCCEEDED)) + .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"); + } } 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 37b49bfd..da943639 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 @@ -6,6 +6,7 @@ import javax.sql.DataSource; import java.lang.reflect.Proxy; import java.sql.Connection; import java.sql.PreparedStatement; +import java.sql.SQLException; import java.sql.Statement; import java.sql.Timestamp; import java.time.Instant; @@ -36,8 +37,8 @@ class TdengineJdbcHistoryWriterTest { void writesRawFramesInBatchesPerChildTable() throws Exception { RecordingJdbc jdbc = new RecordingJdbc(); TdengineJdbcHistoryWriter writer = new TdengineJdbcHistoryWriter(jdbc.dataSource(), schema); - TdengineRawFrameRow first = rawFrame("frame-1", Instant.parse("2026-06-29T05:00:01Z")); - TdengineRawFrameRow second = rawFrame("frame-2", Instant.parse("2026-06-29T05:00:02Z")); + TdengineRawFrameRow first = rawFrame("frame-1", Instant.parse("2026-06-29T05:00:01Z"), "none"); + TdengineRawFrameRow second = rawFrame("frame-2", Instant.parse("2026-06-29T05:00:02Z"), "none"); writer.appendRawFrames(List.of(first, second)); @@ -124,7 +125,42 @@ class TdengineJdbcHistoryWriterTest { assertThat(jdbc.commits).isEqualTo(2); } + @Test + void fallsBackToLiteralInsertWhenPreparedBatchFails() throws Exception { + RecordingJdbc jdbc = new RecordingJdbc(); + jdbc.failPreparedBatches = true; + TdengineJdbcHistoryWriter writer = new TdengineJdbcHistoryWriter(jdbc.dataSource(), schema); + + writer.appendRawFrames(List.of(rawFrame("frame-1", Instant.parse("2026-06-29T05:00:01Z"), "none"))); + + assertThat(jdbc.preparedBatches).hasSize(1); + assertThat(jdbc.executedSql) + .anyMatch(sql -> sql.startsWith("INSERT INTO raw_jt808_") + && sql.contains("parse_error") + && sql.contains("'SUCCEEDED', 'none', '10.0.0.1:808'")); + assertThat(jdbc.commits).isEqualTo(1); + } + + @Test + void usesLiteralInsertForRowsWithEmptyStrings() throws Exception { + RecordingJdbc jdbc = new RecordingJdbc(); + TdengineJdbcHistoryWriter writer = new TdengineJdbcHistoryWriter(jdbc.dataSource(), schema); + + writer.appendRawFrames(List.of(rawFrame("frame-1", Instant.parse("2026-06-29T05:00:01Z")))); + + assertThat(jdbc.preparedBatches).isEmpty(); + assertThat(jdbc.executedSql) + .anyMatch(sql -> sql.startsWith("INSERT INTO raw_jt808_") + && sql.contains("parse_error") + && sql.contains("'SUCCEEDED', '', '10.0.0.1:808'")); + assertThat(jdbc.commits).isEqualTo(1); + } + private static TdengineRawFrameRow rawFrame(String frameId, Instant ts) { + return rawFrame(frameId, ts, ""); + } + + private static TdengineRawFrameRow rawFrame(String frameId, Instant ts, String parseError) { return new TdengineRawFrameRow( ts, frameId, @@ -136,7 +172,7 @@ class TdengineJdbcHistoryWriterTest { "sha256:" + frameId, 67, "SUCCEEDED", - "", + parseError, "10.0.0.1:808", "{}", "JT808", @@ -173,6 +209,7 @@ class TdengineJdbcHistoryWriterTest { private static final class RecordingJdbc { private final List executedSql = new ArrayList<>(); private final Map preparedBatches = new HashMap<>(); + private boolean failPreparedBatches; private int commits; private int rollbacks; @@ -241,7 +278,12 @@ class TdengineJdbcHistoryWriterTest { current.clear(); yield null; } - case "executeBatch" -> new int[batch.rows().size()]; + case "executeBatch" -> { + if (failPreparedBatches) { + throw new SQLException("simulated prepared batch failure"); + } + yield new int[batch.rows().size()]; + } case "unwrap" -> null; case "isWrapperFor" -> false; default -> defaultValue(method.getReturnType());