fix: stabilize jt808 tdengine history writes

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<TdengineTelemetryFieldRow> 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 "";
+    }
 }

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<String> createdChildTables = new LinkedHashSet<>();
             Map<String, PreparedStatement> preparedStatements = new LinkedHashMap<>();
+            Map<String, LiteralBatch> literalBatches = new LinkedHashMap<>();
+            Map<String, LiteralBatch> 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()) {
+                for (Map.Entry<String, PreparedStatement> entry : preparedStatements.entrySet()) {
-                    prepared.executeBatch();
+                    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<Object> 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<Object> row : batch.rows()) {
+                statement.execute(toLiteralInsertSql(batch.insertSql(), row));
+            }
+        }
+    }
+
+    private static String toLiteralInsertSql(String insertSql, List<Object> 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<List<Object>> rows = new java.util.ArrayList<>();
+
+        private LiteralBatch(String insertSql) {
+            this.insertSql = insertSql;
+        }
+
+        private void add(List<Object> row) {
+            rows.add(new java.util.ArrayList<>(row));
+        }
+
+        private String insertSql() {
+            return insertSql;
+        }
+
+        private List<List<Object>> rows() {
+            return rows;
+        }
+    }
 }

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

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 first = rawFrame("frame-1", Instant.parse("2026-06-29T05:00:01Z"), "none");
-        TdengineRawFrameRow second = rawFrame("frame-2", Instant.parse("2026-06-29T05:00:02Z"));
+        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<String> executedSql = new ArrayList<>();
         private final Map<String, PreparedBatch> 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());